diff --git a/crates/blockchain/state_transition/src/beacon/gossip/column.rs b/crates/blockchain/state_transition/src/beacon/gossip/column.rs index 900d83353..b0783b94a 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), + }; + } + // 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; + 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..8ec8a31e9 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,26 @@ 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. + /// + /// `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 }, + /// 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 /// 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/metrics.rs b/crates/storage/src/metrics.rs index 99093a8fb..335875b0b 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 the + /// state it read. + 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_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 51e5f6b0e..a79786582 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, Mutex, OnceLock}; use std::thread::JoinHandle; use std::time::Instant; @@ -66,10 +66,12 @@ 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}; +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. /// @@ -123,6 +125,84 @@ 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. +/// +/// 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 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. +/// +/// 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, the one that ran the rebuild included, 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 share( + &self, + root: H256, + rebuild: impl FnOnce() -> Result>, Error>, + ) -> Result>, Error> { + let cell = self.0.lock().unwrap().entry(root).or_default().clone(); + let mut led = false; + let result = cell + .get_or_init(|| { + led = true; + rebuild().map_err(Arc::new) + }) + .clone(); + 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. /// /// One cache rather than two, so a single capacity bounds the total rather @@ -172,11 +252,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,20 +270,69 @@ pub(crate) fn read_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); + } + }) +} + +/// 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 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, + cache: &StateCache, + root: &H256, +) -> Result>, Error> { let state = match chain { Chain::Lean => { // 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, @@ -217,9 +350,10 @@ pub(crate) fn read_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); @@ -232,10 +366,21 @@ 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)) } +/// [`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 @@ -301,36 +446,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); @@ -377,21 +525,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)) } @@ -405,39 +558,43 @@ 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 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, ) -> 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); }; @@ -450,13 +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. - let view = backend.begin_read().expect("read view"); - let header = view - .read_with(Table::BlockHeaders, &root.to_ssz(), |bytes| { - BlockHeader::from_ssz_bytes(bytes).expect("valid header") - }) - .expect("read"); + // 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(), |bytes| { + BlockHeader::from_ssz_bytes(bytes).expect("valid header") + })?; drop(view); let Some(latest_block_header) = header else { return Ok(None); @@ -600,6 +757,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 +810,7 @@ impl StateWriter { self.chain, &self.cache, &self.pending, + &self.rebuilds, &parent_root, ) .expect("read parent state") @@ -792,6 +953,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 +961,7 @@ impl StateWriterHandle { chain, cache, pending, + rebuilds, snapshot_memo: None, last_base: None, }; @@ -890,6 +1053,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 +1215,336 @@ 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(); + } + } + + /// 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_shares_of_one_root_run_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_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, WAITERS); + return Err(Error::UnexpectedMissingState(root)); + } + 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, "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 14ac9116f..8ea9c9bae 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}; @@ -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. @@ -1170,6 +1179,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 +1836,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 +1859,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 +3262,23 @@ 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`]. + /// + /// `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(), self.chain, &self.state_cache, &self.pending_states, + &self.state_rebuilds, root, ) } @@ -7469,6 +7496,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 0f8d62db8..81a92442a 100644 --- a/docs/data_storage.md +++ b/docs/data_storage.md @@ -698,7 +698,28 @@ 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 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 +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, 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 9ac92fd4f..8183c6ca7 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 (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` @@ -363,6 +364,15 @@ 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 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. + ### Beacon Gossip Validation Every beacon gossip message gets a verdict before gossipsub forwards it