fix(provider): cap storage_v2 unwind history by MDBX tip

crates/stages/api/src/pipeline/mod.rs

@@ -299,13 +299,14 @@ impl<N: ProviderNodeTypes> Pipeline<N> {
         bad_block: Option<BlockNumber>,
     ) -> Result<(), PipelineError> {
         // Add validation before starting unwind
-        let provider = self.provider_factory.provider()?;
-        let latest_block = provider.last_block_number()?;
-
-        // Get the actual pruning configuration
-        let prune_modes = provider.prune_modes_ref();
-
-        let checkpoints = provider.get_prune_checkpoints()?;
+        let (latest_block, prune_modes, checkpoints) = {
+            let provider = self.provider_factory.provider()?;
+            (
+                provider.last_block_number()?,
+                provider.prune_modes_ref().clone(),
+                provider.get_prune_checkpoints()?,
+            )
+        };
         prune_modes.ensure_unwind_target_unpruned(latest_block, to, &checkpoints)?;
 
         // Unwind stages in reverse order of execution

crates/storage/db-api/src/database.rs

@@ -25,6 +25,18 @@ pub trait Database: Send + Sync + Debug {
     /// Returns the path to the database directory.
     fn path(&self) -> PathBuf;
 
+    /// Returns the transaction ID of the oldest active reader, if available.
+    ///
+    /// This is the committed txnid of the snapshot the reader is pinned to, not a unique per-reader
+    /// identifier, so multiple readers can report the same txnid.
+    ///
+    /// Used to check whether stale readers from a previous write transaction have completed.
+    /// Returns `None` if no readers are active or the backend does not support this query.
+    fn oldest_reader_txnid(&self) -> Option<u64>;
+
+    /// Returns the ID of the most recently committed transaction, if available.
+    fn last_txnid(&self) -> Option<u64>;
+
     /// Takes a function and passes a read-only transaction into it, making sure it's closed in the
     /// end of the execution.
     fn view<T, F>(&self, f: F) -> Result<T, DatabaseError>
@@ -69,6 +81,14 @@ impl<DB: Database> Database for Arc<DB> {
     fn path(&self) -> PathBuf {
         <DB as Database>::path(self)
     }
+
+    fn oldest_reader_txnid(&self) -> Option<u64> {
+        <DB as Database>::oldest_reader_txnid(self)
+    }
+
+    fn last_txnid(&self) -> Option<u64> {
+        <DB as Database>::last_txnid(self)
+    }
 }
 
 impl<DB: Database> Database for &DB {
@@ -86,4 +106,29 @@ impl<DB: Database> Database for &DB {
     fn path(&self) -> PathBuf {
         <DB as Database>::path(self)
     }
+
+    fn oldest_reader_txnid(&self) -> Option<u64> {
+        <DB as Database>::oldest_reader_txnid(self)
+    }
+
+    fn last_txnid(&self) -> Option<u64> {
+        <DB as Database>::last_txnid(self)
+    }
+}
+
+/// Object-safe adapter for reader-txn tracking during unwind.
+pub trait ReaderTxnTracker: Send + Sync {
+    /// Waits until all readers older than the latest committed txnid have drained.
+    fn wait_for_pre_commit_readers(&self);
+}
+
+impl<DB: Database> ReaderTxnTracker for DB {
+    fn wait_for_pre_commit_readers(&self) {
+        if let Some(committed_txnid) = Database::last_txnid(self) {
+            while Database::oldest_reader_txnid(self).is_some_and(|oldest| oldest < committed_txnid)
+            {
+                std::thread::sleep(std::time::Duration::from_millis(10));
+            }
+        }
+    }
 }

crates/storage/db-api/src/mock.rs

@@ -54,6 +54,14 @@ impl Database for DatabaseMock {
     fn path(&self) -> PathBuf {
         PathBuf::default()
     }
+
+    fn oldest_reader_txnid(&self) -> Option<u64> {
+        None
+    }
+
+    fn last_txnid(&self) -> Option<u64> {
+        None
+    }
 }
 
 impl DatabaseMetrics for DatabaseMock {}

crates/storage/db/src/implementation/mdbx/mod.rs

@@ -283,6 +283,26 @@ impl Database for DatabaseEnv {
     fn path(&self) -> PathBuf {
         self.path.clone()
     }
+
+    fn oldest_reader_txnid(&self) -> Option<u64> {
+        let info = self.inner.info().ok()?;
+        let txnid = info.latter_reader_txnid();
+        if txnid == 0 {
+            None
+        } else {
+            Some(txnid)
+        }
+    }
+
+    fn last_txnid(&self) -> Option<u64> {
+        let info = self.inner.info().ok()?;
+        let txnid = info.last_txnid();
+        if txnid == 0 {
+            None
+        } else {
+            Some(txnid as u64)
+        }
+    }
 }
 
 impl DatabaseMetrics for DatabaseEnv {

crates/storage/db/src/lib.rs

@@ -144,6 +144,14 @@ pub mod test_utils {
         fn path(&self) -> std::path::PathBuf {
             self.db().path()
         }
+
+        fn oldest_reader_txnid(&self) -> Option<u64> {
+            self.db().oldest_reader_txnid()
+        }
+
+        fn last_txnid(&self) -> Option<u64> {
+            self.db().last_txnid()
+        }
     }
 
     impl<DB: DatabaseMetrics> DatabaseMetrics for TempDatabase<DB> {

crates/storage/libmdbx-rs/src/environment.rs

@@ -424,6 +424,12 @@ impl Info {
         self.0.mi_numreaders as usize
     }
 
+    /// Transaction ID of the oldest active reader.
+    #[inline]
+    pub const fn latter_reader_txnid(&self) -> u64 {
+        self.0.mi_latter_reader_txnid
+    }
+
     /// Return the internal page ops metrics
     #[inline]
     pub const fn page_ops(&self) -> PageOps {

crates/storage/provider/src/either_writer.rs

@@ -848,6 +848,7 @@ where
         storage_key: alloy_primitives::B256,
         block_number: BlockNumber,
         lowest_available_block_number: Option<BlockNumber>,
+        visible_tip: BlockNumber,
     ) -> ProviderResult<HistoryInfo> {
         match self {
             Self::Database(cursor, _) => {
@@ -866,6 +867,7 @@ where
                 storage_key,
                 block_number,
                 lowest_available_block_number,
+                visible_tip,
             ),
         }
     }
@@ -893,6 +895,7 @@ where
         address: Address,
         block_number: BlockNumber,
         lowest_available_block_number: Option<BlockNumber>,
+        visible_tip: BlockNumber,
     ) -> ProviderResult<HistoryInfo> {
         match self {
             Self::Database(cursor, _) => {
@@ -906,9 +909,12 @@ where
                 )
             }
             Self::StaticFile(_, _) => Err(ProviderError::UnsupportedProvider),
-            Self::RocksDB(snapshot) => {
-                snapshot.account_history_info(address, block_number, lowest_available_block_number)
-            }
+            Self::RocksDB(snapshot) => snapshot.account_history_info(
+                address,
+                block_number,
+                lowest_available_block_number,
+                visible_tip,
+            ),
         }
     }
 }
@@ -1438,12 +1444,12 @@ mod rocksdb_tests {
                     PhantomData,
                 );
             let mdbx_result = mdbx_reader
-                .account_history_info(address, query.block_number, query.lowest_available)
+                .account_history_info(address, query.block_number, query.lowest_available, u64::MAX)
                 .unwrap();
 
             // RocksDB query via EitherReader — reuse snapshot for consistent view
             let rocks_result = rocks_snapshot
-                .account_history_info(address, query.block_number, query.lowest_available)
+                .account_history_info(address, query.block_number, query.lowest_available, u64::MAX)
                 .unwrap();
 
             // Assert both backends produce identical results
@@ -1532,6 +1538,7 @@ mod rocksdb_tests {
                     storage_key,
                     query.block_number,
                     query.lowest_available,
+                    u64::MAX,
                 )
                 .unwrap();
 
@@ -1542,6 +1549,7 @@ mod rocksdb_tests {
                     storage_key,
                     query.block_number,
                     query.lowest_available,
+                    u64::MAX,
                 )
                 .unwrap();
 

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

@@ -281,13 +281,17 @@ impl<N: ProviderNodeTypes> ProviderFactory<N> {
                 self.runtime.clone(),
                 self.db.path(),
             )
+            .with_reader_txn_tracker(self.db.clone())
             .with_minimum_pruning_distance(self.minimum_pruning_distance),
         ))
     }
 
     /// Returns a provider with a created `DbTxMut` inside, configured for unwind operations.
     /// Uses unwind commit order (MDBX first, then `RocksDB`, then static files) to allow
     /// recovery by truncating static files on restart if interrupted.
+    ///
+    /// Unwind commits may wait for pre-existing readers to drain before finishing later
+    /// cross-store steps. Drop any long-lived read providers before committing this provider.
     #[track_caller]
     pub fn unwind_provider_rw(
         &self,
@@ -304,6 +308,7 @@ impl<N: ProviderNodeTypes> ProviderFactory<N> {
             self.runtime.clone(),
             self.db.path(),
         )
+        .with_reader_txn_tracker(self.db.clone())
         .with_minimum_pruning_distance(self.minimum_pruning_distance))
     }