From ae6cc3133acac769723e018141c2d751ef26aa80 Mon Sep 17 00:00:00 2001 From: joshieDo <93316087+joshieDo@users.noreply.github.com> Date: Wed, 1 Apr 2026 17:01:20 +0100 Subject: [PATCH 01/12] fix(provider): add unwind reader barrier for storage_v2 Amp-Thread-ID: https://ampcode.com/threads/T-019d4937-7f0e-765b-bf61-489758cb324c Co-authored-by: Amp --- crates/storage/db-api/src/database.rs | 71 +++++++++++ .../storage/db/src/implementation/mdbx/mod.rs | 20 ++++ crates/storage/db/src/lib.rs | 8 ++ crates/storage/libmdbx-rs/src/environment.rs | 6 + .../provider/src/providers/database/mod.rs | 2 + .../src/providers/database/provider.rs | 112 ++++++++++++++++-- 6 files changed, 206 insertions(+), 13 deletions(-) diff --git a/crates/storage/db-api/src/database.rs b/crates/storage/db-api/src/database.rs index 71403505a15..745ca8d42a7 100644 --- a/crates/storage/db-api/src/database.rs +++ b/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,19 @@ 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. + /// + /// 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 { + None + } + + /// Returns the ID of the most recently committed transaction, if available. + fn last_txnid(&self) -> Option { + None + } + /// 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 +83,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 +108,53 @@ 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 and unwind fencing. +pub trait ReaderTxnTracker: Send + Sync { + /// Returns the txnid of the oldest active MDBX reader. + fn oldest_reader_txnid(&self) -> Option; + + /// Returns the latest committed MDBX txnid. + fn last_txnid(&self) -> Option; + + /// Forces a real commit so the latest txnid advances, then returns it. + fn commit_fence(&self) -> Result, DatabaseError>; + + /// Waits until all MDBX readers older than `cutoff_txnid` have drained, + /// polling every 10ms. + fn wait_for_readers_before_txnid(&self, cutoff_txnid: u64) { + while self.oldest_reader_txnid().is_some_and(|oldest| oldest < cutoff_txnid) { + std::thread::sleep(std::time::Duration::from_millis(10)); + } + } +} + +impl ReaderTxnTracker for DB { + fn oldest_reader_txnid(&self) -> Option { + Database::oldest_reader_txnid(self) + } + + fn last_txnid(&self) -> Option { + Database::last_txnid(self) + } + + fn commit_fence(&self) -> Result, DatabaseError> { + let last_txnid = self.last_txnid().unwrap_or_default(); + let tx = self.tx_mut()?; + tx.put::>( + RawKey::::from_vec(vec![0, 1]), + RawValue::from_vec(last_txnid.to_be_bytes().into()), + )?; + tx.commit()?; + Ok(self.last_txnid()) + } } 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/providers/database/mod.rs b/crates/storage/provider/src/providers/database/mod.rs index c2ca29c36ba..4d4dcba6c6c 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), )) } @@ -304,6 +305,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..24aa4a884c1 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,57 @@ impl DatabaseProvider { 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(mut self, db: DB) -> Self + where + DB: Database + 'static, + { + self.reader_txn_tracker = Some(Arc::new(db)); + 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 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() && + let Some(committed_txn_id) = reader_txn_tracker.last_txnid() + { + reader_txn_tracker.wait_for_readers_before_txnid(committed_txn_id); + } + + 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() && + let Some(fence_txnid) = reader_txn_tracker.commit_fence()? + { + reader_txn_tracker.wait_for_readers_before_txnid(fence_txnid); + } + + 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 +433,7 @@ impl DatabaseProvider { commit_order, minimum_pruning_distance: MINIMUM_UNWIND_SAFE_DISTANCE, metrics: metrics::DatabaseProviderMetrics::default(), + reader_txn_tracker: None, } } @@ -1007,6 +1059,7 @@ impl DatabaseProvider { commit_order: CommitOrder::Normal, minimum_pruning_distance: MINIMUM_UNWIND_SAFE_DISTANCE, metrics: metrics::DatabaseProviderMetrics::default(), + reader_txn_tracker: None, } } @@ -3800,19 +3853,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 +3922,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 +3947,47 @@ 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 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(); + let fence_txnid = ReaderTxnTracker::commit_fence(factory.db_ref()).unwrap().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(); From 92fa036dffc57f469b30cb280b7a1eb0669964fe Mon Sep 17 00:00:00 2001 From: joshieDo <93316087+joshieDo@users.noreply.github.com> Date: Wed, 1 Apr 2026 17:11:11 +0100 Subject: [PATCH 02/12] docs(db): clarify reader txnid semantics Amp-Thread-ID: https://ampcode.com/threads/T-019d4937-7f0e-765b-bf61-489758cb324c Co-authored-by: Amp --- crates/storage/db-api/src/database.rs | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/crates/storage/db-api/src/database.rs b/crates/storage/db-api/src/database.rs index 745ca8d42a7..ef74a9c299b 100644 --- a/crates/storage/db-api/src/database.rs +++ b/crates/storage/db-api/src/database.rs @@ -28,6 +28,9 @@ pub trait Database: Send + Sync + Debug { /// 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 { @@ -120,7 +123,9 @@ impl Database for &DB { /// Object-safe adapter for reader-txn tracking and unwind fencing. pub trait ReaderTxnTracker: Send + Sync { - /// Returns the txnid of the oldest active MDBX reader. + /// Returns the txnid of the oldest active reader snapshot. + /// + /// This is not a unique per-reader id; it is the committed txnid the reader is pinned to. fn oldest_reader_txnid(&self) -> Option; /// Returns the latest committed MDBX txnid. @@ -129,7 +134,7 @@ pub trait ReaderTxnTracker: Send + Sync { /// Forces a real commit so the latest txnid advances, then returns it. fn commit_fence(&self) -> Result, DatabaseError>; - /// Waits until all MDBX readers older than `cutoff_txnid` have drained, + /// Waits until all readers pinned to a committed txnid older than `cutoff_txnid` have drained, /// polling every 10ms. fn wait_for_readers_before_txnid(&self, cutoff_txnid: u64) { while self.oldest_reader_txnid().is_some_and(|oldest| oldest < cutoff_txnid) { From 12595052272b4abc9266ea52cfd36f464352c163 Mon Sep 17 00:00:00 2001 From: joshieDo <93316087+joshieDo@users.noreply.github.com> Date: Wed, 1 Apr 2026 17:28:00 +0100 Subject: [PATCH 03/12] refactor(provider): simplify unwind reader tracker Amp-Thread-ID: https://ampcode.com/threads/T-019d4937-7f0e-765b-bf61-489758cb324c Co-authored-by: Amp --- crates/storage/db-api/src/database.rs | 48 +++++++++---------- .../src/providers/database/provider.rs | 17 +++---- 2 files changed, 28 insertions(+), 37 deletions(-) diff --git a/crates/storage/db-api/src/database.rs b/crates/storage/db-api/src/database.rs index ef74a9c299b..427169aec8a 100644 --- a/crates/storage/db-api/src/database.rs +++ b/crates/storage/db-api/src/database.rs @@ -123,43 +123,39 @@ impl Database for &DB { /// Object-safe adapter for reader-txn tracking and unwind fencing. pub trait ReaderTxnTracker: Send + Sync { - /// Returns the txnid of the oldest active reader snapshot. - /// - /// This is not a unique per-reader id; it is the committed txnid the reader is pinned to. - fn oldest_reader_txnid(&self) -> Option; - - /// Returns the latest committed MDBX txnid. - fn last_txnid(&self) -> Option; + /// Waits until all readers older than the latest committed txnid have drained. + fn wait_for_pre_commit_readers(&self); - /// Forces a real commit so the latest txnid advances, then returns it. - fn commit_fence(&self) -> Result, DatabaseError>; - - /// Waits until all readers pinned to a committed txnid older than `cutoff_txnid` have drained, - /// polling every 10ms. - fn wait_for_readers_before_txnid(&self, cutoff_txnid: u64) { - while self.oldest_reader_txnid().is_some_and(|oldest| oldest < cutoff_txnid) { - std::thread::sleep(std::time::Duration::from_millis(10)); - } - } + /// 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 ReaderTxnTracker for DB { - fn oldest_reader_txnid(&self) -> Option { - Database::oldest_reader_txnid(self) - } - - fn last_txnid(&self) -> Option { - Database::last_txnid(self) + 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 commit_fence(&self) -> Result, DatabaseError> { - let last_txnid = self.last_txnid().unwrap_or_default(); + 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::>( RawKey::::from_vec(vec![0, 1]), RawValue::from_vec(last_txnid.to_be_bytes().into()), )?; tx.commit()?; - Ok(self.last_txnid()) + + if let Some(fence_txnid) = Database::last_txnid(self) { + while Database::oldest_reader_txnid(self).is_some_and(|oldest| oldest < fence_txnid) { + std::thread::sleep(std::time::Duration::from_millis(10)); + } + } + + Ok(()) } } diff --git a/crates/storage/provider/src/providers/database/provider.rs b/crates/storage/provider/src/providers/database/provider.rs index 24aa4a884c1..fa99d53557f 100644 --- a/crates/storage/provider/src/providers/database/provider.rs +++ b/crates/storage/provider/src/providers/database/provider.rs @@ -275,11 +275,8 @@ impl DatabaseProvider { 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() && - let Some(committed_txn_id) = reader_txn_tracker.last_txnid() - { - reader_txn_tracker.wait_for_readers_before_txnid(committed_txn_id); + 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()); @@ -287,11 +284,8 @@ impl DatabaseProvider { self.rocksdb_provider.commit_batch(batch)?; } - if storage_v2 && - let Some(reader_txn_tracker) = reader_txn_tracker.as_ref() && - let Some(fence_txnid) = reader_txn_tracker.commit_fence()? - { - reader_txn_tracker.wait_for_readers_before_txnid(fence_txnid); + 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()?; @@ -3983,7 +3977,8 @@ mod tests { provider_rw.commit().unwrap(); let before_fence = Database::last_txnid(factory.db_ref()).unwrap(); - let fence_txnid = ReaderTxnTracker::commit_fence(factory.db_ref()).unwrap().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"); } From dcaf736fef4fa5d6f3a9722a9911bf569ad73fb9 Mon Sep 17 00:00:00 2001 From: joshieDo <93316087+joshieDo@users.noreply.github.com> Date: Wed, 1 Apr 2026 17:29:22 +0100 Subject: [PATCH 04/12] refactor(db): reuse pre-commit reader wait Amp-Thread-ID: https://ampcode.com/threads/T-019d4937-7f0e-765b-bf61-489758cb324c Co-authored-by: Amp --- crates/storage/db-api/src/database.rs | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/crates/storage/db-api/src/database.rs b/crates/storage/db-api/src/database.rs index 427169aec8a..b0c05635bd6 100644 --- a/crates/storage/db-api/src/database.rs +++ b/crates/storage/db-api/src/database.rs @@ -150,11 +150,7 @@ impl ReaderTxnTracker for DB { )?; tx.commit()?; - if let Some(fence_txnid) = Database::last_txnid(self) { - while Database::oldest_reader_txnid(self).is_some_and(|oldest| oldest < fence_txnid) { - std::thread::sleep(std::time::Duration::from_millis(10)); - } - } + self.wait_for_pre_commit_readers(); Ok(()) } From f93fd06913737df6f45a5a75a30e9ef339dad7e2 Mon Sep 17 00:00:00 2001 From: joshieDo <93316087+joshieDo@users.noreply.github.com> Date: Wed, 1 Apr 2026 17:37:44 +0100 Subject: [PATCH 05/12] refactor(db-api): require reader txnid methods Amp-Thread-ID: https://ampcode.com/threads/T-019d4937-7f0e-765b-bf61-489758cb324c Co-authored-by: Amp --- crates/storage/db-api/src/database.rs | 8 ++------ crates/storage/db-api/src/mock.rs | 8 ++++++++ 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/crates/storage/db-api/src/database.rs b/crates/storage/db-api/src/database.rs index b0c05635bd6..05f29a7de3b 100644 --- a/crates/storage/db-api/src/database.rs +++ b/crates/storage/db-api/src/database.rs @@ -33,14 +33,10 @@ pub trait Database: Send + Sync + Debug { /// /// 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 { - None - } + fn oldest_reader_txnid(&self) -> Option; /// Returns the ID of the most recently committed transaction, if available. - fn last_txnid(&self) -> Option { - None - } + 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. 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 {} From 9baccced66fa7e6be741ab19d6cd0486b69b9410 Mon Sep 17 00:00:00 2001 From: joshieDo <93316087+joshieDo@users.noreply.github.com> Date: Wed, 1 Apr 2026 17:40:49 +0100 Subject: [PATCH 06/12] refactor(provider): narrow reader tracker input Amp-Thread-ID: https://ampcode.com/threads/T-019d4937-7f0e-765b-bf61-489758cb324c Co-authored-by: Amp --- .../storage/provider/src/providers/database/provider.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/crates/storage/provider/src/providers/database/provider.rs b/crates/storage/provider/src/providers/database/provider.rs index fa99d53557f..cfaa93508ef 100644 --- a/crates/storage/provider/src/providers/database/provider.rs +++ b/crates/storage/provider/src/providers/database/provider.rs @@ -248,12 +248,12 @@ impl DatabaseProvider { self } - /// Attaches MDBX reader tracking so unwind commits can wait on active readers. - pub(crate) fn with_reader_txn_tracker(mut self, db: DB) -> 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 - DB: Database + 'static, + T: ReaderTxnTracker + 'static, { - self.reader_txn_tracker = Some(Arc::new(db)); + self.reader_txn_tracker = Some(Arc::new(reader_txn_tracker)); self } } From 27430270593ac2d69ded80be600e22458a8c1b6f Mon Sep 17 00:00:00 2001 From: joshieDo <93316087+joshieDo@users.noreply.github.com> Date: Wed, 1 Apr 2026 17:43:42 +0100 Subject: [PATCH 07/12] refactor(provider): keep unwind storage_v2 local Amp-Thread-ID: https://ampcode.com/threads/T-019d4937-7f0e-765b-bf61-489758cb324c Co-authored-by: Amp --- .../src/providers/database/provider.rs | 20 ++++++++++--------- 1 file changed, 11 insertions(+), 9 deletions(-) diff --git a/crates/storage/provider/src/providers/database/provider.rs b/crates/storage/provider/src/providers/database/provider.rs index cfaa93508ef..830fbc77dea 100644 --- a/crates/storage/provider/src/providers/database/provider.rs +++ b/crates/storage/provider/src/providers/database/provider.rs @@ -275,17 +275,19 @@ impl DatabaseProvider { 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(); - } + 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)?; - } + 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()?; + if let Some(reader_txn_tracker) = reader_txn_tracker.as_ref() { + reader_txn_tracker.wait_for_pre_fence_readers()?; + } } self.static_file_provider.commit()?; From 4d62686119b626af1182de02ee951cdddacfbf96 Mon Sep 17 00:00:00 2001 From: joshieDo <93316087+joshieDo@users.noreply.github.com> Date: Wed, 1 Apr 2026 19:55:09 +0100 Subject: [PATCH 08/12] fix(stages): drop ro provider before unwind commit Amp-Thread-ID: https://ampcode.com/threads/T-019d4937-7f0e-765b-bf61-489758cb324c Co-authored-by: Amp --- crates/stages/api/src/pipeline/mod.rs | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) 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 From f2f2bda72a9bb7289ca1ddffbec7cd73c8d3b495 Mon Sep 17 00:00:00 2001 From: joshieDo <93316087+joshieDo@users.noreply.github.com> Date: Wed, 1 Apr 2026 19:59:20 +0100 Subject: [PATCH 09/12] docs(provider): note unwind reader drain Amp-Thread-ID: https://ampcode.com/threads/T-019d4937-7f0e-765b-bf61-489758cb324c Co-authored-by: Amp --- crates/storage/provider/src/providers/database/mod.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/crates/storage/provider/src/providers/database/mod.rs b/crates/storage/provider/src/providers/database/mod.rs index 4d4dcba6c6c..53a5a881b77 100644 --- a/crates/storage/provider/src/providers/database/mod.rs +++ b/crates/storage/provider/src/providers/database/mod.rs @@ -289,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, From b01485af058738c60cec953b8421ad3be1899a9a Mon Sep 17 00:00:00 2001 From: joshieDo <93316087+joshieDo@users.noreply.github.com> Date: Wed, 1 Apr 2026 20:50:51 +0100 Subject: [PATCH 10/12] refactor(provider): cap historical rocksdb lookups by mdbx tip Amp-Thread-ID: https://ampcode.com/threads/T-019d4937-7f0e-765b-bf61-489758cb324c Co-authored-by: Amp --- crates/storage/db-api/src/database.rs | 21 +-- crates/storage/provider/src/either_writer.rs | 18 +- .../src/providers/database/provider.rs | 29 +-- .../src/providers/rocksdb/provider.rs | 172 +++++++++++++++--- .../src/providers/state/historical.rs | 43 +++-- 5 files changed, 193 insertions(+), 90 deletions(-) diff --git a/crates/storage/db-api/src/database.rs b/crates/storage/db-api/src/database.rs index 05f29a7de3b..e9d13d5f101 100644 --- a/crates/storage/db-api/src/database.rs +++ b/crates/storage/db-api/src/database.rs @@ -1,6 +1,5 @@ use crate::{ table::TableImporter, - tables::{self, RawKey, RawTable, RawValue}, transaction::{DbTx, DbTxMut}, DatabaseError, }; @@ -117,14 +116,10 @@ impl Database for &DB { } } -/// Object-safe adapter for reader-txn tracking and unwind fencing. +/// 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); - - /// 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 ReaderTxnTracker for DB { @@ -136,18 +131,4 @@ impl ReaderTxnTracker for DB { } } } - - 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::>( - RawKey::::from_vec(vec![0, 1]), - RawValue::from_vec(last_txnid.to_be_bytes().into()), - )?; - tx.commit()?; - - self.wait_for_pre_commit_readers(); - - Ok(()) - } } 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/provider.rs b/crates/storage/provider/src/providers/database/provider.rs index 830fbc77dea..70eaa89d790 100644 --- a/crates/storage/provider/src/providers/database/provider.rs +++ b/crates/storage/provider/src/providers/database/provider.rs @@ -264,12 +264,11 @@ impl DatabaseProvider { /// 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. + /// For `storage_v2`, this waits after the MDBX commit so readers holding older MDBX-visible + /// views cannot overlap the `RocksDB` unwind. /// - /// 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. + /// 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(); @@ -284,10 +283,6 @@ impl DatabaseProvider { for batch in batches { self.rocksdb_provider.commit_batch(batch)?; } - - if let Some(reader_txn_tracker) = reader_txn_tracker.as_ref() { - reader_txn_tracker.wait_for_pre_fence_readers()?; - } } self.static_file_provider.commit()?; @@ -3969,22 +3964,6 @@ mod tests { 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(); diff --git a/crates/storage/provider/src/providers/rocksdb/provider.rs b/crates/storage/provider/src/providers/rocksdb/provider.rs index a938a0c4424..fd323b15e05 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,24 @@ 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. + /// + /// In particular: + /// - a matching history entry at or after `block_number` and at or below `visible_tip` returns + /// `HistoryInfo::InChangeset` + /// - no later matching history entry at or below `visible_tip`, but some matching history entry + /// for the key at or below `visible_tip`, returns `HistoryInfo::InPlainState` + /// - if every matching history entry is above `visible_tip`, the lookup falls back to + /// `NotYetWritten` or `MaybeInPlainState` 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,27 +1549,65 @@ impl<'db> RocksReadSnapshot<'db> { return fallback(); }; let chunk = BlockNumberList::decompress(value_bytes)?; + + // Only the prefix up to `visible_tip` should influence routing for this reader. Any + // later Rocks history entries are not yet visible in the companion MDBX snapshot. + let visible_count = chunk.rank(visible_tip); let (rank, found_block) = compute_history_rank(&chunk, block_number); - let is_before_first_write = if needs_prev_shard_check(rank, found_block, block_number) { - iter.prev(); - iter.status().map_err(|e| { - ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo { - message: e.to_string().into(), - code: -1, - })) - })?; - let has_prev = iter.valid() && iter.key().is_some_and(&prev_key_matches); - !has_prev - } else { - false + let mut has_previous_visible = || -> ProviderResult { + loop { + iter.prev(); + iter.status().map_err(|e| { + ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo { + message: e.to_string().into(), + code: -1, + })) + })?; + + if !iter.valid() { + return Ok(false) + } + + let Some(prev_key_bytes) = iter.key() else { return Ok(false) }; + if !prev_key_matches(prev_key_bytes) { + return Ok(false) + } + + let Some(prev_value_bytes) = iter.value() else { return Ok(false) }; + // The current shard can be entirely stale (all blocks above `visible_tip`). Walk + // backward across same-key shards until we either find a matching entry at or + // below `visible_tip` or prove there are none for this key. + if BlockNumberList::decompress(prev_value_bytes)?.rank(visible_tip) > 0 { + return Ok(true) + } + } }; - Ok(HistoryInfo::from_lookup( - found_block, - is_before_first_write, - lowest_available_block_number, - )) + // If there is a later matching entry that is still at or below `visible_tip`, we can + // reuse the normal `HistoryInfo::from_lookup` logic against that capped prefix. + if rank < visible_count { + let Some(found_block) = found_block else { return fallback() }; + + let is_before_first_write = + needs_prev_shard_check(rank, Some(found_block), block_number) && + !has_previous_visible()?; + + return Ok(HistoryInfo::from_lookup( + Some(found_block), + is_before_first_write, + lowest_available_block_number, + )) + } + + // No later matching entry remains at or below `visible_tip`. If the key has any matching + // entry at or below `visible_tip`, the lookup should fall through to plain state; + // otherwise keep the existing not-written / maybe-pruned fallback. + if visible_count > 0 || has_previous_visible()? { + return Ok(HistoryInfo::InPlainState) + } + + fallback() } } @@ -2999,7 +3060,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 +3088,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, 200, 300]).unwrap(), + ) + .unwrap(); + + let result = provider.snapshot().account_history_info(address, 150, None, 200).unwrap(); + assert_eq!(result, HistoryInfo::InChangeset(200)); + + let result = provider.snapshot().account_history_info(address, 250, 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..e1c333bd083 100644 --- a/crates/storage/provider/src/providers/state/historical.rs +++ b/crates/storage/provider/src/providers/state/historical.rs @@ -54,20 +54,20 @@ type DbProof<'a, TX, A> = Proof< /// Result of a history lookup for an account or storage slot. /// -/// Indicates where to find the historical value for a given key at a specific block. +/// Indicates where to find the historical value for a given key in the current visible history +/// view. #[derive(Debug, Eq, PartialEq)] pub enum HistoryInfo { - /// The key is written to, but only after our block (not yet written at the target block). Or - /// it has never been written. + /// No visible history entry proves that the key was already written at the target block. NotYetWritten, - /// The chunk contains an entry for a write after our block at the given block number. - /// The value should be looked up in the changeset at this block. + /// The first visible history entry at or after the target block is at this block number. + /// The value should be read from the changeset before that block. InChangeset(u64), - /// The chunk does not contain an entry for a write after our block. This can only - /// happen if this is the last chunk, so we need to look in the plain state. + /// No visible history entry at or after the target block exists, but the key does have some + /// visible history, so plain state matches the target state. InPlainState, - /// The key may have been written, but due to pruning we may not have changesets and - /// history, so we need to make a plain state lookup. + /// History may be incomplete because of pruning, so plain state is the best available + /// fallback. MaybeInPlainState, } @@ -77,10 +77,9 @@ impl HistoryInfo { /// This is a pure function shared by both MDBX and `RocksDB` backends. /// /// # Arguments - /// * `found_block` - The block number from the shard lookup - /// * `is_before_first_write` - True if the target block is before the first write to this key. - /// This should be computed as: `rank == 0 && found_block != Some(block_number) && - /// !has_previous_shard` where `has_previous_shard` comes from a lazy `cursor.prev()` check. + /// * `found_block` - The first visible history entry at or after the target block, if any + /// * `is_before_first_write` - True if the target block is before the first visible history + /// entry for this key in the current lookup view /// * `lowest_available` - Lowest block where history is available (pruning boundary) pub const fn from_lookup( found_block: Option, @@ -89,20 +88,20 @@ impl HistoryInfo { ) -> Self { if is_before_first_write { if let (Some(_), Some(block_number)) = (lowest_available, found_block) { - // The key may have been written, but due to pruning we may not have changesets - // and history, so we need to make a changeset lookup. + // History before the first visible entry may have been pruned, so use the first + // visible changeset as the best available boundary. return Self::InChangeset(block_number) } - // The key is written to, but only after our block. + // The first visible entry is after the target block, so the key is not yet written in + // the current lookup view. return Self::NotYetWritten } if let Some(block_number) = found_block { - // The chunk contains an entry for a write after our block, return it. + // A later visible entry exists, so read the value before that block from changesets. Self::InChangeset(block_number) } else { - // The chunk does not contain an entry for a write after our block. This can only - // happen if this is the last chunk and so we need to look in the plain state. + // No later visible entry exists, so plain state already matches the target state. Self::InPlainState } } @@ -156,12 +155,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 +183,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 +192,7 @@ impl<'b, Provider: DBProvider + ChangeSetReader + StorageChangeSetReader + Block lookup_key, self.block_number, self.lowest_available_blocks.storage_history_block_number, + visible_tip, ) }) } From b5b3657d8e239f782320638a2a7c650144674cf5 Mon Sep 17 00:00:00 2001 From: joshieDo <93316087+joshieDo@users.noreply.github.com> Date: Thu, 2 Apr 2026 11:40:17 +0100 Subject: [PATCH 11/12] refactor(provider): simplify rocksdb history cap logic Amp-Thread-ID: https://ampcode.com/threads/T-019d4937-7f0e-765b-bf61-489758cb324c Co-authored-by: Amp --- .../src/providers/rocksdb/provider.rs | 93 ++++++------------- 1 file changed, 29 insertions(+), 64 deletions(-) diff --git a/crates/storage/provider/src/providers/rocksdb/provider.rs b/crates/storage/provider/src/providers/rocksdb/provider.rs index fd323b15e05..4a1eec47b23 100644 --- a/crates/storage/provider/src/providers/rocksdb/provider.rs +++ b/crates/storage/provider/src/providers/rocksdb/provider.rs @@ -1494,14 +1494,6 @@ impl<'db> RocksReadSnapshot<'db> { /// 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. - /// - /// In particular: - /// - a matching history entry at or after `block_number` and at or below `visible_tip` returns - /// `HistoryInfo::InChangeset` - /// - no later matching history entry at or below `visible_tip`, but some matching history entry - /// for the key at or below `visible_tip`, returns `HistoryInfo::InPlainState` - /// - if every matching history entry is above `visible_tip`, the lookup falls back to - /// `NotYetWritten` or `MaybeInPlainState` fn history_info( &self, encoded_key: &[u8], @@ -1550,64 +1542,37 @@ impl<'db> RocksReadSnapshot<'db> { }; let chunk = BlockNumberList::decompress(value_bytes)?; - // Only the prefix up to `visible_tip` should influence routing for this reader. Any - // later Rocks history entries are not yet visible in the companion MDBX snapshot. - let visible_count = chunk.rank(visible_tip); 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 mut has_previous_visible = || -> ProviderResult { - loop { - iter.prev(); - iter.status().map_err(|e| { - ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo { - message: e.to_string().into(), - code: -1, - })) - })?; - - if !iter.valid() { - return Ok(false) - } - - let Some(prev_key_bytes) = iter.key() else { return Ok(false) }; - if !prev_key_matches(prev_key_bytes) { - return Ok(false) - } + let is_before_first_write = if needs_prev_shard_check(rank, found_block, block_number) { + iter.prev(); + iter.status().map_err(|e| { + ProviderError::Database(DatabaseError::Read(DatabaseErrorInfo { + message: e.to_string().into(), + code: -1, + })) + })?; + let has_prev = iter.valid() && iter.key().is_some_and(&prev_key_matches); - let Some(prev_value_bytes) = iter.value() else { return Ok(false) }; - // The current shard can be entirely stale (all blocks above `visible_tip`). Walk - // backward across same-key shards until we either find a matching entry at or - // below `visible_tip` or prove there are none for this key. - if BlockNumberList::decompress(prev_value_bytes)?.rank(visible_tip) > 0 { - return Ok(true) - } + // 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() } - }; - // If there is a later matching entry that is still at or below `visible_tip`, we can - // reuse the normal `HistoryInfo::from_lookup` logic against that capped prefix. - if rank < visible_count { - let Some(found_block) = found_block else { return fallback() }; - - let is_before_first_write = - needs_prev_shard_check(rank, Some(found_block), block_number) && - !has_previous_visible()?; - - return Ok(HistoryInfo::from_lookup( - Some(found_block), - is_before_first_write, - lowest_available_block_number, - )) - } - - // No later matching entry remains at or below `visible_tip`. If the key has any matching - // entry at or below `visible_tip`, the lookup should fall through to plain state; - // otherwise keep the existing not-written / maybe-pruned fallback. - if visible_count > 0 || has_previous_visible()? { - return Ok(HistoryInfo::InPlainState) - } + !has_prev + } else { + false + }; - fallback() + Ok(HistoryInfo::from_lookup( + found_block, + is_before_first_write, + lowest_available_block_number, + )) } } @@ -3134,14 +3099,14 @@ mod tests { provider .put::( ShardedKey::new(address, u64::MAX), - &IntegerList::new([100, 200, 300]).unwrap(), + &IntegerList::new([100, 150, 300]).unwrap(), ) .unwrap(); - let result = provider.snapshot().account_history_info(address, 150, None, 200).unwrap(); - assert_eq!(result, HistoryInfo::InChangeset(200)); + 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, 250, None, 200).unwrap(); + let result = provider.snapshot().account_history_info(address, 201, None, 200).unwrap(); assert_eq!(result, HistoryInfo::InPlainState); } From 7d8fbd63d73aee09e771f3067bc62aab09c97786 Mon Sep 17 00:00:00 2001 From: joshieDo <93316087+joshieDo@users.noreply.github.com> Date: Thu, 2 Apr 2026 11:48:11 +0100 Subject: [PATCH 12/12] docs(provider): restore historical lookup wording Amp-Thread-ID: https://ampcode.com/threads/T-019d4435-3bfe-7739-b47f-fbcf6473a16a Co-authored-by: Amp --- .../src/providers/state/historical.rs | 37 ++++++++++--------- 1 file changed, 19 insertions(+), 18 deletions(-) diff --git a/crates/storage/provider/src/providers/state/historical.rs b/crates/storage/provider/src/providers/state/historical.rs index e1c333bd083..f0280e4a6a6 100644 --- a/crates/storage/provider/src/providers/state/historical.rs +++ b/crates/storage/provider/src/providers/state/historical.rs @@ -54,20 +54,20 @@ type DbProof<'a, TX, A> = Proof< /// Result of a history lookup for an account or storage slot. /// -/// Indicates where to find the historical value for a given key in the current visible history -/// view. +/// Indicates where to find the historical value for a given key at a specific block. #[derive(Debug, Eq, PartialEq)] pub enum HistoryInfo { - /// No visible history entry proves that the key was already written at the target block. + /// The key is written to, but only after our block (not yet written at the target block). Or + /// it has never been written. NotYetWritten, - /// The first visible history entry at or after the target block is at this block number. - /// The value should be read from the changeset before that block. + /// The chunk contains an entry for a write after our block at the given block number. + /// The value should be looked up in the changeset at this block. InChangeset(u64), - /// No visible history entry at or after the target block exists, but the key does have some - /// visible history, so plain state matches the target state. + /// The chunk does not contain an entry for a write after our block. This can only + /// happen if this is the last chunk, so we need to look in the plain state. InPlainState, - /// History may be incomplete because of pruning, so plain state is the best available - /// fallback. + /// The key may have been written, but due to pruning we may not have changesets and + /// history, so we need to make a plain state lookup. MaybeInPlainState, } @@ -77,9 +77,10 @@ impl HistoryInfo { /// This is a pure function shared by both MDBX and `RocksDB` backends. /// /// # Arguments - /// * `found_block` - The first visible history entry at or after the target block, if any - /// * `is_before_first_write` - True if the target block is before the first visible history - /// entry for this key in the current lookup view + /// * `found_block` - The block number from the shard lookup + /// * `is_before_first_write` - True if the target block is before the first write to this key. + /// This should be computed as: `rank == 0 && found_block != Some(block_number) && + /// !has_previous_shard` where `has_previous_shard` comes from a lazy `cursor.prev()` check. /// * `lowest_available` - Lowest block where history is available (pruning boundary) pub const fn from_lookup( found_block: Option, @@ -88,20 +89,20 @@ impl HistoryInfo { ) -> Self { if is_before_first_write { if let (Some(_), Some(block_number)) = (lowest_available, found_block) { - // History before the first visible entry may have been pruned, so use the first - // visible changeset as the best available boundary. + // The key may have been written, but due to pruning we may not have changesets + // and history, so we need to make a changeset lookup. return Self::InChangeset(block_number) } - // The first visible entry is after the target block, so the key is not yet written in - // the current lookup view. + // The key is written to, but only after our block. return Self::NotYetWritten } if let Some(block_number) = found_block { - // A later visible entry exists, so read the value before that block from changesets. + // The chunk contains an entry for a write after our block, return it. Self::InChangeset(block_number) } else { - // No later visible entry exists, so plain state already matches the target state. + // The chunk does not contain an entry for a write after our block. This can only + // happen if this is the last chunk and so we need to look in the plain state. Self::InPlainState } }