reth_engine_tree/
backfill.rs

1//! It is expected that the node has two sync modes:
2//!
3//!  - Backfill sync: Sync to a certain block height in stages, e.g. download data from p2p then
4//!    execute that range.
5//!  - Live sync: In this mode the node is keeping up with the latest tip and listens for new
6//!    requests from the consensus client.
7//!
8//! These modes are mutually exclusive and the node can only be in one mode at a time.
9
10use alloy_primitives::B256;
11use futures::FutureExt;
12use reth_provider::providers::ProviderNodeTypes;
13use reth_stages_api::{ControlFlow, Pipeline, PipelineError, PipelineTarget, PipelineWithResult};
14use reth_tasks::Runtime;
15use std::task::{ready, Context, Poll};
16use tokio::sync::oneshot;
17use tracing::trace;
18
19/// Represents the state of the backfill synchronization process.
20#[derive(Debug, PartialEq, Eq, Default)]
21pub enum BackfillSyncState {
22    /// The node is not performing any backfill synchronization.
23    /// This is the initial or default state.
24    #[default]
25    Idle,
26    /// A backfill action has been sent to the pipeline, but processing has not started yet.
27    Pending,
28    /// A backfill synchronization must be re-evaluated after persistence catches up.
29    PendingRevalidation,
30    /// The node is actively engaged in backfill synchronization.
31    Active,
32}
33
34impl BackfillSyncState {
35    /// Returns true if the state is idle.
36    pub const fn is_idle(&self) -> bool {
37        matches!(self, Self::Idle)
38    }
39
40    /// Returns true if the state is pending.
41    pub const fn is_pending(&self) -> bool {
42        matches!(self, Self::Pending)
43    }
44
45    /// Returns true if backfill must be re-evaluated before it can start.
46    pub const fn is_pending_revalidation(&self) -> bool {
47        matches!(self, Self::PendingRevalidation)
48    }
49
50    /// Returns true if the state is active.
51    pub const fn is_active(&self) -> bool {
52        matches!(self, Self::Active)
53    }
54}
55
56/// Backfill sync mode functionality.
57pub trait BackfillSync: Send {
58    /// Performs a backfill action.
59    fn on_action(&mut self, action: BackfillAction);
60
61    /// Polls the pipeline for completion.
62    fn poll(&mut self, cx: &mut Context<'_>) -> Poll<BackfillEvent>;
63}
64
65/// The backfill actions that can be performed.
66#[derive(Debug, Clone, PartialEq, Eq)]
67pub enum BackfillAction {
68    /// Start backfilling with the given target.
69    Start(PipelineTarget),
70    /// Moves the target of a running backfill to the new forkchoice head, without starting another
71    /// run. It may arrive when no run is active, in which case it should be ignored.
72    UpdateTarget(B256),
73    /// Reports the new finalized block, which a backfill may anchor irreversible work to.
74    /// Backfills that use finality retain it while idle so the next run can use it.
75    UpdateFinalized(B256),
76}
77
78/// The events that can be emitted on backfill sync.
79#[derive(Debug)]
80pub enum BackfillEvent {
81    /// Backfill sync started.
82    Started(PipelineTarget),
83    /// Backfill sync finished.
84    ///
85    /// If this is returned, backfill sync is idle.
86    Finished(Result<ControlFlow, PipelineError>),
87    /// Sync task was dropped after it was started, unable to receive it because
88    /// channel closed. This would indicate a panicked task.
89    TaskDropped(String),
90}
91
92/// Pipeline sync.
93#[derive(Debug)]
94pub struct PipelineSync<N: ProviderNodeTypes> {
95    /// The type that can spawn the pipeline task.
96    pipeline_task_spawner: Runtime,
97    /// The current state of the pipeline.
98    /// The pipeline is used for large ranges.
99    pipeline_state: PipelineState<N>,
100    /// Pending target block for the pipeline to sync
101    pending_pipeline_target: Option<PipelineTarget>,
102}
103
104impl<N: ProviderNodeTypes> PipelineSync<N> {
105    /// Create a new instance.
106    pub fn new(pipeline: Pipeline<N>, pipeline_task_spawner: Runtime) -> Self {
107        Self {
108            pipeline_task_spawner,
109            pipeline_state: PipelineState::Idle(Some(Box::new(pipeline))),
110            pending_pipeline_target: None,
111        }
112    }
113
114    /// Returns `true` if a pipeline target is queued and will be triggered on the next `poll`.
115    #[expect(dead_code)]
116    const fn is_pipeline_sync_pending(&self) -> bool {
117        self.pending_pipeline_target.is_some() && self.pipeline_state.is_idle()
118    }
119
120    /// Returns `true` if the pipeline is idle.
121    const fn is_pipeline_idle(&self) -> bool {
122        self.pipeline_state.is_idle()
123    }
124
125    /// Returns `true` if the pipeline is active.
126    const fn is_pipeline_active(&self) -> bool {
127        !self.is_pipeline_idle()
128    }
129
130    /// Sets a new target to sync the pipeline to.
131    ///
132    /// But ensures the target is not the zero hash.
133    fn set_pipeline_sync_target(&mut self, target: PipelineTarget) {
134        if target.sync_target().is_some_and(|target| target.is_zero()) {
135            trace!(
136                target: "consensus::engine::sync",
137                "Pipeline target cannot be zero hash."
138            );
139            // precaution to never sync to the zero hash
140            return
141        }
142        self.pending_pipeline_target = Some(target);
143    }
144
145    /// This will spawn the pipeline if it is idle and a target is set or if the pipeline is set to
146    /// run continuously.
147    fn try_spawn_pipeline(&mut self) -> Option<BackfillEvent> {
148        match &mut self.pipeline_state {
149            PipelineState::Idle(pipeline) => {
150                let target = self.pending_pipeline_target.take()?;
151                let (tx, rx) = oneshot::channel();
152
153                let pipeline = pipeline.take().expect("exists");
154                self.pipeline_task_spawner.spawn_critical_blocking_task(
155                    "pipeline task",
156                    async move {
157                        let result = pipeline.run_as_fut(Some(target)).await;
158                        let _ = tx.send(result);
159                    },
160                );
161                self.pipeline_state = PipelineState::Running(rx);
162
163                Some(BackfillEvent::Started(target))
164            }
165            PipelineState::Running(_) => None,
166        }
167    }
168
169    /// Advances the pipeline state.
170    ///
171    /// This checks for the result in the channel, or returns pending if the pipeline is idle.
172    fn poll_pipeline(&mut self, cx: &mut Context<'_>) -> Poll<BackfillEvent> {
173        let res = match self.pipeline_state {
174            PipelineState::Idle(_) => return Poll::Pending,
175            PipelineState::Running(ref mut fut) => {
176                ready!(fut.poll_unpin(cx))
177            }
178        };
179        let ev = match res {
180            Ok((pipeline, result)) => {
181                self.pipeline_state = PipelineState::Idle(Some(Box::new(pipeline)));
182                BackfillEvent::Finished(result)
183            }
184            Err(why) => {
185                // failed to receive the pipeline
186                BackfillEvent::TaskDropped(why.to_string())
187            }
188        };
189        Poll::Ready(ev)
190    }
191}
192
193impl<N: ProviderNodeTypes> BackfillSync for PipelineSync<N> {
194    fn on_action(&mut self, event: BackfillAction) {
195        match event {
196            BackfillAction::Start(target) => self.set_pipeline_sync_target(target),
197            // Ordinary backfill finishes its current range before the engine re-evaluates FCU.
198            BackfillAction::UpdateTarget(_) | BackfillAction::UpdateFinalized(_) => {}
199        }
200    }
201
202    fn poll(&mut self, cx: &mut Context<'_>) -> Poll<BackfillEvent> {
203        // try to spawn a pipeline if a target is set
204        if let Some(event) = self.try_spawn_pipeline() {
205            return Poll::Ready(event)
206        }
207
208        // make sure we poll the pipeline if it's active, and return any ready pipeline events
209        if self.is_pipeline_active() {
210            // advance the pipeline
211            if let Poll::Ready(event) = self.poll_pipeline(cx) {
212                return Poll::Ready(event)
213            }
214        }
215
216        Poll::Pending
217    }
218}
219
220/// The possible pipeline states within the sync controller.
221///
222/// [`PipelineState::Idle`] means that the pipeline is currently idle.
223/// [`PipelineState::Running`] means that the pipeline is currently running.
224///
225/// NOTE: The differentiation between these two states is important, because when the pipeline is
226/// running, it acquires the write lock over the database. This means that we cannot forward to the
227/// blockchain tree any messages that would result in database writes, since it would result in a
228/// deadlock.
229#[derive(Debug)]
230enum PipelineState<N: ProviderNodeTypes> {
231    /// Pipeline is idle.
232    Idle(Option<Box<Pipeline<N>>>),
233    /// Pipeline is running and waiting for a response
234    Running(oneshot::Receiver<PipelineWithResult<N>>),
235}
236
237impl<N: ProviderNodeTypes> PipelineState<N> {
238    /// Returns `true` if the state matches idle.
239    const fn is_idle(&self) -> bool {
240        matches!(self, Self::Idle(_))
241    }
242}
243
244#[cfg(test)]
245mod tests {
246    use super::*;
247    use crate::{
248        chain::{ChainHandler, HandlerEvent},
249        download::{BlockDownloader, DownloadAction, DownloadOutcome},
250        engine::{EngineHandler, EngineRequestHandler, FromEngine, RequestHandlerEvent},
251        test_utils::{insert_headers_into_client, TestPipelineBuilder},
252    };
253    use alloy_consensus::Header;
254    use alloy_eips::eip1559::ETHEREUM_BLOCK_GAS_LIMIT_30M;
255    use alloy_primitives::{BlockNumber, B256};
256    use assert_matches::assert_matches;
257    use futures::poll;
258    use reth_chainspec::{ChainSpecBuilder, MAINNET};
259    use reth_ethereum_primitives::EthPrimitives;
260    use reth_network_p2p::test_utils::TestFullBlockClient;
261    use reth_primitives_traits::{NodePrimitives, SealedHeader};
262    use reth_provider::test_utils::MockNodeTypesWithDB;
263    use reth_stages::ExecOutput;
264    use reth_stages_api::StageCheckpoint;
265    use reth_tasks::Runtime;
266    use std::{
267        collections::VecDeque,
268        future::poll_fn,
269        sync::{
270            atomic::{AtomicUsize, Ordering},
271            Arc,
272        },
273        task::Waker,
274    };
275
276    type TestBlock = <EthPrimitives as NodePrimitives>::Block;
277
278    struct TestHarness {
279        pipeline_sync: PipelineSync<MockNodeTypesWithDB>,
280        tip: B256,
281    }
282
283    impl TestHarness {
284        fn new(total_blocks: usize, pipeline_done_after: u64) -> Self {
285            let chain_spec = Arc::new(
286                ChainSpecBuilder::default()
287                    .chain(MAINNET.chain)
288                    .genesis(MAINNET.genesis.clone())
289                    .paris_activated()
290                    .build(),
291            );
292
293            // force the pipeline to be "done" after `pipeline_done_after` blocks
294            let pipeline = TestPipelineBuilder::new()
295                .with_pipeline_exec_outputs(VecDeque::from([Ok(ExecOutput::done(
296                    StageCheckpoint::new(BlockNumber::from(pipeline_done_after)),
297                ))]))
298                .build(chain_spec);
299
300            let pipeline_sync = PipelineSync::new(pipeline, Runtime::test());
301            let client = TestFullBlockClient::default();
302            let header = Header {
303                base_fee_per_gas: Some(7),
304                gas_limit: ETHEREUM_BLOCK_GAS_LIMIT_30M,
305                ..Default::default()
306            };
307            let header = SealedHeader::seal_slow(header);
308            insert_headers_into_client(&client, header, 0..total_blocks);
309
310            let tip = client.highest_block().expect("there should be blocks here").hash();
311
312            Self { pipeline_sync, tip }
313        }
314    }
315
316    // Emits one queued backfill action per poll.
317    struct QueuedActions(VecDeque<BackfillAction>);
318
319    impl EngineRequestHandler for QueuedActions {
320        type Event = ();
321        type Request = ();
322        type Block = TestBlock;
323
324        fn on_event(&mut self, _event: FromEngine<Self::Request, Self::Block>) {}
325
326        fn poll(&mut self, _cx: &mut Context<'_>) -> Poll<RequestHandlerEvent<Self::Event>> {
327            self.0.pop_front().map_or(Poll::Pending, |action| {
328                Poll::Ready(RequestHandlerEvent::HandlerEvent(HandlerEvent::BackfillAction(action)))
329            })
330        }
331    }
332
333    // Counts how often in-flight downloads are cleared.
334    #[derive(Default)]
335    struct CountedClears(Arc<AtomicUsize>);
336
337    impl BlockDownloader for CountedClears {
338        type Block = TestBlock;
339
340        fn on_action(&mut self, action: DownloadAction) {
341            if matches!(action, DownloadAction::Clear) {
342                self.0.fetch_add(1, Ordering::Relaxed);
343            }
344        }
345
346        fn poll(&mut self, _cx: &mut Context<'_>) -> Poll<DownloadOutcome<Self::Block>> {
347            Poll::Pending
348        }
349    }
350
351    #[tokio::test]
352    async fn pipeline_started_and_finished() {
353        const TOTAL_BLOCKS: usize = 10;
354        const PIPELINE_DONE_AFTER: u64 = 5;
355        let TestHarness { mut pipeline_sync, tip } =
356            TestHarness::new(TOTAL_BLOCKS, PIPELINE_DONE_AFTER);
357
358        let sync_future = poll_fn(|cx| pipeline_sync.poll(cx));
359        let next_event = poll!(sync_future);
360
361        // sync target not set, pipeline not started
362        assert_matches!(next_event, Poll::Pending);
363
364        pipeline_sync.on_action(BackfillAction::Start(PipelineTarget::Sync(tip)));
365
366        let sync_future = poll_fn(|cx| pipeline_sync.poll(cx));
367        let next_event = poll!(sync_future);
368
369        // sync target set, pipeline started
370        assert_matches!(next_event, Poll::Ready(BackfillEvent::Started(target)) => {
371            assert_eq!(target.sync_target().unwrap(), tip);
372        });
373
374        // the next event should be the pipeline finishing in a good state
375        let sync_future = poll_fn(|cx| pipeline_sync.poll(cx));
376        let next_ready = sync_future.await;
377        assert_matches!(next_ready, BackfillEvent::Finished(result) => {
378            assert_matches!(result, Ok(control_flow) => assert_eq!(control_flow, ControlFlow::Continue { block_number: PIPELINE_DONE_AFTER }));
379        });
380    }
381
382    #[tokio::test]
383    async fn target_updates_do_not_start_the_pipeline() {
384        let TestHarness { mut pipeline_sync, tip } = TestHarness::new(10, 5);
385
386        pipeline_sync.on_action(BackfillAction::UpdateTarget(tip));
387
388        assert_matches!(poll!(poll_fn(|cx| pipeline_sync.poll(cx))), Poll::Pending);
389        assert!(pipeline_sync.is_pipeline_idle());
390    }
391
392    #[test]
393    fn only_a_new_backfill_run_clears_downloads() {
394        let head = B256::repeat_byte(1);
395        let actions = VecDeque::from([
396            BackfillAction::UpdateTarget(head),
397            BackfillAction::Start(PipelineTarget::Sync(head)),
398        ]);
399        let downloader = CountedClears::default();
400        let clears = Arc::clone(&downloader.0);
401        let mut handler =
402            EngineHandler::new(QueuedActions(actions), downloader, futures::stream::empty::<()>());
403        let mut cx = Context::from_waker(Waker::noop());
404
405        assert!(matches!(
406            ChainHandler::poll(&mut handler, &mut cx),
407            Poll::Ready(HandlerEvent::BackfillAction(BackfillAction::UpdateTarget(_)))
408        ));
409        assert_eq!(clears.load(Ordering::Relaxed), 0);
410
411        assert!(matches!(
412            ChainHandler::poll(&mut handler, &mut cx),
413            Poll::Ready(HandlerEvent::BackfillAction(BackfillAction::Start(_)))
414        ));
415        assert_eq!(clears.load(Ordering::Relaxed), 1);
416    }
417}
reth_engine_tree/backfill.rs

reth_engine_tree/
backfill.rs
