refactor: store aggregated payloads in memory

CLAUDE.md

@@ -328,21 +328,32 @@ cargo test -p ethlambda-blockchain --test forkchoice_spectests -- --test-threads
 - `LiveChain` table provides fast `(slot||root) → parent_root` index for fork choice
 - Storage uses trait-based API: `StorageBackend` → `StorageReadView` (reads) + `StorageWriteBatch` (atomic writes)
 
-### Storage Tables (10)
+### Storage Tables (8)
 
 | Table | Key → Value | Purpose |
 |-------|-------------|---------|
 | `BlockHeaders` | H256 → BlockHeader | Block headers by root |
 | `BlockBodies` | H256 → BlockBody | Block bodies (empty for genesis) |
 | `BlockSignatures` | H256 → BlockSignaturesWithAttestation | Signatures (absent for genesis) |
 | `States` | H256 → State | Beacon states by root |
-| `LatestKnownAttestations` | u64 → AttestationData | Fork-choice-active attestations |
-| `LatestNewAttestations` | u64 → AttestationData | Pending (pre-promotion) attestations |
 | `GossipSignatures` | SignatureKey → ValidatorSignature | Individual validator signatures |
-| `AggregatedPayloads` | SignatureKey → Vec\<AggregatedSignatureProof\> | Aggregated proofs |
+| `AttestationDataByRoot` | H256 → AttestationData | Content-addressed attestation data |
 | `Metadata` | string → various | Store state (head, config, checkpoints) |
 | `LiveChain` | (slot\|\|root) → parent\_root | Fast fork choice traversal index |
 
+### In-Memory Store Fields
+
+Two ephemeral maps live on `Store` as `Arc<Mutex<HashMap>>` (not persisted to RocksDB):
+
+| Field | Type | Purpose |
+|-------|------|---------|
+| `known_aggregated_payloads` | SignatureKey → Vec\<StoredAggregatedPayload\> | Fork-choice-active aggregated proofs |
+| `new_aggregated_payloads` | SignatureKey → Vec\<StoredAggregatedPayload\> | Pending aggregated proofs (promoted each interval) |
+
+Lock ordering invariant: always acquire `known_aggregated_payloads` before `new_aggregated_payloads`.
+
+**Note**: Existing RocksDB databases with the old `latest_new_aggregated_payloads` / `latest_known_aggregated_payloads` column families must be wiped before running the new binary — `open_cf_descriptors` fails if the DB contains CFs not in the descriptor list.
+
 ### State Root Computation
 - Always computed via `tree_hash_root()` after full state transition
 - Must match proposer's pre-computed `block.state_root`

crates/blockchain/src/store.rs

@@ -100,6 +100,7 @@ fn update_safe_target(store: &mut Store) {
     // node's self-attestation) would be invisible without this merge.
     let all_keys: HashSet<SignatureKey> = store
         .iter_known_aggregated_payload_keys()
+        .into_iter()
         .chain(store.iter_new_aggregated_payload_keys())
         .collect();
     let attestations = store.extract_latest_attestations(all_keys.into_iter());
@@ -765,7 +766,7 @@ pub fn produce_block_with_signatures(
     }
 
     // Single pass over known aggregated payloads: extract both attestation data and proofs
-    let known_payloads: Vec<_> = store.iter_known_aggregated_payloads().collect();
+    let known_payloads = store.iter_known_aggregated_payloads();
 
     let known_attestations =
         store.extract_latest_attestations(known_payloads.iter().map(|(key, _)| *key));

crates/blockchain/tests/forkchoice_spectests.rs

@@ -284,12 +284,16 @@ fn validate_attestation_check(
     let location = check.location.as_str();
 
     let attestations: HashMap<u64, AttestationData> = match location {
-        "new" => {
-            st.extract_latest_attestations(st.iter_new_aggregated_payloads().map(|(key, _)| key))
-        }
-        "known" => {
-            st.extract_latest_attestations(st.iter_known_aggregated_payloads().map(|(key, _)| key))
-        }
+        "new" => st.extract_latest_attestations(
+            st.iter_new_aggregated_payloads()
+                .into_iter()
+                .map(|(key, _)| key),
+        ),
+        "known" => st.extract_latest_attestations(
+            st.iter_known_aggregated_payloads()
+                .into_iter()
+                .map(|(key, _)| key),
+        ),
         other => {
             return Err(
                 format!("Step {}: unknown attestation location: {}", step_idx, other).into(),
@@ -369,8 +373,11 @@ fn validate_lexicographic_head_among(
     }
 
     let blocks = st.get_live_chain();
-    let known_attestations: HashMap<u64, AttestationData> =
-        st.extract_latest_attestations(st.iter_known_aggregated_payloads().map(|(key, _)| key));
+    let known_attestations: HashMap<u64, AttestationData> = st.extract_latest_attestations(
+        st.iter_known_aggregated_payloads()
+            .into_iter()
+            .map(|(key, _)| key),
+    );
 
     // Resolve all fork labels to roots and compute their weights
     // Map: label -> (root, slot, weight)

crates/storage/src/api/tables.rs

@@ -16,12 +16,6 @@ pub enum Table {
     GossipSignatures,
     /// Attestation data indexed by tree hash root: H256 -> AttestationData
     AttestationDataByRoot,
-    /// Pending aggregated payloads (not yet active in fork choice):
-    /// SignatureKey -> Vec<StoredAggregatedPayload>
-    LatestNewAggregatedPayloads,
-    /// Active aggregated payloads (counted in fork choice):
-    /// SignatureKey -> Vec<StoredAggregatedPayload>
-    LatestKnownAggregatedPayloads,
     /// Metadata: string keys -> various scalar values
     Metadata,
     /// Live chain index: (slot || root) -> parent_root
@@ -33,15 +27,13 @@ pub enum Table {
 }
 
 /// All table variants.
-pub const ALL_TABLES: [Table; 10] = [
+pub const ALL_TABLES: [Table; 8] = [
     Table::BlockHeaders,
     Table::BlockBodies,
     Table::BlockSignatures,
     Table::States,
     Table::GossipSignatures,
     Table::AttestationDataByRoot,
-    Table::LatestNewAggregatedPayloads,
-    Table::LatestKnownAggregatedPayloads,
     Table::Metadata,
     Table::LiveChain,
 ];

crates/storage/src/backend/rocksdb.rs

@@ -18,8 +18,6 @@ fn cf_name(table: Table) -> &'static str {
         Table::States => "states",
         Table::GossipSignatures => "gossip_signatures",
         Table::AttestationDataByRoot => "attestation_data_by_root",
-        Table::LatestNewAggregatedPayloads => "latest_new_aggregated_payloads",
-        Table::LatestKnownAggregatedPayloads => "latest_known_aggregated_payloads",
         Table::Metadata => "metadata",
         Table::LiveChain => "live_chain",
     }