From 36e883d57262fc6a7045c0d2ef9cb0ff7fdb1d77 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Mon, 5 Oct 2026 11:39:04 -0300 Subject: [PATCH 1/4] perf(storage): make has_state an existence check that skips the snapshot read On a Hoodi beacon follower the `guards` import phase was 0-1 ms on most slots but median 180 ms (p90 332 ms, max 536 ms) on the block after each epoch's first block. `has_state(parent_root)` read both `States` and `StateDiffs` through `get`, and the epoch-crossing block's state is stored as a full snapshot only (no diff), so RocksDB copied the whole ~150 MB state into a Vec just to test that it exists. On networks with smaller states the same call still costs ~30 ms per block because `States` has no bloom filter, so even a miss is expensive. Three changes, same truth value as before: - Consult the state cache (non-promoting `peek`) after `pending_states`. A cached BlockState is only ever inserted by `insert_state` (also in `pending_states` until committed) or by `read_state` after a backend read, and no path deletes persisted states, so a cache hit implies the state exists. - Check `StateDiffs` before `States`: every non-anchor root has a diff. - Add `StorageReadView::contains`, implemented with `get_pinned_cf` on RocksDB and `contains_key` in memory, so no value is materialized. Tests use a counting backend to show the cached path touches neither table, the diff path skips `States`, and no value read happens. --- crates/storage/src/api/traits.rs | 8 + crates/storage/src/backend/in_memory.rs | 8 + crates/storage/src/backend/rocksdb.rs | 11 ++ crates/storage/src/backend/tests.rs | 24 +++ crates/storage/src/store.rs | 187 +++++++++++++++++++++++- 5 files changed, 233 insertions(+), 5 deletions(-) diff --git a/crates/storage/src/api/traits.rs b/crates/storage/src/api/traits.rs index 5b654bfcd..721c7c4d6 100644 --- a/crates/storage/src/api/traits.rs +++ b/crates/storage/src/api/traits.rs @@ -27,6 +27,14 @@ pub trait StorageReadView { /// Get a value by key from a table. fn get(&self, table: Table, key: &[u8]) -> Result>, Error>; + /// Whether `key` is present in a table. + /// + /// Same answer as `get(..)?.is_some()` but never materializes the value, so + /// the cost does not scale with its size. Prefer it for pure existence + /// checks on large values (full state snapshots are 100+ MB on mainnet-sized + /// beacon chains). + fn contains(&self, table: Table, key: &[u8]) -> Result; + /// Iterate over all entries with a given key prefix. fn prefix_iterator( &self, diff --git a/crates/storage/src/backend/in_memory.rs b/crates/storage/src/backend/in_memory.rs index ec3fcd8f5..fd7d766ed 100644 --- a/crates/storage/src/backend/in_memory.rs +++ b/crates/storage/src/backend/in_memory.rs @@ -75,6 +75,14 @@ impl StorageReadView for InMemoryReadView<'_> { .cloned()) } + fn contains(&self, table: Table, key: &[u8]) -> Result { + Ok(self + .guard + .get(&table) + .expect("table exists") + .contains_key(key)) + } + fn prefix_iterator( &self, table: Table, diff --git a/crates/storage/src/backend/rocksdb.rs b/crates/storage/src/backend/rocksdb.rs index 3440f94ce..ac4aed229 100644 --- a/crates/storage/src/backend/rocksdb.rs +++ b/crates/storage/src/backend/rocksdb.rs @@ -112,6 +112,17 @@ impl StorageReadView for RocksDBReadView { Ok(self.db.get_cf(&cf, key)?) } + fn contains(&self, table: Table, key: &[u8]) -> Result { + let cf = self + .db + .cf_handle(cf_name(table)) + .ok_or_else(|| format!("Column family {} not found", cf_name(table)))?; + + // Pinned: references the block-cache buffer instead of copying the + // value into a `Vec`. + Ok(self.db.get_pinned_cf(&cf, key)?.is_some()) + } + fn prefix_iterator( &self, table: Table, diff --git a/crates/storage/src/backend/tests.rs b/crates/storage/src/backend/tests.rs index 46c908777..ffd3ea308 100644 --- a/crates/storage/src/backend/tests.rs +++ b/crates/storage/src/backend/tests.rs @@ -17,6 +17,7 @@ pub fn run_backend_tests(backend: &dyn StorageBackend) { test_delete(backend); test_prefix_iterator(backend); test_nonexistent_key(backend); + test_contains(backend); test_delete_then_put(backend); test_put_then_delete(backend); test_delete_range(backend); @@ -45,6 +46,29 @@ fn test_put_and_get(backend: &dyn StorageBackend) { } } +fn test_contains(backend: &dyn StorageBackend) { + { + let mut batch = backend.begin_write().unwrap(); + batch + .put_batch( + Table::BlockHeaders, + vec![(b"test_contains_key".to_vec(), b"value1".to_vec())], + ) + .unwrap(); + batch.commit().unwrap(); + } + let view = backend.begin_read().unwrap(); + assert!( + view.contains(Table::BlockHeaders, b"test_contains_key") + .unwrap() + ); + assert!( + !view + .contains(Table::BlockHeaders, b"test_contains_missing") + .unwrap() + ); +} + fn test_delete(backend: &dyn StorageBackend) { // Write data { diff --git a/crates/storage/src/store.rs b/crates/storage/src/store.rs index 96addfbdb..efa50193e 100644 --- a/crates/storage/src/store.rs +++ b/crates/storage/src/store.rs @@ -2660,19 +2660,36 @@ impl Store { /// Returns whether a state is available for the given block root. /// - /// True if `pending_states` holds the state, a snapshot exists, or the - /// state can be reconstructed from a diff. + /// True if `pending_states` or the state cache holds the state, a snapshot + /// exists, or the state can be reconstructed from a diff. Never reads a + /// value: existence checks only. pub fn has_state(&self, root: &H256) -> Result { // Same pending-before-backend order as `read_state`; see its doc for // why the backend never has to consult `pending_states` on its own. if self.pending_states.get(root).is_some() { return Ok(true); } + // A cached block state is always one that was handed to the writer + // (`insert_state`, so in `pending_states` until committed) or read back + // from the backend (`read_state`); nothing deletes a persisted state. + // `peek`, not `get`: an existence check must not reorder the LRU. + // Without this, a state stored as a snapshot only (a beacon epoch + // anchor) costs a full-value read from RocksDB just to be found. + if self + .state_cache + .lock() + .unwrap() + .peek(&CacheKey::BlockState(*root)) + .is_some() + { + return Ok(true); + } let view = self.backend.begin_read().expect("read view"); let key = root.to_ssz(); - let states = view.get(Table::States, &key).expect("get"); - let diffs = view.get(Table::StateDiffs, &key).expect("get"); - Ok(states.is_some() || diffs.is_some()) + // Diffs first: every non-anchor state has one, so most roots never + // touch `States`, where even a miss is costly (no bloom filter). + Ok(view.contains(Table::StateDiffs, &key).expect("contains") + || view.contains(Table::States, &key).expect("contains")) } /// Persist a post-block state. @@ -5242,6 +5259,166 @@ mod tests { assert_eq!(reconstructed.to_ssz(), s1.to_ssz()); } + type BackendError = Box; + + /// Backend wrapper counting value reads and existence checks per table, to + /// show which tables `has_state` touches and that it never copies a value. + #[derive(Default)] + struct ReadCounts { + gets: Mutex>, + contains: Mutex>, + } + + impl ReadCounts { + fn gets(&self, table: Table) -> usize { + self.gets.lock().unwrap().get(&table).copied().unwrap_or(0) + } + fn contains(&self, table: Table) -> usize { + self.contains + .lock() + .unwrap() + .get(&table) + .copied() + .unwrap_or(0) + } + } + + struct CountingBackend { + inner: InMemoryBackend, + counts: Arc, + } + + struct CountingView<'a> { + inner: Box, + counts: Arc, + } + + impl StorageBackend for CountingBackend { + fn begin_read(&self) -> Result, BackendError> { + Ok(Box::new(CountingView { + inner: self.inner.begin_read()?, + counts: self.counts.clone(), + })) + } + + fn begin_write(&self) -> Result, BackendError> { + self.inner.begin_write() + } + } + + impl StorageReadView for CountingView<'_> { + fn get(&self, table: Table, key: &[u8]) -> Result>, BackendError> { + *self.counts.gets.lock().unwrap().entry(table).or_default() += 1; + self.inner.get(table, key) + } + + fn contains(&self, table: Table, key: &[u8]) -> Result { + *self + .counts + .contains + .lock() + .unwrap() + .entry(table) + .or_default() += 1; + self.inner.contains(table, key) + } + + fn prefix_iterator( + &self, + table: Table, + prefix: &[u8], + ) -> Result + '_>, BackendError> { + self.inner.prefix_iterator(table, prefix) + } + } + + fn counting_store() -> (Store, Arc) { + let counts = Arc::new(ReadCounts::default()); + let backend = Arc::new(CountingBackend { + inner: InMemoryBackend::new(), + counts: counts.clone(), + }); + let store = Store::from_parts( + backend, + Arc::new(Config::lean(0, DEFAULT_MILLISECONDS_PER_SLOT)), + Chain::Lean, + 0, + ); + (store, counts) + } + + fn put_raw(store: &Store, table: Table, root: H256) { + let mut batch = store.backend.begin_write().expect("write batch"); + batch + .put_batch(table, vec![(root.to_ssz(), vec![1, 2, 3])]) + .expect("put"); + batch.commit().expect("commit"); + } + + /// A cached state answers without any backend read, so a snapshot-only + /// anchor state is not copied out of the database to test existence. + #[test] + fn has_state_is_answered_by_the_cache_without_touching_the_backend() { + let (store, counts) = counting_store(); + let root = H256::from([7u8; 32]); + let state = BeaconState::Lean(sample_state(1, H256::ZERO, vec![])); + store.cache_state(CacheKey::BlockState(root), Arc::new(state)); + + assert!(store.has_state(&root).expect("has_state")); + for table in [Table::States, Table::StateDiffs] { + assert_eq!(counts.gets(table), 0); + assert_eq!(counts.contains(table), 0); + } + } + + /// The existence check must not promote the entry in the LRU. + #[test] + fn has_state_does_not_reorder_the_state_cache() { + let (store, _counts) = counting_store(); + let first = H256::from([1u8; 32]); + let second = H256::from([2u8; 32]); + for root in [first, second] { + let state = BeaconState::Lean(sample_state(1, H256::ZERO, vec![])); + store.cache_state(CacheKey::BlockState(root), Arc::new(state)); + } + assert!(store.has_state(&first).expect("has_state")); + let cache = store.state_cache.lock().unwrap(); + let (lru_key, _) = cache.iter().next_back().expect("non-empty"); + assert_eq!(*lru_key, CacheKey::BlockState(first)); + } + + /// A root with a diff is found without consulting `States`. + #[test] + fn has_state_checks_diffs_first_and_skips_states_on_a_hit() { + let (store, counts) = counting_store(); + let root = H256::from([3u8; 32]); + put_raw(&store, Table::StateDiffs, root); + + assert!(store.has_state(&root).expect("has_state")); + assert_eq!(counts.contains(Table::StateDiffs), 1); + assert_eq!(counts.contains(Table::States), 0); + // Never a value read. + assert_eq!(counts.gets(Table::StateDiffs), 0); + assert_eq!(counts.gets(Table::States), 0); + } + + /// A snapshot-only root (a beacon epoch anchor has no diff) is found. + #[test] + fn has_state_finds_a_snapshot_only_root() { + let (store, counts) = counting_store(); + let root = H256::from([4u8; 32]); + put_raw(&store, Table::States, root); + + assert!(store.has_state(&root).expect("has_state")); + assert_eq!(counts.gets(Table::States), 0); + } + + #[test] + fn has_state_is_false_for_an_absent_root() { + let (store, _counts) = counting_store(); + assert!(!store.has_state(&H256::from([5u8; 32])).expect("has_state")); + } + /// A state that only the handoff buffer holds is still readable. The LRU /// is cleared first, so a pass here cannot come from the cache, and the /// backend was never written for this root, so it cannot come from disk. From 076b3bd1e311734c2fdfc4bea12c6770ba6ebe2e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Mon, 5 Oct 2026 12:29:16 -0300 Subject: [PATCH 2/4] refactor(storage): make a borrowing read the one required lookup `get` hands back an owned `Vec`, so every lookup copies the whole value before the caller decodes it, and a beacon state snapshot is 100+ MB. RocksDB can lend its own buffer (`get_pinned_cf`), but a generic closure parameter would make `StorageReadView` unusable as a trait object, and every caller holds it as `Box`. `read` takes the callback as `&mut dyn FnMut(&[u8])`, which keeps the trait dyn-compatible. Both backends implement only `read`; `get` and `contains` become default methods on top of it, so a new backend has one lookup to get right instead of three that must agree. --- crates/storage/src/api/traits.rs | 34 ++++++++++++++++--- crates/storage/src/backend/in_memory.rs | 26 +++++++-------- crates/storage/src/backend/rocksdb.rs | 24 +++++++------- crates/storage/src/backend/tests.rs | 44 +++++++++++++++++++++++++ crates/storage/src/store.rs | 11 +++++++ 5 files changed, 107 insertions(+), 32 deletions(-) diff --git a/crates/storage/src/api/traits.rs b/crates/storage/src/api/traits.rs index 721c7c4d6..84230144a 100644 --- a/crates/storage/src/api/traits.rs +++ b/crates/storage/src/api/traits.rs @@ -24,16 +24,40 @@ pub trait StorageBackend: Send + Sync { /// A read-only view of the storage. pub trait StorageReadView { - /// Get a value by key from a table. - fn get(&self, table: Table, key: &[u8]) -> Result>, Error>; + /// Calls `read_fn` with a borrow of the value stored under `key`, if any, + /// and returns whether the key was present. + /// + /// The value is never copied: `read_fn` sees the backend's own buffer, so + /// a caller that only decodes or inspects the bytes avoids allocating a + /// value-sized `Vec` (full state snapshots are 100+ MB on mainnet-sized + /// beacon chains). `read_fn` runs at most once, and its error is returned + /// as-is. A `&mut dyn FnMut` rather than a generic closure, so the trait + /// stays usable as `dyn StorageReadView`. + fn read( + &self, + table: Table, + key: &[u8], + read_fn: &mut dyn FnMut(&[u8]) -> Result<(), Error>, + ) -> Result; + + /// Get a value by key from a table, copied into an owned `Vec`. + fn get(&self, table: Table, key: &[u8]) -> Result>, Error> { + let mut value = None; + self.read(table, key, &mut |bytes| { + value = Some(bytes.to_vec()); + Ok(()) + })?; + Ok(value) + } /// Whether `key` is present in a table. /// /// Same answer as `get(..)?.is_some()` but never materializes the value, so /// the cost does not scale with its size. Prefer it for pure existence - /// checks on large values (full state snapshots are 100+ MB on mainnet-sized - /// beacon chains). - fn contains(&self, table: Table, key: &[u8]) -> Result; + /// checks on large values. + fn contains(&self, table: Table, key: &[u8]) -> Result { + self.read(table, key, &mut |_| Ok(())) + } /// Iterate over all entries with a given key prefix. fn prefix_iterator( diff --git a/crates/storage/src/backend/in_memory.rs b/crates/storage/src/backend/in_memory.rs index fd7d766ed..6c4d303f6 100644 --- a/crates/storage/src/backend/in_memory.rs +++ b/crates/storage/src/backend/in_memory.rs @@ -66,21 +66,17 @@ struct InMemoryReadView<'a> { } impl StorageReadView for InMemoryReadView<'_> { - fn get(&self, table: Table, key: &[u8]) -> Result>, Error> { - Ok(self - .guard - .get(&table) - .expect("table exists") - .get(key) - .cloned()) - } - - fn contains(&self, table: Table, key: &[u8]) -> Result { - Ok(self - .guard - .get(&table) - .expect("table exists") - .contains_key(key)) + fn read( + &self, + table: Table, + key: &[u8], + read_fn: &mut dyn FnMut(&[u8]) -> Result<(), Error>, + ) -> Result { + let Some(value) = self.guard.get(&table).expect("table exists").get(key) else { + return Ok(false); + }; + read_fn(value)?; + Ok(true) } fn prefix_iterator( diff --git a/crates/storage/src/backend/rocksdb.rs b/crates/storage/src/backend/rocksdb.rs index ac4aed229..0a2ae840a 100644 --- a/crates/storage/src/backend/rocksdb.rs +++ b/crates/storage/src/backend/rocksdb.rs @@ -103,24 +103,24 @@ struct RocksDBReadView { } impl StorageReadView for RocksDBReadView { - fn get(&self, table: Table, key: &[u8]) -> Result>, Error> { - let cf = self - .db - .cf_handle(cf_name(table)) - .ok_or_else(|| format!("Column family {} not found", cf_name(table)))?; - - Ok(self.db.get_cf(&cf, key)?) - } - - fn contains(&self, table: Table, key: &[u8]) -> Result { + fn read( + &self, + table: Table, + key: &[u8], + read_fn: &mut dyn FnMut(&[u8]) -> Result<(), Error>, + ) -> Result { let cf = self .db .cf_handle(cf_name(table)) .ok_or_else(|| format!("Column family {} not found", cf_name(table)))?; - // Pinned: references the block-cache buffer instead of copying the + // Pinned: references RocksDB's own buffer instead of copying the // value into a `Vec`. - Ok(self.db.get_pinned_cf(&cf, key)?.is_some()) + let Some(value) = self.db.get_pinned_cf(&cf, key)? else { + return Ok(false); + }; + read_fn(&value)?; + Ok(true) } fn prefix_iterator( diff --git a/crates/storage/src/backend/tests.rs b/crates/storage/src/backend/tests.rs index ffd3ea308..0f9cd962c 100644 --- a/crates/storage/src/backend/tests.rs +++ b/crates/storage/src/backend/tests.rs @@ -18,6 +18,7 @@ pub fn run_backend_tests(backend: &dyn StorageBackend) { test_prefix_iterator(backend); test_nonexistent_key(backend); test_contains(backend); + test_read(backend); test_delete_then_put(backend); test_put_then_delete(backend); test_delete_range(backend); @@ -69,6 +70,49 @@ fn test_contains(backend: &dyn StorageBackend) { ); } +fn test_read(backend: &dyn StorageBackend) { + { + let mut batch = backend.begin_write().unwrap(); + batch + .put_batch( + Table::BlockHeaders, + vec![(b"test_read_key".to_vec(), b"value1".to_vec())], + ) + .unwrap(); + batch.commit().unwrap(); + } + let view = backend.begin_read().unwrap(); + + let mut seen = Vec::new(); + let found = view + .read(Table::BlockHeaders, b"test_read_key", &mut |bytes| { + seen.push(bytes.to_vec()); + Ok(()) + }) + .unwrap(); + assert!(found); + assert_eq!(seen, vec![b"value1".to_vec()]); + + // A missing key never reaches the callback. + let mut calls = 0; + let found = view + .read(Table::BlockHeaders, b"test_read_missing", &mut |_| { + calls += 1; + Ok(()) + }) + .unwrap(); + assert!(!found); + assert_eq!(calls, 0); + + // The callback's own error is what `read` returns. + let err = view + .read(Table::BlockHeaders, b"test_read_key", &mut |_| { + Err("decode failed".into()) + }) + .unwrap_err(); + assert_eq!(err.to_string(), "decode failed"); +} + fn test_delete(backend: &dyn StorageBackend) { // Write data { diff --git a/crates/storage/src/store.rs b/crates/storage/src/store.rs index efa50193e..0bc712e7d 100644 --- a/crates/storage/src/store.rs +++ b/crates/storage/src/store.rs @@ -5307,6 +5307,17 @@ mod tests { } impl StorageReadView for CountingView<'_> { + fn read( + &self, + table: Table, + key: &[u8], + read_fn: &mut dyn FnMut(&[u8]) -> Result<(), BackendError>, + ) -> Result { + self.inner.read(table, key, read_fn) + } + + // `get` and `contains` are overridden, not left to their defaults, so + // the two can be counted apart. fn get(&self, table: Table, key: &[u8]) -> Result>, BackendError> { *self.counts.gets.lock().unwrap().entry(table).or_default() += 1; self.inner.get(table, key) From de6e71c7fdef529ca92ac2e42e019444449e4c87 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Mon, 5 Oct 2026 12:34:54 -0300 Subject: [PATCH 3/4] perf(storage): decode values from the backend's buffer instead of a copy Every decode site fetched an owned `Vec` with `get` and dropped it right after decoding, so each read paid a value-sized allocation and copy first. For a beacon state snapshot that is 100+ MB per cold read. `StorageReadViewExt::read_with` decodes inside `read`'s borrow, and the decode sites use it. Beacon state reconstruction folds its deltas while the snapshot is still borrowed and hands the result over as a `Cow`: a snapshot root decodes straight from RocksDB's buffer, and the writer's parent-bytes path still takes ownership without a second copy. Existence checks written as `get(..).is_some()` become `contains`. `get` stays where the bytes outlive the read: the beacon walk's delta records, the pending-column take, and the stored config, which must not be decoded before the version and preset checks pass. --- crates/storage/src/api/mod.rs | 4 +- crates/storage/src/api/traits.rs | 37 ++++- crates/storage/src/lib.rs | 4 +- crates/storage/src/state_writer.rs | 128 ++++++++++------ crates/storage/src/store.rs | 233 ++++++++++++++++++----------- 5 files changed, 268 insertions(+), 138 deletions(-) diff --git a/crates/storage/src/api/mod.rs b/crates/storage/src/api/mod.rs index 00755269c..703ddfc53 100644 --- a/crates/storage/src/api/mod.rs +++ b/crates/storage/src/api/mod.rs @@ -17,4 +17,6 @@ mod tables; mod traits; pub use tables::{ALL_TABLES, Table}; -pub use traits::{Error, PrefixResult, StorageBackend, StorageReadView, StorageWriteBatch}; +pub use traits::{ + Error, PrefixResult, StorageBackend, StorageReadView, StorageReadViewExt, StorageWriteBatch, +}; diff --git a/crates/storage/src/api/traits.rs b/crates/storage/src/api/traits.rs index 84230144a..35747be3c 100644 --- a/crates/storage/src/api/traits.rs +++ b/crates/storage/src/api/traits.rs @@ -41,13 +41,11 @@ pub trait StorageReadView { ) -> Result; /// Get a value by key from a table, copied into an owned `Vec`. + /// + /// Prefer [`StorageReadViewExt::read_with`] when the bytes are only + /// decoded: it decodes from the backend's buffer without the copy. fn get(&self, table: Table, key: &[u8]) -> Result>, Error> { - let mut value = None; - self.read(table, key, &mut |bytes| { - value = Some(bytes.to_vec()); - Ok(()) - })?; - Ok(value) + self.read_with(table, key, <[u8]>::to_vec) } /// Whether `key` is present in a table. @@ -67,6 +65,33 @@ pub trait StorageReadView { ) -> Result + '_>, Error>; } +/// Generic conveniences over [`StorageReadView::read`]. +/// +/// A separate trait because its methods are generic, which would make +/// `StorageReadView` itself unusable as a trait object. The blanket impl +/// covers every view, `dyn StorageReadView` included. +pub trait StorageReadViewExt: StorageReadView { + /// Decodes the value stored under `key` straight from the backend's + /// buffer, without copying it first. `None` when the key is absent. + fn read_with( + &self, + table: Table, + key: &[u8], + decode: impl FnOnce(&[u8]) -> T, + ) -> Result, Error> { + let mut decode = Some(decode); + let mut value = None; + self.read(table, key, &mut |bytes| { + let decode = decode.take().expect("read calls read_fn at most once"); + value = Some(decode(bytes)); + Ok(()) + })?; + Ok(value) + } +} + +impl StorageReadViewExt for V {} + /// A write batch that can be committed atomically. pub trait StorageWriteBatch: Send { /// Put multiple key-value pairs into a table. diff --git a/crates/storage/src/lib.rs b/crates/storage/src/lib.rs index 8f1fc61b3..bc33414fc 100644 --- a/crates/storage/src/lib.rs +++ b/crates/storage/src/lib.rs @@ -9,7 +9,9 @@ mod state_diff; mod state_writer; mod store; -pub use api::{ALL_TABLES, StorageBackend, StorageReadView, StorageWriteBatch, Table}; +pub use api::{ + ALL_TABLES, StorageBackend, StorageReadView, StorageReadViewExt, StorageWriteBatch, Table, +}; pub use committee_cache::{CommitteeCache, Lookup, ShufflingKey}; /// Error type returned by the fallible [`Store`] operations, exported so /// callers can match on it (e.g. to distinguish [`Error::DbVersionMismatch`]). diff --git a/crates/storage/src/state_writer.rs b/crates/storage/src/state_writer.rs index d82796260..d1103d0a7 100644 --- a/crates/storage/src/state_writer.rs +++ b/crates/storage/src/state_writer.rs @@ -39,6 +39,7 @@ //! Splitting them apart would let the two drift until a reader silently //! failed to reconstruct what a writer had actually written. +use std::borrow::Cow; use std::collections::HashMap; use std::sync::mpsc::{Receiver, SyncSender, sync_channel}; use std::sync::{Arc, Mutex}; @@ -54,7 +55,7 @@ use libssz::{SszDecode, SszEncode}; use lru::LruCache; use tracing::error; -use crate::api::{StorageBackend, Table}; +use crate::api::{StorageBackend, StorageReadViewExt, Table}; use crate::beacon_state_delta; use crate::error::Error; use crate::state_codec::{decode_lean_state_value, decode_state_value, encode_state_value}; @@ -169,9 +170,8 @@ pub(crate) fn read_state( // Anchor snapshot in `States`, otherwise reconstruct from the diff chain. let snapshot = { let view = backend.begin_read().expect("read view"); - view.get(Table::States, &root.to_ssz()) - .expect("get") - .map(|bytes| decode_lean_state_value(&bytes)) + view.read_with(Table::States, &root.to_ssz(), decode_lean_state_value) + .expect("read") }; let state = match snapshot { Some(state) => state, @@ -183,15 +183,17 @@ pub(crate) fn read_state( BeaconState::Lean(state) } Chain::Beacon => { - let Some(bytes) = reconstruct_beacon_state_bytes(backend, root)? else { - return Ok(None); - }; // Decoded exactly once, after every delta in the chain has already // been folded in the byte domain; see // `reconstruct_beacon_state_bytes`'s doc comment for why an SSZ // decode per hop instead would be the whole cost that delta layer - // exists to avoid. - let mut state = decode_state_value(&bytes); + // exists to avoid. A snapshot root decodes straight from the + // backend's buffer. + let Some(mut state) = + reconstruct_beacon_state_bytes(backend, root, |bytes| decode_state_value(&bytes))? + else { + return Ok(None); + }; rebase_onto_resident(cache, &mut state); state } @@ -241,48 +243,78 @@ fn rebase_onto_resident(cache: &StateCache, state: &mut BeaconState) { /// base root read off each hop comes from /// [`beacon_state_delta::unframe`] instead of a `StateDiff`'s own field. /// -/// Returns encoded bytes rather than a decoded [`BeaconState`] so that a -/// caller that only needs the bytes (the writer thread's -/// `StateWriter::encoded_parent_bytes`, for its diff base) is never made to -/// pay for a decode it will not use. [`Store::get_state`](crate::store::Store::get_state)'s -/// beacon arm is the one caller that decodes, and it does so exactly once, -/// after every delta has already been folded. +/// Hands the encoded bytes to `consume` rather than returning a decoded +/// [`BeaconState`], so that a caller that only needs the bytes (the writer +/// thread's `StateWriter::encoded_parent_bytes`, for its diff base) is never +/// made to pay for a decode it will not use. +/// [`Store::get_state`](crate::store::Store::get_state)'s beacon arm is the +/// one caller that decodes, and it does so exactly once, after every delta +/// has already been folded. +/// +/// `consume` gets [`Cow::Borrowed`] when `root` is itself a snapshot: the +/// backend's own buffer, never copied. Otherwise the first delta reads the +/// borrowed snapshot as its base, and `consume` gets the folded result as +/// [`Cow::Owned`], so taking ownership costs no further copy either way +/// beyond the one a borrowed snapshot needs. /// /// `Ok(None)` when `root` is unknown, or the chain runs off the retained /// window before reaching a snapshot: a missing `StateDiffs` record below /// the pruned boundary, matching how the lean walk in /// [`reconstruct_state`] handles both cases. -pub(crate) fn reconstruct_beacon_state_bytes( +pub(crate) fn reconstruct_beacon_state_bytes( backend: &dyn StorageBackend, root: &H256, -) -> Result>, Error> { + consume: impl FnOnce(Cow<'_, [u8]>) -> T, +) -> Result, Error> { let view = backend.begin_read().expect("read view"); let mut records: Vec> = Vec::new(); + let mut consume = Some(consume); let mut cursor = *root; - let snapshot = loop { - if let Some(bytes) = view.get(Table::States, &cursor.to_ssz()).expect("get") { - break bytes; + loop { + let key = cursor.to_ssz(); + // The walk ends at the first snapshot, so the deltas collected so + // far are folded onto it while it is still borrowed. + let folded = view + .read_with(Table::States, &key, |snapshot| { + let consume = consume.take().expect("the walk ends at its first snapshot"); + // `records` runs target -> snapshot child; reverse to snapshot + // child -> target, the order the chain was written in, so + // folding forward replays it correctly. + records.reverse(); + fold_beacon_state_deltas(snapshot, &records, consume) + }) + .expect("read"); + if folded.is_some() { + return Ok(folded); } - let Some(diff_bytes) = view.get(Table::StateDiffs, &cursor.to_ssz()).expect("get") else { + let Some(diff_bytes) = view.get(Table::StateDiffs, &key).expect("get") else { return Ok(None); }; let (base_root, _, _, _) = beacon_state_delta::unframe(&diff_bytes); cursor = base_root; records.push(diff_bytes); - }; - drop(view); - - // `records` runs target -> snapshot child; reverse to snapshot child - // -> target, the order the chain was written in, so folding forward - // replays it correctly. - records.reverse(); + } +} - let mut bytes = snapshot; - for record in &records { +/// Applies `records` (snapshot child first) to a borrowed `snapshot` and +/// hands the result to `consume`: the snapshot itself when there is nothing +/// to apply, otherwise the owned output of the last delta. +fn fold_beacon_state_deltas( + snapshot: &[u8], + records: &[Vec], + consume: impl FnOnce(Cow<'_, [u8]>) -> T, +) -> T { + let mut records = records.iter(); + let Some(first) = records.next() else { + return consume(Cow::Borrowed(snapshot)); + }; + let (_, _, target_len, delta) = beacon_state_delta::unframe(first); + let mut bytes = beacon_state_delta::decode(delta, snapshot, target_len as usize); + for record in records { let (_, _, target_len, delta) = beacon_state_delta::unframe(record); bytes = beacon_state_delta::decode(delta, &bytes, target_len as usize); } - Ok(Some(bytes)) + consume(Cow::Owned(bytes)) } /// Reconstruct a state from diffs and the nearest ancestor snapshot. @@ -312,13 +344,21 @@ pub(crate) fn reconstruct_state( let mut diffs: Vec = Vec::new(); let mut cursor = *root; let snapshot = loop { - if let Some(bytes) = view.get(Table::States, &cursor.to_ssz()).expect("get") { - break decode_lean_state_value(&bytes); + let key = cursor.to_ssz(); + if let Some(snapshot) = view + .read_with(Table::States, &key, decode_lean_state_value) + .expect("read") + { + break snapshot; } - let Some(diff_bytes) = view.get(Table::StateDiffs, &cursor.to_ssz()).expect("get") else { + let Some(diff) = view + .read_with(Table::StateDiffs, &key, |bytes| { + StateDiff::from_ssz_bytes(bytes).expect("valid state diff") + }) + .expect("read") + else { return Ok(None); }; - let diff = StateDiff::from_ssz_bytes(&diff_bytes).expect("valid state diff"); cursor = diff.base_root; diffs.push(diff); }; @@ -330,11 +370,13 @@ pub(crate) fn reconstruct_state( // The latest block header lives in BlockHeaders; the stored state caches // the real state_root there, so it equals the header byte-for-byte. let view = backend.begin_read().expect("read view"); - let header_bytes = view.get(Table::BlockHeaders, &root.to_ssz()).expect("get"); + let header = view + .read_with(Table::BlockHeaders, &root.to_ssz(), |bytes| { + BlockHeader::from_ssz_bytes(bytes).expect("valid header") + }) + .expect("read"); drop(view); - let Some(latest_block_header) = - header_bytes.map(|bytes| BlockHeader::from_ssz_bytes(&bytes).expect("valid header")) - else { + let Some(latest_block_header) = header else { return Ok(None); }; @@ -556,9 +598,11 @@ impl StateWriter { .filter(|(root, _)| *root == parent_root) .map(|(_, bytes)| bytes.clone()) .unwrap_or_else(|| { - reconstruct_beacon_state_bytes(self.backend.as_ref(), &parent_root) - .expect("read parent state") - .expect("parent state must exist to diff against") + reconstruct_beacon_state_bytes(self.backend.as_ref(), &parent_root, |bytes| { + bytes.into_owned() + }) + .expect("read parent state") + .expect("parent state must exist to diff against") }) } } diff --git a/crates/storage/src/store.rs b/crates/storage/src/store.rs index 0bc712e7d..28b147de7 100644 --- a/crates/storage/src/store.rs +++ b/crates/storage/src/store.rs @@ -4,7 +4,7 @@ use std::sync::{Arc, LazyLock, Mutex}; use lru::LruCache; -use crate::api::{StorageBackend, StorageReadView, StorageWriteBatch, Table}; +use crate::api::{StorageBackend, StorageReadView, StorageReadViewExt, StorageWriteBatch, Table}; use crate::committee_cache::CommitteeCache; use crate::error::Error; @@ -779,11 +779,10 @@ fn decode_beacon_block_value(bytes: &[u8]) -> SignedBeaconBlock { /// enforced by having exactly one caller rather than by a runtime check. pub(crate) fn beacon_block_slot(backend: &dyn StorageBackend, root: &H256) -> Option { let view = backend.begin_read().expect("read view"); - let bytes = view - .get(Table::BlockHeaders, &root.to_ssz()) - .expect("get")?; - drop(view); - Some(decode_beacon_block_value(&bytes).slot()) + view.read_with(Table::BlockHeaders, &root.to_ssz(), |bytes| { + decode_beacon_block_value(bytes).slot() + }) + .expect("read") } /// Fork choice store backed by a pluggable storage backend. @@ -982,14 +981,17 @@ impl Store { // Written by both `init_store` and `init_beacon`, so a backend // missing this has never held a chain of either kind. let view = backend.begin_read().expect("read view"); + // Kept as raw bytes: it may only be decoded once the version and + // preset checks below have passed. let Some(bytes) = view.get(Table::Metadata, KEY_CONFIG).expect("get config") else { return Ok(None); }; let found = view - .get(Table::Metadata, KEY_DB_VERSION) - .expect("get db version") - .map(|bytes| u64::from_ssz_bytes(&bytes).expect("valid db version")) + .read_with(Table::Metadata, KEY_DB_VERSION, |bytes| { + u64::from_ssz_bytes(bytes).expect("valid db version") + }) + .expect("read db version") .unwrap_or(0); if found != DB_VERSION { return Err(Error::DbVersionMismatch { @@ -1004,9 +1006,9 @@ impl Store { // shapes back. The version check above cannot stand in for this, // since both presets write the same *layout* at the same version. let found_preset = view - .get(Table::Metadata, KEY_PRESET) - .expect("get preset") - .and_then(|bytes| bytes.first().copied()) + .read_with(Table::Metadata, KEY_PRESET, |bytes| bytes.first().copied()) + .expect("read preset") + .flatten() .and_then(Preset::from_selector); if found_preset != Some(Preset::ACTIVE) { return Err(Error::PresetMismatch { @@ -1016,9 +1018,9 @@ impl Store { } let chain = view - .get(Table::Metadata, KEY_CHAIN) - .expect("get chain") - .and_then(|bytes| bytes.first().copied()) + .read_with(Table::Metadata, KEY_CHAIN, |bytes| bytes.first().copied()) + .expect("read chain") + .flatten() .and_then(Chain::from_selector) .expect("a versioned directory always carries a chain tag"); @@ -1026,9 +1028,10 @@ impl Store { // already turned away every directory written before they did, so // an absent key here is not an old directory but a corrupt one. let anchor_slot = view - .get(Table::Metadata, KEY_ANCHOR_SLOT) - .expect("get anchor slot") - .map(|bytes| u64::from_ssz_bytes(&bytes).expect("valid anchor slot")) + .read_with(Table::Metadata, KEY_ANCHOR_SLOT, |bytes| { + u64::from_ssz_bytes(bytes).expect("valid anchor slot") + }) + .expect("read anchor slot") .expect("a versioned directory always carries an anchor slot"); ( @@ -1527,17 +1530,17 @@ impl Store { /// on this store, and the key is what says which one. pub(crate) fn get_metadata(&self, key: &[u8]) -> T { let view = self.backend.begin_read().expect("read view"); - let bytes = view - .get(Table::Metadata, key) - .expect("get") - .unwrap_or_else(|| { - panic!( - "metadata key {:?} is absent on a {:?} store", - String::from_utf8_lossy(key), - self.chain - ) - }); - T::from_ssz_bytes(&bytes).expect("valid encoding") + view.read_with(Table::Metadata, key, |bytes| { + T::from_ssz_bytes(bytes).expect("valid encoding") + }) + .expect("read") + .unwrap_or_else(|| { + panic!( + "metadata key {:?} is absent on a {:?} store", + String::from_utf8_lossy(key), + self.chain + ) + }) } pub(crate) fn set_metadata(&self, key: &[u8], value: &T) { @@ -2094,9 +2097,10 @@ impl Store { } let view = self.backend.begin_read().expect("read view"); Ok(view - .get(Table::BlockHeaders, &root.to_ssz()) - .expect("get") - .map(|bytes| BlockHeader::from_ssz_bytes(&bytes).expect("valid header"))) + .read_with(Table::BlockHeaders, &root.to_ssz(), |bytes| { + BlockHeader::from_ssz_bytes(bytes).expect("valid header") + }) + .expect("read")) } // ============ Signed Blocks ============ @@ -2387,27 +2391,26 @@ impl Store { /// field is already in hand once the row is decoded. fn block_fields(&self, root: &H256) -> Option<(u64, H256, H256)> { let view = self.backend.begin_read().expect("read view"); - let bytes = view - .get(Table::BlockHeaders, &root.to_ssz()) - .expect("get")?; - Some(match self.chain { - Chain::Lean => { - let header = BlockHeader::from_ssz_bytes(&bytes).expect("valid header"); - (header.slot, header.parent_root, header.state_root) - } - Chain::Beacon => { - let block = decode_beacon_block_value(&bytes); - (block.slot(), block.parent_root(), block.state_root()) + view.read_with(Table::BlockHeaders, &root.to_ssz(), |bytes| { + match self.chain { + Chain::Lean => { + let header = BlockHeader::from_ssz_bytes(bytes).expect("valid header"); + (header.slot, header.parent_root, header.state_root) + } + Chain::Beacon => { + let block = decode_beacon_block_value(bytes); + (block.slot(), block.parent_root(), block.state_root()) + } } }) + .expect("read") } /// Whether a block is stored under `root`. pub fn has_block(&self, root: &H256) -> bool { let view = self.backend.begin_read().expect("read view"); - view.get(Table::BlockHeaders, &root.to_ssz()) - .expect("get") - .is_some() + view.contains(Table::BlockHeaders, &root.to_ssz()) + .expect("contains") } /// Every stored beacon block as `root -> (slot, parent_root)`. @@ -2437,18 +2440,27 @@ impl Store { let view = self.backend.begin_read().expect("read view"); let key = root.to_ssz(); - let Some(header_bytes) = view.get(Table::BlockHeaders, &key).expect("get") else { + let Some(header) = view + .read_with(Table::BlockHeaders, &key, |bytes| { + BlockHeader::from_ssz_bytes(bytes).expect("valid header") + }) + .expect("read") + else { return Ok(None); }; - let header = BlockHeader::from_ssz_bytes(&header_bytes).expect("valid header"); let body = if header.body_root == *EMPTY_BODY_ROOT { BlockBody::default() } else { - let Some(body_bytes) = view.get(Table::BlockBodies, &key).expect("get") else { + let Some(body) = view + .read_with(Table::BlockBodies, &key, |bytes| { + BlockBody::from_ssz_bytes(bytes).expect("valid body") + }) + .expect("read") + else { return Ok(None); }; - BlockBody::from_ssz_bytes(&body_bytes).expect("valid body") + body }; Ok(Some(Block::from_header_and_body(header, body))) @@ -2479,38 +2491,43 @@ impl Store { Chain::Lean => { Ok(Self::signed_block_from_view(view.as_ref(), root).map(SignedBeaconBlock::Lean)) } - Chain::Beacon => { - let Some(bytes) = view.get(Table::BlockHeaders, &root.to_ssz()).expect("get") - else { - return Ok(None); - }; - - let block = decode_beacon_block_value(&bytes); - - Ok(Some(block)) - } + Chain::Beacon => Ok(view + .read_with( + Table::BlockHeaders, + &root.to_ssz(), + decode_beacon_block_value, + ) + .expect("read")), } } fn signed_block_from_view(view: &dyn StorageReadView, root: &H256) -> Option { let key = root.to_ssz(); - let header_bytes = view.get(Table::BlockHeaders, &key).expect("get")?; - let header = BlockHeader::from_ssz_bytes(&header_bytes).expect("valid header"); + let header = view + .read_with(Table::BlockHeaders, &key, |bytes| { + BlockHeader::from_ssz_bytes(bytes).expect("valid header") + }) + .expect("read")?; // Use empty body if header indicates empty, otherwise fetch from DB let body = if header.body_root == *EMPTY_BODY_ROOT { BlockBody::default() } else { - let body_bytes = view.get(Table::BlockBodies, &key).expect("get")?; - BlockBody::from_ssz_bytes(&body_bytes).expect("valid body") + view.read_with(Table::BlockBodies, &key, |bytes| { + BlockBody::from_ssz_bytes(bytes).expect("valid body") + }) + .expect("read")? }; let sig_key = encode_slot_root_key(header.slot, root); - let proof = match view.get(Table::BlockProof, &sig_key).expect("get") { - Some(proof_bytes) => { - MultiMessageAggregate::from_ssz_bytes(&proof_bytes).expect("valid block proof") - } + let proof = view + .read_with(Table::BlockProof, &sig_key, |bytes| { + MultiMessageAggregate::from_ssz_bytes(bytes).expect("valid block proof") + }) + .expect("read"); + let proof = match proof { + Some(proof) => proof, // Synthesis only covers the genesis-style anchor (slot 0). For any // other slot a missing proof (pruned finalized block, or genuine // corruption) surfaces as `None` rather than a fabricated block. @@ -2541,9 +2558,10 @@ impl Store { pub fn canonical_root_at_slot(&self, slot: u64) -> Result, Error> { let view = self.backend.begin_read().expect("read view"); Ok(view - .get(Table::BlockRoots, &encode_block_root_key(slot)) - .expect("get block root") - .map(|bytes| H256::from_ssz_bytes(&bytes).expect("valid block root"))) + .read_with(Table::BlockRoots, &encode_block_root_key(slot), |bytes| { + H256::from_ssz_bytes(bytes).expect("valid block root") + }) + .expect("read block root")) } /// Return canonical signed blocks for the slot range `[start_slot, end_slot]`. @@ -2574,13 +2592,14 @@ impl Store { // `canonical_root_at_slot`, which opens a fresh one per call: a // range must be served from a single snapshot so a head change // partway through cannot splice two branches into one response. - let Some(root_bytes) = view - .get(Table::BlockRoots, &encode_block_root_key(slot)) - .expect("get block root") + let Some(root) = view + .read_with(Table::BlockRoots, &encode_block_root_key(slot), |bytes| { + H256::from_ssz_bytes(bytes).expect("valid block root") + }) + .expect("read block root") else { continue; }; - let root = H256::from_ssz_bytes(&root_bytes).expect("valid block root"); match self.chain { Chain::Lean => { if let Some(block) = Self::signed_block_from_view(view.as_ref(), &root) { @@ -2591,9 +2610,15 @@ impl Store { // lean arm's "header found but proof pruned" `None`: the row is // there or the slot is skipped. Chain::Beacon => { - if let Some(bytes) = view.get(Table::BlockHeaders, &root.to_ssz()).expect("get") + if let Some(block) = view + .read_with( + Table::BlockHeaders, + &root.to_ssz(), + decode_beacon_block_value, + ) + .expect("read") { - blocks.push(decode_beacon_block_value(&bytes)); + blocks.push(block); } } } @@ -3577,9 +3602,8 @@ impl Store { /// gossip column. pub fn has_data_column(&self, slot: u64, root: &H256, index: u64) -> bool { let view = self.backend.begin_read().expect("read view"); - view.get(Table::DataColumns, &data_column_key(slot, root, index)) - .expect("get") - .is_some() + view.contains(Table::DataColumns, &data_column_key(slot, root, index)) + .expect("contains") } /// Which columns of one block this node holds, ascending. @@ -3647,13 +3671,14 @@ impl Store { let view = self.backend.begin_read().expect("read view"); let mut found = Vec::new(); for slot in start_slot..end_slot { - let Some(root_bytes) = view - .get(Table::BlockRoots, &encode_block_root_key(slot)) - .expect("get block root") + let Some(root) = view + .read_with(Table::BlockRoots, &encode_block_root_key(slot), |bytes| { + H256::from_ssz_bytes(bytes).expect("valid block root") + }) + .expect("read block root") else { continue; }; - let root = H256::from_ssz_bytes(&root_bytes).expect("valid block root"); let prefix = data_column_block_prefix(slot, &root); let entries = view .prefix_iterator(Table::DataColumns, &prefix) @@ -3809,15 +3834,14 @@ mod tests { /// Check if a key exists in a table. fn has_key(backend: &dyn StorageBackend, table: Table, root: &H256) -> bool { let view = backend.begin_read().expect("read view"); - view.get(table, &root.to_ssz()).expect("get").is_some() + view.contains(table, &root.to_ssz()).expect("contains") } /// Check whether a block proof exists for a (slot, root) pair. fn has_block_proof(backend: &dyn StorageBackend, slot: u64, root: &H256) -> bool { let view = backend.begin_read().expect("read view"); - view.get(Table::BlockProof, &encode_slot_root_key(slot, root)) - .expect("get") - .is_some() + view.contains(Table::BlockProof, &encode_slot_root_key(slot, root)) + .expect("contains") } /// Canonical block root at `slot`, for storage-index assertions. @@ -5809,6 +5833,39 @@ mod tests { assert_eq!(decoded.to_ssz(), child.to_ssz()); } + #[test] + fn a_cold_beacon_state_folds_every_delta_back_from_its_snapshot() { + let backend: Arc = Arc::new(InMemoryBackend::new()); + let mut store = beacon_test_store(backend.clone()); + let interval = ForkName::Electra.snapshot_interval(); + + // Slot 0 and slot `interval` are snapshots (no parent, then a boundary + // crossing), and every slot between them is a delta on its parent. So + // the reads cover a borrowed snapshot with nothing to fold, and chains + // of 1 up to `interval - 1` deltas applied to the borrowed snapshot. + let mut expected = Vec::new(); + let mut parent = H256::ZERO; + for slot in 0..=interval { + let root = H256::from([(slot + 1) as u8; 32]); + let state = beacon_test_state_with_parent(slot, parent); + store + .insert_signed_block(root, beacon_test_block(slot, parent)) + .expect("insert block"); + expected.push((root, state.to_ssz())); + store.insert_state(root, state).expect("insert state"); + parent = root; + } + + // Dropping the store joins the writer; a fresh one has an empty cache, + // so every read below reconstructs from the backend. + drop(store); + let cold = beacon_test_store(backend); + for (root, encoded) in expected.into_iter().rev() { + let state = cold.get_state(&root).expect("get").expect("present"); + assert_eq!(state.to_ssz(), encoded); + } + } + /// `beacon_test_state` with its parent linked in, the way `insert_state`'s /// beacon arm expects: it reads the base to diff against off the /// post-state's own `latest_block_header.parent_root`, mirroring how the From c2381f7983db1aac969d7a42b604dde56242ef54 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Mon, 5 Oct 2026 12:37:22 -0300 Subject: [PATCH 4/4] perf(storage): keep state snapshots and diffs in RocksDB blob files A beacon snapshot is 100+ MB and lives inline in its SST, so it is its own data block. Every compaction touching that file rewrites it, and with no bloom filter a lookup that misses `States` still reads the data block around the key. On a Plataberget follower that put ~30 ms of `has_state` misses on every block import, one snapshot-sized read per L0 file. `States` and `StateDiffs` now store values of 4 KiB and up (one default data block) in blob files, LZ4-compressed since blob files otherwise default to none while SST blocks get Snappy. Their SSTs shrink to keys and blob references. No blob GC, since nothing deletes or overwrites a state, and no blob cache, since a snapshot-sized entry would evict the whole block cache while the store already caches decoded states. RocksDB applies this to an existing directory as it goes, so a directory written without blob files opens unchanged and no `DB_VERSION` bump is needed; inline values move out as compaction rewrites them. The table-size estimate now counts live blob bytes, which `estimate-live-data-size` leaves out. --- crates/storage/src/backend/rocksdb.rs | 130 +++++++++++++++++++++++++- docs/data_storage.md | 9 +- 2 files changed, 133 insertions(+), 6 deletions(-) diff --git a/crates/storage/src/backend/rocksdb.rs b/crates/storage/src/backend/rocksdb.rs index 0a2ae840a..cccdb39af 100644 --- a/crates/storage/src/backend/rocksdb.rs +++ b/crates/storage/src/backend/rocksdb.rs @@ -4,8 +4,8 @@ use crate::api::{ ALL_TABLES, Error, PrefixResult, StorageBackend, StorageReadView, StorageWriteBatch, Table, }; use rocksdb::{ - BlockBasedOptions, Cache, ColumnFamilyDescriptor, DBWithThreadMode, MultiThreaded, Options, - WriteBatch, WriteOptions, + BlockBasedOptions, Cache, ColumnFamilyDescriptor, DBCompressionType, DBWithThreadMode, + MultiThreaded, Options, WriteBatch, WriteOptions, }; use std::path::Path; use std::sync::Arc; @@ -18,6 +18,43 @@ fn cf_name(table: Table) -> &'static str { table.name() } +/// The smallest value a blob-file table stores out of line. +/// +/// RocksDB's default data block size: a larger value would get an oversized +/// block of its own anyway, so it gains nothing from staying inline. +const MIN_BLOB_SIZE: u64 = 4 * 1024; + +/// Whether a table keeps its values in blob files rather than inline in SSTs. +/// +/// `States` holds full state snapshots (100+ MB on mainnet-sized beacon +/// chains) and `StateDiffs` the deltas between them (hundreds of KB). Inline, +/// every compaction that touches a file rewrites those values, and a lookup +/// that misses still reads the data block around the key, which here can be +/// a whole snapshot. In a blob file a value is written once, and the SSTs +/// hold only small references to it. +fn stores_values_in_blob_files(table: Table) -> bool { + matches!(table, Table::States | Table::StateDiffs) +} + +/// Moves a column family's large values into blob files. +/// +/// RocksDB applies the change to an existing database as it goes: new +/// writes land in blob files, and inline values move out as compaction +/// rewrites their SSTs. So a data directory written without it opens +/// unchanged, and no `DB_VERSION` bump is needed. +fn enable_blob_files(cf_opts: &mut Options) { + cf_opts.set_enable_blob_files(true); + cf_opts.set_min_blob_size(MIN_BLOB_SIZE); + // SST blocks get RocksDB's default Snappy, but blob files default to no + // compression, so moving the values out would otherwise grow the tables + // on disk. The values are raw SSZ. + cf_opts.set_blob_compression_type(DBCompressionType::Lz4); + // No blob garbage collection: nothing deletes or overwrites a state, so it + // would only relocate live blobs during compaction. Revisit if states are + // ever pruned. No blob cache either: the store caches decoded states + // itself, and a snapshot-sized entry would evict the whole block cache. +} + /// RocksDB storage backend. #[derive(Clone)] pub struct RocksDBBackend { @@ -52,6 +89,9 @@ impl RocksDBBackend { .map(|t| { let mut cf_opts = Options::default(); cf_opts.set_block_based_table_factory(&block_opts); + if stores_values_in_blob_files(*t) { + enable_blob_files(&mut cf_opts); + } ColumnFamilyDescriptor::new(cf_name(*t), cf_opts) }) .collect(); @@ -93,7 +133,15 @@ impl StorageBackend for RocksDBBackend { .ok() .flatten() .unwrap_or(0); - sst_bytes + memtable_bytes + // `estimate-live-data-size` counts SST files only, so a blob-file + // table's values would otherwise vanish from the estimate. + let blob_bytes = self + .db + .property_int_value_cf(&cf, "rocksdb.live-blob-file-size") + .ok() + .flatten() + .unwrap_or(0); + sst_bytes + memtable_bytes + blob_bytes } } @@ -211,6 +259,82 @@ mod tests { run_backend_tests(&backend); } + /// A value big enough for a blob file, and the property counting them. + const BLOB_VALUE_LEN: usize = 64 * 1024; + const NUM_BLOB_FILES: &str = "rocksdb.num-blob-files"; + + fn num_blob_files(backend: &RocksDBBackend, table: Table) -> u64 { + let cf = backend.db.cf_handle(cf_name(table)).unwrap(); + backend + .db + .property_int_value_cf(&cf, NUM_BLOB_FILES) + .unwrap() + .unwrap() + } + + fn put_and_flush(backend: &RocksDBBackend, table: Table, key: &[u8], value: Vec) { + let mut batch = backend.begin_write().unwrap(); + batch.put_batch(table, vec![(key.to_vec(), value)]).unwrap(); + batch.commit().unwrap(); + let cf = backend.db.cf_handle(cf_name(table)).unwrap(); + backend.db.flush_cf(&cf).unwrap(); + } + + #[test] + fn large_state_values_live_in_blob_files() { + let dir = tempdir().unwrap(); + let backend = RocksDBBackend::open(dir.path()).unwrap(); + let value: Vec = (0..BLOB_VALUE_LEN).map(|i| (i % 251) as u8).collect(); + + for table in [Table::States, Table::StateDiffs] { + put_and_flush(&backend, table, b"big", value.clone()); + assert_eq!(num_blob_files(&backend, table), 1, "{table:?}"); + + let view = backend.begin_read().unwrap(); + assert_eq!(view.get(table, b"big").unwrap(), Some(value.clone())); + assert!(view.contains(table, b"big").unwrap()); + assert!(!view.contains(table, b"absent").unwrap()); + } + assert!(backend.estimate_table_bytes(Table::States) > 0); + + // Small values, and every other table, stay inline. + put_and_flush(&backend, Table::States, b"small", vec![7; 16]); + assert_eq!(num_blob_files(&backend, Table::States), 1); + put_and_flush(&backend, Table::BlockHeaders, b"big", value); + assert_eq!(num_blob_files(&backend, Table::BlockHeaders), 0); + } + + #[test] + fn a_directory_written_without_blob_files_still_reads() { + let dir = tempdir().unwrap(); + let value: Vec = (0..BLOB_VALUE_LEN).map(|i| (i % 251) as u8).collect(); + + // The layout a data directory had before blob files: every table + // with plain options. + { + let mut opts = Options::default(); + opts.create_if_missing(true); + opts.create_missing_column_families(true); + let cfs = ALL_TABLES.iter().map(|t| cf_name(*t)); + let db = DBWithThreadMode::::open_cf(&opts, dir.path(), cfs).unwrap(); + let cf = db.cf_handle(cf_name(Table::States)).unwrap(); + db.put_cf(&cf, b"old", &value).unwrap(); + db.flush_cf(&cf).unwrap(); + } + + let backend = RocksDBBackend::open(dir.path()).unwrap(); + assert_eq!(num_blob_files(&backend, Table::States), 0); + put_and_flush(&backend, Table::States, b"new", value.clone()); + assert_eq!(num_blob_files(&backend, Table::States), 1); + + let view = backend.begin_read().unwrap(); + assert_eq!( + view.get(Table::States, b"old").unwrap(), + Some(value.clone()) + ); + assert_eq!(view.get(Table::States, b"new").unwrap(), Some(value)); + } + #[test] fn test_persistence() { let dir = tempdir().unwrap(); diff --git a/docs/data_storage.md b/docs/data_storage.md index 4a524b097..cc238c836 100644 --- a/docs/data_storage.md +++ b/docs/data_storage.md @@ -56,8 +56,11 @@ Everything persisted is SSZ-encoded bytes. The `StorageBackend` trait knows nothing about consensus types. It moves raw bytes in and out of named tables: -- `begin_read()` returns a `StorageReadView` with `get(table, key)` and - `prefix_iterator(table, prefix)`. +- `begin_read()` returns a `StorageReadView`. Its one required lookup is + `read(table, key, read_fn)`, which lends the value to `read_fn` without + copying it; `get` (an owned copy) and `contains` are built on it, and + `StorageReadViewExt::read_with` decodes straight from the borrowed bytes. + It also has `prefix_iterator(table, prefix)`. - `begin_write()` returns a `StorageWriteBatch` with `put_batch`, `delete_batch`, and `commit()`. A batch stages puts and deletes across **multiple tables** and applies them **atomically** on commit. @@ -66,7 +69,7 @@ The two implementations live in `crates/storage/src/backend/`: | Backend | Details | | ----------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `RocksDBBackend` | One column family per table. Writes go through a native `WriteBatch` with `sync=false` (no fsync per commit). | +| `RocksDBBackend` | One column family per table. Writes go through a native `WriteBatch` with `sync=false` (no fsync per commit). `States` and `StateDiffs` keep values of 4 KiB and up in LZ4-compressed blob files, so compaction never rewrites a snapshot and a lookup miss never reads one. | | `InMemoryBackend` | A `HashMap` per table behind an `RwLock`. Its `prefix_iterator` sorts keys lexicographically to match RocksDB's iteration order, because pruning relies on slot-ordered early-stop scans (see [Key encoding](#key-encoding)). | ### Store: all the semantics