reth_stages/stages/
era.rs

1use crate::{StageCheckpoint, StageId};
2use alloy_primitives::{BlockHash, BlockNumber};
3use futures_util::{Stream, StreamExt};
4use reqwest::{Client, Url};
5use reth_config::config::EtlConfig;
6use reth_db_api::{table::Value, transaction::DbTxMut};
7use reth_era::{
8    common::file_ops::{EraFileType, StreamReader},
9    era::file::EraReader,
10    era1::file::Era1Reader,
11    ere::file::EreReader,
12};
13use reth_era_downloader::{read_dir, read_era_dir, EraClient, EraMeta, EraStream, EraStreamConfig};
14use reth_era_utils as era;
15use reth_etl::Collector;
16use reth_primitives_traits::{FullBlockBody, FullBlockHeader, NodePrimitives};
17use reth_provider::{
18    BlockReader, BlockWriter, DBProvider, StageCheckpointWriter, StaticFileProviderFactory,
19    StaticFileWriter,
20};
21use reth_stages_api::{ExecInput, ExecOutput, Stage, StageError, UnwindInput, UnwindOutput};
22use reth_static_file_types::StaticFileSegment;
23use std::{
24    fmt::{Debug, Formatter},
25    iter,
26    path::Path,
27    task::{ready, Context, Poll},
28};
29
30type Item<Header, Body> =
31    Box<dyn Iterator<Item = eyre::Result<(Header, Body)>> + Send + Sync + Unpin>;
32type ThreadSafeEraStream<Header, Body> =
33    Box<dyn Stream<Item = eyre::Result<Item<Header, Body>>> + Send + Sync + Unpin>;
34
35/// The ERA history stage.
36///
37/// Imports block headers and bodies from
38/// [ERA1](https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era1.md) files,
39/// which hold pre-merge execution blocks, and
40/// [ERE](https://github.com/eth-clients/e2store-format-specs/blob/main/formats/ere.md) files,
41/// which hold both pre-merge and post-merge execution blocks, and consensus-layer
42/// [ERA](https://github.com/eth-clients/e2store-format-specs/blob/main/formats/era.md) files, whose
43/// post-merge beacon blocks embed an execution payload. The reader is selected per file from its
44/// extension (`.era` vs `.era1` vs `.ere`/`.erae`).
45///
46/// Imports block headers and bodies up to the configured target block. Receipts are generated by
47/// execution, which is not done in this stage.
48pub struct EraStage<Header, Body, StreamFactory> {
49    /// The `source` creates `stream`.
50    source: Option<StreamFactory>,
51    /// A map of block hash to block height collected when processing headers and inserted into
52    /// database afterward.
53    hash_collector: Collector<BlockHash, BlockNumber>,
54    /// Last extracted iterator of block `Header` and `Body` pairs.
55    item: Option<Item<Header, Body>>,
56    /// A stream of [`Item`]s, i.e. iterators over block `Header` and `Body` pairs.
57    stream: Option<ThreadSafeEraStream<Header, Body>>,
58}
59
60trait EraStreamFactory<Header, Body> {
61    fn create(self, input: ExecInput) -> Result<ThreadSafeEraStream<Header, Body>, StageError>;
62}
63
64impl<Header, Body> EraStreamFactory<Header, Body> for EraImportSource
65where
66    Header: FullBlockHeader + Value,
67    Body: FullBlockBody<OmmerHeader = Header>,
68{
69    fn create(self, input: ExecInput) -> Result<ThreadSafeEraStream<Header, Body>, StageError> {
70        match self {
71            Self::Path(path) => {
72                // Consensus `.era` files are slot-indexed and ship no `checksums.txt`, so they use
73                // a different directory reader than the block-indexed `.era1`/`.ere` files.
74                if EraFileType::from_dir(&path).map_err(|e| StageError::Fatal(e.into()))? ==
75                    Some(EraFileType::Era)
76                {
77                    Self::convert(read_era_dir(path).map_err(|e| StageError::Fatal(e.into()))?)
78                } else {
79                    Self::convert(
80                        read_dir(path, input.next_block())
81                            .map_err(|e| StageError::Fatal(e.into()))?,
82                    )
83                }
84            }
85            Self::Url(url, folder) => {
86                let _ = reth_fs_util::create_dir_all(&folder);
87                // `start_from` maps a block number to a file index, valid only for block-indexed
88                // `.era1`/`.ere`. Consensus `.era` files are slot-indexed, so stream from 0 and let
89                // the pipeline skip already-imported blocks.
90                let is_era = EraFileType::from_url(url.as_str()) == EraFileType::Era;
91                let client = EraClient::new(Client::new(), url, folder);
92                let mut config = EraStreamConfig::default();
93                if !is_era {
94                    config = config.start_from(input.next_block());
95                }
96
97                Self::convert(EraStream::new(client, config))
98            }
99        }
100    }
101}
102
103impl EraImportSource {
104    fn convert<Header, Body>(
105        stream: impl Stream<Item = eyre::Result<impl EraMeta + Send + Sync + 'static + Unpin>>
106            + Send
107            + Sync
108            + 'static
109            + Unpin,
110    ) -> Result<ThreadSafeEraStream<Header, Body>, StageError>
111    where
112        Header: FullBlockHeader + Value,
113        Body: FullBlockBody<OmmerHeader = Header>,
114    {
115        Ok(Box::new(Box::pin(stream.map(|meta| {
116            meta.and_then(|meta| {
117                // The reader is selected per file from its extension: consensus `.era` (beacon
118                // blocks with an embedded execution payload), `.ere`/`.erae`, or `.era1`.
119                let file = reth_fs_util::open(meta.path())?;
120                let iter = match meta
121                    .path()
122                    .file_name()
123                    .and_then(|name| name.to_str())
124                    .and_then(EraFileType::from_filename)
125                {
126                    Some(EraFileType::Era) => {
127                        // `.era` blocks decode to an `Option` (pre-merge slots carry no payload and
128                        // are skipped); `buf` is scratch space reused for the RLP re-encode.
129                        let mut buf = Vec::new();
130                        Box::new(
131                            EraReader::new(file).iter().filter_map(move |block| {
132                                era::Era::decode(block, &mut buf).transpose()
133                            }),
134                        ) as Item<Header, Body>
135                    }
136                    Some(EraFileType::Ere) => {
137                        Box::new(EreReader::new(file).iter().map(era::Ere::decode))
138                            as Item<Header, Body>
139                    }
140                    _ => Box::new(Era1Reader::new(file).iter().map(era::decode))
141                        as Item<Header, Body>,
142                };
143
144                let iter = iter.chain(
145                    iter::once_with(move || match meta.mark_as_processed() {
146                        Ok(..) => None,
147                        Err(e) => Some(Err(e)),
148                    })
149                    .flatten(),
150                );
151
152                Ok(Box::new(iter) as Item<Header, Body>)
153            })
154        }))))
155    }
156}
157
158impl<Header: Debug, Body: Debug, F: Debug> Debug for EraStage<Header, Body, F> {
159    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
160        f.debug_struct("EraStage")
161            .field("source", &self.source)
162            .field("hash_collector", &self.hash_collector)
163            .field("item", &self.item.is_some())
164            .field("stream", &"dyn Stream")
165            .finish()
166    }
167}
168
169impl<Header, Body, F> EraStage<Header, Body, F> {
170    /// Creates a new [`EraStage`].
171    pub fn new(source: Option<F>, etl_config: EtlConfig) -> Self {
172        Self {
173            source,
174            item: None,
175            stream: None,
176            hash_collector: Collector::new(etl_config.file_size, etl_config.dir),
177        }
178    }
179}
180
181impl<Provider, N, F> Stage<Provider> for EraStage<N::BlockHeader, N::BlockBody, F>
182where
183    Provider: DBProvider<Tx: DbTxMut>
184        + StaticFileProviderFactory<Primitives = N>
185        + BlockWriter<Block = N::Block>
186        + BlockReader<Block = N::Block>
187        + StageCheckpointWriter,
188    F: EraStreamFactory<N::BlockHeader, N::BlockBody> + Send + Sync + Clone,
189    N: NodePrimitives<BlockHeader: Value>,
190{
191    fn id(&self) -> StageId {
192        StageId::Era
193    }
194
195    fn poll_execute_ready(
196        &mut self,
197        cx: &mut Context<'_>,
198        input: ExecInput,
199    ) -> Poll<Result<(), StageError>> {
200        if input.target_reached() || self.item.is_some() {
201            return Poll::Ready(Ok(()));
202        }
203
204        if self.stream.is_none() &&
205            let Some(source) = self.source.clone()
206        {
207            self.stream.replace(source.create(input)?);
208        }
209        if let Some(stream) = &mut self.stream &&
210            let Some(next) = ready!(stream.poll_next_unpin(cx))
211                .transpose()
212                .map_err(|e| StageError::Fatal(e.into()))?
213        {
214            self.item.replace(next);
215        }
216
217        Poll::Ready(Ok(()))
218    }
219
220    fn execute(&mut self, provider: &Provider, input: ExecInput) -> Result<ExecOutput, StageError> {
221        let height = if let Some(era) = self.item.take() {
222            let static_file_provider = provider.static_file_provider();
223
224            // Consistency check of expected headers in static files vs DB is done on
225            // provider::sync_gap when poll_execute_ready is polled.
226            let last_header_number = static_file_provider
227                .get_highest_static_file_block(StaticFileSegment::Headers)
228                .unwrap_or_default();
229
230            // Although headers were downloaded in reverse order, the collector iterates it in
231            // ascending order
232            let mut writer = static_file_provider.latest_writer(StaticFileSegment::Headers)?;
233
234            // This stage never persists receipts, execution does that separately.
235            let height = era::process_iter(
236                era.map(|r| r.map(|(header, body)| (header, body, None))),
237                &mut writer,
238                None,
239                provider,
240                &mut self.hash_collector,
241                last_header_number..=input.target(),
242                era::ImportPolicy {
243                    headers_tip: last_header_number,
244                    is_receipt_verifiable: &|_| false,
245                },
246            )
247            .map_err(|e| StageError::Fatal(e.into()))?;
248
249            if !self.hash_collector.is_empty() {
250                era::build_index(provider, &mut self.hash_collector)
251                    .map_err(|e| StageError::Recoverable(e.into()))?;
252                self.hash_collector.clear();
253            }
254
255            era::save_stage_checkpoints(
256                provider,
257                input.checkpoint().block_number,
258                height,
259                height,
260                input.target(),
261            )?;
262
263            height
264        } else {
265            // No era files to process. Return the highest block we're aware of to avoid
266            // limiting subsequent stages with an outdated checkpoint.
267            //
268            // This can happen when:
269            // 1. Era import is complete (all pre-merge blocks imported)
270            // 2. No era import source was configured
271            //
272            // We return max(checkpoint, highest_header, target) to ensure we don't return
273            // a stale checkpoint that could limit subsequent stages like Headers.
274            let highest_header = provider
275                .static_file_provider()
276                .get_highest_static_file_block(StaticFileSegment::Headers)
277                .unwrap_or_default();
278
279            let checkpoint = input.checkpoint().block_number;
280            let from_target = input.target.unwrap_or(checkpoint);
281
282            checkpoint.max(highest_header).max(from_target)
283        };
284
285        Ok(ExecOutput { checkpoint: StageCheckpoint::new(height), done: height >= input.target() })
286    }
287
288    fn unwind(
289        &mut self,
290        _provider: &Provider,
291        input: UnwindInput,
292    ) -> Result<UnwindOutput, StageError> {
293        Ok(UnwindOutput { checkpoint: input.checkpoint.with_block_number(input.unwind_to) })
294    }
295}
296
297/// Describes where to get the era files from.
298#[derive(Debug, Clone)]
299pub enum EraImportSource {
300    /// Remote HTTP accessible host.
301    Url(Url, Box<Path>),
302    /// Local directory.
303    Path(Box<Path>),
304}
305
306impl EraImportSource {
307    /// Maybe constructs a new `EraImportSource` depending on the arguments.
308    ///
309    /// Only one of `url` or `path` should be provided, but upholding this invariant is delegated
310    /// above so that both parameters can be accepted.
311    ///
312    /// # Arguments
313    /// * The `path` uses a directory as the import source. It and its contents must be readable.
314    /// * The `url` uses an HTTP client to list and download files.
315    /// * The `default` gives the default [`Url`] if none of the previous parameters are provided.
316    /// * For any [`Url`] the `folder` is used as the download directory for storing files
317    ///   temporarily. It and its contents must be readable and writable.
318    pub fn maybe_new(
319        path: Option<Box<Path>>,
320        url: Option<Url>,
321        default: impl FnOnce() -> Option<Url>,
322        folder: impl FnOnce() -> Box<Path>,
323    ) -> Option<Self> {
324        path.map(Self::Path).or_else(|| url.or_else(default).map(|url| Self::Url(url, folder())))
325    }
326}
327
328#[cfg(test)]
329mod tests {
330    use super::*;
331    use crate::test_utils::{
332        stage_test_suite, ExecuteStageTestRunner, StageTestRunner, UnwindStageTestRunner,
333    };
334    use alloy_consensus::{BlockBody, Header};
335    use alloy_primitives::B256;
336    use assert_matches::assert_matches;
337    use futures_util::stream;
338    use reth_db_api::tables;
339    use reth_era::{
340        common::file_ops::{EraFileFormat, StreamWriter},
341        ere::{
342            file::{EreFile, EreWriter},
343            types::{
344                execution::{BlockTuple, CompressedBody, CompressedHeader},
345                group::{DynamicBlockIndex, EreGroup, EreId},
346            },
347        },
348    };
349    use reth_ethereum_primitives::{Block, TransactionSigned};
350    use reth_primitives_traits::SealedBlock;
351    use reth_provider::BlockHashReader;
352    use reth_testing_utils::generators::{
353        self, random_block_range, random_header, BlockRangeParams,
354    };
355    use std::{fs::File, path::PathBuf};
356    use test_runner::EraTestRunner;
357
358    #[tokio::test]
359    async fn test_era_range_ends_below_target() {
360        let era_cap = 2;
361        let target = 20000;
362
363        let mut runner = EraTestRunner::default();
364
365        let input = ExecInput { target: Some(era_cap), checkpoint: None };
366        runner.seed_execution(input).unwrap();
367
368        let input = ExecInput { target: Some(target), checkpoint: None };
369        let output = runner.execute(input).await.unwrap();
370
371        runner.commit();
372
373        assert_matches!(
374            output,
375            Ok(ExecOutput {
376                checkpoint: StageCheckpoint { block_number, stage_checkpoint: None },
377                done: false
378            }) if block_number == era_cap
379        );
380
381        let output = output.unwrap();
382        let validation_output = runner.validate_execution(input, Some(output.clone()));
383
384        assert_matches!(validation_output, Ok(()));
385
386        runner.take_responses();
387
388        let input = ExecInput { target: Some(target), checkpoint: Some(output.checkpoint) };
389        let output = runner.execute(input).await.unwrap();
390
391        runner.commit();
392
393        assert_matches!(
394            output,
395            Ok(ExecOutput {
396                checkpoint: StageCheckpoint { block_number, stage_checkpoint: None },
397                done: true
398            }) if block_number == target
399        );
400
401        let validation_output = runner.validate_execution(input, output.ok());
402
403        assert_matches!(validation_output, Ok(()));
404    }
405
406    /// Writes `blocks` to a header+body-only ERE file at `path`.
407    fn write_ere_file(path: &Path, blocks: &[SealedBlock<Block>]) {
408        let tuples = blocks
409            .iter()
410            .map(|block| {
411                BlockTuple::new(
412                    CompressedHeader::from_header(block.header()).unwrap(),
413                    CompressedBody::from_body(block.body()).unwrap(),
414                )
415            })
416            .collect::<Vec<_>>();
417
418        let start = blocks[0].number;
419        // Header + body only is component-count 2. The index offsets are stored verbatim and not
420        // interpreted when assembling blocks, so placeholder values are fine.
421        let component_count = 2;
422        let offsets = vec![0; blocks.len() * component_count as usize];
423        let index = DynamicBlockIndex::new(start, component_count, offsets);
424        let group = EreGroup::new(tuples, None, index);
425        let file = EreFile::new(group, EreId::new("mainnet", start, blocks.len() as u32));
426
427        EreWriter::new(File::create(path).unwrap()).write_file(&file).unwrap();
428    }
429
430    #[derive(Debug)]
431    struct EreTestMeta {
432        path: PathBuf,
433    }
434
435    impl EraMeta for EreTestMeta {
436        fn mark_as_processed(&self) -> eyre::Result<()> {
437            Ok(())
438        }
439
440        fn path(&self) -> &Path {
441            &self.path
442        }
443    }
444
445    #[tokio::test]
446    async fn convert_decodes_ere_files() {
447        let mut rng = generators::rng();
448        let blocks = random_block_range(
449            &mut rng,
450            1..=3,
451            BlockRangeParams { tx_count: 1..3, ..Default::default() },
452        );
453
454        let dir = tempfile::tempdir().unwrap();
455
456        // ERE files are published with either the `.ere` or `.erae` extension; both must be
457        // recognized and routed to the ERE reader rather than decoded as era1.
458        for ext in ["ere", "erae"] {
459            let path = dir.path().join(format!("mainnet-00000-abcd1234.{ext}"));
460            write_ere_file(&path, &blocks);
461
462            let stream =
463                stream::iter(vec![Ok::<_, eyre::Error>(EreTestMeta { path: path.clone() })]);
464            let mut stream =
465                EraImportSource::convert::<Header, BlockBody<TransactionSigned>>(stream).unwrap();
466
467            let item = stream.next().await.expect("a file to decode").expect("decoding to succeed");
468            let decoded = item.collect::<eyre::Result<Vec<_>>>().unwrap();
469
470            assert_eq!(
471                decoded.len(),
472                blocks.len(),
473                "ERE file with `.{ext}` extension should decode every block"
474            );
475            for ((header, body), block) in decoded.iter().zip(&blocks) {
476                assert_eq!(header, block.header());
477                assert_eq!(body, block.body());
478            }
479        }
480    }
481
482    mod test_runner {
483        use super::*;
484        use crate::test_utils::{TestRunnerError, TestStageDB};
485        use alloy_consensus::{BlockBody, Header};
486        use futures_util::stream;
487        use reth_db_api::{
488            cursor::DbCursorRO,
489            models::{StoredBlockBodyIndices, StoredBlockOmmers},
490            transaction::DbTx,
491        };
492        use reth_ethereum_primitives::TransactionSigned;
493        use reth_primitives_traits::{SealedBlock, SealedHeader};
494        use reth_provider::{BlockNumReader, HeaderProvider, TransactionsProvider};
495        use reth_testing_utils::generators::{
496            random_block_range, random_signed_tx, BlockRangeParams,
497        };
498        use tokio::sync::watch;
499
500        pub(crate) struct EraTestRunner {
501            channel: (watch::Sender<B256>, watch::Receiver<B256>),
502            db: TestStageDB,
503            responses: Option<Vec<(Header, BlockBody<TransactionSigned>)>>,
504        }
505
506        impl Default for EraTestRunner {
507            fn default() -> Self {
508                Self {
509                    channel: watch::channel(B256::ZERO),
510                    db: TestStageDB::default(),
511                    responses: Default::default(),
512                }
513            }
514        }
515
516        impl StageTestRunner for EraTestRunner {
517            type S = EraStage<Header, BlockBody<TransactionSigned>, StubResponses>;
518
519            fn db(&self) -> &TestStageDB {
520                &self.db
521            }
522
523            fn stage(&self) -> Self::S {
524                EraStage::new(self.responses.clone().map(StubResponses), EtlConfig::default())
525            }
526        }
527
528        impl ExecuteStageTestRunner for EraTestRunner {
529            type Seed = Vec<SealedBlock<reth_ethereum_primitives::Block>>;
530
531            fn seed_execution(&mut self, input: ExecInput) -> Result<Self::Seed, TestRunnerError> {
532                let start = input.checkpoint().block_number;
533                let end = input.target();
534
535                let static_file_provider = self.db.factory.static_file_provider();
536
537                let mut rng = generators::rng();
538
539                // Static files do not support gaps in headers, so we need to generate 0 to end
540                let blocks = random_block_range(
541                    &mut rng,
542                    0..=end,
543                    BlockRangeParams {
544                        parent: Some(B256::ZERO),
545                        tx_count: 0..2,
546                        ..Default::default()
547                    },
548                );
549                self.db.insert_headers(blocks.iter().map(|block| block.sealed_header()))?;
550                if let Some(progress) = blocks.get(start as usize) {
551                    // Insert last progress data
552                    {
553                        let tx = self.db.factory.provider_rw()?.into_tx();
554                        let mut static_file_producer = static_file_provider
555                            .get_writer(start, StaticFileSegment::Transactions)?;
556
557                        let body = StoredBlockBodyIndices {
558                            first_tx_num: 0,
559                            tx_count: progress.transaction_count() as u64,
560                        };
561
562                        static_file_producer.set_block_range(0..=progress.number);
563
564                        body.tx_num_range().try_for_each(|tx_num| {
565                            let transaction = random_signed_tx(&mut rng);
566                            static_file_producer.append_transaction(tx_num, &transaction).map(drop)
567                        })?;
568
569                        if body.tx_count != 0 {
570                            tx.put::<tables::TransactionBlocks>(
571                                body.last_tx_num(),
572                                progress.number,
573                            )?;
574                        }
575
576                        tx.put::<tables::BlockBodyIndices>(progress.number, body)?;
577
578                        if !progress.ommers_hash_is_empty() {
579                            tx.put::<tables::BlockOmmers>(
580                                progress.number,
581                                StoredBlockOmmers { ommers: progress.body().ommers.clone() },
582                            )?;
583                        }
584
585                        static_file_producer.commit()?;
586                        tx.commit()?;
587                    }
588                }
589                self.responses.replace(
590                    blocks.iter().map(|v| (v.header().clone(), v.body().clone())).collect(),
591                );
592                Ok(blocks)
593            }
594
595            /// Validate stored headers and bodies
596            fn validate_execution(
597                &self,
598                input: ExecInput,
599                output: Option<ExecOutput>,
600            ) -> Result<(), TestRunnerError> {
601                let initial_checkpoint = input.checkpoint().block_number;
602                match output {
603                    Some(output) if output.checkpoint.block_number > initial_checkpoint => {
604                        let provider = self.db.factory.provider()?;
605
606                        for block_num in initial_checkpoint..
607                            output
608                                .checkpoint
609                                .block_number
610                                .min(self.responses.as_ref().map(|v| v.len()).unwrap_or_default()
611                                    as BlockNumber)
612                        {
613                            // look up the header hash
614                            let hash = provider.block_hash(block_num)?.expect("no header hash");
615
616                            // validate the header number
617                            assert_eq!(provider.block_number(hash)?, Some(block_num));
618
619                            // validate the header
620                            let header = provider.header_by_number(block_num)?;
621                            assert!(header.is_some());
622                            let header = SealedHeader::seal_slow(header.unwrap());
623                            assert_eq!(header.hash(), hash);
624                        }
625
626                        self.validate_db_blocks(
627                            output.checkpoint.block_number,
628                            output.checkpoint.block_number,
629                        )?;
630                    }
631                    _ => self.check_no_header_entry_above(initial_checkpoint)?,
632                };
633                Ok(())
634            }
635
636            async fn after_execution(&self, headers: Self::Seed) -> Result<(), TestRunnerError> {
637                let tip = if headers.is_empty() {
638                    let tip = random_header(&mut generators::rng(), 0, None);
639                    self.db.insert_headers(iter::once(&tip))?;
640                    tip.hash()
641                } else {
642                    headers.last().unwrap().hash()
643                };
644                self.send_tip(tip);
645                Ok(())
646            }
647        }
648
649        impl UnwindStageTestRunner for EraTestRunner {
650            fn validate_unwind(&self, _input: UnwindInput) -> Result<(), TestRunnerError> {
651                Ok(())
652            }
653        }
654
655        impl EraTestRunner {
656            pub(crate) fn check_no_header_entry_above(
657                &self,
658                block: BlockNumber,
659            ) -> Result<(), TestRunnerError> {
660                self.db
661                    .ensure_no_entry_above_by_value::<tables::HeaderNumbers, _>(block, |val| val)?;
662                self.db.ensure_no_entry_above::<tables::CanonicalHeaders, _>(block, |key| key)?;
663                self.db.ensure_no_entry_above::<tables::Headers, _>(block, |key| key)?;
664                Ok(())
665            }
666
667            pub(crate) fn send_tip(&self, tip: B256) {
668                self.channel.0.send(tip).expect("failed to send tip");
669            }
670
671            /// Validate that the inserted block data is valid
672            pub(crate) fn validate_db_blocks(
673                &self,
674                prev_progress: BlockNumber,
675                highest_block: BlockNumber,
676            ) -> Result<(), TestRunnerError> {
677                let static_file_provider = self.db.factory.static_file_provider();
678
679                self.db.query(|tx| {
680                    // Acquire cursors on body related tables
681                    let mut bodies_cursor = tx.cursor_read::<tables::BlockBodyIndices>()?;
682                    let mut ommers_cursor = tx.cursor_read::<tables::BlockOmmers>()?;
683                    let mut tx_block_cursor = tx.cursor_read::<tables::TransactionBlocks>()?;
684
685                    let first_body_key = match bodies_cursor.first()? {
686                        Some((key, _)) => key,
687                        None => return Ok(()),
688                    };
689
690                    let mut prev_number: Option<BlockNumber> = None;
691
692
693                    for entry in bodies_cursor.walk(Some(first_body_key))? {
694                        let (number, body) = entry?;
695
696                        // Validate sequentiality only after prev progress,
697                        // since the data before is mocked and can contain gaps
698                        if number > prev_progress
699                            && let Some(prev_key) = prev_number {
700                                assert_eq!(prev_key + 1, number, "Body entries must be sequential");
701                            }
702
703                        // Validate that the current entry is below or equals to the highest allowed block
704                        assert!(
705                            number <= highest_block,
706                            "We wrote a block body outside of our synced range. Found block with number {number}, highest block according to stage is {highest_block}",
707                        );
708
709                        let header = static_file_provider.header_by_number(number)?.expect("to be present");
710                        // Validate that ommers exist if any
711                        let stored_ommers =  ommers_cursor.seek_exact(number)?;
712                        if header.ommers_hash_is_empty() {
713                            assert!(stored_ommers.is_none(), "Unexpected ommers entry");
714                        } else {
715                            assert!(stored_ommers.is_some(), "Missing ommers entry");
716                        }
717
718                        let tx_block_id = tx_block_cursor.seek_exact(body.last_tx_num())?.map(|(_,b)| b);
719                        if body.tx_count == 0 {
720                            assert_ne!(tx_block_id,Some(number));
721                        } else {
722                            assert_eq!(tx_block_id, Some(number));
723                        }
724
725                        for tx_id in body.tx_num_range() {
726                            assert!(static_file_provider.transaction_by_id(tx_id)?.is_some(), "Transaction is missing.");
727                        }
728
729                        prev_number = Some(number);
730                    }
731                    Ok(())
732                })?;
733                Ok(())
734            }
735
736            pub(crate) fn take_responses(&mut self) {
737                self.responses.take();
738            }
739
740            pub(crate) fn commit(&self) {
741                self.db.factory.static_file_provider().commit().unwrap();
742            }
743        }
744
745        #[derive(Clone)]
746        pub(crate) struct StubResponses(Vec<(Header, BlockBody<TransactionSigned>)>);
747
748        impl EraStreamFactory<Header, BlockBody<TransactionSigned>> for StubResponses {
749            fn create(
750                self,
751                _input: ExecInput,
752            ) -> Result<ThreadSafeEraStream<Header, BlockBody<TransactionSigned>>, StageError>
753            {
754                let stream = stream::iter(vec![self.0]);
755
756                Ok(Box::new(Box::pin(stream.map(|meta| {
757                    Ok(Box::new(meta.into_iter().map(Ok))
758                        as Item<Header, BlockBody<TransactionSigned>>)
759                }))))
760            }
761        }
762    }
763
764    stage_test_suite!(EraTestRunner, era);
765}
reth_stages/stages/era.rs

reth_stages/stages/
era.rs
