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

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

@@ -1,5 +1,6 @@
 use crate::{
     table::TableImporter,
+    tables::{self, RawKey, RawTable, RawValue},
     transaction::{DbTx, DbTxMut},
     DatabaseError,
 };
@@ -25,6 +26,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 +82,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 +107,47 @@ 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 and unwind fencing.
+pub trait ReaderTxnTracker: Send + Sync {
+    /// Waits until all readers older than the latest committed txnid have drained.
+    fn wait_for_pre_commit_readers(&self);
+
+    /// Forces a real commit boundary and waits until readers older than that new txnid have
+    /// drained.
+    fn wait_for_pre_fence_readers(&self) -> Result<(), DatabaseError>;
+}
+
+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));
+            }
+        }
+    }
+
+    fn wait_for_pre_fence_readers(&self) -> Result<(), DatabaseError> {
+        let last_txnid = Database::last_txnid(self).unwrap_or_default();
+        let tx = self.tx_mut()?;
+        tx.put::<RawTable<tables::Metadata>>(
+            RawKey::<String>::from_vec(vec![0, 1]),
+            RawValue::from_vec(last_txnid.to_be_bytes().into()),
+        )?;
+        tx.commit()?;
+
+        self.wait_for_pre_commit_readers();
+
+        Ok(())
+    }
 }

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/providers/database/mod.rs

@@ -281,6 +281,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),
         ))
     }
@@ -304,6 +305,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))
     }
 

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

@@ -37,7 +37,7 @@ use reth_chain_state::{ComputedTrieData, ExecutedBlock};
 use reth_chainspec::{ChainInfo, ChainSpecProvider, EthChainSpec};
 use reth_db_api::{
     cursor::{DbCursorRO, DbCursorRW, DbDupCursorRO, DbDupCursorRW},
-    database::Database,
+    database::{Database, ReaderTxnTracker},
     models::{
         sharded_key, storage_sharded_key::StorageShardedKey, AccountBeforeTx, BlockNumberAddress,
         BlockNumberAddressRange, ShardedKey, StorageBeforeTx, StorageSettings,
@@ -212,6 +212,8 @@ pub struct DatabaseProvider<TX, N: NodeTypes> {
     minimum_pruning_distance: u64,
     /// Database provider metrics
     metrics: metrics::DatabaseProviderMetrics,
+    /// Database handle used to inspect active MDBX readers during unwind commits.
+    reader_txn_tracker: Option<Arc<dyn ReaderTxnTracker>>,
 }
 
 impl<TX: Debug, N: NodeTypes> Debug for DatabaseProvider<TX, N> {
@@ -229,6 +231,7 @@ impl<TX: Debug, N: NodeTypes> Debug for DatabaseProvider<TX, N> {
             .field("pending_rocksdb_batches", &"<pending batches>")
             .field("commit_order", &self.commit_order)
             .field("minimum_pruning_distance", &self.minimum_pruning_distance)
+            .field("reader_txn_tracker", &"<reader txn tracker>")
             .finish()
     }
 }
@@ -244,9 +247,51 @@ impl<TX, N: NodeTypes> DatabaseProvider<TX, N> {
         self.minimum_pruning_distance = distance;
         self
     }
+
+    /// Attaches MDBX reader tracking so unwind commits can wait on active readers.
+    pub(crate) fn with_reader_txn_tracker<DB>(mut self, db: DB) -> Self
+    where
+        DB: Database + 'static,
+    {
+        self.reader_txn_tracker = Some(Arc::new(db));
+        self
+    }
 }
 
 impl<TX: DbTx + 'static, N: NodeTypes> DatabaseProvider<TX, N> {
+    /// Commits unwind writes in MDBX -> `RocksDB` -> static-file order.
+    ///
+    /// This keeps MDBX as the first durable step so an interrupted unwind can be recovered by
+    /// truncating static files from checkpoints on the next startup.
+    ///
+    /// For `storage_v2`, this waits after the MDBX commit and again after the `RocksDB` commit so
+    /// readers holding older MDBX-visible views, including readers opened before the `RocksDB`
+    /// commit, cannot overlap the later cross-store steps.
+    ///
+    /// Example: a reader that still sees pre-unwind `RocksDB` history must not survive long enough
+    /// to route a lookup into changesets (SF) that the unwind is about to make unreachable.
+    fn commit_unwind(self) -> ProviderResult<()> {
+        let storage_v2 = self.cached_storage_settings().storage_v2;
+        let reader_txn_tracker = self.reader_txn_tracker.clone();
+        self.tx.commit()?;
+
+        if storage_v2 && let Some(reader_txn_tracker) = reader_txn_tracker.as_ref() {
+            reader_txn_tracker.wait_for_pre_commit_readers();
+        }
+
+        let batches = std::mem::take(&mut *self.pending_rocksdb_batches.lock());
+        for batch in batches {
+            self.rocksdb_provider.commit_batch(batch)?;
+        }
+
+        if storage_v2 && let Some(reader_txn_tracker) = reader_txn_tracker.as_ref() {
+            reader_txn_tracker.wait_for_pre_fence_readers()?;
+        }
+
+        self.static_file_provider.commit()?;
+        Ok(())
+    }
+
     /// State provider for latest state
     pub fn latest<'a>(&'a self) -> Box<dyn StateProvider + 'a> {
         trace!(target: "providers::db", "Returning latest state provider");
@@ -382,6 +427,7 @@ impl<TX: DbTxMut, N: NodeTypes> DatabaseProvider<TX, N> {
             commit_order,
             minimum_pruning_distance: MINIMUM_UNWIND_SAFE_DISTANCE,
             metrics: metrics::DatabaseProviderMetrics::default(),
+            reader_txn_tracker: None,
         }
     }
 
@@ -1007,6 +1053,7 @@ impl<TX: DbTx + 'static, N: NodeTypesForProvider> DatabaseProvider<TX, N> {
             commit_order: CommitOrder::Normal,
             minimum_pruning_distance: MINIMUM_UNWIND_SAFE_DISTANCE,
             metrics: metrics::DatabaseProviderMetrics::default(),
+            reader_txn_tracker: None,
         }
     }
 
@@ -3800,19 +3847,8 @@ impl<TX: DbTx + 'static, N: NodeTypes + 'static> DBProvider for DatabaseProvider
         skip_all
     )]
     fn commit(self) -> ProviderResult<()> {
-        // For unwinding it makes more sense to commit the database first, since if
-        // it is interrupted before the static files commit, we can just
-        // truncate the static files according to the
-        // checkpoints on the next start-up.
         if self.static_file_provider.has_unwind_queued() || self.commit_order.is_unwind() {
-            self.tx.commit()?;
-
-            let batches = std::mem::take(&mut *self.pending_rocksdb_batches.lock());
-            for batch in batches {
-                self.rocksdb_provider.commit_batch(batch)?;
-            }
-
-            self.static_file_provider.commit()?;
+            self.commit_unwind()?;
         } else {
             // Normal path: finalize() will call sync_all() if not already synced
             let mut timings = metrics::CommitTimings::default();
@@ -3880,15 +3916,18 @@ mod tests {
         U256,
     };
     use reth_chain_state::ExecutedBlock;
+    use reth_db_api::models::StorageSettings;
     use reth_ethereum_primitives::Receipt;
     use reth_execution_types::{AccountRevertInit, BlockExecutionOutput, BlockExecutionResult};
     use reth_primitives_traits::SealedBlock;
+    use reth_storage_api::MetadataWriter;
     use reth_testing_utils::generators::{self, random_block, BlockParams};
     use reth_trie::{
         HashedPostState, KeccakKeyHasher, Nibbles, StoredNibbles, StoredNibblesSubKey,
     };
     use revm_database::BundleState;
     use revm_state::AccountInfo;
+    use std::{sync::mpsc, time::Duration};
 
     #[test]
     fn test_receipts_by_block_range_empty_range() {
@@ -3902,6 +3941,48 @@ mod tests {
         assert_eq!(result, Vec::<Vec<reth_ethereum_primitives::Receipt>>::new());
     }
 
+    #[test]
+    fn unwind_commit_waits_for_pre_commit_readers() {
+        let factory = create_test_provider_factory();
+        factory.set_storage_settings_cache(StorageSettings::v2());
+
+        let reader = factory.provider().unwrap();
+        let provider_rw = factory.unwind_provider_rw().unwrap();
+        provider_rw.write_metadata("unwind-wait-test", vec![1]).unwrap();
+        let (done_tx, done_rx) = mpsc::channel();
+
+        let handle = std::thread::spawn(move || {
+            let result = provider_rw.commit();
+            done_tx.send(result).unwrap();
+        });
+
+        assert!(
+            done_rx.recv_timeout(Duration::from_millis(50)).is_err(),
+            "unwind commit should wait while an older read transaction is still open"
+        );
+
+        drop(reader);
+
+        done_rx.recv_timeout(Duration::from_secs(1)).unwrap().unwrap();
+        handle.join().unwrap();
+    }
+
+    #[test]
+    fn unwind_fence_commit_advances_txnid() {
+        let factory = create_test_provider_factory();
+        factory.set_storage_settings_cache(StorageSettings::v2());
+
+        let provider_rw = factory.unwind_provider_rw().unwrap();
+        provider_rw.write_metadata("unwind-fence-txnid-test", vec![1]).unwrap();
+        provider_rw.commit().unwrap();
+
+        let before_fence = Database::last_txnid(factory.db_ref()).unwrap();
+        ReaderTxnTracker::wait_for_pre_fence_readers(factory.db_ref()).unwrap();
+        let fence_txnid = Database::last_txnid(factory.db_ref()).unwrap();
+
+        assert!(fence_txnid > before_fence, "sentinel fence should advance the MDBX txnid");
+    }
+
     #[test]
     fn test_receipts_by_block_range_nonexistent_blocks() {
         let factory = create_test_provider_factory();