arb_node/
payload.rs

1//! Arbitrum payload service builder.
2//!
3//! Spawns a minimal payload service that handles Subscribe commands
4//! by returning a valid broadcast receiver. Block building is driven
5//! externally by the sequencer via RPC, not by the payload builder.
6
7use std::{
8    future::Future,
9    pin::Pin,
10    task::{Context, Poll},
11};
12
13use futures_util::{StreamExt, ready};
14use reth_node_builder::{
15    BuilderContext, FullNodeTypes, NodeTypes, components::PayloadServiceBuilder,
16};
17use reth_payload_builder::{PayloadBuilderHandle, PayloadServiceCommand};
18use reth_payload_primitives::{PayloadBuilderAttributes, PayloadTypes};
19use reth_transaction_pool::TransactionPool;
20use tokio::sync::{broadcast, mpsc};
21use tokio_stream::wrappers::UnboundedReceiverStream;
22use tracing::info;
23
24/// Payload builder service that handles Subscribe commands properly.
25///
26/// The noop service drops Subscribe senders, which causes reth's engine
27/// tree to fail with "ChannelClosed". This service keeps a broadcast
28/// channel alive and responds to Subscribe with a valid receiver.
29struct ArbPayloadService<T: PayloadTypes> {
30    command_rx: UnboundedReceiverStream<PayloadServiceCommand<T>>,
31    events_tx: broadcast::Sender<reth_payload_builder_primitives::Events<T>>,
32}
33
34impl<T: PayloadTypes> ArbPayloadService<T> {
35    fn new() -> (Self, PayloadBuilderHandle<T>) {
36        let (service_tx, command_rx) = mpsc::unbounded_channel();
37        let (events_tx, _) = broadcast::channel(16);
38        (
39            Self {
40                command_rx: UnboundedReceiverStream::new(command_rx),
41                events_tx,
42            },
43            PayloadBuilderHandle::new(service_tx),
44        )
45    }
46}
47
48impl<T: PayloadTypes> Future for ArbPayloadService<T> {
49    type Output = ();
50
51    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
52        let this = self.get_mut();
53        loop {
54            let Some(cmd) = ready!(this.command_rx.poll_next_unpin(cx)) else {
55                return Poll::Ready(());
56            };
57            match cmd {
58                PayloadServiceCommand::BuildNewPayload(attr, tx) => {
59                    let id = attr.payload_id();
60                    let _ = tx.send(Ok(id));
61                }
62                PayloadServiceCommand::BestPayload(_, tx) => {
63                    let _ = tx.send(None);
64                }
65                PayloadServiceCommand::PayloadTimestamp(_, tx) => {
66                    let _ = tx.send(None);
67                }
68                PayloadServiceCommand::Resolve(_, _, tx) => {
69                    let _ = tx.send(None);
70                }
71                PayloadServiceCommand::Subscribe(tx) => {
72                    let rx = this.events_tx.subscribe();
73                    let _ = tx.send(rx);
74                }
75            }
76        }
77    }
78}
79
80/// Builder for the Arbitrum payload service.
81///
82/// Spawns a minimal payload builder service. Block building is driven
83/// by the sequencer through RPC calls, not through the payload service.
84#[derive(Debug, Default, Clone, Copy)]
85pub struct ArbPayloadServiceBuilder;
86
87impl<Node, Pool, Evm> PayloadServiceBuilder<Node, Pool, Evm> for ArbPayloadServiceBuilder
88where
89    Node: FullNodeTypes,
90    Pool: TransactionPool + Unpin + 'static,
91    Evm: Send + 'static,
92{
93    async fn spawn_payload_builder_service(
94        self,
95        ctx: &BuilderContext<Node>,
96        _pool: Pool,
97        _evm_config: Evm,
98    ) -> eyre::Result<PayloadBuilderHandle<<Node::Types as NodeTypes>::Payload>> {
99        let (service, handle) = ArbPayloadService::<<Node::Types as NodeTypes>::Payload>::new();
100        ctx.task_executor()
101            .spawn_critical_task("payload builder service", Box::pin(service));
102        info!(target: "reth::cli", "Payload builder service initialized");
103        Ok(handle)
104    }
105}
arb_node/payload.rs

arb_node/
payload.rs
