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            let height = era::process_iter(
235                era,
236                &mut writer,
237                provider,
238                &mut self.hash_collector,
239                last_header_number..=input.target(),
240            )
241            .map_err(|e| StageError::Fatal(e.into()))?;
242
243            if !self.hash_collector.is_empty() {
244                era::build_index(provider, &mut self.hash_collector)
245                    .map_err(|e| StageError::Recoverable(e.into()))?;
246                self.hash_collector.clear();
247            }
248
249            era::save_stage_checkpoints(
250                provider,
251                input.checkpoint().block_number,
252                height,
253                height,
254                input.target(),
255            )?;
256
257            height
258        } else {
259            // No era files to process. Return the highest block we're aware of to avoid
260            // limiting subsequent stages with an outdated checkpoint.
261            //
262            // This can happen when:
263            // 1. Era import is complete (all pre-merge blocks imported)
264            // 2. No era import source was configured
265            //
266            // We return max(checkpoint, highest_header, target) to ensure we don't return
267            // a stale checkpoint that could limit subsequent stages like Headers.
268            let highest_header = provider
269                .static_file_provider()
270                .get_highest_static_file_block(StaticFileSegment::Headers)
271                .unwrap_or_default();
272
273            let checkpoint = input.checkpoint().block_number;
274            let from_target = input.target.unwrap_or(checkpoint);
275
276            checkpoint.max(highest_header).max(from_target)
277        };
278
279        Ok(ExecOutput { checkpoint: StageCheckpoint::new(height), done: height >= input.target() })
280    }
281
282    fn unwind(
283        &mut self,
284        _provider: &Provider,
285        input: UnwindInput,
286    ) -> Result<UnwindOutput, StageError> {
287        Ok(UnwindOutput { checkpoint: input.checkpoint.with_block_number(input.unwind_to) })
288    }
289}
290
291/// Describes where to get the era files from.
292#[derive(Debug, Clone)]
293pub enum EraImportSource {
294    /// Remote HTTP accessible host.
295    Url(Url, Box<Path>),
296    /// Local directory.
297    Path(Box<Path>),
298}
299
300impl EraImportSource {
301    /// Maybe constructs a new `EraImportSource` depending on the arguments.
302    ///
303    /// Only one of `url` or `path` should be provided, but upholding this invariant is delegated
304    /// above so that both parameters can be accepted.
305    ///
306    /// # Arguments
307    /// * The `path` uses a directory as the import source. It and its contents must be readable.
308    /// * The `url` uses an HTTP client to list and download files.
309    /// * The `default` gives the default [`Url`] if none of the previous parameters are provided.
310    /// * For any [`Url`] the `folder` is used as the download directory for storing files
311    ///   temporarily. It and its contents must be readable and writable.
312    pub fn maybe_new(
313        path: Option<Box<Path>>,
314        url: Option<Url>,
315        default: impl FnOnce() -> Option<Url>,
316        folder: impl FnOnce() -> Box<Path>,
317    ) -> Option<Self> {
318        path.map(Self::Path).or_else(|| url.or_else(default).map(|url| Self::Url(url, folder())))
319    }
320}
321
322#[cfg(test)]
323mod tests {
324    use super::*;
325    use crate::test_utils::{
326        stage_test_suite, ExecuteStageTestRunner, StageTestRunner, UnwindStageTestRunner,
327    };
328    use alloy_consensus::{BlockBody, Header};
329    use alloy_primitives::B256;
330    use assert_matches::assert_matches;
331    use futures_util::stream;
332    use reth_db_api::tables;
333    use reth_era::{
334        common::file_ops::{EraFileFormat, StreamWriter},
335        ere::{
336            file::{EreFile, EreWriter},
337            types::{
338                execution::{BlockTuple, CompressedBody, CompressedHeader},
339                group::{DynamicBlockIndex, EreGroup, EreId},
340            },
341        },
342    };
343    use reth_ethereum_primitives::{Block, TransactionSigned};
344    use reth_primitives_traits::SealedBlock;
345    use reth_provider::BlockHashReader;
346    use reth_testing_utils::generators::{
347        self, random_block_range, random_header, BlockRangeParams,
348    };
349    use std::{fs::File, path::PathBuf};
350    use test_runner::EraTestRunner;
351
352    #[tokio::test]
353    async fn test_era_range_ends_below_target() {
354        let era_cap = 2;
355        let target = 20000;
356
357        let mut runner = EraTestRunner::default();
358
359        let input = ExecInput { target: Some(era_cap), checkpoint: None };
360        runner.seed_execution(input).unwrap();
361
362        let input = ExecInput { target: Some(target), checkpoint: None };
363        let output = runner.execute(input).await.unwrap();
364
365        runner.commit();
366
367        assert_matches!(
368            output,
369            Ok(ExecOutput {
370                checkpoint: StageCheckpoint { block_number, stage_checkpoint: None },
371                done: false
372            }) if block_number == era_cap
373        );
374
375        let output = output.unwrap();
376        let validation_output = runner.validate_execution(input, Some(output.clone()));
377
378        assert_matches!(validation_output, Ok(()));
379
380        runner.take_responses();
381
382        let input = ExecInput { target: Some(target), checkpoint: Some(output.checkpoint) };
383        let output = runner.execute(input).await.unwrap();
384
385        runner.commit();
386
387        assert_matches!(
388            output,
389            Ok(ExecOutput {
390                checkpoint: StageCheckpoint { block_number, stage_checkpoint: None },
391                done: true
392            }) if block_number == target
393        );
394
395        let validation_output = runner.validate_execution(input, output.ok());
396
397        assert_matches!(validation_output, Ok(()));
398    }
399
400    /// Writes `blocks` to a header+body-only ERE file at `path`.
401    fn write_ere_file(path: &Path, blocks: &[SealedBlock<Block>]) {
402        let tuples = blocks
403            .iter()
404            .map(|block| {
405                BlockTuple::new(
406                    CompressedHeader::from_header(block.header()).unwrap(),
407                    CompressedBody::from_body(block.body()).unwrap(),
408                )
409            })
410            .collect::<Vec<_>>();
411
412        let start = blocks[0].number;
413        // Header + body only is component-count 2. The index offsets are stored verbatim and not
414        // interpreted when assembling blocks, so placeholder values are fine.
415        let component_count = 2;
416        let offsets = vec![0; blocks.len() * component_count as usize];
417        let index = DynamicBlockIndex::new(start, component_count, offsets);
418        let group = EreGroup::new(tuples, None, index);
419        let file = EreFile::new(group, EreId::new("mainnet", start, blocks.len() as u32));
420
421        EreWriter::new(File::create(path).unwrap()).write_file(&file).unwrap();
422    }
423
424    #[derive(Debug)]
425    struct EreTestMeta {
426        path: PathBuf,
427    }
428
429    impl EraMeta for EreTestMeta {
430        fn mark_as_processed(&self) -> eyre::Result<()> {
431            Ok(())
432        }
433
434        fn path(&self) -> &Path {
435            &self.path
436        }
437    }
438
439    #[tokio::test]
440    async fn convert_decodes_ere_files() {
441        let mut rng = generators::rng();
442        let blocks = random_block_range(
443            &mut rng,
444            1..=3,
445            BlockRangeParams { tx_count: 1..3, ..Default::default() },
446        );
447
448        let dir = tempfile::tempdir().unwrap();
449
450        // ERE files are published with either the `.ere` or `.erae` extension; both must be
451        // recognized and routed to the ERE reader rather than decoded as era1.
452        for ext in ["ere", "erae"] {
453            let path = dir.path().join(format!("mainnet-00000-abcd1234.{ext}"));
454            write_ere_file(&path, &blocks);
455
456            let stream =
457                stream::iter(vec![Ok::<_, eyre::Error>(EreTestMeta { path: path.clone() })]);
458            let mut stream =
459                EraImportSource::convert::<Header, BlockBody<TransactionSigned>>(stream).unwrap();
460
461            let item = stream.next().await.expect("a file to decode").expect("decoding to succeed");
462            let decoded = item.collect::<eyre::Result<Vec<_>>>().unwrap();
463
464            assert_eq!(
465                decoded.len(),
466                blocks.len(),
467                "ERE file with `.{ext}` extension should decode every block"
468            );
469            for ((header, body), block) in decoded.iter().zip(&blocks) {
470                assert_eq!(header, block.header());
471                assert_eq!(body, block.body());
472            }
473        }
474    }
475
476    mod test_runner {
477        use super::*;
478        use crate::test_utils::{TestRunnerError, TestStageDB};
479        use alloy_consensus::{BlockBody, Header};
480        use futures_util::stream;
481        use reth_db_api::{
482            cursor::DbCursorRO,
483            models::{StoredBlockBodyIndices, StoredBlockOmmers},
484            transaction::DbTx,
485        };
486        use reth_ethereum_primitives::TransactionSigned;
487        use reth_primitives_traits::{SealedBlock, SealedHeader};
488        use reth_provider::{BlockNumReader, HeaderProvider, TransactionsProvider};
489        use reth_testing_utils::generators::{
490            random_block_range, random_signed_tx, BlockRangeParams,
491        };
492        use tokio::sync::watch;
493
494        pub(crate) struct EraTestRunner {
495            channel: (watch::Sender<B256>, watch::Receiver<B256>),
496            db: TestStageDB,
497            responses: Option<Vec<(Header, BlockBody<TransactionSigned>)>>,
498        }
499
500        impl Default for EraTestRunner {
501            fn default() -> Self {
502                Self {
503                    channel: watch::channel(B256::ZERO),
504                    db: TestStageDB::default(),
505                    responses: Default::default(),
506                }
507            }
508        }
509
510        impl StageTestRunner for EraTestRunner {
511            type S = EraStage<Header, BlockBody<TransactionSigned>, StubResponses>;
512
513            fn db(&self) -> &TestStageDB {
514                &self.db
515            }
516
517            fn stage(&self) -> Self::S {
518                EraStage::new(self.responses.clone().map(StubResponses), EtlConfig::default())
519            }
520        }
521
522        impl ExecuteStageTestRunner for EraTestRunner {
523            type Seed = Vec<SealedBlock<reth_ethereum_primitives::Block>>;
524
525            fn seed_execution(&mut self, input: ExecInput) -> Result<Self::Seed, TestRunnerError> {
526                let start = input.checkpoint().block_number;
527                let end = input.target();
528
529                let static_file_provider = self.db.factory.static_file_provider();
530
531                let mut rng = generators::rng();
532
533                // Static files do not support gaps in headers, so we need to generate 0 to end
534                let blocks = random_block_range(
535                    &mut rng,
536                    0..=end,
537                    BlockRangeParams {
538                        parent: Some(B256::ZERO),
539                        tx_count: 0..2,
540                        ..Default::default()
541                    },
542                );
543                self.db.insert_headers(blocks.iter().map(|block| block.sealed_header()))?;
544                if let Some(progress) = blocks.get(start as usize) {
545                    // Insert last progress data
546                    {
547                        let tx = self.db.factory.provider_rw()?.into_tx();
548                        let mut static_file_producer = static_file_provider
549                            .get_writer(start, StaticFileSegment::Transactions)?;
550
551                        let body = StoredBlockBodyIndices {
552                            first_tx_num: 0,
553                            tx_count: progress.transaction_count() as u64,
554                        };
555
556                        static_file_producer.set_block_range(0..=progress.number);
557
558                        body.tx_num_range().try_for_each(|tx_num| {
559                            let transaction = random_signed_tx(&mut rng);
560                            static_file_producer.append_transaction(tx_num, &transaction).map(drop)
561                        })?;
562
563                        if body.tx_count != 0 {
564                            tx.put::<tables::TransactionBlocks>(
565                                body.last_tx_num(),
566                                progress.number,
567                            )?;
568                        }
569
570                        tx.put::<tables::BlockBodyIndices>(progress.number, body)?;
571
572                        if !progress.ommers_hash_is_empty() {
573                            tx.put::<tables::BlockOmmers>(
574                                progress.number,
575                                StoredBlockOmmers { ommers: progress.body().ommers.clone() },
576                            )?;
577                        }
578
579                        static_file_producer.commit()?;
580                        tx.commit()?;
581                    }
582                }
583                self.responses.replace(
584                    blocks.iter().map(|v| (v.header().clone(), v.body().clone())).collect(),
585                );
586                Ok(blocks)
587            }
588
589            /// Validate stored headers and bodies
590            fn validate_execution(
591                &self,
592                input: ExecInput,
593                output: Option<ExecOutput>,
594            ) -> Result<(), TestRunnerError> {
595                let initial_checkpoint = input.checkpoint().block_number;
596                match output {
597                    Some(output) if output.checkpoint.block_number > initial_checkpoint => {
598                        let provider = self.db.factory.provider()?;
599
600                        for block_num in initial_checkpoint..
601                            output
602                                .checkpoint
603                                .block_number
604                                .min(self.responses.as_ref().map(|v| v.len()).unwrap_or_default()
605                                    as BlockNumber)
606                        {
607                            // look up the header hash
608                            let hash = provider.block_hash(block_num)?.expect("no header hash");
609
610                            // validate the header number
611                            assert_eq!(provider.block_number(hash)?, Some(block_num));
612
613                            // validate the header
614                            let header = provider.header_by_number(block_num)?;
615                            assert!(header.is_some());
616                            let header = SealedHeader::seal_slow(header.unwrap());
617                            assert_eq!(header.hash(), hash);
618                        }
619
620                        self.validate_db_blocks(
621                            output.checkpoint.block_number,
622                            output.checkpoint.block_number,
623                        )?;
624                    }
625                    _ => self.check_no_header_entry_above(initial_checkpoint)?,
626                };
627                Ok(())
628            }
629
630            async fn after_execution(&self, headers: Self::Seed) -> Result<(), TestRunnerError> {
631                let tip = if headers.is_empty() {
632                    let tip = random_header(&mut generators::rng(), 0, None);
633                    self.db.insert_headers(iter::once(&tip))?;
634                    tip.hash()
635                } else {
636                    headers.last().unwrap().hash()
637                };
638                self.send_tip(tip);
639                Ok(())
640            }
641        }
642
643        impl UnwindStageTestRunner for EraTestRunner {
644            fn validate_unwind(&self, _input: UnwindInput) -> Result<(), TestRunnerError> {
645                Ok(())
646            }
647        }
648
649        impl EraTestRunner {
650            pub(crate) fn check_no_header_entry_above(
651                &self,
652                block: BlockNumber,
653            ) -> Result<(), TestRunnerError> {
654                self.db
655                    .ensure_no_entry_above_by_value::<tables::HeaderNumbers, _>(block, |val| val)?;
656                self.db.ensure_no_entry_above::<tables::CanonicalHeaders, _>(block, |key| key)?;
657                self.db.ensure_no_entry_above::<tables::Headers, _>(block, |key| key)?;
658                Ok(())
659            }
660
661            pub(crate) fn send_tip(&self, tip: B256) {
662                self.channel.0.send(tip).expect("failed to send tip");
663            }
664
665            /// Validate that the inserted block data is valid
666            pub(crate) fn validate_db_blocks(
667                &self,
668                prev_progress: BlockNumber,
669                highest_block: BlockNumber,
670            ) -> Result<(), TestRunnerError> {
671                let static_file_provider = self.db.factory.static_file_provider();
672
673                self.db.query(|tx| {
674                    // Acquire cursors on body related tables
675                    let mut bodies_cursor = tx.cursor_read::<tables::BlockBodyIndices>()?;
676                    let mut ommers_cursor = tx.cursor_read::<tables::BlockOmmers>()?;
677                    let mut tx_block_cursor = tx.cursor_read::<tables::TransactionBlocks>()?;
678
679                    let first_body_key = match bodies_cursor.first()? {
680                        Some((key, _)) => key,
681                        None => return Ok(()),
682                    };
683
684                    let mut prev_number: Option<BlockNumber> = None;
685
686
687                    for entry in bodies_cursor.walk(Some(first_body_key))? {
688                        let (number, body) = entry?;
689
690                        // Validate sequentiality only after prev progress,
691                        // since the data before is mocked and can contain gaps
692                        if number > prev_progress
693                            && let Some(prev_key) = prev_number {
694                                assert_eq!(prev_key + 1, number, "Body entries must be sequential");
695                            }
696
697                        // Validate that the current entry is below or equals to the highest allowed block
698                        assert!(
699                            number <= highest_block,
700                            "We wrote a block body outside of our synced range. Found block with number {number}, highest block according to stage is {highest_block}",
701                        );
702
703                        let header = static_file_provider.header_by_number(number)?.expect("to be present");
704                        // Validate that ommers exist if any
705                        let stored_ommers =  ommers_cursor.seek_exact(number)?;
706                        if header.ommers_hash_is_empty() {
707                            assert!(stored_ommers.is_none(), "Unexpected ommers entry");
708                        } else {
709                            assert!(stored_ommers.is_some(), "Missing ommers entry");
710                        }
711
712                        let tx_block_id = tx_block_cursor.seek_exact(body.last_tx_num())?.map(|(_,b)| b);
713                        if body.tx_count == 0 {
714                            assert_ne!(tx_block_id,Some(number));
715                        } else {
716                            assert_eq!(tx_block_id, Some(number));
717                        }
718
719                        for tx_id in body.tx_num_range() {
720                            assert!(static_file_provider.transaction_by_id(tx_id)?.is_some(), "Transaction is missing.");
721                        }
722
723                        prev_number = Some(number);
724                    }
725                    Ok(())
726                })?;
727                Ok(())
728            }
729
730            pub(crate) fn take_responses(&mut self) {
731                self.responses.take();
732            }
733
734            pub(crate) fn commit(&self) {
735                self.db.factory.static_file_provider().commit().unwrap();
736            }
737        }
738
739        #[derive(Clone)]
740        pub(crate) struct StubResponses(Vec<(Header, BlockBody<TransactionSigned>)>);
741
742        impl EraStreamFactory<Header, BlockBody<TransactionSigned>> for StubResponses {
743            fn create(
744                self,
745                _input: ExecInput,
746            ) -> Result<ThreadSafeEraStream<Header, BlockBody<TransactionSigned>>, StageError>
747            {
748                let stream = stream::iter(vec![self.0]);
749
750                Ok(Box::new(Box::pin(stream.map(|meta| {
751                    Ok(Box::new(meta.into_iter().map(Ok))
752                        as Item<Header, BlockBody<TransactionSigned>>)
753                }))))
754            }
755        }
756    }
757
758    stage_test_suite!(EraTestRunner, era);
759}
reth_stages/stages/era.rs

reth_stages/stages/
era.rs
