diff --git a/crates/stages/api/src/pipeline/mod.rs b/crates/stages/api/src/pipeline/mod.rs index 37506a65b76..374a79ccb5e 100644 --- a/crates/stages/api/src/pipeline/mod.rs +++ b/crates/stages/api/src/pipeline/mod.rs @@ -299,13 +299,14 @@ impl Pipeline { bad_block: Option, ) -> 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 diff --git a/crates/storage/db-api/src/database.rs b/crates/storage/db-api/src/database.rs index 71403505a15..e9d13d5f101 100644 --- a/crates/storage/db-api/src/database.rs +++ b/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; + + /// Returns the ID of the most recently committed transaction, if available. + fn last_txnid(&self) -> Option; + /// Takes a function and passes a read-only transaction into it, making sure it's closed in the /// end of the execution. fn view(&self, f: F) -> Result @@ -69,6 +81,14 @@ impl Database for Arc { fn path(&self) -> PathBuf { ::path(self) } + + fn oldest_reader_txnid(&self) -> Option { + ::oldest_reader_txnid(self) + } + + fn last_txnid(&self) -> Option { + ::last_txnid(self) + } } impl Database for &DB { @@ -86,4 +106,29 @@ impl Database for &DB { fn path(&self) -> PathBuf { ::path(self) } + + fn oldest_reader_txnid(&self) -> Option { + ::oldest_reader_txnid(self) + } + + fn last_txnid(&self) -> Option { + ::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 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)); + } + } + } } diff --git a/crates/storage/db-api/src/mock.rs b/crates/storage/db-api/src/mock.rs index 324f3cddac6..688161c01db 100644 --- a/crates/storage/db-api/src/mock.rs +++ b/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 { + None + } + + fn last_txnid(&self) -> Option { + None + } } impl DatabaseMetrics for DatabaseMock {} diff --git a/crates/storage/db/src/implementation/mdbx/mod.rs b/crates/storage/db/src/implementation/mdbx/mod.rs index 325a7918ded..d9c63b3bd68 100644 --- a/crates/storage/db/src/implementation/mdbx/mod.rs +++ b/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 { + let info = self.inner.info().ok()?; + let txnid = info.latter_reader_txnid(); + if txnid == 0 { + None + } else { + Some(txnid) + } + } + + fn last_txnid(&self) -> Option { + let info = self.inner.info().ok()?; + let txnid = info.last_txnid(); + if txnid == 0 { + None + } else { + Some(txnid as u64) + } + } } impl DatabaseMetrics for DatabaseEnv { diff --git a/crates/storage/db/src/lib.rs b/crates/storage/db/src/lib.rs index 97a33b138c3..f6062551f72 100644 --- a/crates/storage/db/src/lib.rs +++ b/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 { + self.db().oldest_reader_txnid() + } + + fn last_txnid(&self) -> Option { + self.db().last_txnid() + } } impl DatabaseMetrics for TempDatabase { diff --git a/crates/storage/libmdbx-rs/src/environment.rs b/crates/storage/libmdbx-rs/src/environment.rs index 0154d097036..1c5edc003d1 100644 --- a/crates/storage/libmdbx-rs/src/environment.rs +++ b/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 { diff --git a/crates/storage/provider/src/either_writer.rs b/crates/storage/provider/src/either_writer.rs index 7e5c9765def..90f6b77ac91 100644 --- a/crates/storage/provider/src/either_writer.rs +++ b/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, + visible_tip: BlockNumber, ) -> ProviderResult { 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, + visible_tip: BlockNumber, ) -> ProviderResult { 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(); diff --git a/crates/storage/provider/src/providers/database/mod.rs b/crates/storage/provider/src/providers/database/mod.rs index c2ca29c36ba..53a5a881b77 100644 --- a/crates/storage/provider/src/providers/database/mod.rs +++ b/crates/storage/provider/src/providers/database/mod.rs @@ -281,6 +281,7 @@ impl ProviderFactory { self.runtime.clone(), self.db.path(), ) + .with_reader_txn_tracker(self.db.clone()) .with_minimum_pruning_distance(self.minimum_pruning_distance), )) } @@ -288,6 +289,9 @@ impl ProviderFactory { /// 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 ProviderFactory { self.runtime.clone(), self.db.path(), ) + .with_reader_txn_tracker(self.db.clone()) .with_minimum_pruning_distance(self.minimum_pruning_distance)) } diff --git a/crates/storage/provider/src/providers/database/provider.rs b/crates/storage/provider/src/providers/database/provider.rs index 08df81b27c1..70eaa89d790 100644 --- a/crates/storage/provider/src/providers/database/provider.rs +++ b/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 { 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>, } impl Debug for DatabaseProvider { @@ -229,6 +231,7 @@ impl Debug for DatabaseProvider { .field("pending_rocksdb_batches", &"") .field("commit_order", &self.commit_order) .field("minimum_pruning_distance", &self.minimum_pruning_distance) + .field("reader_txn_tracker", &"") .finish() } } @@ -244,9 +247,48 @@ impl DatabaseProvider { self.minimum_pruning_distance = distance; self } + + /// Attaches reader tracking so unwind commits can wait on active readers. + pub(crate) fn with_reader_txn_tracker(mut self, reader_txn_tracker: T) -> Self + where + T: ReaderTxnTracker + 'static, + { + self.reader_txn_tracker = Some(Arc::new(reader_txn_tracker)); + self + } } impl DatabaseProvider { + /// 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 so readers holding older MDBX-visible + /// views cannot overlap the `RocksDB` unwind. + /// + /// Historical `storage_v2` reads ignore `RocksDB` history entries above their MDBX-visible tip, + /// so no additional post-`RocksDB` wait is needed before static-file commit. + 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 { + if 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)?; + } + } + + self.static_file_provider.commit()?; + Ok(()) + } + /// State provider for latest state pub fn latest<'a>(&'a self) -> Box { trace!(target: "providers::db", "Returning latest state provider"); @@ -382,6 +424,7 @@ impl DatabaseProvider { commit_order, minimum_pruning_distance: MINIMUM_UNWIND_SAFE_DISTANCE, metrics: metrics::DatabaseProviderMetrics::default(), + reader_txn_tracker: None, } } @@ -1007,6 +1050,7 @@ impl DatabaseProvider { commit_order: CommitOrder::Normal, minimum_pruning_distance: MINIMUM_UNWIND_SAFE_DISTANCE, metrics: metrics::DatabaseProviderMetrics::default(), + reader_txn_tracker: None, } } @@ -3800,19 +3844,8 @@ impl 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 +3913,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 +3938,32 @@ mod tests { assert_eq!(result, Vec::>::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 test_receipts_by_block_range_nonexistent_blocks() { let factory = create_test_provider_factory(); diff --git a/crates/storage/provider/src/providers/rocksdb/provider.rs b/crates/storage/provider/src/providers/rocksdb/provider.rs index a938a0c4424..4a1eec47b23 100644 --- a/crates/storage/provider/src/providers/rocksdb/provider.rs +++ b/crates/storage/provider/src/providers/rocksdb/provider.rs @@ -1434,17 +1434,22 @@ impl<'db> RocksReadSnapshot<'db> { } /// Lookup account history and return [`HistoryInfo`] directly. + /// + /// `visible_tip` is the highest block considered visible from the companion MDBX snapshot. + /// History entries above it are ignored even if they already exist in `RocksDB`. pub fn account_history_info( &self, address: Address, block_number: BlockNumber, lowest_available_block_number: Option, + visible_tip: BlockNumber, ) -> ProviderResult { let key = ShardedKey::new(address, block_number); self.history_info::( key.encode().as_ref(), block_number, lowest_available_block_number, + visible_tip, |key_bytes| Ok( as Decode>::decode(key_bytes)?.key == address), |prev_bytes| { as Decode>::decode(prev_bytes) @@ -1455,18 +1460,23 @@ impl<'db> RocksReadSnapshot<'db> { } /// Lookup storage history and return [`HistoryInfo`] directly. + /// + /// `visible_tip` is the highest block considered visible from the companion MDBX snapshot. + /// History entries above it are ignored even if they already exist in `RocksDB`. pub fn storage_history_info( &self, address: Address, storage_key: B256, block_number: BlockNumber, lowest_available_block_number: Option, + visible_tip: BlockNumber, ) -> ProviderResult { let key = StorageShardedKey::new(address, storage_key, block_number); self.history_info::( key.encode().as_ref(), block_number, lowest_available_block_number, + visible_tip, |key_bytes| { let k = ::decode(key_bytes)?; Ok(k.address == address && k.sharded_key.key == storage_key) @@ -1480,11 +1490,16 @@ impl<'db> RocksReadSnapshot<'db> { } /// Generic history lookup using the snapshot's raw iterator. + /// + /// The result is derived from the history that is visible through `visible_tip`, not from the + /// full contents of `RocksDB`. This lets a reader combine an older MDBX snapshot with a newer + /// Rocks snapshot without routing through history entries that MDBX cannot see yet. fn history_info( &self, encoded_key: &[u8], block_number: BlockNumber, lowest_available_block_number: Option, + visible_tip: BlockNumber, key_matches: impl FnOnce(&[u8]) -> Result, prev_key_matches: impl Fn(&[u8]) -> bool, ) -> ProviderResult @@ -1526,7 +1541,10 @@ impl<'db> RocksReadSnapshot<'db> { return fallback(); }; let chunk = BlockNumberList::decompress(value_bytes)?; + let (rank, found_block) = compute_history_rank(&chunk, block_number); + // Ignore later Rocks history that is ahead of the companion MDBX snapshot. + let found_block = found_block.filter(|block| *block <= visible_tip); let is_before_first_write = if needs_prev_shard_check(rank, found_block, block_number) { iter.prev(); @@ -1537,6 +1555,14 @@ impl<'db> RocksReadSnapshot<'db> { })) })?; let has_prev = iter.valid() && iter.key().is_some_and(&prev_key_matches); + + // If the current shard only contains history above `visible_tip`, there is no usable + // later change. Without a previous shard for the same key, fall back to the existing + // not-written / maybe-pruned result instead of routing into plain state. + if found_block.is_none() && !has_prev { + return fallback() + } + !has_prev } else { false @@ -2999,7 +3025,8 @@ mod tests { // This simulates a pruned state where data before block 100 is not available. // Since we're before the first write AND pruning boundary is set, we need to // check the changeset at the first write block. - let result = provider.snapshot().account_history_info(address, 50, Some(100)).unwrap(); + let result = + provider.snapshot().account_history_info(address, 50, Some(100), u64::MAX).unwrap(); assert_eq!(result, HistoryInfo::InChangeset(100)); } @@ -3026,16 +3053,84 @@ mod tests { .build() .unwrap(); - let result = ro_provider.snapshot().account_history_info(address, 200, None).unwrap(); + let result = + ro_provider.snapshot().account_history_info(address, 200, None, u64::MAX).unwrap(); assert_eq!(result, HistoryInfo::InChangeset(200)); - let result = ro_provider.snapshot().account_history_info(address, 50, None).unwrap(); + let result = + ro_provider.snapshot().account_history_info(address, 50, None, u64::MAX).unwrap(); assert_eq!(result, HistoryInfo::NotYetWritten); - let result = ro_provider.snapshot().account_history_info(address, 400, None).unwrap(); + let result = + ro_provider.snapshot().account_history_info(address, 400, None, u64::MAX).unwrap(); assert_eq!(result, HistoryInfo::InPlainState); } + #[test] + fn test_account_history_info_ignores_blocks_above_visible_tip() { + let temp_dir = TempDir::new().unwrap(); + let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap(); + + let address = Address::from([0x42; 20]); + + provider + .put::( + ShardedKey::new(address, 110), + &IntegerList::new([100, 110]).unwrap(), + ) + .unwrap(); + provider + .put::( + ShardedKey::new(address, u64::MAX), + &IntegerList::new([200, 210]).unwrap(), + ) + .unwrap(); + + let result = provider.snapshot().account_history_info(address, 150, None, 150).unwrap(); + assert_eq!(result, HistoryInfo::InPlainState); + } + + #[test] + fn test_account_history_info_mixed_shard_respects_visible_tip() { + let temp_dir = TempDir::new().unwrap(); + let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap(); + + let address = Address::from([0x42; 20]); + provider + .put::( + ShardedKey::new(address, u64::MAX), + &IntegerList::new([100, 150, 300]).unwrap(), + ) + .unwrap(); + + let result = provider.snapshot().account_history_info(address, 120, None, 200).unwrap(); + assert_eq!(result, HistoryInfo::InChangeset(150)); + + let result = provider.snapshot().account_history_info(address, 201, None, 200).unwrap(); + assert_eq!(result, HistoryInfo::InPlainState); + } + + #[test] + fn test_account_history_info_only_stale_entries_use_fallback() { + let temp_dir = TempDir::new().unwrap(); + let provider = RocksDBBuilder::new(temp_dir.path()).with_default_tables().build().unwrap(); + + let address = Address::from([0x42; 20]); + provider + .put::( + ShardedKey::new(address, u64::MAX), + &IntegerList::new([200, 210]).unwrap(), + ) + .unwrap(); + + let result = provider.snapshot().account_history_info(address, 150, None, 150).unwrap(); + assert_eq!(result, HistoryInfo::NotYetWritten); + + let result = + provider.snapshot().account_history_info(address, 150, Some(100), 150).unwrap(); + assert_eq!(result, HistoryInfo::MaybeInPlainState); + } + #[test] fn test_account_history_shard_split_at_boundary() { let temp_dir = TempDir::new().unwrap(); diff --git a/crates/storage/provider/src/providers/state/historical.rs b/crates/storage/provider/src/providers/state/historical.rs index ee2a85fae60..f0280e4a6a6 100644 --- a/crates/storage/provider/src/providers/state/historical.rs +++ b/crates/storage/provider/src/providers/state/historical.rs @@ -156,12 +156,15 @@ impl<'b, Provider: DBProvider + ChangeSetReader + StorageChangeSetReader + Block return Err(ProviderError::StateAtBlockPruned(self.block_number)) } + let visible_tip = self.provider.best_block_number()?; + self.provider.with_rocksdb_snapshot(|rocksdb_ref| { let mut reader = EitherReader::new_accounts_history(self.provider, rocksdb_ref)?; reader.account_history_info( address, self.block_number, self.lowest_available_blocks.account_history_block_number, + visible_tip, ) }) } @@ -181,6 +184,8 @@ impl<'b, Provider: DBProvider + ChangeSetReader + StorageChangeSetReader + Block return Err(ProviderError::StateAtBlockPruned(self.block_number)) } + let visible_tip = self.provider.best_block_number()?; + self.provider.with_rocksdb_snapshot(|rocksdb_ref| { let mut reader = EitherReader::new_storages_history(self.provider, rocksdb_ref)?; reader.storage_history_info( @@ -188,6 +193,7 @@ impl<'b, Provider: DBProvider + ChangeSetReader + StorageChangeSetReader + Block lookup_key, self.block_number, self.lowest_available_blocks.storage_history_block_number, + visible_tip, ) }) }