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(mut 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                            let leases = input.resources.take_leases();
459                            let job_result = {
460                                let _entered = job_span.enter();
461                                this.generator.new_payload_job(*input, id)
462                            };
463
464                            match job_result {
465                                Ok(job) => {
466                                    this.metrics.new_job_duration_seconds.record(start.elapsed());
467                                    info!(target: "payload_builder", %id, %parent, "New payload job created");
468                                    this.metrics.inc_initiated_jobs();
469                                    new_job = true;
470                                    this.payload_jobs.push(PayloadJobEntry {
471                                        job,
472                                        id,
473                                        span: job_span,
474                                        leases,
475                                    });
476                                    this.payload_events.send(Events::Attributes(attributes)).ok();
477
478                                    // Clear stale cached payload for this id so
479                                    // resolve() never returns an outdated result
480                                    // from a previous job with the same id.
481                                    if this
482                                        .cached_payload_rx
483                                        .borrow()
484                                        .as_ref()
485                                        .is_some_and(|(cached_id, _, _)| *cached_id == id)
486                                    {
487                                        trace!(target: "payload_builder", %id, "clearing stale cached payload for reused payload id");
488                                        let _ = this.cached_payload_tx.send(None);
489                                    }
490                                }
491                                Err(err) => {
492                                    this.metrics.new_job_duration_seconds.record(start.elapsed());
493                                    this.metrics.inc_failed_jobs();
494                                    warn!(target: "payload_builder", %err, %id, "Failed to create payload builder job");
495                                    res = Err(err);
496                                }
497                            }
498                        }
499
500                        let _ = tx.send(res);
501                    }
502                    PayloadServiceCommand::BestPayload(id, tx) => {
503                        let _ = tx.send(this.best_payload(id));
504                    }
505                    PayloadServiceCommand::PayloadTimestamp(id, tx) => {
506                        let timestamp = this.payload_timestamp(id);
507                        let _ = tx.send(timestamp);
508                    }
509                    PayloadServiceCommand::Resolve(id, strategy, tx) => {
510                        let (payload_fut, resolved_job) = this.resolve(id, strategy);
511                        let _ = tx.send(payload_fut);
512
513                        if let Some(entry) = resolved_job {
514                            debug!(target: "payload_builder", id = %entry.id, "terminated resolved job");
515                        }
516                    }
517                    PayloadServiceCommand::Subscribe(tx) => {
518                        let new_rx = this.payload_events.subscribe();
519                        let _ = tx.send(new_rx);
520                    }
521                }
522            }
523
524            if !new_job {
525                return Poll::Pending
526            }
527        }
528    }
529}
530
531/// Message type for the [`PayloadBuilderService`].
532#[derive(derive_more::Debug)]
533pub enum PayloadServiceCommand<T: PayloadTypes> {
534    /// Start building a new payload.
535    ///
536    /// Carries the caller's [`Span`] so the service can parent payload-building work under the
537    /// originating Engine API trace.
538    BuildNewPayload(
539        Box<BuildNewPayload<T::PayloadAttributes>>,
540        Span,
541        oneshot::Sender<Result<PayloadId, PayloadBuilderError>>,
542    ),
543    /// Get the best payload so far
544    BestPayload(PayloadId, oneshot::Sender<Option<Result<T::BuiltPayload, PayloadBuilderError>>>),
545    /// Get the payload timestamp for the given payload
546    PayloadTimestamp(PayloadId, oneshot::Sender<Option<Result<u64, PayloadBuilderError>>>),
547    /// Resolve the payload and return the payload
548    Resolve(
549        PayloadId,
550        /* kind: */ PayloadKind,
551        #[debug(skip)] oneshot::Sender<Option<PayloadFuture<T::BuiltPayload>>>,
552    ),
553    /// Payload service events
554    Subscribe(oneshot::Sender<broadcast::Receiver<Events<T>>>),
555}
556
557/// A request to build a new payload.
558#[derive(Debug)]
559pub struct BuildNewPayload<T> {
560    /// The attributes for the new payload
561    pub attributes: T,
562    /// The parent hash of the new payload
563    pub parent_hash: B256,
564    /// Resources loaned to the payload builder for this job.
565    pub resources: PayloadBuilderResources,
566}
567
568impl<T: PayloadAttributes> BuildNewPayload<T> {
569    /// Returns the payload id for the new payload.
570    pub fn payload_id(&self) -> PayloadId {
571        self.attributes.payload_id(&self.parent_hash)
572    }
573}
574
575/// Resources loaned to a payload builder job by the engine.
576#[derive(Debug, Default)]
577pub struct PayloadBuilderResources {
578    /// Optional execution cache to use for the payload.
579    ///
580    /// Only provided if `--engine.share-execution-cache-with-payload-builder` is enabled.
581    execution_cache: Option<SavedCache>,
582    /// Optional handle to a background state-root task.
583    state_root_handle: Option<PayloadStateRootHandle>,
584    /// Lifecycle leases retained by the service while the payload job is active.
585    leases: Vec<PayloadBuilderLease>,
586}
587
588impl PayloadBuilderResources {
589    /// Creates a new payload builder resource bundle.
590    pub const fn new(
591        execution_cache: Option<SavedCache>,
592        state_root_handle: Option<PayloadStateRootHandle>,
593    ) -> Self {
594        Self { execution_cache, state_root_handle, leases: Vec::new() }
595    }
596
597    /// Adds a lease that remains active for the lifetime of the payload job.
598    pub fn with_lease(mut self, lease: PayloadBuilderLease) -> Self {
599        self.leases.push(lease);
600        self
601    }
602
603    /// Returns the loaned execution cache, if any.
604    pub const fn execution_cache(&self) -> Option<&SavedCache> {
605        self.execution_cache.as_ref()
606    }
607
608    /// Takes the loaned execution cache, if any.
609    pub const fn take_execution_cache(&mut self) -> Option<SavedCache> {
610        self.execution_cache.take()
611    }
612
613    /// Returns the loaned state-root task handle, if any.
614    pub const fn state_root_handle(&self) -> Option<&PayloadStateRootHandle> {
615        self.state_root_handle.as_ref()
616    }
617
618    /// Takes the loaned state-root task handle, if any.
619    pub const fn take_state_root_handle(&mut self) -> Option<PayloadStateRootHandle> {
620        self.state_root_handle.take()
621    }
622
623    /// Takes the lifecycle leases that the service must retain for this job.
624    fn take_leases(&mut self) -> Vec<PayloadBuilderLease> {
625        std::mem::take(&mut self.leases)
626    }
627}
628
629/// Keeps a loaned resource active for the lifetime of a payload job.
630pub struct PayloadBuilderLease {
631    _lease: Box<dyn Send>,
632}
633
634impl PayloadBuilderLease {
635    /// Wraps a lease that releases its resource when dropped.
636    pub fn new(lease: impl Send + 'static) -> Self {
637        Self { _lease: Box::new(lease) }
638    }
639}
640
641impl std::fmt::Debug for PayloadBuilderLease {
642    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
643        f.debug_struct("PayloadBuilderLease").finish_non_exhaustive()
644    }
645}
646
647/// An active payload job and its service metadata.
648#[derive(Debug)]
649struct PayloadJobEntry<Job> {
650    job: Job,
651    id: PayloadId,
652    span: Span,
653    leases: Vec<PayloadBuilderLease>,
654}
655
656#[cfg(test)]
657mod tests {
658    use super::*;
659    use crate::test_utils::test_payload_service;
660    use alloy_primitives::Address;
661    use reth_ethereum_engine_primitives::{EthEngineTypes, EthPayloadAttributes};
662    use std::sync::atomic::{AtomicBool, Ordering};
663
664    struct DropProbe(Arc<AtomicBool>);
665
666    impl Drop for DropProbe {
667        fn drop(&mut self) {
668            self.0.store(true, Ordering::Release);
669        }
670    }
671
672    #[test]
673    fn payload_builder_lease_is_held_until_resolve_finishes() {
674        tokio::runtime::Builder::new_current_thread().build().unwrap().block_on(async {
675            let (service, handle) = test_payload_service::<EthEngineTypes>();
676            let service = tokio::spawn(service);
677            let dropped = Arc::new(AtomicBool::new(false));
678            let lease = PayloadBuilderLease::new(DropProbe(Arc::clone(&dropped)));
679            let input = BuildNewPayload {
680                attributes: EthPayloadAttributes {
681                    timestamp: 1,
682                    prev_randao: B256::ZERO,
683                    suggested_fee_recipient: Address::ZERO,
684                    withdrawals: None,
685                    parent_beacon_block_root: None,
686                    slot_number: None,
687                    target_gas_limit: None,
688                },
689                parent_hash: B256::ZERO,
690                resources: PayloadBuilderResources::default().with_lease(lease),
691            };
692
693            let id = handle.send_new_payload(input).await.unwrap().unwrap();
694            assert!(!dropped.load(Ordering::Acquire));
695
696            handle.resolve_kind(id, PayloadKind::Earliest).await.unwrap().unwrap();
697            assert!(dropped.load(Ordering::Acquire));
698            service.abort();
699        });
700    }
701}
reth_payload_builder/service.rs

reth_payload_builder/
service.rs
