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_api::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 [`StaticFileProducerInner`] 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;
259    use assert_matches::assert_matches;
260    use reth_provider::{
261        providers::StaticFileWriter, test_utils::MockNodeTypesWithDB, ProviderError,
262        ProviderFactory, StaticFileProviderFactory,
263    };
264    use reth_prune_types::PruneModes;
265    use reth_stages::test_utils::{StorageKind, TestStageDB};
266    use reth_static_file_types::{HighestStaticFiles, StaticFileSegment};
267    use reth_testing_utils::generators::{
268        self, random_block_range, random_receipt, BlockRangeParams,
269    };
270    use std::{sync::mpsc::channel, time::Duration};
271    use tempfile::TempDir;
272
273    fn setup() -> (ProviderFactory<MockNodeTypesWithDB>, TempDir) {
274        let mut rng = generators::rng();
275        let db = TestStageDB::default();
276
277        let blocks = random_block_range(
278            &mut rng,
279            0..=3,
280            BlockRangeParams { parent: Some(B256::ZERO), tx_count: 2..3, ..Default::default() },
281        );
282        db.insert_blocks(blocks.iter(), StorageKind::Database(None)).expect("insert blocks");
283        // Unwind headers from static_files and manually insert them into the database, so we're
284        // able to check that static_file_producer works
285        let static_file_provider = db.factory.static_file_provider();
286        let mut static_file_writer = static_file_provider
287            .latest_writer(StaticFileSegment::Headers)
288            .expect("get static file writer for headers");
289        static_file_writer.prune_headers(blocks.len() as u64).unwrap();
290        static_file_writer.commit().expect("prune headers");
291        drop(static_file_writer);
292
293        db.insert_blocks(blocks.iter(), StorageKind::Database(None)).expect("insert blocks");
294
295        let mut receipts = Vec::new();
296        for block in &blocks {
297            for transaction in &block.body().transactions {
298                receipts.push((
299                    receipts.len() as u64,
300                    random_receipt(&mut rng, transaction, Some(0), None),
301                ));
302            }
303        }
304        db.insert_receipts(receipts).expect("insert receipts");
305
306        let provider_factory = db.factory;
307        (provider_factory, db.temp_static_files_dir)
308    }
309
310    #[test]
311    fn run() {
312        let (provider_factory, _temp_static_files_dir) = setup();
313
314        let static_file_producer =
315            StaticFileProducerInner::new(provider_factory.clone(), PruneModes::default());
316
317        let targets = static_file_producer
318            .get_static_file_targets(HighestStaticFiles {
319                headers: Some(1),
320                receipts: Some(1),
321                transactions: Some(1),
322            })
323            .expect("get static file targets");
324        assert_eq!(
325            targets,
326            StaticFileTargets {
327                headers: Some(0..=1),
328                receipts: Some(0..=1),
329                transactions: Some(0..=1)
330            }
331        );
332        assert_matches!(static_file_producer.run(targets), Ok(_));
333        assert_eq!(
334            provider_factory.static_file_provider().get_highest_static_files(),
335            HighestStaticFiles { headers: Some(1), receipts: Some(1), transactions: Some(1) }
336        );
337
338        let targets = static_file_producer
339            .get_static_file_targets(HighestStaticFiles {
340                headers: Some(3),
341                receipts: Some(3),
342                transactions: Some(3),
343            })
344            .expect("get static file targets");
345        assert_eq!(
346            targets,
347            StaticFileTargets {
348                headers: Some(2..=3),
349                receipts: Some(2..=3),
350                transactions: Some(2..=3)
351            }
352        );
353        assert_matches!(static_file_producer.run(targets), Ok(_));
354        assert_eq!(
355            provider_factory.static_file_provider().get_highest_static_files(),
356            HighestStaticFiles { headers: Some(3), receipts: Some(3), transactions: Some(3) }
357        );
358
359        let targets = static_file_producer
360            .get_static_file_targets(HighestStaticFiles {
361                headers: Some(4),
362                receipts: Some(4),
363                transactions: Some(4),
364            })
365            .expect("get static file targets");
366        assert_eq!(
367            targets,
368            StaticFileTargets {
369                headers: Some(4..=4),
370                receipts: Some(4..=4),
371                transactions: Some(4..=4)
372            }
373        );
374        assert_matches!(
375            static_file_producer.run(targets),
376            Err(ProviderError::BlockBodyIndicesNotFound(4))
377        );
378        assert_eq!(
379            provider_factory.static_file_provider().get_highest_static_files(),
380            HighestStaticFiles { headers: Some(3), receipts: Some(3), transactions: Some(3) }
381        );
382    }
383
384    /// Tests that a cloneable [`StaticFileProducer`] type is not susceptible to any race condition.
385    #[test]
386    fn only_one() {
387        let (provider_factory, _temp_static_files_dir) = setup();
388
389        let static_file_producer = StaticFileProducer::new(provider_factory, PruneModes::default());
390
391        let (tx, rx) = channel();
392
393        for i in 0..5 {
394            let producer = static_file_producer.clone();
395            let tx = tx.clone();
396
397            std::thread::spawn(move || {
398                let locked_producer = producer.lock();
399                if i == 0 {
400                    // Let other threads spawn as well.
401                    std::thread::sleep(Duration::from_millis(100));
402                }
403                let targets = locked_producer
404                    .get_static_file_targets(HighestStaticFiles {
405                        headers: Some(1),
406                        receipts: Some(1),
407                        transactions: Some(1),
408                    })
409                    .expect("get static file targets");
410                assert_matches!(locked_producer.run(targets.clone()), Ok(_));
411                tx.send(targets).unwrap();
412            });
413        }
414
415        drop(tx);
416
417        let mut only_one = Some(());
418        for target in rx {
419            // Only the first spawn should have any meaningful target.
420            assert!(only_one.take().is_some_and(|_| target.any()) || !target.any())
421        }
422    }
423}
reth_static_file/static_file_producer.rs

reth_static_file/
static_file_producer.rs
