chore: add safety check on StaticFileProviderRW::increment_block (#7137)

Co-authored-by: Alexey Shekhirin <a.shekhirin@gmail.com>

crates/storage/provider/src/bundle_state/bundle_state_with_receipts.rs

@@ -324,7 +324,7 @@ impl BundleStateWithReceipts {
 
             if let Some(static_file_producer) = &mut static_file_producer {
                 // Increment block on static file header.
-                static_file_producer.increment_block(StaticFileSegment::Receipts)?;
+                static_file_producer.increment_block(StaticFileSegment::Receipts, block_number)?;
 
                 for (tx_idx, receipt) in receipts.into_iter().enumerate() {
                     let receipt = receipt

crates/storage/provider/src/providers/database/provider.rs

@@ -181,6 +181,17 @@ impl<TX: DbTxMut + DbTx> DatabaseProvider<TX> {
         };
 
         let mut writer = self.static_file_provider.latest_writer(StaticFileSegment::Headers)?;
+
+        // Backfill: some tests start at a forward block number, but static files require no gaps.
+        let segment_header = writer.user_header();
+        if segment_header.block_end().is_none() && segment_header.expected_block_start() == 0 {
+            for block_number in 0..block.number {
+                let mut prev = block.header.clone().unseal();
+                prev.number = block_number;
+                writer.append_header(prev, U256::ZERO, B256::ZERO)?;
+            }
+        }
+
         writer.append_header(block.header.as_ref().clone(), ttd, block.hash())?;
 
         self.insert_block(block, prune_modes)

crates/storage/provider/src/providers/static_file/writer.rs

@@ -185,7 +185,13 @@ impl StaticFileProviderRW {
     /// and create the next one if we are past the end range.
     ///
     /// Returns the current [`BlockNumber`] as seen in the static file.
-    pub fn increment_block(&mut self, segment: StaticFileSegment) -> ProviderResult<BlockNumber> {
+    pub fn increment_block(
+        &mut self,
+        segment: StaticFileSegment,
+        expected_block_number: BlockNumber,
+    ) -> ProviderResult<BlockNumber> {
+        self.check_next_block_number(expected_block_number, segment)?;
+
         let start = Instant::now();
         if let Some(last_block) = self.writer.user_header().block_end() {
             // We have finished the previous static file and must freeze it
@@ -216,6 +222,33 @@ impl StaticFileProviderRW {
         Ok(block)
     }
 
+    /// Verifies if the incoming block number matches the next expected block number
+    /// for a static file. This ensures data continuity when adding new blocks.
+    fn check_next_block_number(
+        &mut self,
+        expected_block_number: u64,
+        segment: StaticFileSegment,
+    ) -> ProviderResult<()> {
+        // The next static file block number can be found by checking the one after block_end.
+        // However if it's a new file that hasn't been added any data, its block range will actually
+        // be None. In that case, the next block will be found on `expected_block_start`.
+        let next_static_file_block = self
+            .writer
+            .user_header()
+            .block_end()
+            .map(|b| b + 1)
+            .unwrap_or(self.writer.user_header().expected_block_start());
+
+        if expected_block_number != next_static_file_block {
+            return Err(ProviderError::UnexpectedStaticFileBlockNumber(
+                segment,
+                expected_block_number,
+                next_static_file_block,
+            ))
+        }
+        Ok(())
+    }
+
     /// Truncates a number of rows from disk. It deletes and loads an older static file if block
     /// goes beyond the start of the current block range.
     ///
@@ -342,7 +375,7 @@ impl StaticFileProviderRW {
 
         debug_assert!(self.writer.user_header().segment() == StaticFileSegment::Headers);
 
-        let block_number = self.increment_block(StaticFileSegment::Headers)?;
+        let block_number = self.increment_block(StaticFileSegment::Headers, header.number)?;
 
         self.append_column(header)?;
         self.append_column(CompactU256::from(terminal_difficulty))?;
@@ -511,6 +544,12 @@ impl StaticFileProviderRW {
     pub fn set_block_range(&mut self, block_range: std::ops::RangeInclusive<BlockNumber>) {
         self.writer.user_header_mut().set_block_range(*block_range.start(), *block_range.end())
     }
+
+    #[cfg(any(test, feature = "test-utils"))]
+    /// Helper function to access [`SegmentHeader`].
+    pub fn user_header(&self) -> &SegmentHeader {
+        self.writer.user_header()
+    }
 }
 
 fn create_jar(