reth_static_file/
static_file_producer.rs

1//! Support for producing static files.
2
3use crate::{segments, segments::Segment, StaticFileProducerEvent};
4use alloy_primitives::BlockNumber;
5use parking_lot::Mutex;
6use rayon::prelude::*;
7use reth_codecs::Compact;
8use reth_db::table::Value;
9use reth_primitives_traits::NodePrimitives;
10use reth_provider::{
11    providers::StaticFileWriter, BlockReader, ChainStateBlockReader, DBProvider,
12    DatabaseProviderFactory, StageCheckpointReader, StaticFileProviderFactory,
13};
14use reth_prune_types::PruneModes;
15use reth_stages_types::StageId;
16use reth_static_file_types::{HighestStaticFiles, StaticFileTargets};
17use reth_storage_errors::provider::ProviderResult;
18use reth_tokio_util::{EventSender, EventStream};
19use std::{
20    ops::{Deref, RangeInclusive},
21    sync::Arc,
22    time::Instant,
23};
24use tracing::{debug, trace};
25
26/// Result of [`StaticFileProducerInner::run`] execution.
27pub type StaticFileProducerResult = ProviderResult<StaticFileTargets>;
28
29/// The [`StaticFileProducer`] instance itself with the result of [`StaticFileProducerInner::run`]
30pub type StaticFileProducerWithResult<Provider> =
31    (StaticFileProducer<Provider>, StaticFileProducerResult);
32
33/// Static File producer. It's a wrapper around [`StaticFileProducer`] that allows to share it
34/// between threads.
35#[derive(Debug)]
36pub struct StaticFileProducer<Provider>(Arc<Mutex<StaticFileProducerInner<Provider>>>);
37
38impl<Provider> StaticFileProducer<Provider> {
39    /// Creates a new [`StaticFileProducer`].
40    pub fn new(provider: Provider, prune_modes: PruneModes) -> Self {
41        Self(Arc::new(Mutex::new(StaticFileProducerInner::new(provider, prune_modes))))
42    }
43}
44
45impl<Provider> Clone for StaticFileProducer<Provider> {
46    fn clone(&self) -> Self {
47        Self(self.0.clone())
48    }
49}
50
51impl<Provider> Deref for StaticFileProducer<Provider> {
52    type Target = Arc<Mutex<StaticFileProducerInner<Provider>>>;
53
54    fn deref(&self) -> &Self::Target {
55        &self.0
56    }
57}
58
59/// Static File producer routine. See [`StaticFileProducerInner::run`] for more detailed
60/// description.
61#[derive(Debug)]
62pub struct StaticFileProducerInner<Provider> {
63    /// Provider factory
64    provider: Provider,
65    /// Pruning configuration for every part of the data that can be pruned. Set by user, and
66    /// needed in [`StaticFileProducerInner`] to prevent attempting to move prunable data to static
67    /// files. See [`StaticFileProducerInner::get_static_file_targets`].
68    prune_modes: PruneModes,
69    event_sender: EventSender<StaticFileProducerEvent>,
70}
71
72impl<Provider> StaticFileProducerInner<Provider> {
73    fn new(provider: Provider, prune_modes: PruneModes) -> Self {
74        Self { provider, prune_modes, event_sender: Default::default() }
75    }
76}
77
78impl<Provider> StaticFileProducerInner<Provider>
79where
80    Provider: StaticFileProviderFactory + DatabaseProviderFactory<Provider: ChainStateBlockReader>,
81{
82    /// Returns the last finalized block number on disk.
83    pub fn last_finalized_block(&self) -> ProviderResult<Option<BlockNumber>> {
84        self.provider.database_provider_ro()?.last_finalized_block_number()
85    }
86}
87
88impl<Provider> StaticFileProducerInner<Provider>
89where
90    Provider: StaticFileProviderFactory
91        + DatabaseProviderFactory<
92            Provider: StaticFileProviderFactory<
93                Primitives: NodePrimitives<
94                    SignedTx: Value + Compact,
95                    BlockHeader: Value + Compact,
96                    Receipt: Value + Compact,
97                >,
98            > + StageCheckpointReader
99                          + BlockReader,
100        >,
101{
102    /// Listen for events on the `static_file_producer`.
103    pub fn events(&self) -> EventStream<StaticFileProducerEvent> {
104        self.event_sender.new_listener()
105    }
106
107    /// Run the `static_file_producer`.
108    ///
109    /// For each [Some] target in [`StaticFileTargets`], initializes a corresponding [Segment] and
110    /// runs it with the provided block range using [`reth_provider::providers::StaticFileProvider`]
111    /// and a read-only database transaction from [`DatabaseProviderFactory`]. All segments are run
112    /// in parallel.
113    ///
114    /// NOTE: it doesn't delete the data from database, and the actual deleting (aka pruning) logic
115    /// lives in the `prune` crate.
116    pub fn run(&self, targets: StaticFileTargets) -> StaticFileProducerResult {
117        // If there are no targets, do not produce any static files and return early
118        if !targets.any() {
119            return Ok(targets)
120        }
121
122        debug_assert!(targets.is_contiguous_to_highest_static_files(
123            self.provider.static_file_provider().get_highest_static_files()
124        ));
125
126        self.event_sender.notify(StaticFileProducerEvent::Started { targets: targets.clone() });
127
128        debug!(target: "static_file", ?targets, "StaticFileProducer started");
129        let start = Instant::now();
130
131        let mut segments =
132            Vec::<(Box<dyn Segment<Provider::Provider>>, RangeInclusive<BlockNumber>)>::new();
133
134        if let Some(block_range) = targets.transactions.clone() {
135            segments.push((Box::new(segments::Transactions), block_range));
136        }
137        if let Some(block_range) = targets.headers.clone() {
138            segments.push((Box::new(segments::Headers), block_range));
139        }
140        if let Some(block_range) = targets.receipts.clone() {
141            segments.push((Box::new(segments::Receipts), block_range));
142        }
143
144        segments.par_iter().try_for_each(|(segment, block_range)| -> ProviderResult<()> {
145            debug!(target: "static_file", segment = %segment.segment(), ?block_range, "StaticFileProducer segment");
146            let start = Instant::now();
147
148            // Create a new database transaction on every segment to prevent long-lived read-only
149            // transactions
150            let provider = self.provider.database_provider_ro()?.disable_long_read_transaction_safety();
151            segment.copy_to_static_files(provider,  block_range.clone())?;
152
153            let elapsed = start.elapsed(); // TODO(alexey): track in metrics
154            debug!(target: "static_file", segment = %segment.segment(), ?block_range, ?elapsed, "Finished StaticFileProducer segment");
155
156            Ok(())
157        })?;
158
159        self.provider.static_file_provider().commit()?;
160        for (segment, block_range) in segments {
161            self.provider
162                .static_file_provider()
163                .update_index(segment.segment(), Some(*block_range.end()))?;
164        }
165
166        let elapsed = start.elapsed(); // TODO(alexey): track in metrics
167        debug!(target: "static_file", ?targets, ?elapsed, "StaticFileProducer finished");
168
169        self.event_sender
170            .notify(StaticFileProducerEvent::Finished { targets: targets.clone(), elapsed });
171
172        Ok(targets)
173    }
174
175    /// Copies data from database to static files according to
176    /// [stage checkpoints](reth_stages_types::StageCheckpoint).
177    ///
178    /// Returns highest block numbers for all static file segments.
179    pub fn copy_to_static_files(&self) -> ProviderResult<HighestStaticFiles> {
180        let provider = self.provider.database_provider_ro()?;
181        let stages_checkpoints = [StageId::Headers, StageId::Execution, StageId::Bodies]
182            .into_iter()
183            .map(|stage| provider.get_stage_checkpoint(stage).map(|c| c.map(|c| c.block_number)))
184            .collect::<Result<Vec<_>, _>>()?;
185
186        let highest_static_files = HighestStaticFiles {
187            headers: stages_checkpoints[0],
188            receipts: stages_checkpoints[1],
189            transactions: stages_checkpoints[2],
190        };
191        let targets = self.get_static_file_targets(highest_static_files)?;
192        self.run(targets)?;
193
194        Ok(highest_static_files)
195    }
196
197    /// Returns a static file targets at the provided finalized block numbers per segment.
198    /// The target is determined by the check against highest `static_files` using
199    /// [`reth_provider::providers::StaticFileProvider::get_highest_static_files`].
200    pub fn get_static_file_targets(
201        &self,
202        finalized_block_numbers: HighestStaticFiles,
203    ) -> ProviderResult<StaticFileTargets> {
204        let highest_static_files = self.provider.static_file_provider().get_highest_static_files();
205
206        let targets = StaticFileTargets {
207            headers: finalized_block_numbers.headers.and_then(|finalized_block_number| {
208                self.get_static_file_target(highest_static_files.headers, finalized_block_number)
209            }),
210            // StaticFile receipts only if they're not pruned according to the user configuration
211            receipts: if self.prune_modes.receipts.is_none() &&
212                self.prune_modes.receipts_log_filter.is_empty()
213            {
214                finalized_block_numbers.receipts.and_then(|finalized_block_number| {
215                    self.get_static_file_target(
216                        highest_static_files.receipts,
217                        finalized_block_number,
218                    )
219                })
220            } else {
221                None
222            },
223            transactions: finalized_block_numbers.transactions.and_then(|finalized_block_number| {
224                self.get_static_file_target(
225                    highest_static_files.transactions,
226                    finalized_block_number,
227                )
228            }),
229        };
230
231        trace!(
232            target: "static_file",
233            ?finalized_block_numbers,
234            ?highest_static_files,
235            ?targets,
236            any = %targets.any(),
237            "StaticFile targets"
238        );
239
240        Ok(targets)
241    }
242
243    fn get_static_file_target(
244        &self,
245        highest_static_file: Option<BlockNumber>,
246        finalized_block_number: BlockNumber,
247    ) -> Option<RangeInclusive<BlockNumber>> {
248        let range = highest_static_file.map_or(0, |block| block + 1)..=finalized_block_number;
249        (!range.is_empty()).then_some(range)
250    }
251}
252
253#[cfg(test)]
254mod tests {
255    use crate::static_file_producer::{
256        StaticFileProducer, StaticFileProducerInner, StaticFileTargets,
257    };
258    use alloy_primitives::{B256, U256};
259    use assert_matches::assert_matches;
260    use reth_db_api::{database::Database, transaction::DbTx};
261    use reth_provider::{
262        providers::StaticFileWriter, test_utils::MockNodeTypesWithDB, ProviderError,
263        ProviderFactory, StaticFileProviderFactory,
264    };
265    use reth_prune_types::PruneModes;
266    use reth_stages::test_utils::{StorageKind, TestStageDB};
267    use reth_static_file_types::{HighestStaticFiles, StaticFileSegment};
268    use reth_testing_utils::generators::{
269        self, random_block_range, random_receipt, BlockRangeParams,
270    };
271    use std::{sync::mpsc::channel, time::Duration};
272    use tempfile::TempDir;
273
274    fn setup() -> (ProviderFactory<MockNodeTypesWithDB>, TempDir) {
275        let mut rng = generators::rng();
276        let db = TestStageDB::default();
277
278        let blocks = random_block_range(
279            &mut rng,
280            0..=3,
281            BlockRangeParams { parent: Some(B256::ZERO), tx_count: 2..3, ..Default::default() },
282        );
283        db.insert_blocks(blocks.iter(), StorageKind::Database(None)).expect("insert blocks");
284        // Unwind headers from static_files and manually insert them into the database, so we're
285        // able to check that static_file_producer works
286        let static_file_provider = db.factory.static_file_provider();
287        let mut static_file_writer = static_file_provider
288            .latest_writer(StaticFileSegment::Headers)
289            .expect("get static file writer for headers");
290        static_file_writer.prune_headers(blocks.len() as u64).unwrap();
291        static_file_writer.commit().expect("prune headers");
292
293        let tx = db.factory.db_ref().tx_mut().expect("init tx");
294        for block in &blocks {
295            TestStageDB::insert_header(None, &tx, &block.header, U256::ZERO)
296                .expect("insert block header");
297        }
298        tx.commit().expect("commit tx");
299
300        let mut receipts = Vec::new();
301        for block in &blocks {
302            for transaction in &block.body.transactions {
303                receipts
304                    .push((receipts.len() as u64, random_receipt(&mut rng, transaction, Some(0))));
305            }
306        }
307        db.insert_receipts(receipts).expect("insert receipts");
308
309        let provider_factory = db.factory;
310        (provider_factory, db.temp_static_files_dir)
311    }
312
313    #[test]
314    fn run() {
315        let (provider_factory, _temp_static_files_dir) = setup();
316
317        let static_file_producer =
318            StaticFileProducerInner::new(provider_factory.clone(), PruneModes::default());
319
320        let targets = static_file_producer
321            .get_static_file_targets(HighestStaticFiles {
322                headers: Some(1),
323                receipts: Some(1),
324                transactions: Some(1),
325            })
326            .expect("get static file targets");
327        assert_eq!(
328            targets,
329            StaticFileTargets {
330                headers: Some(0..=1),
331                receipts: Some(0..=1),
332                transactions: Some(0..=1)
333            }
334        );
335        assert_matches!(static_file_producer.run(targets), Ok(_));
336        assert_eq!(
337            provider_factory.static_file_provider().get_highest_static_files(),
338            HighestStaticFiles { headers: Some(1), receipts: Some(1), transactions: Some(1) }
339        );
340
341        let targets = static_file_producer
342            .get_static_file_targets(HighestStaticFiles {
343                headers: Some(3),
344                receipts: Some(3),
345                transactions: Some(3),
346            })
347            .expect("get static file targets");
348        assert_eq!(
349            targets,
350            StaticFileTargets {
351                headers: Some(2..=3),
352                receipts: Some(2..=3),
353                transactions: Some(2..=3)
354            }
355        );
356        assert_matches!(static_file_producer.run(targets), Ok(_));
357        assert_eq!(
358            provider_factory.static_file_provider().get_highest_static_files(),
359            HighestStaticFiles { headers: Some(3), receipts: Some(3), transactions: Some(3) }
360        );
361
362        let targets = static_file_producer
363            .get_static_file_targets(HighestStaticFiles {
364                headers: Some(4),
365                receipts: Some(4),
366                transactions: Some(4),
367            })
368            .expect("get static file targets");
369        assert_eq!(
370            targets,
371            StaticFileTargets {
372                headers: Some(4..=4),
373                receipts: Some(4..=4),
374                transactions: Some(4..=4)
375            }
376        );
377        assert_matches!(
378            static_file_producer.run(targets),
379            Err(ProviderError::BlockBodyIndicesNotFound(4))
380        );
381        assert_eq!(
382            provider_factory.static_file_provider().get_highest_static_files(),
383            HighestStaticFiles { headers: Some(3), receipts: Some(3), transactions: Some(3) }
384        );
385    }
386
387    /// Tests that a cloneable [`StaticFileProducer`] type is not susceptible to any race condition.
388    #[test]
389    fn only_one() {
390        let (provider_factory, _temp_static_files_dir) = setup();
391
392        let static_file_producer = StaticFileProducer::new(provider_factory, PruneModes::default());
393
394        let (tx, rx) = channel();
395
396        for i in 0..5 {
397            let producer = static_file_producer.clone();
398            let tx = tx.clone();
399
400            std::thread::spawn(move || {
401                let locked_producer = producer.lock();
402                if i == 0 {
403                    // Let other threads spawn as well.
404                    std::thread::sleep(Duration::from_millis(100));
405                }
406                let targets = locked_producer
407                    .get_static_file_targets(HighestStaticFiles {
408                        headers: Some(1),
409                        receipts: Some(1),
410                        transactions: Some(1),
411                    })
412                    .expect("get static file targets");
413                assert_matches!(locked_producer.run(targets.clone()), Ok(_));
414                tx.send(targets).unwrap();
415            });
416        }
417
418        drop(tx);
419
420        let mut only_one = Some(());
421        for target in rx {
422            // Only the first spawn should have any meaningful target.
423            assert!(only_one.take().is_some_and(|_| target.any()) || !target.any())
424        }
425    }
426}
reth_static_file/static_file_producer.rs

reth_static_file/
static_file_producer.rs
