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.receipts.clone() {
135            segments.push((Box::new(segments::Receipts), block_range));
136        }
137
138        segments.par_iter().try_for_each(|(segment, block_range)| -> ProviderResult<()> {
139            debug!(target: "static_file", segment = %segment.segment(), ?block_range, "StaticFileProducer segment");
140            let start = Instant::now();
141
142            // Create a new database transaction on every segment to prevent long-lived read-only
143            // transactions
144            let provider = self.provider.database_provider_ro()?.disable_long_read_transaction_safety();
145            segment.copy_to_static_files(provider,  block_range.clone())?;
146
147            let elapsed = start.elapsed(); // TODO(alexey): track in metrics
148            debug!(target: "static_file", segment = %segment.segment(), ?block_range, ?elapsed, "Finished StaticFileProducer segment");
149
150            Ok(())
151        })?;
152
153        self.provider.static_file_provider().commit()?;
154        for (segment, block_range) in segments {
155            self.provider
156                .static_file_provider()
157                .update_index(segment.segment(), Some(*block_range.end()))?;
158        }
159
160        let elapsed = start.elapsed(); // TODO(alexey): track in metrics
161        debug!(target: "static_file", ?targets, ?elapsed, "StaticFileProducer finished");
162
163        self.event_sender
164            .notify(StaticFileProducerEvent::Finished { targets: targets.clone(), elapsed });
165
166        Ok(targets)
167    }
168
169    /// Copies data from database to static files according to
170    /// [stage checkpoints](reth_stages_types::StageCheckpoint).
171    ///
172    /// Returns highest block numbers for all static file segments.
173    pub fn copy_to_static_files(&self) -> ProviderResult<HighestStaticFiles> {
174        let provider = self.provider.database_provider_ro()?;
175        let stages_checkpoints = std::iter::once(StageId::Execution)
176            .map(|stage| provider.get_stage_checkpoint(stage).map(|c| c.map(|c| c.block_number)))
177            .collect::<Result<Vec<_>, _>>()?;
178
179        let highest_static_files = HighestStaticFiles { receipts: stages_checkpoints[0] };
180        let targets = self.get_static_file_targets(highest_static_files)?;
181        self.run(targets)?;
182
183        Ok(highest_static_files)
184    }
185
186    /// Returns a static file targets at the provided finalized block numbers per segment.
187    /// The target is determined by the check against highest `static_files` using
188    /// [`reth_provider::providers::StaticFileProvider::get_highest_static_files`].
189    pub fn get_static_file_targets(
190        &self,
191        finalized_block_numbers: HighestStaticFiles,
192    ) -> ProviderResult<StaticFileTargets> {
193        let highest_static_files = self.provider.static_file_provider().get_highest_static_files();
194
195        let targets = StaticFileTargets {
196            // StaticFile receipts only if they're not pruned according to the user configuration
197            receipts: if self.prune_modes.receipts.is_none() {
198                finalized_block_numbers.receipts.and_then(|finalized_block_number| {
199                    self.get_static_file_target(
200                        highest_static_files.receipts,
201                        finalized_block_number,
202                    )
203                })
204            } else {
205                None
206            },
207        };
208
209        trace!(
210            target: "static_file",
211            ?finalized_block_numbers,
212            ?highest_static_files,
213            ?targets,
214            any = %targets.any(),
215            "StaticFile targets"
216        );
217
218        Ok(targets)
219    }
220
221    fn get_static_file_target(
222        &self,
223        highest_static_file: Option<BlockNumber>,
224        finalized_block_number: BlockNumber,
225    ) -> Option<RangeInclusive<BlockNumber>> {
226        let range = highest_static_file.map_or(0, |block| block + 1)..=finalized_block_number;
227        (!range.is_empty()).then_some(range)
228    }
229}
230
231#[cfg(test)]
232mod tests {
233    use crate::static_file_producer::{
234        StaticFileProducer, StaticFileProducerInner, StaticFileTargets,
235    };
236    use alloy_primitives::B256;
237    use assert_matches::assert_matches;
238    use reth_provider::{
239        providers::StaticFileWriter, test_utils::MockNodeTypesWithDB, ProviderError,
240        ProviderFactory, StaticFileProviderFactory,
241    };
242    use reth_prune_types::PruneModes;
243    use reth_stages::test_utils::{StorageKind, TestStageDB};
244    use reth_static_file_types::{HighestStaticFiles, StaticFileSegment};
245    use reth_testing_utils::generators::{
246        self, random_block_range, random_receipt, BlockRangeParams,
247    };
248    use std::{sync::mpsc::channel, time::Duration};
249    use tempfile::TempDir;
250
251    fn setup() -> (ProviderFactory<MockNodeTypesWithDB>, TempDir) {
252        let mut rng = generators::rng();
253        let db = TestStageDB::default();
254
255        let blocks = random_block_range(
256            &mut rng,
257            0..=3,
258            BlockRangeParams { parent: Some(B256::ZERO), tx_count: 2..3, ..Default::default() },
259        );
260        db.insert_blocks(blocks.iter(), StorageKind::Database(None)).expect("insert blocks");
261        // Unwind headers from static_files and manually insert them into the database, so we're
262        // able to check that static_file_producer works
263        let static_file_provider = db.factory.static_file_provider();
264        let mut static_file_writer = static_file_provider
265            .latest_writer(StaticFileSegment::Headers)
266            .expect("get static file writer for headers");
267        static_file_writer.prune_headers(blocks.len() as u64).unwrap();
268        static_file_writer.commit().expect("prune headers");
269        drop(static_file_writer);
270
271        db.insert_blocks(blocks.iter(), StorageKind::Database(None)).expect("insert blocks");
272
273        let mut receipts = Vec::new();
274        for block in &blocks {
275            for transaction in &block.body().transactions {
276                receipts.push((
277                    receipts.len() as u64,
278                    random_receipt(&mut rng, transaction, Some(0), None),
279                ));
280            }
281        }
282        db.insert_receipts(receipts).expect("insert receipts");
283
284        let provider_factory = db.factory;
285        (provider_factory, db.temp_static_files_dir)
286    }
287
288    #[test]
289    fn run() {
290        let (provider_factory, _temp_static_files_dir) = setup();
291
292        let static_file_producer =
293            StaticFileProducerInner::new(provider_factory.clone(), PruneModes::default());
294
295        let targets = static_file_producer
296            .get_static_file_targets(HighestStaticFiles { receipts: Some(1) })
297            .expect("get static file targets");
298        assert_eq!(targets, StaticFileTargets { receipts: Some(0..=1) });
299        assert_matches!(static_file_producer.run(targets), Ok(_));
300        assert_eq!(
301            provider_factory.static_file_provider().get_highest_static_files(),
302            HighestStaticFiles { receipts: Some(1) }
303        );
304
305        let targets = static_file_producer
306            .get_static_file_targets(HighestStaticFiles { receipts: Some(3) })
307            .expect("get static file targets");
308        assert_eq!(targets, StaticFileTargets { receipts: Some(2..=3) });
309        assert_matches!(static_file_producer.run(targets), Ok(_));
310        assert_eq!(
311            provider_factory.static_file_provider().get_highest_static_files(),
312            HighestStaticFiles { receipts: Some(3) }
313        );
314
315        let targets = static_file_producer
316            .get_static_file_targets(HighestStaticFiles { receipts: Some(4) })
317            .expect("get static file targets");
318        assert_eq!(targets, StaticFileTargets { receipts: Some(4..=4) });
319        assert_matches!(
320            static_file_producer.run(targets),
321            Err(ProviderError::BlockBodyIndicesNotFound(4))
322        );
323        assert_eq!(
324            provider_factory.static_file_provider().get_highest_static_files(),
325            HighestStaticFiles { receipts: Some(3) }
326        );
327    }
328
329    /// Tests that a cloneable [`StaticFileProducer`] type is not susceptible to any race condition.
330    #[test]
331    fn only_one() {
332        let (provider_factory, _temp_static_files_dir) = setup();
333
334        let static_file_producer = StaticFileProducer::new(provider_factory, PruneModes::default());
335
336        let (tx, rx) = channel();
337
338        for i in 0..5 {
339            let producer = static_file_producer.clone();
340            let tx = tx.clone();
341
342            std::thread::spawn(move || {
343                let locked_producer = producer.lock();
344                if i == 0 {
345                    // Let other threads spawn as well.
346                    std::thread::sleep(Duration::from_millis(100));
347                }
348                let targets = locked_producer
349                    .get_static_file_targets(HighestStaticFiles { receipts: Some(1) })
350                    .expect("get static file targets");
351                assert_matches!(locked_producer.run(targets.clone()), Ok(_));
352                tx.send(targets).unwrap();
353            });
354        }
355
356        drop(tx);
357
358        let mut only_one = Some(());
359        for target in rx {
360            // Only the first spawn should have any meaningful target.
361            assert!(only_one.take().is_some_and(|_| target.any()) || !target.any())
362        }
363    }
364}
reth_static_file/static_file_producer.rs

reth_static_file/
static_file_producer.rs
