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 futures::FutureExt;
11use reth_provider::providers::ProviderNodeTypes;
12use reth_stages_api::{ControlFlow, Pipeline, PipelineError, PipelineTarget, PipelineWithResult};
13use reth_tasks::Runtime;
14use std::task::{ready, Context, Poll};
15use tokio::sync::oneshot;
16use tracing::trace;
17
18/// Represents the state of the backfill synchronization process.
19#[derive(Debug, PartialEq, Eq, Default)]
20pub enum BackfillSyncState {
21    /// The node is not performing any backfill synchronization.
22    /// This is the initial or default state.
23    #[default]
24    Idle,
25    /// A backfill action has been sent to the pipeline, but processing has not started yet.
26    Pending,
27    /// A backfill synchronization must be re-evaluated after persistence catches up.
28    PendingRevalidation,
29    /// The node is actively engaged in backfill synchronization.
30    Active,
31}
32
33impl BackfillSyncState {
34    /// Returns true if the state is idle.
35    pub const fn is_idle(&self) -> bool {
36        matches!(self, Self::Idle)
37    }
38
39    /// Returns true if the state is pending.
40    pub const fn is_pending(&self) -> bool {
41        matches!(self, Self::Pending)
42    }
43
44    /// Returns true if backfill must be re-evaluated before it can start.
45    pub const fn is_pending_revalidation(&self) -> bool {
46        matches!(self, Self::PendingRevalidation)
47    }
48
49    /// Returns true if the state is active.
50    pub const fn is_active(&self) -> bool {
51        matches!(self, Self::Active)
52    }
53}
54
55/// Backfill sync mode functionality.
56pub trait BackfillSync: Send {
57    /// Performs a backfill action.
58    fn on_action(&mut self, action: BackfillAction);
59
60    /// Polls the pipeline for completion.
61    fn poll(&mut self, cx: &mut Context<'_>) -> Poll<BackfillEvent>;
62}
63
64/// The backfill actions that can be performed.
65#[derive(Debug, Clone, PartialEq, Eq)]
66pub enum BackfillAction {
67    /// Start backfilling with the given target.
68    Start(PipelineTarget),
69}
70
71/// The events that can be emitted on backfill sync.
72#[derive(Debug)]
73pub enum BackfillEvent {
74    /// Backfill sync started.
75    Started(PipelineTarget),
76    /// Backfill sync finished.
77    ///
78    /// If this is returned, backfill sync is idle.
79    Finished(Result<ControlFlow, PipelineError>),
80    /// Sync task was dropped after it was started, unable to receive it because
81    /// channel closed. This would indicate a panicked task.
82    TaskDropped(String),
83}
84
85/// Pipeline sync.
86#[derive(Debug)]
87pub struct PipelineSync<N: ProviderNodeTypes> {
88    /// The type that can spawn the pipeline task.
89    pipeline_task_spawner: Runtime,
90    /// The current state of the pipeline.
91    /// The pipeline is used for large ranges.
92    pipeline_state: PipelineState<N>,
93    /// Pending target block for the pipeline to sync
94    pending_pipeline_target: Option<PipelineTarget>,
95}
96
97impl<N: ProviderNodeTypes> PipelineSync<N> {
98    /// Create a new instance.
99    pub fn new(pipeline: Pipeline<N>, pipeline_task_spawner: Runtime) -> Self {
100        Self {
101            pipeline_task_spawner,
102            pipeline_state: PipelineState::Idle(Some(Box::new(pipeline))),
103            pending_pipeline_target: None,
104        }
105    }
106
107    /// Returns `true` if a pipeline target is queued and will be triggered on the next `poll`.
108    #[expect(dead_code)]
109    const fn is_pipeline_sync_pending(&self) -> bool {
110        self.pending_pipeline_target.is_some() && self.pipeline_state.is_idle()
111    }
112
113    /// Returns `true` if the pipeline is idle.
114    const fn is_pipeline_idle(&self) -> bool {
115        self.pipeline_state.is_idle()
116    }
117
118    /// Returns `true` if the pipeline is active.
119    const fn is_pipeline_active(&self) -> bool {
120        !self.is_pipeline_idle()
121    }
122
123    /// Sets a new target to sync the pipeline to.
124    ///
125    /// But ensures the target is not the zero hash.
126    fn set_pipeline_sync_target(&mut self, target: PipelineTarget) {
127        if target.sync_target().is_some_and(|target| target.is_zero()) {
128            trace!(
129                target: "consensus::engine::sync",
130                "Pipeline target cannot be zero hash."
131            );
132            // precaution to never sync to the zero hash
133            return
134        }
135        self.pending_pipeline_target = Some(target);
136    }
137
138    /// This will spawn the pipeline if it is idle and a target is set or if the pipeline is set to
139    /// run continuously.
140    fn try_spawn_pipeline(&mut self) -> Option<BackfillEvent> {
141        match &mut self.pipeline_state {
142            PipelineState::Idle(pipeline) => {
143                let target = self.pending_pipeline_target.take()?;
144                let (tx, rx) = oneshot::channel();
145
146                let pipeline = pipeline.take().expect("exists");
147                self.pipeline_task_spawner.spawn_critical_blocking_task(
148                    "pipeline task",
149                    async move {
150                        let result = pipeline.run_as_fut(Some(target)).await;
151                        let _ = tx.send(result);
152                    },
153                );
154                self.pipeline_state = PipelineState::Running(rx);
155
156                Some(BackfillEvent::Started(target))
157            }
158            PipelineState::Running(_) => None,
159        }
160    }
161
162    /// Advances the pipeline state.
163    ///
164    /// This checks for the result in the channel, or returns pending if the pipeline is idle.
165    fn poll_pipeline(&mut self, cx: &mut Context<'_>) -> Poll<BackfillEvent> {
166        let res = match self.pipeline_state {
167            PipelineState::Idle(_) => return Poll::Pending,
168            PipelineState::Running(ref mut fut) => {
169                ready!(fut.poll_unpin(cx))
170            }
171        };
172        let ev = match res {
173            Ok((pipeline, result)) => {
174                self.pipeline_state = PipelineState::Idle(Some(Box::new(pipeline)));
175                BackfillEvent::Finished(result)
176            }
177            Err(why) => {
178                // failed to receive the pipeline
179                BackfillEvent::TaskDropped(why.to_string())
180            }
181        };
182        Poll::Ready(ev)
183    }
184}
185
186impl<N: ProviderNodeTypes> BackfillSync for PipelineSync<N> {
187    fn on_action(&mut self, event: BackfillAction) {
188        match event {
189            BackfillAction::Start(target) => self.set_pipeline_sync_target(target),
190        }
191    }
192
193    fn poll(&mut self, cx: &mut Context<'_>) -> Poll<BackfillEvent> {
194        // try to spawn a pipeline if a target is set
195        if let Some(event) = self.try_spawn_pipeline() {
196            return Poll::Ready(event)
197        }
198
199        // make sure we poll the pipeline if it's active, and return any ready pipeline events
200        if self.is_pipeline_active() {
201            // advance the pipeline
202            if let Poll::Ready(event) = self.poll_pipeline(cx) {
203                return Poll::Ready(event)
204            }
205        }
206
207        Poll::Pending
208    }
209}
210
211/// The possible pipeline states within the sync controller.
212///
213/// [`PipelineState::Idle`] means that the pipeline is currently idle.
214/// [`PipelineState::Running`] means that the pipeline is currently running.
215///
216/// NOTE: The differentiation between these two states is important, because when the pipeline is
217/// running, it acquires the write lock over the database. This means that we cannot forward to the
218/// blockchain tree any messages that would result in database writes, since it would result in a
219/// deadlock.
220#[derive(Debug)]
221enum PipelineState<N: ProviderNodeTypes> {
222    /// Pipeline is idle.
223    Idle(Option<Box<Pipeline<N>>>),
224    /// Pipeline is running and waiting for a response
225    Running(oneshot::Receiver<PipelineWithResult<N>>),
226}
227
228impl<N: ProviderNodeTypes> PipelineState<N> {
229    /// Returns `true` if the state matches idle.
230    const fn is_idle(&self) -> bool {
231        matches!(self, Self::Idle(_))
232    }
233}
234
235#[cfg(test)]
236mod tests {
237    use super::*;
238    use crate::test_utils::{insert_headers_into_client, TestPipelineBuilder};
239    use alloy_consensus::Header;
240    use alloy_eips::eip1559::ETHEREUM_BLOCK_GAS_LIMIT_30M;
241    use alloy_primitives::{BlockNumber, B256};
242    use assert_matches::assert_matches;
243    use futures::poll;
244    use reth_chainspec::{ChainSpecBuilder, MAINNET};
245    use reth_network_p2p::test_utils::TestFullBlockClient;
246    use reth_primitives_traits::SealedHeader;
247    use reth_provider::test_utils::MockNodeTypesWithDB;
248    use reth_stages::ExecOutput;
249    use reth_stages_api::StageCheckpoint;
250    use reth_tasks::Runtime;
251    use std::{collections::VecDeque, future::poll_fn, sync::Arc};
252
253    struct TestHarness {
254        pipeline_sync: PipelineSync<MockNodeTypesWithDB>,
255        tip: B256,
256    }
257
258    impl TestHarness {
259        fn new(total_blocks: usize, pipeline_done_after: u64) -> Self {
260            let chain_spec = Arc::new(
261                ChainSpecBuilder::default()
262                    .chain(MAINNET.chain)
263                    .genesis(MAINNET.genesis.clone())
264                    .paris_activated()
265                    .build(),
266            );
267
268            // force the pipeline to be "done" after `pipeline_done_after` blocks
269            let pipeline = TestPipelineBuilder::new()
270                .with_pipeline_exec_outputs(VecDeque::from([Ok(ExecOutput {
271                    checkpoint: StageCheckpoint::new(BlockNumber::from(pipeline_done_after)),
272                    done: true,
273                })]))
274                .build(chain_spec);
275
276            let pipeline_sync = PipelineSync::new(pipeline, Runtime::test());
277            let client = TestFullBlockClient::default();
278            let header = Header {
279                base_fee_per_gas: Some(7),
280                gas_limit: ETHEREUM_BLOCK_GAS_LIMIT_30M,
281                ..Default::default()
282            };
283            let header = SealedHeader::seal_slow(header);
284            insert_headers_into_client(&client, header, 0..total_blocks);
285
286            let tip = client.highest_block().expect("there should be blocks here").hash();
287
288            Self { pipeline_sync, tip }
289        }
290    }
291
292    #[tokio::test]
293    async fn pipeline_started_and_finished() {
294        const TOTAL_BLOCKS: usize = 10;
295        const PIPELINE_DONE_AFTER: u64 = 5;
296        let TestHarness { mut pipeline_sync, tip } =
297            TestHarness::new(TOTAL_BLOCKS, PIPELINE_DONE_AFTER);
298
299        let sync_future = poll_fn(|cx| pipeline_sync.poll(cx));
300        let next_event = poll!(sync_future);
301
302        // sync target not set, pipeline not started
303        assert_matches!(next_event, Poll::Pending);
304
305        pipeline_sync.on_action(BackfillAction::Start(PipelineTarget::Sync(tip)));
306
307        let sync_future = poll_fn(|cx| pipeline_sync.poll(cx));
308        let next_event = poll!(sync_future);
309
310        // sync target set, pipeline started
311        assert_matches!(next_event, Poll::Ready(BackfillEvent::Started(target)) => {
312            assert_eq!(target.sync_target().unwrap(), tip);
313        });
314
315        // the next event should be the pipeline finishing in a good state
316        let sync_future = poll_fn(|cx| pipeline_sync.poll(cx));
317        let next_ready = sync_future.await;
318        assert_matches!(next_ready, BackfillEvent::Finished(result) => {
319            assert_matches!(result, Ok(control_flow) => assert_eq!(control_flow, ControlFlow::Continue { block_number: PIPELINE_DONE_AFTER }));
320        });
321    }
322}
reth_engine_tree/backfill.rs

reth_engine_tree/
backfill.rs
