reth_payload_builder/
service.rs

1//! Support for building payloads.
2//!
3//! The payload builder is responsible for building payloads.
4//! Once a new payload is created, it is continuously updated.
5
6use crate::{
7    metrics::PayloadBuilderServiceMetrics, traits::PayloadJobGenerator, KeepPayloadJobAlive,
8    PayloadJob,
9};
10use alloy_consensus::BlockHeader;
11use alloy_primitives::{BlockTimestamp, B256};
12use alloy_rpc_types::engine::PayloadId;
13use futures_util::{future::FutureExt, Stream, StreamExt};
14use reth_chain_state::CanonStateNotification;
15use reth_execution_cache::SavedCache;
16use reth_payload_builder_primitives::{Events, PayloadBuilderError, PayloadEvents};
17use reth_payload_primitives::{BuiltPayload, PayloadAttributes, PayloadKind, PayloadTypes};
18use reth_primitives_traits::{FastInstant as Instant, NodePrimitives};
19use reth_trie_parallel::state_root_task::PayloadStateRootHandle;
20use std::{
21    future::Future,
22    pin::Pin,
23    sync::Arc,
24    task::{Context, Poll},
25};
26use tokio::sync::{
27    broadcast, mpsc,
28    oneshot::{self, Receiver},
29    watch,
30};
31use tokio_stream::wrappers::UnboundedReceiverStream;
32use tracing::{debug, debug_span, info, trace, warn, Span};
33
34type PayloadFuture<P> = Pin<Box<dyn Future<Output = Result<P, PayloadBuilderError>> + Send>>;
35type ResolvePayloadResult<P, Job> = (Option<PayloadFuture<P>>, Option<PayloadJobEntry<Job>>);
36
37/// A communication channel to the [`PayloadBuilderService`] that can retrieve payloads.
38///
39/// This type is intended to be used to retrieve payloads from the service (e.g. from the engine
40/// API).
41#[derive(Debug)]
42pub struct PayloadStore<T: PayloadTypes> {
43    inner: Arc<PayloadBuilderHandle<T>>,
44}
45
46impl<T> PayloadStore<T>
47where
48    T: PayloadTypes,
49{
50    /// Resolves the payload job and returns the best payload that has been built so far.
51    ///
52    /// Note: depending on the installed [`PayloadJobGenerator`], this may or may not terminate the
53    /// job, See [`PayloadJob::resolve`].
54    pub fn resolve_kind(
55        &self,
56        id: PayloadId,
57        kind: PayloadKind,
58    ) -> impl Future<Output = Option<Result<T::BuiltPayload, PayloadBuilderError>>> {
59        self.inner.resolve_kind(id, kind)
60    }
61
62    /// Resolves the payload job and returns the best payload that has been built so far.
63    pub async fn resolve(
64        &self,
65        id: PayloadId,
66    ) -> Option<Result<T::BuiltPayload, PayloadBuilderError>> {
67        self.resolve_kind(id, PayloadKind::Earliest).await
68    }
69
70    /// Returns the best payload for the given identifier.
71    ///
72    /// Note: this merely returns the best payload so far and does not resolve the job.
73    pub async fn best_payload(
74        &self,
75        id: PayloadId,
76    ) -> Option<Result<T::BuiltPayload, PayloadBuilderError>> {
77        self.inner.best_payload(id).await
78    }
79
80    /// Returns the payload timestamp associated with the given identifier.
81    ///
82    /// Note: this returns the timestamp of the payload and does not resolve the job.
83    pub async fn payload_timestamp(
84        &self,
85        id: PayloadId,
86    ) -> Option<Result<u64, PayloadBuilderError>> {
87        self.inner.payload_timestamp(id).await
88    }
89
90    /// Create a new instance
91    pub fn new(inner: PayloadBuilderHandle<T>) -> Self {
92        Self { inner: Arc::new(inner) }
93    }
94}
95
96impl<T> From<PayloadBuilderHandle<T>> for PayloadStore<T>
97where
98    T: PayloadTypes,
99{
100    fn from(inner: PayloadBuilderHandle<T>) -> Self {
101        Self::new(inner)
102    }
103}
104
105/// A communication channel to the [`PayloadBuilderService`].
106///
107/// This is the API used to create new payloads and to get the current state of existing ones.
108#[derive(Debug)]
109pub struct PayloadBuilderHandle<T: PayloadTypes> {
110    /// Sender half of the message channel to the [`PayloadBuilderService`].
111    to_service: mpsc::UnboundedSender<PayloadServiceCommand<T>>,
112}
113
114impl<T: PayloadTypes> PayloadBuilderHandle<T> {
115    /// Creates a new payload builder handle for the given channel.
116    ///
117    /// Note: this is only used internally by the [`PayloadBuilderService`] to manage the payload
118    /// building flow See [`PayloadBuilderService::poll`] for implementation details.
119    pub const fn new(to_service: mpsc::UnboundedSender<PayloadServiceCommand<T>>) -> Self {
120        Self { to_service }
121    }
122
123    /// Sends a message to the service to start building a new payload for the given payload.
124    ///
125    /// Returns a receiver that will receive the payload id.
126    pub fn send_new_payload(
127        &self,
128        input: BuildNewPayload<T::PayloadAttributes>,
129    ) -> Receiver<Result<PayloadId, PayloadBuilderError>> {
130        let (tx, rx) = oneshot::channel();
131        let span = debug_span!(parent: Span::current(), "payload_job");
132        let _ =
133            self.to_service.send(PayloadServiceCommand::BuildNewPayload(input.into(), span, tx));
134        rx
135    }
136
137    /// Returns the best payload for the given identifier.
138    /// Note: this does not resolve the job if it's still in progress.
139    pub async fn best_payload(
140        &self,
141        id: PayloadId,
142    ) -> Option<Result<T::BuiltPayload, PayloadBuilderError>> {
143        let (tx, rx) = oneshot::channel();
144        self.to_service.send(PayloadServiceCommand::BestPayload(id, tx)).ok()?;
145        rx.await.ok()?
146    }
147
148    /// Resolves the payload job and returns the best payload that has been built so far.
149    ///
150    /// # Cancellation safety
151    ///
152    /// The future returned by this method is not cancellation-safe. This method sends the resolve
153    /// command before returning the future, so dropping the returned future drops the response
154    /// receiver and cancels the job identified by `id`.
155    pub fn resolve_kind(
156        &self,
157        id: PayloadId,
158        kind: PayloadKind,
159    ) -> impl Future<Output = Option<Result<T::BuiltPayload, PayloadBuilderError>>> {
160        let (tx, rx) = oneshot::channel();
161        let sent = self.to_service.send(PayloadServiceCommand::Resolve(id, kind, tx)).is_ok();
162        async move {
163            if !sent {
164                return None
165            }
166
167            match rx.await.transpose()? {
168                Ok(fut) => Some(fut.await),
169                Err(e) => Some(Err(e.into())),
170            }
171        }
172    }
173
174    /// Sends a message to the service to subscribe to payload events.
175    /// Returns a receiver that will receive them.
176    pub async fn subscribe(&self) -> Result<PayloadEvents<T>, PayloadBuilderError> {
177        let (tx, rx) = oneshot::channel();
178        let _ = self.to_service.send(PayloadServiceCommand::Subscribe(tx));
179        Ok(PayloadEvents { receiver: rx.await? })
180    }
181
182    /// Returns the payload timestamp associated with the given identifier.
183    ///
184    /// Note: this returns the timestamp of the payload and does not resolve the job.
185    pub async fn payload_timestamp(
186        &self,
187        id: PayloadId,
188    ) -> Option<Result<u64, PayloadBuilderError>> {
189        let (tx, rx) = oneshot::channel();
190        self.to_service.send(PayloadServiceCommand::PayloadTimestamp(id, tx)).ok()?;
191        rx.await.ok()?
192    }
193}
194
195impl<T> Clone for PayloadBuilderHandle<T>
196where
197    T: PayloadTypes,
198{
199    fn clone(&self) -> Self {
200        Self { to_service: self.to_service.clone() }
201    }
202}
203
204/// A service that manages payload building tasks.
205///
206/// This type is an endless future that manages the building of payloads.
207///
208/// It tracks active payloads and their build jobs that run in a worker pool.
209///
210/// By design, this type relies entirely on the [`PayloadJobGenerator`] to create new payloads and
211/// does know nothing about how to build them, it just drives their jobs to completion.
212#[derive(Debug)]
213#[must_use = "futures do nothing unless you `.await` or poll them"]
214pub struct PayloadBuilderService<Gen, St, T>
215where
216    T: PayloadTypes,
217    Gen: PayloadJobGenerator,
218    Gen::Job: PayloadJob<PayloadAttributes = T::PayloadAttributes>,
219{
220    /// The type that knows how to create new payloads.
221    generator: Gen,
222    /// All active payload jobs, each accompanied by its id and the caller's tracing span
223    /// propagated across the channel so that poll and resolve work appears as children of the
224    /// original Engine API request.
225    payload_jobs: Vec<PayloadJobEntry<Gen::Job>>,
226    /// Copy of the sender half, so new [`PayloadBuilderHandle`] can be created on demand.
227    service_tx: mpsc::UnboundedSender<PayloadServiceCommand<T>>,
228    /// Receiver half of the command channel.
229    command_rx: UnboundedReceiverStream<PayloadServiceCommand<T>>,
230    /// Metrics for the payload builder service
231    metrics: PayloadBuilderServiceMetrics,
232    /// Chain events notification stream
233    chain_events: St,
234    /// Payload events handler, used to broadcast and subscribe to payload events.
235    payload_events: broadcast::Sender<Events<T>>,
236    /// We retain latest resolved payload just to make sure that we can handle repeating
237    /// requests for it gracefully.
238    cached_payload_rx: watch::Receiver<Option<(PayloadId, BlockTimestamp, T::BuiltPayload)>>,
239    /// Sender half of the cached payload channel.
240    cached_payload_tx: watch::Sender<Option<(PayloadId, BlockTimestamp, T::BuiltPayload)>>,
241}
242
243const PAYLOAD_EVENTS_BUFFER_SIZE: usize = 20;
244
245// === impl PayloadBuilderService ===
246
247impl<Gen, St, T> PayloadBuilderService<Gen, St, T>
248where
249    T: PayloadTypes,
250    Gen: PayloadJobGenerator,
251    Gen::Job: PayloadJob<PayloadAttributes = T::PayloadAttributes>,
252    <Gen::Job as PayloadJob>::BuiltPayload: Into<T::BuiltPayload>,
253{
254    /// Creates a new payload builder service and returns the [`PayloadBuilderHandle`] to interact
255    /// with it.
256    ///
257    /// This also takes a stream of chain events that will be forwarded to the generator to apply
258    /// additional logic when new state is committed. See also
259    /// [`PayloadJobGenerator::on_new_state`].
260    pub fn new(generator: Gen, chain_events: St) -> (Self, PayloadBuilderHandle<T>) {
261        let (service_tx, command_rx) = mpsc::unbounded_channel();
262        let (payload_events, _) = broadcast::channel(PAYLOAD_EVENTS_BUFFER_SIZE);
263
264        let (cached_payload_tx, cached_payload_rx) = watch::channel(None);
265
266        let service = Self {
267            generator,
268            payload_jobs: Vec::new(),
269            service_tx,
270            command_rx: UnboundedReceiverStream::new(command_rx),
271            metrics: Default::default(),
272            chain_events,
273            payload_events,
274            cached_payload_rx,
275            cached_payload_tx,
276        };
277
278        let handle = service.handle();
279        (service, handle)
280    }
281
282    /// Returns a handle to the service.
283    pub fn handle(&self) -> PayloadBuilderHandle<T> {
284        PayloadBuilderHandle::new(self.service_tx.clone())
285    }
286
287    /// Create clone on `payload_events` sending handle that could be used by builder to produce
288    /// additional events during block building
289    pub fn payload_events_handle(&self) -> broadcast::Sender<Events<T>> {
290        self.payload_events.clone()
291    }
292
293    /// Returns true if the given payload is currently being built.
294    fn contains_payload(&self, id: PayloadId) -> bool {
295        self.payload_jobs.iter().any(|entry| entry.id == id)
296    }
297
298    /// Returns the best payload for the given identifier that has been built so far.
299    fn best_payload(&self, id: PayloadId) -> Option<Result<T::BuiltPayload, PayloadBuilderError>> {
300        let res = self
301            .payload_jobs
302            .iter()
303            .find(|entry| entry.id == id)
304            .map(|entry| entry.job.best_payload().map(|payload| payload.into()));
305        if let Some(Ok(ref best)) = res {
306            self.metrics.set_best_revenue(best.block().number(), f64::from(best.fees()));
307        }
308
309        res
310    }
311
312    /// Returns the best payload for the given identifier that has been built so far.
313    ///
314    /// If the job should be terminated, this removes it from active polling and returns it so the
315    /// caller can drop it after the response is sent.
316    fn resolve(
317        &mut self,
318        id: PayloadId,
319        kind: PayloadKind,
320    ) -> ResolvePayloadResult<T::BuiltPayload, Gen::Job> {
321        let start = Instant::now();
322        debug!(target: "payload_builder", %id, "resolving payload job");
323
324        if let Some((cached, _, payload)) = &*self.cached_payload_rx.borrow() &&
325            *cached == id
326        {
327            self.metrics.resolve_duration_seconds.record(start.elapsed());
328            return (Some(Box::pin(core::future::ready(Ok(payload.clone())))), None);
329        }
330
331        let Some(job) = self.payload_jobs.iter().position(|entry| entry.id == id) else {
332            return (None, None)
333        };
334        let (fut, keep_alive) = self.payload_jobs[job].job.resolve_kind(kind);
335        let payload_timestamp = self.payload_jobs[job].job.payload_timestamp();
336
337        let mut resolved_job =
338            (keep_alive == KeepPayloadJobAlive::No).then(|| self.payload_jobs.swap_remove(job));
339        let leases = resolved_job
340            .as_mut()
341            .map(|entry| std::mem::take(&mut entry.leases))
342            .unwrap_or_default();
343
344        // Since the fees will not be known until the payload future is resolved / awaited, we wrap
345        // the future in a new future that will update the metrics.
346        let resolved_metrics = self.metrics.clone();
347        let payload_events = self.payload_events.clone();
348        let cached_payload_tx = self.cached_payload_tx.clone();
349
350        let fut = async move {
351            let _leases = leases;
352            let res = fut.await;
353            resolved_metrics.resolve_duration_seconds.record(start.elapsed());
354            if let Ok(payload) = &res {
355                if payload_events.receiver_count() > 0 {
356                    payload_events.send(Events::BuiltPayload(payload.clone().into())).ok();
357                }
358
359                if let Ok(timestamp) = payload_timestamp {
360                    let _ = cached_payload_tx.send(Some((id, timestamp, payload.clone().into())));
361                }
362
363                resolved_metrics
364                    .set_resolved_revenue(payload.block().number(), f64::from(payload.fees()));
365            }
366            res.map(|p| p.into())
367        };
368
369        (Some(Box::pin(fut)), resolved_job)
370    }
371
372    /// Returns the payload timestamp for the given payload.
373    fn payload_timestamp(&self, id: PayloadId) -> Option<Result<u64, PayloadBuilderError>> {
374        if let Some((cached_id, timestamp, _)) = *self.cached_payload_rx.borrow() &&
375            cached_id == id
376        {
377            return Some(Ok(timestamp));
378        }
379
380        let timestamp = self
381            .payload_jobs
382            .iter()
383            .find(|entry| entry.id == id)
384            .map(|entry| entry.job.payload_timestamp());
385
386        if timestamp.is_none() {
387            trace!(target: "payload_builder", %id, "no matching payload job found to get timestamp for");
388        }
389
390        timestamp
391    }
392}
393
394impl<Gen, St, T, N> Future for PayloadBuilderService<Gen, St, T>
395where
396    T: PayloadTypes,
397    N: NodePrimitives,
398    Gen: PayloadJobGenerator + Unpin + 'static,
399    <Gen as PayloadJobGenerator>::Job: Unpin + 'static,
400    St: Stream<Item = CanonStateNotification<N>> + Send + Unpin + 'static,
401    Gen::Job: PayloadJob<PayloadAttributes = T::PayloadAttributes>,
402    <Gen::Job as PayloadJob>::BuiltPayload: Into<T::BuiltPayload>,
403{
404    type Output = ();
405
406    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
407        let this = self.get_mut();
408        loop {
409            // notify the generator of new chain events
410            while let Poll::Ready(Some(new_head)) = this.chain_events.poll_next_unpin(cx) {
411                this.generator.on_new_state(new_head);
412            }
413
414            // we poll all jobs first, so we always have the latest payload that we can report if
415            // requests
416            // we don't care about the order of the jobs, so we can just swap_remove them
417            for idx in (0..this.payload_jobs.len()).rev() {
418                let PayloadJobEntry { mut job, id, span, leases } =
419                    this.payload_jobs.swap_remove(idx);
420
421                let poll_result = {
422                    let _entered = span.enter();
423                    job.poll_unpin(cx)
424                };
425
426                match poll_result {
427                    Poll::Ready(Ok(_)) => {
428                        this.metrics.set_active_jobs(this.payload_jobs.len());
429                        trace!(target: "payload_builder", %id, "payload job finished");
430                    }
431                    Poll::Ready(Err(err)) => {
432                        warn!(target: "payload_builder",%err, ?id, "Payload builder job failed; resolving payload");
433                        this.metrics.inc_failed_jobs();
434                        this.metrics.set_active_jobs(this.payload_jobs.len());
435                    }
436                    Poll::Pending => {
437                        this.payload_jobs.push(PayloadJobEntry { job, id, span, leases });
438                    }
439                }
440            }
441
442            // marker for exit condition
443            let mut new_job = false;
444
445            // drain all requests
446            while let Poll::Ready(Some(cmd)) = this.command_rx.poll_next_unpin(cx) {
447                match cmd {
448                    PayloadServiceCommand::BuildNewPayload(input, job_span, tx) => {
449                        let id = input.payload_id();
450                        let mut res = Ok(id);
451                        let parent = input.parent_hash;
452
453                        if this.contains_payload(id) {
454                            debug!(target: "payload_builder", %id, %parent, "Payload job already in progress, ignoring.");
455                        } else {
456                            let start = Instant::now();
457                            let attributes = input.attributes.clone();
458                            // Keep a service-owned reference for generic payload jobs. Builders
459                            // that move work into detached tasks can retain their own references
460                            // from `resources` until those tasks exit.
461                            let leases = input.resources.clone_leases();
462                            let job_result = {
463                                let _entered = job_span.enter();
464                                this.generator.new_payload_job(*input, id)
465                            };
466
467                            match job_result {
468                                Ok(job) => {
469                                    this.metrics.new_job_duration_seconds.record(start.elapsed());
470                                    info!(target: "payload_builder", %id, %parent, "New payload job created");
471                                    this.metrics.inc_initiated_jobs();
472                                    new_job = true;
473                                    this.payload_jobs.push(PayloadJobEntry {
474                                        job,
475                                        id,
476                                        span: job_span,
477                                        leases,
478                                    });
479                                    this.payload_events.send(Events::Attributes(attributes)).ok();
480
481                                    // Clear stale cached payload for this id so
482                                    // resolve() never returns an outdated result
483                                    // from a previous job with the same id.
484                                    if this
485                                        .cached_payload_rx
486                                        .borrow()
487                                        .as_ref()
488                                        .is_some_and(|(cached_id, _, _)| *cached_id == id)
489                                    {
490                                        trace!(target: "payload_builder", %id, "clearing stale cached payload for reused payload id");
491                                        let _ = this.cached_payload_tx.send(None);
492                                    }
493                                }
494                                Err(err) => {
495                                    this.metrics.new_job_duration_seconds.record(start.elapsed());
496                                    this.metrics.inc_failed_jobs();
497                                    warn!(target: "payload_builder", %err, %id, "Failed to create payload builder job");
498                                    res = Err(err);
499                                }
500                            }
501                        }
502
503                        let _ = tx.send(res);
504                    }
505                    PayloadServiceCommand::BestPayload(id, tx) => {
506                        let _ = tx.send(this.best_payload(id));
507                    }
508                    PayloadServiceCommand::PayloadTimestamp(id, tx) => {
509                        let timestamp = this.payload_timestamp(id);
510                        let _ = tx.send(timestamp);
511                    }
512                    PayloadServiceCommand::Resolve(id, strategy, tx) => {
513                        let (payload_fut, resolved_job) = this.resolve(id, strategy);
514                        let _ = tx.send(payload_fut);
515
516                        if let Some(entry) = resolved_job {
517                            debug!(target: "payload_builder", id = %entry.id, "terminated resolved job");
518                        }
519                    }
520                    PayloadServiceCommand::Subscribe(tx) => {
521                        let new_rx = this.payload_events.subscribe();
522                        let _ = tx.send(new_rx);
523                    }
524                }
525            }
526
527            if !new_job {
528                return Poll::Pending
529            }
530        }
531    }
532}
533
534/// Message type for the [`PayloadBuilderService`].
535#[derive(derive_more::Debug)]
536pub enum PayloadServiceCommand<T: PayloadTypes> {
537    /// Start building a new payload.
538    ///
539    /// Carries the caller's [`Span`] so the service can parent payload-building work under the
540    /// originating Engine API trace.
541    ///
542    /// The sender receives the id of the started payload job. Answering with
543    /// [`PayloadBuilderError::InvalidAttributes`] rejects the payload attributes; any other error,
544    /// or dropping the sender, is reported to the CL as an internal error.
545    BuildNewPayload(
546        Box<BuildNewPayload<T::PayloadAttributes>>,
547        Span,
548        oneshot::Sender<Result<PayloadId, PayloadBuilderError>>,
549    ),
550    /// Get the best payload so far
551    BestPayload(PayloadId, oneshot::Sender<Option<Result<T::BuiltPayload, PayloadBuilderError>>>),
552    /// Get the payload timestamp for the given payload
553    PayloadTimestamp(PayloadId, oneshot::Sender<Option<Result<u64, PayloadBuilderError>>>),
554    /// Resolve the payload and return the payload
555    Resolve(
556        PayloadId,
557        /* kind: */ PayloadKind,
558        #[debug(skip)] oneshot::Sender<Option<PayloadFuture<T::BuiltPayload>>>,
559    ),
560    /// Payload service events
561    Subscribe(oneshot::Sender<broadcast::Receiver<Events<T>>>),
562}
563
564/// A request to build a new payload.
565#[derive(Debug)]
566pub struct BuildNewPayload<T> {
567    /// The attributes for the new payload
568    pub attributes: T,
569    /// The parent hash of the new payload
570    pub parent_hash: B256,
571    /// Resources loaned to the payload builder for this job.
572    pub resources: PayloadBuilderResources,
573}
574
575impl<T: PayloadAttributes> BuildNewPayload<T> {
576    /// Returns the payload id for the new payload.
577    pub fn payload_id(&self) -> PayloadId {
578        self.attributes.payload_id(&self.parent_hash)
579    }
580}
581
582/// Resources loaned to a payload builder job by the engine.
583#[derive(Debug, Default)]
584pub struct PayloadBuilderResources {
585    /// Optional execution cache to use for the payload.
586    ///
587    /// Only provided if `--engine.share-execution-cache-with-payload-builder` is enabled.
588    execution_cache: Option<SavedCache>,
589    /// Optional handle to a background state-root task.
590    state_root_handle: Option<PayloadStateRootHandle>,
591    /// Lifecycle leases retained by the service or by detached payload build tasks.
592    leases: Vec<PayloadBuilderLease>,
593}
594
595impl PayloadBuilderResources {
596    /// Creates a new payload builder resource bundle.
597    pub const fn new(
598        execution_cache: Option<SavedCache>,
599        state_root_handle: Option<PayloadStateRootHandle>,
600    ) -> Self {
601        Self { execution_cache, state_root_handle, leases: Vec::new() }
602    }
603
604    /// Adds a lease for this payload build.
605    pub fn with_lease(mut self, lease: PayloadBuilderLease) -> Self {
606        self.leases.push(lease);
607        self
608    }
609
610    /// Returns the loaned execution cache, if any.
611    pub const fn execution_cache(&self) -> Option<&SavedCache> {
612        self.execution_cache.as_ref()
613    }
614
615    /// Takes the loaned execution cache, if any.
616    pub const fn take_execution_cache(&mut self) -> Option<SavedCache> {
617        self.execution_cache.take()
618    }
619
620    /// Returns the loaned state-root task handle, if any.
621    pub const fn state_root_handle(&self) -> Option<&PayloadStateRootHandle> {
622        self.state_root_handle.as_ref()
623    }
624
625    /// Takes the loaned state-root task handle, if any.
626    pub const fn take_state_root_handle(&mut self) -> Option<PayloadStateRootHandle> {
627        self.state_root_handle.take()
628    }
629
630    /// Takes lifecycle leases for a payload job that owns detached work.
631    pub fn take_leases(&mut self) -> Vec<PayloadBuilderLease> {
632        std::mem::take(&mut self.leases)
633    }
634
635    /// Clones lifecycle leases for the payload builder service to retain.
636    fn clone_leases(&self) -> Vec<PayloadBuilderLease> {
637        self.leases.clone()
638    }
639}
640
641/// Keeps a loaned resource active until the last lease clone is dropped.
642#[derive(Clone)]
643pub struct PayloadBuilderLease {
644    _lease: Arc<dyn Send + Sync>,
645}
646
647impl PayloadBuilderLease {
648    /// Wraps a lease that releases its resource when the last clone is dropped.
649    pub fn new(lease: impl Send + Sync + 'static) -> Self {
650        Self { _lease: Arc::new(lease) }
651    }
652}
653
654impl std::fmt::Debug for PayloadBuilderLease {
655    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
656        f.debug_struct("PayloadBuilderLease").finish_non_exhaustive()
657    }
658}
659
660/// An active payload job and its service metadata.
661#[derive(Debug)]
662struct PayloadJobEntry<Job> {
663    job: Job,
664    id: PayloadId,
665    span: Span,
666    leases: Vec<PayloadBuilderLease>,
667}
668
669#[cfg(test)]
670mod tests {
671    use super::*;
672    use crate::test_utils::test_payload_service;
673    use alloy_primitives::Address;
674    use reth_ethereum_engine_primitives::{EthEngineTypes, EthPayloadAttributes};
675    use std::sync::atomic::{AtomicBool, Ordering};
676
677    struct DropProbe(Arc<AtomicBool>);
678
679    impl Drop for DropProbe {
680        fn drop(&mut self) {
681            self.0.store(true, Ordering::Release);
682        }
683    }
684
685    #[test]
686    fn payload_builder_lease_is_held_until_resolve_finishes() {
687        tokio::runtime::Builder::new_current_thread().build().unwrap().block_on(async {
688            let (service, handle) = test_payload_service::<EthEngineTypes>();
689            let service = tokio::spawn(service);
690            let dropped = Arc::new(AtomicBool::new(false));
691            let lease = PayloadBuilderLease::new(DropProbe(Arc::clone(&dropped)));
692            let input = BuildNewPayload {
693                attributes: EthPayloadAttributes {
694                    timestamp: 1,
695                    prev_randao: B256::ZERO,
696                    suggested_fee_recipient: Address::ZERO,
697                    withdrawals: None,
698                    parent_beacon_block_root: None,
699                    slot_number: None,
700                    ..Default::default()
701                },
702                parent_hash: B256::ZERO,
703                resources: PayloadBuilderResources::default().with_lease(lease),
704            };
705
706            let id = handle.send_new_payload(input).await.unwrap().unwrap();
707            assert!(!dropped.load(Ordering::Acquire));
708
709            handle.resolve_kind(id, PayloadKind::Earliest).await.unwrap().unwrap();
710            assert!(dropped.load(Ordering::Acquire));
711            service.abort();
712        });
713    }
714}
reth_payload_builder/service.rs

reth_payload_builder/
service.rs
