feat(storage): diff-layer state storage with bounded pruning

CLAUDE.md

@@ -362,7 +362,13 @@ cargo test -p ethlambda-blockchain --test forkchoice_spectests -- --test-threads
 - Blocks are split into three tables: `BlockHeaders`, `BlockBodies`, `BlockSignatures`
 - Genesis/anchor blocks have empty bodies (detected via `EMPTY_BODY_ROOT`) — no entry in `BlockBodies`
 - Genesis block has no signatures — no entry in `BlockSignatures`
-- All other blocks must have entries in all three tables
+- Non-genesis blocks have a `BlockSignatures` entry until finalized: once below the
+  finalized boundary, signatures are pruned (`prune_old_block_signatures`) while
+  headers and bodies are kept forever. `get_signed_block` returns `None` for a
+  pruned finalized block
+- States are stored as parent-linked diffs (`StateDiffs`, never pruned) plus
+  full-state snapshots (`States`) at anchors/hot states; `get_state`
+  reconstructs from diffs when no snapshot exists
 - `LiveChain` table provides fast `(slot||root) → parent_root` index for fork choice
 - Storage uses trait-based API: `StorageBackend` → `StorageReadView` (reads) + `StorageWriteBatch` (atomic writes)
 

crates/blockchain/src/store.rs

@@ -1,7 +1,7 @@
 use std::collections::HashSet;
 
 use ethlambda_state_transition::{is_proposer, slot_is_justifiable_after};
-use ethlambda_storage::{ForkCheckpoints, Store};
+use ethlambda_storage::{DiffBase, ForkCheckpoints, Store};
 use ethlambda_types::{
     ShortRoot,
     attestation::{
@@ -556,6 +556,10 @@ fn on_block_core(
 
     let block = signed_block.message.clone();
 
+    // Capture the diff base before the parent is consumed into the post-state
+    // (avoids cloning the multi-MB historical_block_hashes list).
+    let diff_base = DiffBase::from_state(block.parent_root, &parent_state);
+
     // Execute state transition function to compute post-block state
     let state_transition_start = std::time::Instant::now();
     let mut post_state = parent_state;
@@ -576,9 +580,9 @@ fn on_block_core(
         store.update_checkpoints(ForkCheckpoints::new(store.head(), Some(justified), None));
     }
 
-    // Store signed block and state
+    // Store signed block and state (as a parent-linked diff + snapshot)
     store.insert_signed_block(block_root, signed_block.clone());
-    store.insert_state(block_root, post_state);
+    store.insert_state_with_diff(block_root, diff_base, post_state);
 
     for att in block.body.attestations.iter() {
         // Count each participating validator as a valid attestation.

crates/storage/Cargo.toml

@@ -18,6 +18,7 @@ thiserror.workspace = true
 
 libssz.workspace = true
 libssz-derive.workspace = true
+libssz-types.workspace = true
 
 [dev-dependencies]
 tempfile = "3"

crates/storage/src/api/tables.rs

@@ -11,7 +11,21 @@ pub enum Table {
     /// All other blocks must have an entry in this table.
     BlockSignatures,
     /// State storage: H256 -> State
+    ///
+    /// Holds full-state snapshots only: anchors, hot states (recent window,
+    /// finalized/justified/head). Non-snapshot states live in `StateDiffs` and
+    /// are reconstructed on demand.
     States,
+    /// State diffs: H256 -> StateDiff
+    ///
+    /// Parent-linked diff written for every non-genesis state. Never pruned, so
+    /// it preserves full state history. See `get_state` for reconstruction.
+    StateDiffs,
+    /// State snapshot anchors: H256 (root) -> u64 (slot)
+    ///
+    /// Roots whose snapshots are retained permanently to bound diff-walk depth
+    /// (one per 1024-slot window). Protected from snapshot eviction.
+    StateAnchors,
     /// Metadata: string keys -> various scalar values
     Metadata,
     /// Live chain index: (slot || root) -> parent_root
@@ -23,11 +37,13 @@ pub enum Table {
 }
 
 /// All table variants.
-pub const ALL_TABLES: [Table; 6] = [
+pub const ALL_TABLES: [Table; 8] = [
     Table::BlockHeaders,
     Table::BlockBodies,
     Table::BlockSignatures,
     Table::States,
+    Table::StateDiffs,
+    Table::StateAnchors,
     Table::Metadata,
     Table::LiveChain,
 ];
@@ -40,6 +56,8 @@ impl Table {
             Table::BlockBodies => "block_bodies",
             Table::BlockSignatures => "block_signatures",
             Table::States => "states",
+            Table::StateDiffs => "state_diffs",
+            Table::StateAnchors => "state_anchors",
             Table::Metadata => "metadata",
             Table::LiveChain => "live_chain",
         }

crates/storage/src/backend/rocksdb.rs

@@ -11,15 +11,11 @@ use std::path::Path;
 use std::sync::Arc;
 
 /// Returns the column family name for a table.
+///
+/// Delegates to [`Table::name`] so the CF name and the metrics label share a
+/// single source of truth (and a new table only needs one mapping).
 fn cf_name(table: Table) -> &'static str {
-    match table {
-        Table::BlockHeaders => "block_headers",
-        Table::BlockBodies => "block_bodies",
-        Table::BlockSignatures => "block_signatures",
-        Table::States => "states",
-        Table::Metadata => "metadata",
-        Table::LiveChain => "live_chain",
-    }
+    table.name()
 }
 
 /// RocksDB storage backend.

crates/storage/src/lib.rs

@@ -1,6 +1,8 @@
 mod api;
 pub mod backend;
+mod state_diff;
 mod store;
 
 pub use api::{ALL_TABLES, StorageBackend, StorageReadView, StorageWriteBatch, Table};
+pub use state_diff::DiffBase;
 pub use store::{ForkCheckpoints, GetForkchoiceStoreError, MAX_RESUMABLE_DB_STATE_AGE, Store};

crates/storage/src/state_diff.rs

@@ -0,0 +1,242 @@
+//! Parent-linked state diffs for diff-layer state storage.
+//!
+//! A [`StateDiff`] captures the change from a base state (the parent block's
+//! post-state) to a target state, storing only what cannot be recovered from a
+//! snapshot plus the parent relationship.
+//!
+//! Field handling:
+//! - `config`, `validators`: never change; omitted (taken from the snapshot).
+//! - `latest_block_header`: omitted; reconstructed from the `BlockHeaders` table.
+//! - `historical_block_hashes`: pure-append in the STF, so only the appended
+//!   tail (`hbh_appended`) is stored.
+//! - everything else: stored verbatim (the justification fields are bounded by
+//!   the non-finalized window, so they stay small under healthy finality).
+
+use ethlambda_types::{
+    block::BlockHeader,
+    checkpoint::Checkpoint,
+    primitives::H256,
+    state::{
+        HISTORICAL_ROOTS_LIMIT, JustificationRoots, JustificationValidators, JustifiedSlots, State,
+    },
+};
+use libssz_derive::{SszDecode, SszEncode};
+use libssz_types::SszList;
+
+/// Appended tail of `historical_block_hashes`, bounded by the same limit as the
+/// full list.
+pub type HistoricalBlockHashesTail = SszList<H256, HISTORICAL_ROOTS_LIMIT>;
+
+/// Describes the parent state a new state's diff is built against.
+///
+/// Captured by the caller before the parent is consumed into the post-state, so
+/// the store can build the diff and decide anchoring without re-reading it.
+/// Construct via [`DiffBase::from_state`]; fields are crate-internal.
+pub struct DiffBase {
+    /// Block root of the parent state (the diff's `base_root`).
+    pub(crate) root: H256,
+    /// Parent state's `historical_block_hashes` length.
+    pub(crate) hbh_len: usize,
+    /// Parent state's slot (used for the anchor-boundary check).
+    pub(crate) slot: u64,
+}
+
+impl DiffBase {
+    /// Build the diff base from the parent state and its block root.
+    ///
+    /// `root` is the parent block root (the child's `parent_root`), passed in
+    /// since the caller already has it; `hbh_len` and `slot` are read from
+    /// `state`. Call this before the parent is consumed into the child.
+    pub fn from_state(root: H256, state: &State) -> Self {
+        Self {
+            root,
+            hbh_len: state.historical_block_hashes.len(),
+            slot: state.slot,
+        }
+    }
+}
+
+/// The change from a base (parent) state to a target state.
+///
+/// Reconstruct the target with [`StateDiff`] applied against the nearest
+/// ancestor snapshot; see the storage layer's `get_state` for the walk.
+#[derive(Debug, Clone, PartialEq, Eq, SszEncode, SszDecode)]
+pub struct StateDiff {
+    /// Block root of the base state this diff is relative to (`block.parent_root`).
+    pub base_root: H256,
+    /// Target state's slot.
+    pub slot: u64,
+    /// Target state's latest justified checkpoint.
+    pub latest_justified: Checkpoint,
+    /// Target state's latest finalized checkpoint.
+    pub latest_finalized: Checkpoint,
+    /// Target state's `justified_slots` (stored in full).
+    pub justified_slots: JustifiedSlots,
+    /// Target state's `justifications_roots` (stored in full).
+    pub justifications_roots: JustificationRoots,
+    /// Target state's `justifications_validators` (stored in full).
+    pub justifications_validators: JustificationValidators,
+    /// Elements appended to `historical_block_hashes` relative to the base.
+    pub hbh_appended: HistoricalBlockHashesTail,
+}
+
+impl StateDiff {
+    /// Build a diff from a consumed target state against a base identified by its
+    /// `historical_block_hashes` length.
+    ///
+    /// Takes `target` by value so the multi-MB justification fields are moved
+    /// into the diff rather than cloned. On the block-import path the base state
+    /// has already been consumed into `target`, so only its length is retained;
+    /// `base_hbh_len` is that length (`target`'s historical list strictly extends
+    /// the base's, since the state transition only appends to it).
+    ///
+    /// # Panics
+    ///
+    /// Panics if `target.historical_block_hashes` is shorter than `base_hbh_len`.
+    pub fn from_base(base_root: H256, base_hbh_len: usize, target: State) -> Self {
+        let State {
+            slot,
+            latest_justified,
+            latest_finalized,
+            historical_block_hashes,
+            justified_slots,
+            justifications_roots,
+            justifications_validators,
+            ..
+        } = target;
+
+        let hbh = historical_block_hashes.into_inner();
+        assert!(
+            hbh.len() >= base_hbh_len,
+            "target historical_block_hashes shorter than base: {} < {base_hbh_len}",
+            hbh.len()
+        );
+        let hbh_appended = HistoricalBlockHashesTail::try_from(hbh[base_hbh_len..].to_vec())
+            .expect("appended tail cannot exceed HISTORICAL_ROOTS_LIMIT");
+
+        Self {
+            base_root,
+            slot,
+            latest_justified,
+            latest_finalized,
+            justified_slots,
+            justifications_roots,
+            justifications_validators,
+            hbh_appended,
+        }
+    }
+}
+
+/// Rebuild a state from a base snapshot and the diffs leading to the target.
+///
+/// `diffs` are ordered from the snapshot's child up to the target (inclusive,
+/// non-empty). `latest_block_header` is the target's header (kept in the
+/// `BlockHeaders` table rather than the diff). `config`/`validators` come from
+/// `snapshot` (they never change), `historical_block_hashes` is replayed from
+/// the appended tails, and the remaining fields come from the last diff.
+///
+/// # Panics
+///
+/// Panics if `diffs` is empty.
+pub(crate) fn reconstruct(
+    snapshot: State,
+    diffs: &[StateDiff],
+    latest_block_header: BlockHeader,
+) -> State {
+    let target = diffs
+        .last()
+        .expect("reconstruct requires at least one diff");
+
+    let mut hbh: Vec<H256> = snapshot.historical_block_hashes.to_vec();
+    for diff in diffs {
+        hbh.extend_from_slice(&diff.hbh_appended);
+    }
+    let historical_block_hashes = hbh
+        .try_into()
+        .expect("reconstructed historical_block_hashes within limit");
+
+    State {
+        config: snapshot.config,
+        slot: target.slot,
+        latest_block_header,
+        latest_justified: target.latest_justified,
+        latest_finalized: target.latest_finalized,
+        historical_block_hashes,
+        justified_slots: target.justified_slots.clone(),
+        validators: snapshot.validators,
+        justifications_roots: target.justifications_roots.clone(),
+        justifications_validators: target.justifications_validators.clone(),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use ethlambda_types::state::{State, Validator};
+    use libssz::{SszDecode, SszEncode};
+
+    use super::*;
+
+    fn h256(byte: u8) -> H256 {
+        H256::from([byte; 32])
+    }
+
+    /// A minimal genesis-like base state with two validators.
+    fn base_state() -> State {
+        let validators = vec![
+            Validator {
+                attestation_pubkey: [1u8; 52],
+                proposal_pubkey: [2u8; 52],
+                index: 0,
+            },
+            Validator {
+                attestation_pubkey: [3u8; 52],
+                proposal_pubkey: [4u8; 52],
+                index: 1,
+            },
+        ];
+        State::from_genesis(1_000, validators)
+    }
+
+    #[test]
+    fn from_base_captures_appended_tail_and_absolute_fields() {
+        let base = base_state();
+        let base_len = base.historical_block_hashes.len();
+
+        let mut target = base.clone();
+        target.slot = 5;
+        let expected_justified = Checkpoint {
+            root: h256(7),
+            slot: 4,
+        };
+        target.latest_justified = expected_justified;
+        // Append three roots (one real parent + two zero-filled empty slots).
+        let mut hbh: Vec<H256> = base.historical_block_hashes.to_vec();
+        hbh.extend([h256(9), H256::ZERO, H256::ZERO]);
+        target.historical_block_hashes = hbh.try_into().unwrap();
+
+        let diff = StateDiff::from_base(h256(1), base_len, target);
+
+        assert_eq!(diff.base_root, h256(1));
+        assert_eq!(diff.slot, 5);
+        assert_eq!(diff.latest_justified, expected_justified);
+        assert_eq!(diff.hbh_appended.len(), 3);
+        assert_eq!(diff.hbh_appended[0], h256(9));
+        assert_eq!(diff.hbh_appended[1], H256::ZERO);
+    }
+
+    #[test]
+    fn ssz_roundtrips() {
+        let base = base_state();
+        let base_len = base.historical_block_hashes.len();
+        let mut target = base.clone();
+        target.slot = 2;
+        let mut hbh: Vec<H256> = base.historical_block_hashes.to_vec();
+        hbh.push(h256(9));
+        target.historical_block_hashes = hbh.try_into().unwrap();
+
+        let diff = StateDiff::from_base(h256(1), base_len, target);
+        let bytes = diff.to_ssz();
+        let decoded = StateDiff::from_ssz_bytes(&bytes).expect("decodes");
+        assert_eq!(diff, decoded);
+    }
+}