From e63bbd835a886909b9359dd51814d16c93211b13 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Wed, 11 Nov 2026 12:04:11 -0300 Subject: [PATCH 1/6] perf(storage): share one rebuild between concurrent get_state misses on a root A beacon state rebuild reads a 100+ MB snapshot, folds a delta into a second full-size buffer and decodes a fresh registry: about a second of CPU and several hundred MB at peak. Concurrent misses on one root each paid that in full. A late block's custody columns are chain-checked one sidecar per blocking task, so a parent state that had left the 32-entry cache was rebuilt eight times at once. On the Hoodi follower that showed up as +1 to +4 GiB resident-memory spikes. read_state now routes a miss through StateRebuilds, a per-root registry shared by the Store and the writer thread. The first miss leads: it re-checks the cache and the write buffer, rebuilds, caches the result, then deregisters the root and wakes the waiters, who take the same Arc. A leader that errors or panics hands its waiters no result; they retry, one of them as the new leader. New counter lean_state_rebuilds_total{role="led"|"joined"} tells real storage reads apart from shared ones. --- crates/storage/src/metrics.rs | 36 ++++ crates/storage/src/state_writer.rs | 295 ++++++++++++++++++++++++++++- crates/storage/src/store.rs | 17 +- docs/data_storage.md | 8 + docs/metrics.md | 9 + 5 files changed, 361 insertions(+), 4 deletions(-) diff --git a/crates/storage/src/metrics.rs b/crates/storage/src/metrics.rs index 99093a8fb..4ad6b0caf 100644 --- a/crates/storage/src/metrics.rs +++ b/crates/storage/src/metrics.rs @@ -90,6 +90,42 @@ impl StateCacheMethod { } } +static LEAN_STATE_REBUILDS_TOTAL: LazyLock = LazyLock::new(|| { + register_int_counter_vec!( + "lean_state_rebuilds_total", + "Block states read back from storage after missing the cache and the write buffer, \ + by whether the read did the rebuild or waited on one already under way for that root", + &["role"] + ) + .unwrap() +}); + +/// How a `read_state` miss got its state from storage. +#[derive(Debug, Clone, Copy)] +pub(crate) enum RebuildRole { + /// Read and decoded the state itself. + Led, + /// Waited on a rebuild of the same root already under way and took its + /// result. + Joined, +} + +impl RebuildRole { + fn label(self) -> &'static str { + match self { + Self::Led => "led", + Self::Joined => "joined", + } + } +} + +/// Count one storage read of a block state, by `role`. +pub(crate) fn inc_state_rebuilds(role: RebuildRole) { + LEAN_STATE_REBUILDS_TOTAL + .with_label_values(&[role.label()]) + .inc(); +} + /// Count one state-cache lookup for `key`, made by `method`. pub(crate) fn inc_state_cache_lookups(method: StateCacheMethod, key: &CacheKey, hit: bool) { let kind = match key { diff --git a/crates/storage/src/state_writer.rs b/crates/storage/src/state_writer.rs index 51e5f6b0e..074d9b720 100644 --- a/crates/storage/src/state_writer.rs +++ b/crates/storage/src/state_writer.rs @@ -49,7 +49,7 @@ use std::borrow::Cow; use std::collections::HashMap; use std::sync::mpsc::{Receiver, SyncSender, sync_channel}; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Condvar, Mutex, PoisonError}; use std::thread::JoinHandle; use std::time::Instant; @@ -66,7 +66,7 @@ use tracing::error; use crate::api::{StorageBackend, StorageReadView, StorageReadViewExt, Table}; use crate::beacon_state_delta; use crate::error::Error; -use crate::metrics::StateCacheMethod; +use crate::metrics::{RebuildRole, StateCacheMethod}; use crate::state_codec::{decode_lean_state_value, decode_state_value, encode_state_value}; use crate::state_diff::StateDiff; use crate::store::{Chain, beacon_block_slot, block_slot, block_slot_in, encode_slot_root_key}; @@ -123,6 +123,131 @@ impl PendingStates { } } +/// Block-state rebuilds under way, keyed by block root, so that concurrent +/// [`read_state`] misses for one root share a single rebuild. +/// +/// A beacon rebuild reads a snapshot of over a hundred megabytes, folds a +/// delta into a second buffer of the same size and decodes a fresh registry: +/// about a second of CPU and several hundred megabytes at its peak. Callers +/// that miss on the same root at once used to each pay that in full. A late +/// block's custody columns are checked one sidecar per blocking task, so a +/// parent state that had left the cache was rebuilt once per column, eight at +/// a time, and the node's resident memory jumped by gigabytes for it. +/// +/// The first miss on a root leads: it registers the root, rebuilds, caches +/// the result, and only then deregisters the root and wakes the others. Every +/// other miss on that root in the meantime waits and takes the leader's +/// result, the same `Arc`. A miss that arrives after the leader deregistered +/// leads a rebuild of its own, but finds the state in the cache on +/// [`read_state`]'s re-check instead of reading it again. +/// +/// A leader that fails, with an `Err` or a panic out of one of the backend +/// reads' `.expect()`s, hands its waiters no result. Each of them retries, +/// one as the new leader, so a failure is never shared and never strands a +/// waiter. +#[derive(Default)] +pub(crate) struct StateRebuilds(Mutex>>); + +/// One rebuild under way: its outcome once there is one, and the condition +/// its waiters sleep on until then. +#[derive(Default)] +struct Rebuild { + outcome: Mutex, + settled: Condvar, +} + +#[derive(Default)] +enum RebuildOutcome { + /// The leader is still rebuilding. + #[default] + Running, + /// The leader's result: the state, or `None` for a root with no state. + Built(Option>), + /// The leader returned an error or unwound; a waiter retries. + Abandoned, +} + +impl StateRebuilds { + /// Runs `rebuild` for `root` as the leader, or waits for the rebuild + /// already under way for `root` and returns its result. + /// + /// `Fn`, not `FnOnce`: a caller whose leader failed retries, and may be + /// the one to lead the retry. + fn share( + &self, + root: H256, + rebuild: impl Fn() -> Result>, Error>, + ) -> Result>, Error> { + loop { + let (flight, leads) = { + let mut flights = self.0.lock().unwrap(); + match flights.get(&root) { + Some(flight) => (flight.clone(), false), + None => { + let flight = Arc::new(Rebuild::default()); + flights.insert(root, flight.clone()); + (flight, true) + } + } + }; + if leads { + // Settles on drop, so an unwinding `rebuild` still wakes the + // waiters, as `Abandoned`. + let mut lead = Lead { + rebuilds: self, + root, + flight, + outcome: RebuildOutcome::Abandoned, + }; + let result = rebuild(); + if let Ok(state) = &result { + lead.outcome = RebuildOutcome::Built(state.clone()); + } + return result; + } + let outcome = flight + .settled + .wait_while(flight.outcome.lock().unwrap(), |outcome| { + matches!(outcome, RebuildOutcome::Running) + }) + .unwrap(); + if let RebuildOutcome::Built(state) = &*outcome { + crate::metrics::inc_state_rebuilds(RebuildRole::Joined); + return Ok(state.clone()); + } + } + } +} + +/// The leader's hold on a [`Rebuild`]. Dropping it deregisters the root and +/// wakes the waiters, whether the rebuild returned or unwound. +struct Lead<'a> { + rebuilds: &'a StateRebuilds, + root: H256, + flight: Arc, + outcome: RebuildOutcome, +} + +impl Drop for Lead<'_> { + fn drop(&mut self) { + // Tolerates poisoning rather than unwrapping: this runs during the + // unwind of a panicking rebuild, where a second panic would abort. + // Neither lock is ever held across a rebuild, so poisoning is not + // expected here; this only keeps it from turning into an abort. + self.rebuilds + .0 + .lock() + .unwrap_or_else(PoisonError::into_inner) + .remove(&self.root); + *self + .flight + .outcome + .lock() + .unwrap_or_else(PoisonError::into_inner) = std::mem::take(&mut self.outcome); + self.flight.settled.notify_all(); + } +} + /// What a cached state is keyed by. /// /// One cache rather than two, so a single capacity bounds the total rather @@ -172,11 +297,15 @@ pub(crate) fn cache_get( /// The second half is what a single writer thread gives and a pool would not: /// a grandchild committed while its ancestor is still pending would make this /// walk return `Ok(None)` for a state that exists. +/// +/// A miss on both goes to the backend through `rebuilds`, so concurrent +/// misses on one root read and decode it once; see [`StateRebuilds`]. pub(crate) fn read_state( backend: &dyn StorageBackend, chain: Chain, cache: &StateCache, pending: &PendingStates, + rebuilds: &StateRebuilds, root: &H256, ) -> Result>, Error> { let key = CacheKey::BlockState(*root); @@ -186,7 +315,31 @@ pub(crate) fn read_state( if let Some(state) = pending.get(root) { return Ok(Some(state)); } + rebuilds.share(*root, || { + // Checked again under the lead: a rebuild that finished between the + // miss above and taking the lead has already cached its state, and a + // second read would be the very cost the lead exists to avoid. Not + // counted as a lookup, since the miss above already was. + if let Some(state) = cache.lock().unwrap().get(&key).cloned() { + return Ok(Some(state)); + } + if let Some(state) = pending.get(root) { + return Ok(Some(state)); + } + crate::metrics::inc_state_rebuilds(RebuildRole::Led); + rebuild_state(backend, chain, cache, root) + }) +} +/// Reads `root`'s state from the backend and caches it: [`read_state`]'s +/// path past the cache and the write buffer, which [`StateRebuilds`] runs for +/// one caller per root at a time. +fn rebuild_state( + backend: &dyn StorageBackend, + chain: Chain, + cache: &StateCache, + root: &H256, +) -> Result>, Error> { let state = match chain { Chain::Lean => { // Anchor snapshot in `States`, otherwise reconstruct from the diff chain. @@ -232,6 +385,9 @@ pub(crate) fn read_state( let mut state = state; state.apply_pending_mutations(); let state = Arc::new(state); + // Cached before the lead is released, which is what lets a miss arriving + // after the release find it on `read_state`'s re-check. + let key = CacheKey::BlockState(*root); cache.lock().unwrap().put(key, state.clone()); Ok(Some(state)) } @@ -600,6 +756,9 @@ struct StateWriter { chain: Chain, cache: Arc, pending: Arc, + /// Shared with the `Store`'s own reads, so a parent read here joins a + /// rebuild a reader already has under way rather than starting another. + rebuilds: Arc, /// Root and [`encode_state_value`] bytes of the snapshot the newest beacon /// diffs are taken against, so a diff does not re-read a snapshot of over a /// hundred megabytes on every import. @@ -650,6 +809,7 @@ impl StateWriter { self.chain, &self.cache, &self.pending, + &self.rebuilds, &parent_root, ) .expect("read parent state") @@ -792,6 +952,7 @@ impl StateWriterHandle { chain: Chain, cache: Arc, pending: Arc, + rebuilds: Arc, ) -> Self { let (tx, rx) = sync_channel(STATE_WRITE_QUEUE_CAPACITY); let writer = StateWriter { @@ -799,6 +960,7 @@ impl StateWriterHandle { chain, cache, pending, + rebuilds, snapshot_memo: None, last_base: None, }; @@ -890,6 +1052,8 @@ impl Drop for StateWriterHandle { #[cfg(test)] mod tests { use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + use ethlambda_types::block::BlockHeader; use ethlambda_types::primitives::HashTreeRoot as _; use ethlambda_types::state::{PUBLIC_KEY_SIZE, Validator}; @@ -1050,4 +1214,131 @@ mod tests { pending.remove(&root); assert!(pending.get(&root).is_none()); } + + /// Blocks until `waiters` callers other than the leader are waiting on + /// `root`'s rebuild: each holds a clone of its `Arc`, next to the + /// map's and the leader's own. + fn wait_for_waiters(rebuilds: &StateRebuilds, root: &H256, waiters: usize) { + let deadline = Instant::now() + std::time::Duration::from_secs(10); + loop { + let holders = rebuilds + .0 + .lock() + .unwrap() + .get(root) + .map_or(0, Arc::strong_count); + if holders == waiters + 2 { + return; + } + assert!( + Instant::now() < deadline, + "only {holders} holders of the rebuild, expected {}", + waiters + 2 + ); + std::thread::yield_now(); + } + } + + #[test] + fn concurrent_misses_on_one_root_share_one_rebuild() { + const WAITERS: usize = 7; + let rebuilds = StateRebuilds::default(); + let root = H256([5u8; 32]); + let built = Arc::new(BeaconState::Lean(base_state())); + let calls = AtomicUsize::new(0); + let rebuild = || { + calls.fetch_add(1, Ordering::SeqCst); + // Held until every other caller is waiting, so none of them can + // arrive after the lead is released and start a second rebuild. + wait_for_waiters(&rebuilds, &root, WAITERS); + Ok(Some(built.clone())) + }; + + let results: Vec<_> = std::thread::scope(|scope| { + let handles: Vec<_> = (0..=WAITERS) + .map(|_| scope.spawn(|| rebuilds.share(root, rebuild))) + .collect(); + handles.into_iter().map(|h| h.join().unwrap()).collect() + }); + + assert_eq!(calls.load(Ordering::SeqCst), 1, "one caller rebuilt"); + for result in results { + let state = result.expect("no error").expect("a state"); + assert!( + Arc::ptr_eq(&state, &built), + "every caller got the leader's own Arc, not a copy" + ); + } + assert!( + rebuilds.0.lock().unwrap().is_empty(), + "a settled rebuild deregisters its root" + ); + } + + #[test] + fn a_leader_that_panics_hands_the_rebuild_to_a_waiter() { + let rebuilds = StateRebuilds::default(); + let root = H256([6u8; 32]); + let built = Arc::new(BeaconState::Lean(base_state())); + let calls = AtomicUsize::new(0); + let rebuild = || { + if calls.fetch_add(1, Ordering::SeqCst) == 0 { + wait_for_waiters(&rebuilds, &root, 1); + panic!("the first rebuild fails"); + } + Ok(Some(built.clone())) + }; + + let results: Vec<_> = std::thread::scope(|scope| { + let handles: Vec<_> = (0..2) + .map(|_| scope.spawn(|| rebuilds.share(root, rebuild))) + .collect(); + handles.into_iter().map(|h| h.join()).collect() + }); + + assert_eq!( + calls.load(Ordering::SeqCst), + 2, + "the waiter rebuilt after the leader unwound" + ); + let (panicked, returned): (Vec<_>, Vec<_>) = results.into_iter().partition(Result::is_err); + assert_eq!(panicked.len(), 1, "the leader's own panic is its own"); + let state = returned + .into_iter() + .next() + .expect("the waiter returned") + .unwrap() + .expect("no error") + .expect("a state"); + assert!(Arc::ptr_eq(&state, &built)); + assert!(rebuilds.0.lock().unwrap().is_empty()); + } + + #[test] + fn a_leader_error_is_not_handed_to_its_waiters() { + let rebuilds = StateRebuilds::default(); + let root = H256([7u8; 32]); + let built = Arc::new(BeaconState::Lean(base_state())); + let calls = AtomicUsize::new(0); + let rebuild = || { + if calls.fetch_add(1, Ordering::SeqCst) == 0 { + wait_for_waiters(&rebuilds, &root, 1); + return Err(Error::UnexpectedMissingState(root)); + } + Ok(Some(built.clone())) + }; + + let results: Vec<_> = std::thread::scope(|scope| { + let handles: Vec<_> = (0..2) + .map(|_| scope.spawn(|| rebuilds.share(root, rebuild))) + .collect(); + handles.into_iter().map(|h| h.join().unwrap()).collect() + }); + + assert_eq!(calls.load(Ordering::SeqCst), 2, "the waiter retried"); + let (errors, states): (Vec<_>, Vec<_>) = results.into_iter().partition(Result::is_err); + assert_eq!(errors.len(), 1, "only the leader saw its error"); + let state = states[0].as_ref().unwrap().as_ref().expect("a state"); + assert!(Arc::ptr_eq(state, &built)); + } } diff --git a/crates/storage/src/store.rs b/crates/storage/src/store.rs index 14ac9116f..34adf7417 100644 --- a/crates/storage/src/store.rs +++ b/crates/storage/src/store.rs @@ -47,8 +47,8 @@ use libssz::{SszDecode, SszEncode}; use crate::metrics::StateCacheMethod; use crate::state_codec::encode_state_value; use crate::state_writer::{ - CacheKey, PendingStates, STATE_WRITE_QUEUE_CAPACITY, StateCache, StateWriteRequest, - StateWriterHandle, cache_get, read_state, + CacheKey, PendingStates, STATE_WRITE_QUEUE_CAPACITY, StateCache, StateRebuilds, + StateWriteRequest, StateWriterHandle, cache_get, read_state, }; use thiserror::Error; use tracing::{info, warn}; @@ -1170,6 +1170,10 @@ pub struct Store { /// States handed to the writer but not yet committed; see /// [`PendingStates`]. pending_states: Arc, + /// Block-state rebuilds under way, shared with the writer thread, so + /// concurrent misses on one root read it from storage once; see + /// [`StateRebuilds`]. + state_rebuilds: Arc, /// Committee shufflings shared across everything that asks a beacon state /// which validators attest at a slot: the state transition as it /// processes a block's attestations, fork choice as it replays those same @@ -1823,11 +1827,13 @@ impl Store { ) -> Self { let state_cache = new_state_cache(); let pending_states = Arc::new(PendingStates::default()); + let state_rebuilds = Arc::new(StateRebuilds::default()); let state_writer = Arc::new(StateWriterHandle::spawn( backend.clone(), chain, state_cache.clone(), pending_states.clone(), + state_rebuilds.clone(), )); Self { backend, @@ -1844,6 +1850,7 @@ impl Store { operation_pool: Default::default(), state_cache, pending_states, + state_rebuilds, committee_cache: Arc::new(CommitteeCache::default()), observed_liveness: Arc::new(ObservedLiveness::default()), active_balance_cache: Arc::new(ActiveBalanceCache::default()), @@ -3246,12 +3253,18 @@ impl Store { /// [`read_state`](crate::state_writer::read_state), which this calls /// directly rather than restating: one copy of the algorithm is one copy /// that can go stale. + /// + /// Safe to call from many threads for the same root at once: a miss + /// that finds a rebuild of `root` already under way waits for it and + /// returns its result rather than reading the state again. See + /// [`StateRebuilds`]. pub fn get_state(&self, root: &H256) -> Result>, Error> { read_state( self.backend.as_ref(), self.chain, &self.state_cache, &self.pending_states, + &self.state_rebuilds, root, ) } diff --git a/docs/data_storage.md b/docs/data_storage.md index 0f8d62db8..240cd6914 100644 --- a/docs/data_storage.md +++ b/docs/data_storage.md @@ -700,6 +700,14 @@ Reads go through `get_state`, which tries three levels: If the diff chain is broken or the target's header is missing, `get_state` returns `None` rather than a partial state. +Concurrent misses on one root rebuild it once (`StateRebuilds` in +`crates/storage/src/state_writer.rs`). The first caller leads. Every other +caller that misses on that root while the rebuild is under way waits for it +and gets the same `Arc`, so p2p's per-sidecar column checks, the chain actor +and the RPC asking for one evicted parent state cost one snapshot read and one +decode, not one each. A leader that errors or panics hands its waiters no +result; they retry, one of them as the new leader. + ## Write Paths: What a Block Import Persists Block import (`on_block` in `crates/blockchain/src/store.rs`) commits a diff --git a/docs/metrics.md b/docs/metrics.md index 9ac92fd4f..6881ed557 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -335,6 +335,7 @@ In practice the distribution is bimodal and dominated by production rather than | `lean_state_write_queue_depth` | Gauge | States handed to the background writer but not yet committed | On every hand-off and on every commit | | | `lean_state_write_seconds` | Histogram | Time the background writer spends encoding, diffing and committing one state | Per state written | | | `lean_state_cache_lookups_total` | Counter | State cache lookups, by the `Store` method that made them, the kind of state asked for, and whether the cache held it | On every `get_state` (and the writer's own parent read), `has_state` and `cached_state` call | method=get_state,has_state,cached_state
kind=block,checkpoint
result=hit,miss | +| `lean_state_rebuilds_total` | Counter | Block states read back from storage after missing the cache and the write buffer, by whether the read did the rebuild or waited on one already under way for that root | On every `get_state` miss that reaches storage | role=led,joined | **On a beacon follower, watch `lean_table_bytes{table="data_columns"}`.** Every other table's series either stays flat or is bounded by pruning; `data_columns` @@ -363,6 +364,14 @@ buffer or RocksDB. A `cached_state` miss with `kind="checkpoint"` means fork choice's `checkpoint_state` rebuilt the state, running `process_slots` up to the epoch boundary when the checkpoint's block precedes it. +**`lean_state_rebuilds_total{role="led"}` is the number of states actually +read and decoded from storage.** A beacon rebuild costs about a second of CPU +and several hundred megabytes at its peak, so this is the series to watch for +resident-memory spikes. `role="joined"` counts misses that found a rebuild of +the same root already under way and waited for it instead. A burst of joins +next to a single lead is the pattern a late block's custody columns produce, +each checked on its own task against a parent state that has left the cache. + ### Beacon Gossip Validation Every beacon gossip message gets a verdict before gossipsub forwards it From 622dd5143755fafc68dca43d8fe53d877206f368 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Wed, 11 Nov 2026 12:26:43 -0300 Subject: [PATCH 2/6] refactor(storage): share state rebuilds through a OnceLock per root OnceLock::get_or_init already blocks concurrent callers until one initializer finishes and hands them all its value, and a panicking initializer leaves the cell for a blocked caller to fill. That replaces the Condvar, the outcome enum, the drop guard and the retry loop. The rebuild path has no Err return today (every failure is Ok(None) or an .expect() panic), so the cell stores Err(()) for an error instead of sharing a non-Clone Error, and a caller that finds it reads the state itself. --- crates/storage/src/state_writer.rs | 140 +++++++++-------------------- docs/data_storage.md | 13 +-- 2 files changed, 50 insertions(+), 103 deletions(-) diff --git a/crates/storage/src/state_writer.rs b/crates/storage/src/state_writer.rs index 074d9b720..6b3b536cc 100644 --- a/crates/storage/src/state_writer.rs +++ b/crates/storage/src/state_writer.rs @@ -49,7 +49,7 @@ use std::borrow::Cow; use std::collections::HashMap; use std::sync::mpsc::{Receiver, SyncSender, sync_channel}; -use std::sync::{Arc, Condvar, Mutex, PoisonError}; +use std::sync::{Arc, Mutex, OnceLock}; use std::thread::JoinHandle; use std::time::Instant; @@ -134,120 +134,66 @@ impl PendingStates { /// parent state that had left the cache was rebuilt once per column, eight at /// a time, and the node's resident memory jumped by gigabytes for it. /// -/// The first miss on a root leads: it registers the root, rebuilds, caches -/// the result, and only then deregisters the root and wakes the others. Every -/// other miss on that root in the meantime waits and takes the leader's -/// result, the same `Arc`. A miss that arrives after the leader deregistered -/// leads a rebuild of its own, but finds the state in the cache on -/// [`read_state`]'s re-check instead of reading it again. -/// -/// A leader that fails, with an `Err` or a panic out of one of the backend -/// reads' `.expect()`s, hands its waiters no result. Each of them retries, -/// one as the new leader, so a failure is never shared and never strands a -/// waiter. +/// Each root under way has a [`OnceLock`]. The first miss runs the rebuild +/// inside `get_or_init`; every other miss on that root blocks in +/// `get_or_init` until it is done and takes the same `Arc`. A rebuild that +/// panics leaves its cell empty, and `OnceLock` lets one of the blocked +/// callers run its own rebuild instead, so a panic never strands a waiter. +/// The entry is removed once the rebuild settles, so the map holds rebuilds +/// under way and never pins a state. #[derive(Default)] -pub(crate) struct StateRebuilds(Mutex>>); +pub(crate) struct StateRebuilds(Mutex>>); -/// One rebuild under way: its outcome once there is one, and the condition -/// its waiters sleep on until then. -#[derive(Default)] -struct Rebuild { - outcome: Mutex, - settled: Condvar, -} - -#[derive(Default)] -enum RebuildOutcome { - /// The leader is still rebuilding. - #[default] - Running, - /// The leader's result: the state, or `None` for a root with no state. - Built(Option>), - /// The leader returned an error or unwound; a waiter retries. - Abandoned, -} +/// One rebuild's shared result. `Err(())` when the rebuild returned an error, +/// which is not shared (`Error` is not `Clone`): a caller that finds it reads +/// the state itself. +type RebuildCell = OnceLock>, ()>>; impl StateRebuilds { - /// Runs `rebuild` for `root` as the leader, or waits for the rebuild - /// already under way for `root` and returns its result. + /// Runs `rebuild` for `root`, or waits for the rebuild of `root` already + /// under way and returns its result. /// - /// `Fn`, not `FnOnce`: a caller whose leader failed retries, and may be - /// the one to lead the retry. + /// `Fn`, not `FnOnce`: a caller that finds a failed rebuild runs its own. fn share( &self, root: H256, rebuild: impl Fn() -> Result>, Error>, ) -> Result>, Error> { - loop { - let (flight, leads) = { - let mut flights = self.0.lock().unwrap(); - match flights.get(&root) { - Some(flight) => (flight.clone(), false), - None => { - let flight = Arc::new(Rebuild::default()); - flights.insert(root, flight.clone()); - (flight, true) - } - } - }; - if leads { - // Settles on drop, so an unwinding `rebuild` still wakes the - // waiters, as `Abandoned`. - let mut lead = Lead { - rebuilds: self, - root, - flight, - outcome: RebuildOutcome::Abandoned, - }; + let cell = self.0.lock().unwrap().entry(root).or_default().clone(); + let mut own = None; + let shared = cell + .get_or_init(|| { let result = rebuild(); - if let Ok(state) = &result { - lead.outcome = RebuildOutcome::Built(state.clone()); - } - return result; + let shared = result.as_ref().map(Option::clone).map_err(|_| ()); + own = Some(result); + shared + }) + .clone(); + // The first caller out removes the cell. A later miss can then + // register a fresh one for `root`, which a caller still leaving this + // cell must not remove. A rebuild that panicked with no one waiting + // skips this and leaves an empty cell, which the next miss on `root` + // fills and removes. + { + let mut cells = self.0.lock().unwrap(); + if cells + .get(&root) + .is_some_and(|current| Arc::ptr_eq(current, &cell)) + { + cells.remove(&root); } - let outcome = flight - .settled - .wait_while(flight.outcome.lock().unwrap(), |outcome| { - matches!(outcome, RebuildOutcome::Running) - }) - .unwrap(); - if let RebuildOutcome::Built(state) = &*outcome { + } + match (own, shared) { + (Some(result), _) => result, + (None, Ok(state)) => { crate::metrics::inc_state_rebuilds(RebuildRole::Joined); - return Ok(state.clone()); + Ok(state) } + (None, Err(())) => rebuild(), } } } -/// The leader's hold on a [`Rebuild`]. Dropping it deregisters the root and -/// wakes the waiters, whether the rebuild returned or unwound. -struct Lead<'a> { - rebuilds: &'a StateRebuilds, - root: H256, - flight: Arc, - outcome: RebuildOutcome, -} - -impl Drop for Lead<'_> { - fn drop(&mut self) { - // Tolerates poisoning rather than unwrapping: this runs during the - // unwind of a panicking rebuild, where a second panic would abort. - // Neither lock is ever held across a rebuild, so poisoning is not - // expected here; this only keeps it from turning into an abort. - self.rebuilds - .0 - .lock() - .unwrap_or_else(PoisonError::into_inner) - .remove(&self.root); - *self - .flight - .outcome - .lock() - .unwrap_or_else(PoisonError::into_inner) = std::mem::take(&mut self.outcome); - self.flight.settled.notify_all(); - } -} - /// What a cached state is keyed by. /// /// One cache rather than two, so a single capacity bounds the total rather diff --git a/docs/data_storage.md b/docs/data_storage.md index 240cd6914..45ceff792 100644 --- a/docs/data_storage.md +++ b/docs/data_storage.md @@ -701,12 +701,13 @@ If the diff chain is broken or the target's header is missing, `get_state` returns `None` rather than a partial state. Concurrent misses on one root rebuild it once (`StateRebuilds` in -`crates/storage/src/state_writer.rs`). The first caller leads. Every other -caller that misses on that root while the rebuild is under way waits for it -and gets the same `Arc`, so p2p's per-sidecar column checks, the chain actor -and the RPC asking for one evicted parent state cost one snapshot read and one -decode, not one each. A leader that errors or panics hands its waiters no -result; they retry, one of them as the new leader. +`crates/storage/src/state_writer.rs`, a `OnceLock` per root under way). The +first caller rebuilds. Every other caller that misses on that root while the +rebuild is under way waits for it and gets the same `Arc`, so p2p's +per-sidecar column checks, the chain actor and the RPC asking for one evicted +parent state cost one snapshot read and one decode, not one each. A rebuild +that fails hands the waiting callers no result: after a panic one of them +rebuilds for the rest, and after an error each reads the state itself. ## Write Paths: What a Block Import Persists From 0bbdec3f6edc3f69add924975e92dcbeded7fb11 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Wed, 11 Nov 2026 12:52:57 -0300 Subject: [PATCH 3/6] fix(storage): return state rebuild failures as errors A state rebuild that could not read or decode what it found used to panic through an `.expect()`. It now returns `Error::Storage` for a failed read and `Error::CorruptState` for a snapshot, delta, lean diff or header that does not decode, so a caller can give up on one request instead of unwinding. `StateRebuilds` hands that error to every caller waiting on the rebuild, as `Error::StateRebuild`, so a failing root is read once rather than once per caller. Neither a state nor an error outlives the rebuild: the next miss reads storage again. The fulu column check used to treat an `Err` like a missing parent and park the sidecar. It now logs the error and drops the sidecar as `Ignore(Internal)`, since parking would wait on an import that has already happened. --- .../src/beacon/gossip/column.rs | 75 ++++++- .../state_transition/src/beacon/gossip/mod.rs | 2 +- crates/net/p2p/src/beacon/column_checks.rs | 8 +- crates/storage/src/beacon_state_delta.rs | 29 ++- crates/storage/src/error.rs | 14 ++ crates/storage/src/state_codec.rs | 35 +-- crates/storage/src/state_writer.rs | 209 +++++++++++------- crates/storage/src/store.rs | 49 +++- docs/data_storage.md | 17 +- 9 files changed, 310 insertions(+), 128 deletions(-) diff --git a/crates/blockchain/state_transition/src/beacon/gossip/column.rs b/crates/blockchain/state_transition/src/beacon/gossip/column.rs index 900d83353..48075e0c1 100644 --- a/crates/blockchain/state_transition/src/beacon/gossip/column.rs +++ b/crates/blockchain/state_transition/src/beacon/gossip/column.rs @@ -11,6 +11,8 @@ //! again, so these are the only checks such a sidecar gets. use ethlambda_storage::CacheKey; +use ethlambda_types::ShortRoot; +use tracing::error; use super::{ IgnoreReason, Outcome, QueueReason, RejectReason, SeenBlockColumns, SeenColumns, @@ -162,15 +164,33 @@ pub fn chain_checks(store: &Store, sidecar: &DataColumnSidecar, now_ms: u64) -> if store.has_data_column(header.slot, &header.hash_tree_root(), sidecar.index) { return ChainVerdict::Drop(Outcome::Ignore(IgnoreReason::AlreadyStored)); } - let Ok(Some(parent_state)) = store.get_state(&header.parent_root) else { + let parent_state = match store.get_state(&header.parent_root) { + Ok(Some(parent_state)) => parent_state, // Parked rather than dropped: the specification's "MAY be queued" // for a sidecar whose parent has not been seen. The head-state // signature check keeps a forged header from taking a // `PendingDataColumns` row while it waits. - return match queue_unless_forged(store, sidecar, QueueReason::ParentUnknown) { - Outcome::Queue(_) => ChainVerdict::AwaitParent, - outcome => ChainVerdict::Drop(outcome), - }; + Ok(None) => { + return match queue_unless_forged(store, sidecar, QueueReason::ParentUnknown) { + Outcome::Queue(_) => ChainVerdict::AwaitParent, + outcome => ChainVerdict::Drop(outcome), + }; + } + // The parent's state is stored but could not be read back, so + // parking would wait on an import that has already happened. Dropped + // instead, as a panic in this read used to drop it. + Err(err) => { + let slot = header.slot; + let column = sidecar.index; + error!( + slot, + column, + parent_root = %ShortRoot(&header.parent_root.0), + %err, + "Could not read a data column's parent state; dropping the column" + ); + return ChainVerdict::Drop(Outcome::Ignore(IgnoreReason::Internal)); + } }; match judge_against_parent(store, sidecar, &parent_state, PastLookahead::Advance) { Outcome::Accept => ChainVerdict::Keep, @@ -599,13 +619,19 @@ fn header_signature_is_valid( mod tests { use std::sync::Arc; + use ethlambda_storage::backend::InMemoryBackend; + use ethlambda_storage::{StorageBackend, Table}; + use ethlambda_types::checkpoint::Checkpoint; + use super::*; use crate::beacon::constants::{ DATA_COLUMN_SIDECAR_SUBNET_COUNT, MAXIMUM_GOSSIP_CLOCK_DISPARITY as DISPARITY, }; use crate::beacon::containers::{SignedBeaconBlock, electra, fulu, shared}; use crate::beacon::fork::ForkName; - use crate::beacon::gossip::test_support::{fulu_parent, seen_columns, slot_start_ms, store}; + use crate::beacon::gossip::test_support::{ + GENESIS_TIME, fulu_parent, seen_columns, slot_start_ms, store, + }; use crate::beacon::gossip::{IgnoreReason, Outcome, QueueReason, RejectReason}; use crate::beacon::helpers::misc::compute_start_slot_at_epoch; use crate::beacon::helpers::test_state::secret_key_for; @@ -1073,6 +1099,43 @@ mod tests { ); } + /// A parent whose state is stored but does not read back is not waiting + /// on an import, so the sidecar is dropped rather than parked, and the + /// failed read is an error rather than a panic. + #[test] + fn chain_checks_drop_a_sidecar_whose_parent_state_does_not_read_back() { + let backend = Arc::new(InMemoryBackend::new()); + let mut store = Store::init_beacon( + backend.clone(), + GENESIS_TIME, + Config::mainnet().with_fork_epoch(ForkName::Fulu, 0), + Root::ZERO, + Checkpoint { + root: Root::ZERO, + slot: 0, + }, + 0, + ); + let parent_slot = 4; + let card = sidecar(parent_slot + 1, 1, 0); + let parent_root = card.signed_block_header.message.parent_root; + store + .insert_pending_block(parent_root, parent_block(parent_slot)) + .expect("insert pending parent"); + // `States` is keyed by the block's slot (big-endian), then its root. + // A fork selector with no state after it decodes as no fork's state. + let key = [parent_slot.to_be_bytes().as_slice(), parent_root.as_slice()].concat(); + let corrupt = vec![(key, vec![ForkName::Fulu.selector()])]; + let mut batch = backend.begin_write().expect("write batch"); + batch.put_batch(Table::States, corrupt).expect("put state"); + batch.commit().expect("commit"); + + assert_eq!( + chain_checks(&store, &card, slot_start_ms(&store, parent_slot + 1)), + ChainVerdict::Drop(Outcome::Ignore(IgnoreReason::Internal)) + ); + } + /// The head-state signature check gossip runs before queuing runs before /// parking here too, so a forged header takes no `PendingDataColumns` row. #[test] diff --git a/crates/blockchain/state_transition/src/beacon/gossip/mod.rs b/crates/blockchain/state_transition/src/beacon/gossip/mod.rs index 701fa705c..355302fa6 100644 --- a/crates/blockchain/state_transition/src/beacon/gossip/mod.rs +++ b/crates/blockchain/state_transition/src/beacon/gossip/mod.rs @@ -124,7 +124,7 @@ pub enum IgnoreReason { NoConsumer, /// Every stateful-validation permit was taken. Overloaded, - /// Validation panicked. + /// Validation panicked, or a store read it needed failed. Internal, /// The message's own fork is one this build does not validate or decode /// (currently gloas): not the sender's fault, and not malformed, so this diff --git a/crates/net/p2p/src/beacon/column_checks.rs b/crates/net/p2p/src/beacon/column_checks.rs index 91fbd1efe..cb6029347 100644 --- a/crates/net/p2p/src/beacon/column_checks.rs +++ b/crates/net/p2p/src/beacon/column_checks.rs @@ -60,9 +60,11 @@ pub(crate) fn check_and_forward(server: &P2PServer, sidecars: Vec checked, Err(err) => { diff --git a/crates/storage/src/beacon_state_delta.rs b/crates/storage/src/beacon_state_delta.rs index d03d8a88d..46ca0f987 100644 --- a/crates/storage/src/beacon_state_delta.rs +++ b/crates/storage/src/beacon_state_delta.rs @@ -21,7 +21,7 @@ //! `insert_state`/`get_state` (in `store.rs`) call into this module for the //! beacon arm: `insert_state` writes a snapshot at anchors and a //! [`frame`]d [`encode`] otherwise, and `get_state` follows the frame's base -//! and [`decode`]s it back, via [`unframe`]. +//! and decodes it back ([`try_decode`]), via [`unframe`]. //! //! The base of a delta is the nearest snapshot on the state's branch, not its //! parent. A read is then one snapshot plus one delta, whatever the state's @@ -63,25 +63,30 @@ pub(crate) fn encode(target: &[u8], base: &[u8]) -> Vec { }) } -/// The inverse of [`encode`]. +/// The inverse of [`encode`], for a delta known to be sound: the writer's own +/// round trip and tests. Panics where [`try_decode`] returns `Err`. +pub(crate) fn decode(delta: &[u8], base: &[u8], target_len: usize) -> Vec { + try_decode(delta, base, target_len).unwrap_or_else(|err| panic!("{err}")) +} + +/// The inverse of [`encode`], for a delta read back from storage. /// /// `target_len` comes from the frame, so the output buffer is sized exactly -/// rather than at `(delta + base) * 2`. Panics on a delta this build cannot +/// rather than at `(delta + base) * 2`. `Err` on a delta this build cannot /// apply: `from_db_state` has already rejected a directory of the wrong -/// format version, so anything reaching here is corruption rather than an old -/// database, which matches how every other read in this crate treats a bad -/// value. +/// format version, so anything failing here is corruption rather than an old +/// database, and the state read reports it rather than panicking. /// /// A `target_len` of zero is [`encode`]'s empty-target case; the empty /// output stands in directly for the same reason encode short-circuits it. -pub(crate) fn decode(delta: &[u8], base: &[u8], target_len: usize) -> Vec { +pub(crate) fn try_decode(delta: &[u8], base: &[u8], target_len: usize) -> Result, String> { if target_len == 0 { - return Vec::new(); + return Ok(Vec::new()); } - let output_buffer_len = - u32::try_from(target_len).expect("target length fits a u32 at beacon scale"); - xdelta3::decode_with_output_len(delta, base, output_buffer_len).unwrap_or_else(|err| { - panic!("xdelta3 decode failed against exact target length {target_len}: {err:?}") + let output_buffer_len = u32::try_from(target_len) + .map_err(|_| format!("target length {target_len} does not fit a u32"))?; + xdelta3::decode_with_output_len(delta, base, output_buffer_len).map_err(|err| { + format!("xdelta3 decode failed against exact target length {target_len}: {err:?}") }) } diff --git a/crates/storage/src/error.rs b/crates/storage/src/error.rs index 7e35c2709..1c4b00341 100644 --- a/crates/storage/src/error.rs +++ b/crates/storage/src/error.rs @@ -1,3 +1,5 @@ +use std::sync::Arc; + use ethlambda_types::checkpoint::Checkpoint; use ethlambda_types::primitives::H256; @@ -9,6 +11,18 @@ pub enum Error { UnexpectedMissingBlockHeader(H256), #[error("unexpected missing state for root {0}")] UnexpectedMissingState(H256), + /// A stored state did not decode, or a delta on the way to it did not + /// apply to its base. [`Store::from_db_state`](crate::store::Store::from_db_state) + /// has already refused a directory of another format version, so this is + /// corruption at rest rather than an old database. + #[error("the stored state for root {root} is corrupt: {reason}")] + CorruptState { root: H256, reason: String }, + /// A state rebuild's error, handed to every concurrent read of the same + /// root that waited on that rebuild rather than reading the state itself + /// (`StateRebuilds` in `state_writer.rs`). Reference-counted because + /// `Error` is not `Clone`; it displays as the error it wraps. + #[error(transparent)] + StateRebuild(Arc), /// The data directory was written by a build with a different on-disk /// format. There is no migration: the `States` value layout changed, so /// every state already written would decode as the wrong shape. diff --git a/crates/storage/src/state_codec.rs b/crates/storage/src/state_codec.rs index 3e0255d89..d457866c1 100644 --- a/crates/storage/src/state_codec.rs +++ b/crates/storage/src/state_codec.rs @@ -32,26 +32,29 @@ pub(crate) fn encode_state_value(state: &BeaconState) -> Vec { /// The inverse of [`encode_state_value`]. /// -/// Panics on a value this build cannot tag-decode, matching every other state -/// and block read in the crate: +/// `Err` names what is wrong with a value this build cannot tag-decode. /// [`Store::from_db_state`](crate::store::Store::from_db_state) has already -/// rejected a directory of the wrong format version, so anything reaching -/// here is corruption rather than an old database. -pub(crate) fn decode_state_value(bytes: &[u8]) -> BeaconState { - let (tag, ssz) = bytes.split_first().expect("value is never empty"); - let fork = ForkName::from_selector(*tag).expect("value carries a known fork selector"); - BeaconState::from_ssz(fork, ssz).expect("valid state value") +/// rejected a directory of the wrong format version, so anything failing +/// here is corruption rather than an old database. The caller knows which +/// root it was reading and turns this into +/// [`Error::CorruptState`](crate::error::Error::CorruptState). +pub(crate) fn decode_state_value(bytes: &[u8]) -> Result { + let (tag, ssz) = bytes.split_first().ok_or("the value is empty")?; + let fork = + ForkName::from_selector(*tag).ok_or_else(|| format!("unknown fork selector {tag}"))?; + BeaconState::from_ssz(fork, ssz).map_err(|err| format!("not a valid {fork} state: {err}")) } /// [`decode_state_value`] for the lean reader, which has no beacon shape to do -/// anything with. -pub(crate) fn decode_lean_state_value(bytes: &[u8]) -> State { - match decode_state_value(bytes) { - BeaconState::Lean(state) => state, - beacon => panic!( - "lean read a {} state out of the States table; a data directory holds one chain", +/// anything with: a beacon value in a lean directory is as corrupt as one +/// that does not decode, since a data directory holds one chain. +pub(crate) fn decode_lean_state_value(bytes: &[u8]) -> Result { + match decode_state_value(bytes)? { + BeaconState::Lean(state) => Ok(state), + beacon => Err(format!( + "a {} state in a lean directory", beacon.fork_name() - ), + )), } } @@ -63,7 +66,7 @@ mod tests { fn a_tagged_state_value_round_trips() { let state = BeaconState::Lean(State::from_genesis(7, vec![])); let bytes = encode_state_value(&state); - assert_eq!(decode_state_value(&bytes), state); + assert_eq!(decode_state_value(&bytes), Ok(state)); } #[test] diff --git a/crates/storage/src/state_writer.rs b/crates/storage/src/state_writer.rs index 6b3b536cc..0cc3c2bab 100644 --- a/crates/storage/src/state_writer.rs +++ b/crates/storage/src/state_writer.rs @@ -69,7 +69,9 @@ use crate::error::Error; use crate::metrics::{RebuildRole, StateCacheMethod}; use crate::state_codec::{decode_lean_state_value, decode_state_value, encode_state_value}; use crate::state_diff::StateDiff; -use crate::store::{Chain, beacon_block_slot, block_slot, block_slot_in, encode_slot_root_key}; +use crate::store::{ + Chain, beacon_block_slot, block_slot_in, encode_slot_root_key, try_block_slot_in, +}; /// What a state insert will write, decided but not yet committed. /// @@ -136,37 +138,33 @@ impl PendingStates { /// /// Each root under way has a [`OnceLock`]. The first miss runs the rebuild /// inside `get_or_init`; every other miss on that root blocks in -/// `get_or_init` until it is done and takes the same `Arc`. A rebuild that -/// panics leaves its cell empty, and `OnceLock` lets one of the blocked -/// callers run its own rebuild instead, so a panic never strands a waiter. -/// The entry is removed once the rebuild settles, so the map holds rebuilds -/// under way and never pins a state. +/// `get_or_init` until it is done and takes the same result: the same `Arc`, +/// or the same error. A rebuild that panics leaves its cell empty, and +/// `OnceLock` lets one of the blocked callers run its own rebuild instead, so +/// a panic never strands a waiter. The entry is removed once the rebuild +/// settles, so the map holds rebuilds under way, never pins a state, and +/// never keeps an error for a later miss: that one reads storage again. #[derive(Default)] pub(crate) struct StateRebuilds(Mutex>>); -/// One rebuild's shared result. `Err(())` when the rebuild returned an error, -/// which is not shared (`Error` is not `Clone`): a caller that finds it reads -/// the state itself. -type RebuildCell = OnceLock>, ()>>; +/// One rebuild's shared result. The error is behind an `Arc` because `Error` +/// is not `Clone`; every caller gets it as [`Error::StateRebuild`]. +type RebuildCell = OnceLock>, Arc>>; impl StateRebuilds { /// Runs `rebuild` for `root`, or waits for the rebuild of `root` already /// under way and returns its result. - /// - /// `Fn`, not `FnOnce`: a caller that finds a failed rebuild runs its own. fn share( &self, root: H256, - rebuild: impl Fn() -> Result>, Error>, + rebuild: impl FnOnce() -> Result>, Error>, ) -> Result>, Error> { let cell = self.0.lock().unwrap().entry(root).or_default().clone(); - let mut own = None; - let shared = cell + let mut led = false; + let result = cell .get_or_init(|| { - let result = rebuild(); - let shared = result.as_ref().map(Option::clone).map_err(|_| ()); - own = Some(result); - shared + led = true; + rebuild().map_err(Arc::new) }) .clone(); // The first caller out removes the cell. A later miss can then @@ -183,14 +181,10 @@ impl StateRebuilds { cells.remove(&root); } } - match (own, shared) { - (Some(result), _) => result, - (None, Ok(state)) => { - crate::metrics::inc_state_rebuilds(RebuildRole::Joined); - Ok(state) - } - (None, Err(())) => rebuild(), + if !led { + crate::metrics::inc_state_rebuilds(RebuildRole::Joined); } + result.map_err(Error::StateRebuild) } } @@ -280,6 +274,10 @@ pub(crate) fn read_state( /// Reads `root`'s state from the backend and caches it: [`read_state`]'s /// path past the cache and the write buffer, which [`StateRebuilds`] runs for /// one caller per root at a time. +/// +/// `Err` when a read fails ([`Error::Storage`]) or what it read does not +/// decode ([`Error::CorruptState`]), rather than a panic, so that a caller +/// such as a gossip check can give up on one message instead of unwinding. fn rebuild_state( backend: &dyn StorageBackend, chain: Chain, @@ -291,14 +289,15 @@ fn rebuild_state( // Anchor snapshot in `States`, otherwise reconstruct from the diff chain. // `States` is keyed by block slot then root, so a root with no // block row has no state to find. - let Some(slot) = block_slot(backend, chain, root) else { - return Ok(None); - }; let snapshot = { - let view = backend.begin_read().expect("read view"); + let view = backend.begin_read()?; + let Some(slot) = try_block_slot_in(view.as_ref(), chain, root)? else { + return Ok(None); + }; let key = encode_slot_root_key(slot, root); - view.read_with(Table::States, &key, decode_lean_state_value) - .expect("read") + view.read_with(Table::States, &key, decode_lean_state_value)? + .transpose() + .map_err(|reason| corrupt_state(root, reason))? }; let state = match snapshot { Some(state) => state, @@ -316,9 +315,10 @@ fn rebuild_state( // decode per hop instead would be the whole cost that delta layer // 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 { + let decode = |bytes: Cow<'_, [u8]>| { + decode_state_value(&bytes).map_err(|reason| corrupt_state(root, reason)) + }; + let Some(mut state) = reconstruct_beacon_state_bytes(backend, root, decode)? else { return Ok(None); }; rebase_onto_resident(cache, &mut state); @@ -338,6 +338,14 @@ fn rebuild_state( Ok(Some(state)) } +/// [`Error::CorruptState`] for the state of `root`, from a decoder's reason. +fn corrupt_state(root: &H256, reason: String) -> Error { + Error::CorruptState { + root: *root, + reason, + } +} + /// Makes a state just decoded from storage share memory with a resident one. /// /// A decoded state's tree-backed fields (see `BeaconState::rebase_on`) are @@ -403,36 +411,39 @@ fn rebase_onto_resident(cache: &StateCache, state: &mut BeaconState) { /// [`Store::prune_finalized_states`](crate::store::Store::prune_finalized_states)), /// matching how the lean walk in [`reconstruct_state`] handles both cases. A /// hop whose block row is absent ends the walk the same way. +/// +/// `Err` when a read fails, when a delta does not apply to its base +/// ([`Error::CorruptState`]), or when `consume` fails. A frame too short to +/// hold its own header still panics in [`beacon_state_delta::unframe`]: no +/// store writes one. pub(crate) fn reconstruct_beacon_state_bytes( backend: &dyn StorageBackend, root: &H256, - consume: impl FnOnce(Cow<'_, [u8]>) -> T, + consume: impl FnOnce(Cow<'_, [u8]>) -> Result, ) -> Result, Error> { - let view = backend.begin_read().expect("read view"); + let view = backend.begin_read()?; let mut records: Vec> = Vec::new(); let mut consume = Some(consume); let mut cursor = *root; loop { - let Some(slot) = block_slot_in(view.as_ref(), Chain::Beacon, &cursor) else { + let Some(slot) = try_block_slot_in(view.as_ref(), Chain::Beacon, &cursor)? else { return Ok(None); }; let key = encode_slot_root_key(slot, &cursor); // 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 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(root, snapshot, &records, consume) + })?; + if let Some(folded) = folded { + return folded.map(Some); } - let Some(diff_bytes) = view.get(Table::StateDiffs, &key).expect("get") else { + let Some(diff_bytes) = view.get(Table::StateDiffs, &key)? else { return Ok(None); }; let (base_root, _, _, _) = beacon_state_delta::unframe(&diff_bytes); @@ -479,21 +490,26 @@ pub(crate) fn nearest_snapshot_root(view: &dyn StorageReadView, root: &H256) -> /// 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. +/// to apply, otherwise the owned output of the last delta. A delta that does +/// not apply is [`Error::CorruptState`] for `root`, the state being read. fn fold_beacon_state_deltas( + root: &H256, snapshot: &[u8], records: &[Vec], - consume: impl FnOnce(Cow<'_, [u8]>) -> T, -) -> T { + consume: impl FnOnce(Cow<'_, [u8]>) -> Result, +) -> Result { + let apply = |record: &[u8], base: &[u8]| { + let (_, _, target_len, delta) = beacon_state_delta::unframe(record); + beacon_state_delta::try_decode(delta, base, target_len as usize) + .map_err(|reason| corrupt_state(root, reason)) + }; 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); + let mut bytes = apply(first, snapshot)?; for record in records { - let (_, _, target_len, delta) = beacon_state_delta::unframe(record); - bytes = beacon_state_delta::decode(delta, &bytes, target_len as usize); + bytes = apply(record, &bytes)?; } consume(Cow::Owned(bytes)) } @@ -507,39 +523,41 @@ fn fold_beacon_state_deltas( /// Lean directories only: the inlined `BlockHeaders` read below assumes a /// bare [`BlockHeader`], which is what a lean directory stores there. A /// beacon directory tags that table with a fork selector ahead of a -/// `SignedBeaconBlock`, so calling this on one would decode the wrong shape -/// and hit the `expect("valid header")` below instead of an explicit panic. +/// `SignedBeaconBlock`, so calling this on one would fail decoding the wrong +/// shape (the slot lookup's own header decode panics first) rather than +/// refuse explicitly. /// Nothing does today: [`read_state`]'s `Chain::Lean` arm is this function's /// only caller, so that invariant is enforced by having exactly one caller /// rather than by a runtime check, unlike /// [`Store::get_block_header`](crate::store::Store::get_block_header)'s /// `lean_only` guard. /// -/// Returns `Ok(None)` when the root is unknown or the diff chain is broken. +/// Returns `Ok(None)` when the root is unknown or the diff chain is broken, +/// and `Err` when a read fails or a row on the way does not decode. pub(crate) fn reconstruct_state( backend: &dyn StorageBackend, root: &H256, ) -> Result, Error> { // Walk back collecting diffs until we reach a snapshot. - let view = backend.begin_read().expect("read view"); + let view = backend.begin_read()?; let mut diffs: Vec = Vec::new(); let mut cursor = *root; let snapshot = loop { - let Some(slot) = block_slot_in(view.as_ref(), Chain::Lean, &cursor) else { + let Some(slot) = try_block_slot_in(view.as_ref(), Chain::Lean, &cursor)? else { return Ok(None); }; let key = encode_slot_root_key(slot, &cursor); if let Some(snapshot) = view - .read_with(Table::States, &key, decode_lean_state_value) - .expect("read") + .read_with(Table::States, &key, decode_lean_state_value)? + .transpose() + .map_err(|reason| corrupt_state(root, reason))? { break snapshot; } let Some(diff) = view - .read_with(Table::StateDiffs, &key, |bytes| { - StateDiff::from_ssz_bytes(bytes).expect("valid state diff") - }) - .expect("read") + .read_with(Table::StateDiffs, &key, StateDiff::from_ssz_bytes)? + .transpose() + .map_err(|err| corrupt_state(root, format!("a state diff does not decode: {err:?}")))? else { return Ok(None); }; @@ -553,12 +571,15 @@ 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 view = backend.begin_read()?; let header = view - .read_with(Table::BlockHeaders, &root.to_ssz(), |bytes| { - BlockHeader::from_ssz_bytes(bytes).expect("valid header") - }) - .expect("read"); + .read_with( + Table::BlockHeaders, + &root.to_ssz(), + BlockHeader::from_ssz_bytes, + )? + .transpose() + .map_err(|err| corrupt_state(root, format!("its block header does not decode: {err:?}")))?; drop(view); let Some(latest_block_header) = header else { return Ok(None); @@ -1162,8 +1183,8 @@ mod tests { } /// Blocks until `waiters` callers other than the leader are waiting on - /// `root`'s rebuild: each holds a clone of its `Arc`, next to the - /// map's and the leader's own. + /// `root`'s rebuild: each holds a clone of its `Arc`, next to + /// the map's and the leader's own. fn wait_for_waiters(rebuilds: &StateRebuilds, root: &H256, waiters: usize) { let deadline = Instant::now() + std::time::Duration::from_secs(10); loop { @@ -1261,30 +1282,50 @@ mod tests { } #[test] - fn a_leader_error_is_not_handed_to_its_waiters() { + fn a_leader_error_is_handed_to_its_waiters_and_not_kept() { + const WAITERS: usize = 3; let rebuilds = StateRebuilds::default(); let root = H256([7u8; 32]); let built = Arc::new(BeaconState::Lean(base_state())); let calls = AtomicUsize::new(0); + // Only the first call fails, so a waiter that read the state itself + // instead of taking the leader's error would come back with a state. let rebuild = || { if calls.fetch_add(1, Ordering::SeqCst) == 0 { - wait_for_waiters(&rebuilds, &root, 1); + wait_for_waiters(&rebuilds, &root, WAITERS); return Err(Error::UnexpectedMissingState(root)); } Ok(Some(built.clone())) }; let results: Vec<_> = std::thread::scope(|scope| { - let handles: Vec<_> = (0..2) + let handles: Vec<_> = (0..=WAITERS) .map(|_| scope.spawn(|| rebuilds.share(root, rebuild))) .collect(); handles.into_iter().map(|h| h.join().unwrap()).collect() }); - assert_eq!(calls.load(Ordering::SeqCst), 2, "the waiter retried"); - let (errors, states): (Vec<_>, Vec<_>) = results.into_iter().partition(Result::is_err); - assert_eq!(errors.len(), 1, "only the leader saw its error"); - let state = states[0].as_ref().unwrap().as_ref().expect("a state"); - assert!(Arc::ptr_eq(state, &built)); + assert_eq!(calls.load(Ordering::SeqCst), 1, "no waiter read it again"); + for result in results { + let err = result.expect_err("every caller got the leader's error"); + assert!( + matches!( + &err, + Error::StateRebuild(shared) + if matches!(**shared, Error::UnexpectedMissingState(r) if r == root) + ), + "got {err:?}" + ); + } + assert!( + rebuilds.0.lock().unwrap().is_empty(), + "a failed rebuild deregisters its root" + ); + + let state = rebuilds + .share(root, rebuild) + .expect("a later miss reads storage again rather than the old error") + .expect("a state"); + assert!(Arc::ptr_eq(&state, &built)); } } diff --git a/crates/storage/src/store.rs b/crates/storage/src/store.rs index 34adf7417..7371b3f74 100644 --- a/crates/storage/src/store.rs +++ b/crates/storage/src/store.rs @@ -1076,6 +1076,16 @@ pub(crate) fn block_slot(backend: &dyn StorageBackend, chain: Chain, root: &H256 /// second read view while holding the first can deadlock against a writer /// queued between the two on a backend whose read lock is not reentrant. pub(crate) fn block_slot_in(view: &dyn StorageReadView, chain: Chain, root: &H256) -> Option { + try_block_slot_in(view, chain, root).expect("read") +} + +/// [`block_slot_in`], returning a failed read rather than panicking on it, +/// for the state rebuild in `read_state`, whose callers handle a storage error. +pub(crate) fn try_block_slot_in( + view: &dyn StorageReadView, + chain: Chain, + root: &H256, +) -> Result, crate::api::Error> { match chain { Chain::Beacon => { view.read_with(Table::BlockHeaders, &root.to_ssz(), peek_beacon_block_slot) @@ -1086,7 +1096,6 @@ pub(crate) fn block_slot_in(view: &dyn StorageReadView, chain: Chain, root: &H25 .slot }), } - .expect("read") } /// Fork choice store backed by a pluggable storage backend. @@ -7482,6 +7491,44 @@ mod tests { } } + #[test] + fn a_corrupt_beacon_state_reads_back_as_an_error_and_is_not_cached() { + let backend: Arc = Arc::new(InMemoryBackend::new()); + let mut store = beacon_test_store(backend.clone()); + // Slot 0 is a snapshot and slot 1 a delta on it, so the two reads + // below cover a snapshot that does not decode and a delta that has + // the wrong base to apply to. + let tip = insert_beacon_chain(&mut store, 0..=1, H256::ZERO); + drop(store); + + let snapshot_root = beacon_chain_root(0); + let truncated = vec![( + encode_slot_root_key(0, &snapshot_root), + vec![ForkName::Electra.selector(), 0, 1, 2], + )]; + let mut batch = backend.begin_write().expect("write batch"); + batch + .put_batch(Table::States, truncated) + .expect("put state"); + batch.commit().expect("commit"); + + let cold = beacon_test_store(backend); + for root in [snapshot_root, tip] { + let err = cold + .get_state(&root) + .expect_err("a corrupt state is an error, not a panic"); + assert!( + matches!( + &err, + Error::StateRebuild(shared) + if matches!(**shared, Error::CorruptState { root: r, .. } if r == root) + ), + "got {err:?}" + ); + assert!(cold.cached_state(CacheKey::BlockState(root)).is_none()); + } + } + /// `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 diff --git a/docs/data_storage.md b/docs/data_storage.md index 45ceff792..c93f74f91 100644 --- a/docs/data_storage.md +++ b/docs/data_storage.md @@ -698,16 +698,23 @@ Reads go through `get_state`, which tries three levels: ``` If the diff chain is broken or the target's header is missing, `get_state` -returns `None` rather than a partial state. +returns `None` rather than a partial state. If a read fails, or a row on the +way does not decode (a snapshot, a delta that does not apply to its base, a +lean diff or header), it returns an error (`Error::Storage` or +`Error::CorruptState`) rather than panicking. Each caller decides what that +costs: p2p's column checks drop the one sidecar and log it, while fork choice +and the state writer still panic on it. Concurrent misses on one root rebuild it once (`StateRebuilds` in `crates/storage/src/state_writer.rs`, a `OnceLock` per root under way). The first caller rebuilds. Every other caller that misses on that root while the -rebuild is under way waits for it and gets the same `Arc`, so p2p's +rebuild is under way waits for it and gets the same result, so p2p's per-sidecar column checks, the chain actor and the RPC asking for one evicted -parent state cost one snapshot read and one decode, not one each. A rebuild -that fails hands the waiting callers no result: after a panic one of them -rebuilds for the rest, and after an error each reads the state itself. +parent state cost one snapshot read and one decode, not one each. A failed +rebuild's error is shared the same way, as `Error::StateRebuild`, so a failing +root is read once rather than once per caller. Neither result is kept once the +rebuild settles: the next miss reads storage again. A rebuild that panics +hands the waiting callers nothing, and one of them rebuilds for the rest. ## Write Paths: What a Block Import Persists From 9006389d821e8c7dfdb8755da6001fd91d41f09f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Wed, 11 Nov 2026 12:59:10 -0300 Subject: [PATCH 4/6] docs(storage): say when get_state errs and why a failed parent read drops the column --- .../blockchain/state_transition/src/beacon/gossip/column.rs | 4 ++-- crates/storage/src/store.rs | 5 +++++ 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/crates/blockchain/state_transition/src/beacon/gossip/column.rs b/crates/blockchain/state_transition/src/beacon/gossip/column.rs index 48075e0c1..b0783b94a 100644 --- a/crates/blockchain/state_transition/src/beacon/gossip/column.rs +++ b/crates/blockchain/state_transition/src/beacon/gossip/column.rs @@ -176,8 +176,8 @@ pub fn chain_checks(store: &Store, sidecar: &DataColumnSidecar, now_ms: u64) -> outcome => ChainVerdict::Drop(outcome), }; } - // The parent's state is stored but could not be read back, so - // parking would wait on an import that has already happened. Dropped + // A failed read or a corrupt row is not a parent still to come, so + // parking would wait for an import that need never happen. Dropped // instead, as a panic in this read used to drop it. Err(err) => { let slot = header.slot; diff --git a/crates/storage/src/store.rs b/crates/storage/src/store.rs index 7371b3f74..8ea9c9bae 100644 --- a/crates/storage/src/store.rs +++ b/crates/storage/src/store.rs @@ -3267,6 +3267,11 @@ impl Store { /// that finds a rebuild of `root` already under way waits for it and /// returns its result rather than reading the state again. See /// [`StateRebuilds`]. + /// + /// `Ok(None)` when no state is stored for `root`. `Err` when a read from + /// the backend fails or a stored row does not decode + /// ([`Error::CorruptState`]), wrapped in [`Error::StateRebuild`] since + /// every caller of that rebuild shares it. pub fn get_state(&self, root: &H256) -> Result>, Error> { read_state( self.backend.as_ref(), From 0a7cfeaa6a4223918789f81742937ea782fdce74 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Wed, 11 Nov 2026 14:55:35 -0300 Subject: [PATCH 5/6] fix(storage): count only real state reads as led, and test sharing through read_state `role="led"` was counted before the rebuild ran, so a lookup for a root with no state (one header read) and a failed read both counted as a 100+ MB rebuild. A parked column's parent lookups then produced the same "1 led + N joined" shape as a real late-block rebuild. Now `led` counts only a state read back from storage, and `joined` only a waiter that took one. The lean header read at the end of `reconstruct_state` mapped a decode failure to `CorruptState`, but the walk's slot lookup decodes that same row first and panics, so the mapping could never run. It is an `expect` again, and the docs no longer list a lean header among the rows that return an error. The docs also said only waiting callers get `Error::StateRebuild`; every caller does. New tests: eight concurrent `read_state` misses behind a gated backend open one read view; the leader's cache and write-buffer re-check reads nothing; and a caller leaving an old rebuild does not remove a newer rebuild's entry. Each fails when the code it covers is removed. --- crates/storage/src/error.rs | 16 +- crates/storage/src/metrics.rs | 4 +- crates/storage/src/state_writer.rs | 309 ++++++++++++++++++++++++----- docs/data_storage.md | 25 ++- docs/metrics.md | 5 +- 5 files changed, 296 insertions(+), 63 deletions(-) diff --git a/crates/storage/src/error.rs b/crates/storage/src/error.rs index 1c4b00341..8ec8a31e9 100644 --- a/crates/storage/src/error.rs +++ b/crates/storage/src/error.rs @@ -15,12 +15,20 @@ pub enum Error { /// apply to its base. [`Store::from_db_state`](crate::store::Store::from_db_state) /// has already refused a directory of another format version, so this is /// corruption at rest rather than an old database. + /// + /// `root` is the state being read, which for a delta can be built on the + /// corrupt row rather than be it: a snapshot that does not decode is + /// reported against every state read through it. #[error("the stored state for root {root} is corrupt: {reason}")] CorruptState { root: H256, reason: String }, - /// A state rebuild's error, handed to every concurrent read of the same - /// root that waited on that rebuild rather than reading the state itself - /// (`StateRebuilds` in `state_writer.rs`). Reference-counted because - /// `Error` is not `Clone`; it displays as the error it wraps. + /// Every error a state rebuild returns, as + /// [`Store::get_state`](crate::store::Store::get_state) gives it: the + /// caller that ran the rebuild and every concurrent read of the same root + /// that waited on it get the same one (`StateRebuilds` in + /// `state_writer.rs`). It wraps [`Error::Storage`] or + /// [`Error::CorruptState`], so match through it rather than on those + /// directly. Reference-counted because `Error` is not `Clone`; it displays + /// as the error it wraps. #[error(transparent)] StateRebuild(Arc), /// The data directory was written by a build with a different on-disk diff --git a/crates/storage/src/metrics.rs b/crates/storage/src/metrics.rs index 4ad6b0caf..335875b0b 100644 --- a/crates/storage/src/metrics.rs +++ b/crates/storage/src/metrics.rs @@ -105,8 +105,8 @@ static LEAN_STATE_REBUILDS_TOTAL: LazyLock = LazyLock::new(|| { pub(crate) enum RebuildRole { /// Read and decoded the state itself. Led, - /// Waited on a rebuild of the same root already under way and took its - /// result. + /// Waited on a rebuild of the same root already under way and took the + /// state it read. Joined, } diff --git a/crates/storage/src/state_writer.rs b/crates/storage/src/state_writer.rs index 0cc3c2bab..a79786582 100644 --- a/crates/storage/src/state_writer.rs +++ b/crates/storage/src/state_writer.rs @@ -141,14 +141,22 @@ impl PendingStates { /// `get_or_init` until it is done and takes the same result: the same `Arc`, /// or the same error. A rebuild that panics leaves its cell empty, and /// `OnceLock` lets one of the blocked callers run its own rebuild instead, so -/// a panic never strands a waiter. The entry is removed once the rebuild -/// settles, so the map holds rebuilds under way, never pins a state, and -/// never keeps an error for a later miss: that one reads storage again. +/// a panic never strands a waiter. +/// +/// Sharing an error means one failed read fails every caller waiting on it: +/// a late block's eight custody columns are all dropped, not one, when their +/// parent's state does not read back. +/// +/// The first caller out of a settled rebuild removes its entry, so the map +/// holds rebuilds under way and never pins a state. A result outlives the +/// rebuild only for the moment between the cell being set and that removal: +/// a miss arriving then takes it, and any later miss reads storage again. #[derive(Default)] pub(crate) struct StateRebuilds(Mutex>>); /// One rebuild's shared result. The error is behind an `Arc` because `Error` -/// is not `Clone`; every caller gets it as [`Error::StateRebuild`]. +/// is not `Clone`; every caller, the one that ran the rebuild included, gets +/// it as [`Error::StateRebuild`]. type RebuildCell = OnceLock>, Arc>>; impl StateRebuilds { @@ -167,25 +175,32 @@ impl StateRebuilds { rebuild().map_err(Arc::new) }) .clone(); - // The first caller out removes the cell. A later miss can then - // register a fresh one for `root`, which a caller still leaving this - // cell must not remove. A rebuild that panicked with no one waiting - // skips this and leaves an empty cell, which the next miss on `root` - // fills and removes. - { - let mut cells = self.0.lock().unwrap(); - if cells - .get(&root) - .is_some_and(|current| Arc::ptr_eq(current, &cell)) - { - cells.remove(&root); - } - } - if !led { + self.release(root, &cell); + // Only a waiter that took a state counts: one that took a miss or + // an error saved nothing. + if !led && matches!(result, Ok(Some(_))) { crate::metrics::inc_state_rebuilds(RebuildRole::Joined); } result.map_err(Error::StateRebuild) } + + /// Removes `root`'s entry if it is still `cell`. + /// + /// The first caller out removes the cell. A later miss can then register + /// a fresh one for `root`, which a caller still leaving this cell must + /// not remove: that would let a third miss start a second rebuild of + /// `root` beside the fresh one. A rebuild that panicked with no one + /// waiting skips this and leaves an empty cell, which the next miss on + /// `root` fills and removes. + fn release(&self, root: H256, cell: &Arc) { + let mut cells = self.0.lock().unwrap(); + if cells + .get(&root) + .is_some_and(|current| Arc::ptr_eq(current, cell)) + { + cells.remove(&root); + } + } } /// What a cached state is keyed by. @@ -255,19 +270,37 @@ pub(crate) fn read_state( if let Some(state) = pending.get(root) { return Ok(Some(state)); } - rebuilds.share(*root, || { - // Checked again under the lead: a rebuild that finished between the - // miss above and taking the lead has already cached its state, and a - // second read would be the very cost the lead exists to avoid. Not - // counted as a lookup, since the miss above already was. - if let Some(state) = cache.lock().unwrap().get(&key).cloned() { - return Ok(Some(state)); - } - if let Some(state) = pending.get(root) { - return Ok(Some(state)); + rebuilds.share(*root, || lead_rebuild(backend, chain, cache, pending, root)) +} + +/// What the one caller leading a rebuild of `root` runs: the cache and the +/// write buffer once more, then storage. +/// +/// Checked again under the lead because a rebuild that finished between the +/// caller's miss and its taking the lead has already cached its state, and a +/// second read would be the very cost the lead exists to avoid. Not counted +/// as a lookup, since the miss before it already was. +/// +/// Counts a `led` rebuild only when storage gave back a state: a root with +/// no state costs one block-header read, and is not what that series is for. +fn lead_rebuild( + backend: &dyn StorageBackend, + chain: Chain, + cache: &StateCache, + pending: &PendingStates, + root: &H256, +) -> Result>, Error> { + let key = CacheKey::BlockState(*root); + if let Some(state) = cache.lock().unwrap().get(&key).cloned() { + return Ok(Some(state)); + } + if let Some(state) = pending.get(root) { + return Ok(Some(state)); + } + rebuild_state(backend, chain, cache, root).inspect(|state| { + if state.is_some() { + crate::metrics::inc_state_rebuilds(RebuildRole::Led); } - crate::metrics::inc_state_rebuilds(RebuildRole::Led); - rebuild_state(backend, chain, cache, root) }) } @@ -275,9 +308,11 @@ pub(crate) fn read_state( /// path past the cache and the write buffer, which [`StateRebuilds`] runs for /// one caller per root at a time. /// -/// `Err` when a read fails ([`Error::Storage`]) or what it read does not -/// decode ([`Error::CorruptState`]), rather than a panic, so that a caller -/// such as a gossip check can give up on one message instead of unwinding. +/// `Err` when a read fails ([`Error::Storage`]) or a state, delta or diff it +/// read does not decode ([`Error::CorruptState`]), rather than a panic, so +/// that a caller such as a gossip check can give up on one message instead +/// of unwinding. A `BlockHeaders` row whose slot does not decode still panics +/// in the slot lookup, on both chains. fn rebuild_state( backend: &dyn StorageBackend, chain: Chain, @@ -533,7 +568,9 @@ fn fold_beacon_state_deltas( /// `lean_only` guard. /// /// Returns `Ok(None)` when the root is unknown or the diff chain is broken, -/// and `Err` when a read fails or a row on the way does not decode. +/// and `Err` when a read fails or a snapshot or diff on the way does not +/// decode. A `BlockHeaders` row that does not decode still panics, in the +/// slot lookup that reads it first. pub(crate) fn reconstruct_state( backend: &dyn StorageBackend, root: &H256, @@ -570,16 +607,13 @@ pub(crate) fn reconstruct_state( diffs.reverse(); // The latest block header lives in BlockHeaders; the stored state caches - // the real state_root there, so it equals the header byte-for-byte. + // the real state_root there, so it equals the header byte-for-byte. The + // walk's first slot lookup decoded this same row, so one that does not + // decode has already panicked there. let view = backend.begin_read()?; - let header = view - .read_with( - Table::BlockHeaders, - &root.to_ssz(), - BlockHeader::from_ssz_bytes, - )? - .transpose() - .map_err(|err| corrupt_state(root, format!("its block header does not decode: {err:?}")))?; + let header = view.read_with(Table::BlockHeaders, &root.to_ssz(), |bytes| { + BlockHeader::from_ssz_bytes(bytes).expect("valid header") + })?; drop(view); let Some(latest_block_header) = header else { return Ok(None); @@ -1206,8 +1240,193 @@ mod tests { } } + /// An in-memory backend that counts read views and holds each one until + /// [`Self::open`], so a test can line callers up behind one rebuild's + /// read before letting it finish. + struct GatedBackend { + inner: crate::backend::InMemoryBackend, + open: Mutex, + opened: std::sync::Condvar, + reads: AtomicUsize, + } + + impl GatedBackend { + fn new(open: bool) -> Self { + Self { + inner: crate::backend::InMemoryBackend::new(), + open: Mutex::new(open), + opened: std::sync::Condvar::new(), + reads: AtomicUsize::new(0), + } + } + + fn open(&self) { + *self.open.lock().unwrap() = true; + self.opened.notify_all(); + } + + fn reads(&self) -> usize { + self.reads.load(Ordering::SeqCst) + } + + /// A lean `state` stored as a snapshot under `root`, with the + /// `BlockHeaders` row a read finds its slot in. + fn store_lean_state(&self, root: H256, state: &State) { + let header = BlockHeader { + slot: state.slot, + proposer_index: 0, + parent_root: H256::ZERO, + state_root: H256::ZERO, + body_root: H256::ZERO, + }; + let headers = vec![(root.to_ssz(), header.to_ssz())]; + let key = encode_slot_root_key(state.slot, &root); + let states = vec![(key, encode_state_value(&BeaconState::Lean(state.clone())))]; + let mut batch = self.inner.begin_write().expect("write batch"); + batch + .put_batch(Table::BlockHeaders, headers) + .expect("put header"); + batch.put_batch(Table::States, states).expect("put state"); + batch.commit().expect("commit"); + } + } + + impl StorageBackend for GatedBackend { + fn begin_read(&self) -> Result, crate::api::Error> { + self.reads.fetch_add(1, Ordering::SeqCst); + let mut open = self.open.lock().unwrap(); + while !*open { + open = self.opened.wait(open).unwrap(); + } + drop(open); + self.inner.begin_read() + } + + fn begin_write( + &self, + ) -> Result, crate::api::Error> { + self.inner.begin_write() + } + } + + /// Opens the gate when dropped, so a failed assertion that unwinds past it + /// releases the callers held behind it rather than leaving the thread + /// scope waiting on them forever. + struct OpenOnDrop<'a>(&'a GatedBackend); + + impl Drop for OpenOnDrop<'_> { + fn drop(&mut self) { + self.0.open(); + } + } + + fn state_cache() -> StateCache { + Mutex::new(LruCache::new( + std::num::NonZeroUsize::new(32).expect("non-zero"), + )) + } + + #[test] + fn concurrent_get_state_misses_on_one_root_read_storage_once() { + const WAITERS: usize = 7; + let backend = GatedBackend::new(false); + let root = H256([8u8; 32]); + backend.store_lean_state(root, &base_state()); + let cache = state_cache(); + let pending = PendingStates::default(); + let rebuilds = StateRebuilds::default(); + + let states: Vec<_> = std::thread::scope(|scope| { + let handles: Vec<_> = (0..=WAITERS) + .map(|_| { + scope.spawn(|| { + read_state(&backend, Chain::Lean, &cache, &pending, &rebuilds, &root) + }) + }) + .collect(); + // The leader is held in its read view until every other caller is + // waiting on its rebuild, so none of them can miss the cache after + // the rebuild settles and read storage a second time. + let gate = OpenOnDrop(&backend); + wait_for_waiters(&rebuilds, &root, WAITERS); + drop(gate); + handles + .into_iter() + .map(|h| h.join().unwrap().expect("no error").expect("a state")) + .collect() + }); + + assert_eq!(backend.reads(), 1, "eight misses, one read view"); + for state in &states { + assert!(Arc::ptr_eq(state, &states[0]), "one state, not copies"); + } + assert!(rebuilds.0.lock().unwrap().is_empty()); + } + + #[test] + fn a_lead_takes_a_state_settled_since_its_miss_without_reading() { + let backend = GatedBackend::new(true); + let root = H256([9u8; 32]); + let state = Arc::new(BeaconState::Lean(base_state())); + let cache = state_cache(); + let pending = PendingStates::default(); + + // Cached by a rebuild that settled after this caller missed the + // cache and before it took the lead. + cache + .lock() + .unwrap() + .put(CacheKey::BlockState(root), state.clone()); + let found = lead_rebuild(&backend, Chain::Lean, &cache, &pending, &root) + .expect("no error") + .expect("the cached state"); + assert!(Arc::ptr_eq(&found, &state)); + + // Handed to the write buffer in that window, and already evicted. + cache.lock().unwrap().clear(); + pending.insert(root, state.clone()); + let found = lead_rebuild(&backend, Chain::Lean, &cache, &pending, &root) + .expect("no error") + .expect("the pending state"); + assert!(Arc::ptr_eq(&found, &state)); + + assert_eq!(backend.reads(), 0, "neither lead went to storage"); + } + + #[test] + fn a_caller_leaving_an_old_rebuild_keeps_a_newer_one_registered() { + let rebuilds = StateRebuilds::default(); + let root = H256([10u8; 32]); + let built = Arc::new(BeaconState::Lean(base_state())); + let calls = AtomicUsize::new(0); + let rebuild = || { + calls.fetch_add(1, Ordering::SeqCst); + wait_for_waiters(&rebuilds, &root, 1); + Ok(Some(built.clone())) + }; + + std::thread::scope(|scope| { + let leader = scope.spawn(|| rebuilds.share(root, rebuild)); + wait_for_waiters(&rebuilds, &root, 0); + // The last caller out of an earlier, settled rebuild of `root` + // leaves only now, after the leader above registered a new one. + rebuilds.release(root, &Arc::new(RebuildCell::new())); + let waiter = scope.spawn(|| rebuilds.share(root, rebuild)); + for caller in [leader, waiter] { + let state = caller.join().unwrap().expect("no error").expect("a state"); + assert!(Arc::ptr_eq(&state, &built)); + } + }); + + assert_eq!( + calls.load(Ordering::SeqCst), + 1, + "the waiter joined the leader's rebuild rather than starting a second" + ); + } + #[test] - fn concurrent_misses_on_one_root_share_one_rebuild() { + fn concurrent_shares_of_one_root_run_one_rebuild() { const WAITERS: usize = 7; let rebuilds = StateRebuilds::default(); let root = H256([5u8; 32]); diff --git a/docs/data_storage.md b/docs/data_storage.md index c93f74f91..81a92442a 100644 --- a/docs/data_storage.md +++ b/docs/data_storage.md @@ -698,12 +698,14 @@ Reads go through `get_state`, which tries three levels: ``` If the diff chain is broken or the target's header is missing, `get_state` -returns `None` rather than a partial state. If a read fails, or a row on the -way does not decode (a snapshot, a delta that does not apply to its base, a -lean diff or header), it returns an error (`Error::Storage` or -`Error::CorruptState`) rather than panicking. Each caller decides what that -costs: p2p's column checks drop the one sidecar and log it, while fork choice -and the state writer still panic on it. +returns `None` rather than a partial state. If a read fails, or a state on +the way does not decode (a snapshot, a delta that does not apply to its base, +a lean snapshot or diff), it returns an error rather than panicking: +`Error::StateRebuild` wrapping `Error::Storage` or `Error::CorruptState`. A +`BlockHeaders` row that does not decode still panics, in the slot lookup that +reads it before any state row. Each caller decides what an error costs: p2p's +column checks drop the one sidecar and log it, while fork choice and the state +writer still panic on it. Concurrent misses on one root rebuild it once (`StateRebuilds` in `crates/storage/src/state_writer.rs`, a `OnceLock` per root under way). The @@ -711,10 +713,13 @@ first caller rebuilds. Every other caller that misses on that root while the rebuild is under way waits for it and gets the same result, so p2p's per-sidecar column checks, the chain actor and the RPC asking for one evicted parent state cost one snapshot read and one decode, not one each. A failed -rebuild's error is shared the same way, as `Error::StateRebuild`, so a failing -root is read once rather than once per caller. Neither result is kept once the -rebuild settles: the next miss reads storage again. A rebuild that panics -hands the waiting callers nothing, and one of them rebuilds for the rest. +rebuild's error is shared the same way, so a failing root is read once rather +than once per caller, and every caller waiting on it fails: a late block's +eight custody columns are all dropped when their parent's state does not read +back. The first caller out removes the rebuild's entry; a miss arriving +before that takes the settled result, and any later miss reads storage +again. A rebuild that panics hands the waiting callers nothing, and one of +them rebuilds for the rest. ## Write Paths: What a Block Import Persists diff --git a/docs/metrics.md b/docs/metrics.md index 6881ed557..7bf920937 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -335,7 +335,7 @@ In practice the distribution is bimodal and dominated by production rather than | `lean_state_write_queue_depth` | Gauge | States handed to the background writer but not yet committed | On every hand-off and on every commit | | | `lean_state_write_seconds` | Histogram | Time the background writer spends encoding, diffing and committing one state | Per state written | | | `lean_state_cache_lookups_total` | Counter | State cache lookups, by the `Store` method that made them, the kind of state asked for, and whether the cache held it | On every `get_state` (and the writer's own parent read), `has_state` and `cached_state` call | method=get_state,has_state,cached_state
kind=block,checkpoint
result=hit,miss | -| `lean_state_rebuilds_total` | Counter | Block states read back from storage after missing the cache and the write buffer, by whether the read did the rebuild or waited on one already under way for that root | On every `get_state` miss that reaches storage | role=led,joined | +| `lean_state_rebuilds_total` | Counter | Block states read back from storage after missing the cache and the write buffer, by whether the read did the rebuild or waited on one already under way for that root | On every `get_state` miss that reads a state back from storage, or takes one from a rebuild under way; a miss that finds no state or fails counts as neither | role=led,joined | **On a beacon follower, watch `lean_table_bytes{table="data_columns"}`.** Every other table's series either stays flat or is bounded by pruning; `data_columns` @@ -368,7 +368,8 @@ epoch boundary when the checkpoint's block precedes it. read and decoded from storage.** A beacon rebuild costs about a second of CPU and several hundred megabytes at its peak, so this is the series to watch for resident-memory spikes. `role="joined"` counts misses that found a rebuild of -the same root already under way and waited for it instead. A burst of joins +the same root already under way and took its state instead. A lookup that +finds no state for its root, or fails, counts under neither. A burst of joins next to a single lead is the pattern a late block's custody columns produce, each checked on its own task against a parent state that has left the cache. From 65f632f3e329eb08b5aed4ca25abebd38d8559d9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Wed, 11 Nov 2026 15:46:51 -0300 Subject: [PATCH 6/6] docs(metrics): count the writer's parent read under lean_state_rebuilds_total --- docs/metrics.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/metrics.md b/docs/metrics.md index 7bf920937..8183c6ca7 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -335,7 +335,7 @@ In practice the distribution is bimodal and dominated by production rather than | `lean_state_write_queue_depth` | Gauge | States handed to the background writer but not yet committed | On every hand-off and on every commit | | | `lean_state_write_seconds` | Histogram | Time the background writer spends encoding, diffing and committing one state | Per state written | | | `lean_state_cache_lookups_total` | Counter | State cache lookups, by the `Store` method that made them, the kind of state asked for, and whether the cache held it | On every `get_state` (and the writer's own parent read), `has_state` and `cached_state` call | method=get_state,has_state,cached_state
kind=block,checkpoint
result=hit,miss | -| `lean_state_rebuilds_total` | Counter | Block states read back from storage after missing the cache and the write buffer, by whether the read did the rebuild or waited on one already under way for that root | On every `get_state` miss that reads a state back from storage, or takes one from a rebuild under way; a miss that finds no state or fails counts as neither | role=led,joined | +| `lean_state_rebuilds_total` | Counter | Block states read back from storage after missing the cache and the write buffer, by whether the read did the rebuild or waited on one already under way for that root | On every `get_state` miss (and the writer's own parent read) that reads a state back from storage, or takes one from a rebuild under way; a miss that finds no state or fails counts as neither | role=led,joined | **On a beacon follower, watch `lean_table_bytes{table="data_columns"}`.** Every other table's series either stays flat or is bounded by pruning; `data_columns`