perf(storage): share one rebuild between concurrent get_state misses on a root

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]

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

crates/net/p2p/src/beacon/column_checks.rs

@@ -60,9 +60,11 @@ pub(crate) fn check_and_forward(server: &P2PServer, sidecars: Vec<DataColumnSide
         let mut keep = Vec::new();
         let mut awaiting_parent = Vec::new();
         for check in checks {
-            // A panic inside the checks (a `Store` read's `.expect()` on a DB
-            // error, say) costs that one sidecar, which is what dropping it
-            // would have done anyway. Tokio has already caught it.
+            // A panic inside the checks (one of the `Store` reads that still
+            // `.expect()` on a DB error, say) costs that one sidecar, which is
+            // what dropping it would have done anyway. Tokio has already
+            // caught it. The parent-state read returns its error instead, and
+            // `chain_checks` drops the sidecar for it without unwinding.
             let (sidecar, verdict) = match check.await {
                 Ok(checked) => checked,
                 Err(err) => {

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<u8> {
     })
 }
 
-/// 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<u8> {
+    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<u8> {
+pub(crate) fn try_decode(delta: &[u8], base: &[u8], target_len: usize) -> Result<Vec<u8>, 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:?}")
     })
 }
 

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<Error>),
     /// 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.

crates/storage/src/metrics.rs

@@ -90,6 +90,42 @@ impl StateCacheMethod {
     }
 }
 
+static LEAN_STATE_REBUILDS_TOTAL: LazyLock<IntCounterVec> = 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 {

crates/storage/src/state_codec.rs

@@ -32,26 +32,29 @@ pub(crate) fn encode_state_value(state: &BeaconState) -> Vec<u8> {
 
 /// 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<BeaconState, String> {
+    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<State, String> {
+    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]