Generalize cold DB: ColdStore trait + slot-keyed static archive

.github/workflows/test-suite.yml

@@ -148,6 +148,31 @@ jobs:
           cache-provider: warpbuild
     - name: Run beacon_chain tests for all known forks
       run: make test-beacon-chain
+  beacon-chain-store-tests-static-cold:
+    name: beacon-chain-store-tests-static-cold
+    needs: [check-labels]
+    if: needs.check-labels.outputs.skip_ci != 'true'
+    runs-on: ${{ github.repository == 'sigp/lighthouse' && 'warp-ubuntu-latest-x64-8x;snapshot.key=lighthouse-ubuntu-latest-v1' || 'ubuntu-latest' }}
+    env:
+        GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+    steps:
+    - uses: actions/checkout@v5
+    - if: github.repository != 'sigp/lighthouse'
+      name: Get latest version of stable Rust
+      uses: moonrepo/setup-rust@v1
+      with:
+          channel: stable
+          cache-target: release
+          bins: cargo-nextest
+    - if: github.repository == 'sigp/lighthouse'
+      uses: Swatinem/rust-cache@v2
+      with:
+          cache-provider: warpbuild
+    - name: Run beacon_chain store_tests against the static cold backend
+      env:
+        COLD_BACKEND: static
+        FORK_NAME: fulu
+      run: cargo nextest run --release --features "fork_from_env,slasher/lmdb,$TEST_FEATURES" -p beacon_chain --test beacon_chain_tests -E 'test(/^store_tests::/)' --no-fail-fast
   http-api-tests:
     name: http-api-tests
     needs: [check-labels]
@@ -493,6 +518,7 @@ jobs:
       'forbidden-files-check',
       'release-tests-ubuntu',
       'beacon-chain-tests',
+      'beacon-chain-store-tests-static-cold',
       'op-pool-tests',
       'network-tests',
       'slasher-tests',

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -320,7 +320,7 @@ pub enum StateSkipConfig {
 
 pub trait BeaconChainTypes: Send + Sync + 'static {
     type HotStore: store::ItemStore<Self::EthSpec>;
-    type ColdStore: store::ItemStore<Self::EthSpec>;
+    type ColdStore: store::ColdStore<Self::EthSpec>;
     type SlotClock: slot_clock::SlotClock;
     type EthSpec: types::EthSpec;
 }

beacon_node/beacon_chain/src/beacon_fork_choice_store.rs

@@ -14,7 +14,7 @@ use ssz_derive::{Decode, Encode};
 use std::collections::BTreeSet;
 use std::marker::PhantomData;
 use std::sync::Arc;
-use store::{Error as StoreError, HotColdDB, ItemStore};
+use store::{ColdStore, Error as StoreError, HotColdDB, ItemStore};
 use superstruct::superstruct;
 use types::{
     AbstractExecPayload, BeaconBlockRef, BeaconState, BeaconStateError, Checkpoint, Epoch, EthSpec,
@@ -129,8 +129,8 @@ impl BalancesCache {
 /// Implements `fork_choice::ForkChoiceStore` in order to provide a persistent backing to the
 /// `fork_choice::ForkChoice` struct.
 #[derive(Debug, Educe)]
-#[educe(PartialEq(bound(E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>)))]
-pub struct BeaconForkChoiceStore<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> {
+#[educe(PartialEq(bound(E: EthSpec, Hot: ItemStore<E>, Cold: ColdStore<E>)))]
+pub struct BeaconForkChoiceStore<E: EthSpec, Hot: ItemStore<E>, Cold: ColdStore<E>> {
     #[educe(PartialEq(ignore))]
     store: Arc<HotColdDB<E, Hot, Cold>>,
     balances_cache: BalancesCache,
@@ -151,7 +151,7 @@ impl<E, Hot, Cold> BeaconForkChoiceStore<E, Hot, Cold>
 where
     E: EthSpec,
     Hot: ItemStore<E>,
-    Cold: ItemStore<E>,
+    Cold: ColdStore<E>,
 {
     /// Initialize `Self` from some `anchor` checkpoint which may or may not be the genesis state.
     ///
@@ -268,7 +268,7 @@ impl<E, Hot, Cold> ForkChoiceStore<E> for BeaconForkChoiceStore<E, Hot, Cold>
 where
     E: EthSpec,
     Hot: ItemStore<E>,
-    Cold: ItemStore<E>,
+    Cold: ColdStore<E>,
 {
     type Error = Error;
 

beacon_node/beacon_chain/src/builder.rs

@@ -40,7 +40,7 @@ use state_processing::per_slot_processing;
 use std::marker::PhantomData;
 use std::sync::Arc;
 use std::time::Duration;
-use store::{Error as StoreError, HotColdDB, ItemStore, KeyValueStoreOp};
+use store::{ColdStore, DBColumnCold, Error as StoreError, HotColdDB, ItemStore, KeyValueStoreOp};
 use task_executor::{ShutdownReason, TaskExecutor};
 use tracing::{debug, error, info, warn};
 use tree_hash::TreeHash;
@@ -60,7 +60,7 @@ impl<TSlotClock, E, THotStore, TColdStore> BeaconChainTypes
     for Witness<TSlotClock, E, THotStore, TColdStore>
 where
     THotStore: ItemStore<E> + 'static,
-    TColdStore: ItemStore<E> + 'static,
+    TColdStore: ColdStore<E> + 'static,
     TSlotClock: SlotClock + 'static,
     E: EthSpec + 'static,
 {
@@ -115,7 +115,7 @@ impl<TSlotClock, E, THotStore, TColdStore>
     BeaconChainBuilder<Witness<TSlotClock, E, THotStore, TColdStore>>
 where
     THotStore: ItemStore<E> + 'static,
-    TColdStore: ItemStore<E> + 'static,
+    TColdStore: ColdStore<E> + 'static,
     TSlotClock: SlotClock + 'static,
     E: EthSpec + 'static,
 {
@@ -340,7 +340,7 @@ where
             .map_err(|e| format!("Failed to store genesis block: {:?}", e))?;
         store
             .store_frozen_block_root_at_skip_slots(Slot::new(0), Slot::new(1), beacon_block_root)
-            .and_then(|ops| store.cold_db.do_atomically(ops))
+            .and_then(|ops| store.cold_db.put_batch(DBColumnCold::BlockRoots, ops))
             .map_err(|e| format!("Failed to store genesis block root: {e:?}"))?;
 
         // Store the genesis block under the `ZERO_HASH` key.
@@ -435,6 +435,21 @@ where
             .clone()
             .ok_or("weak_subjectivity_state requires a store")?;
 
+        // The static cold backend is append-only in ascending slot order. A
+        // checkpoint / weak-subjectivity start writes the anchor state in the
+        // middle of the chain and then backfills earlier slots, which the
+        // static format can't represent. Refuse the combination at startup
+        // rather than failing later with an out-of-order put.
+        if matches!(
+            store.get_config().cold_backend,
+            store::config::ColdBackendKind::Static
+        ) {
+            return Err("static cold backend only supports starting from genesis; \
+                 checkpoint sync and weak subjectivity sync require the kv \
+                 cold backend"
+                .to_string());
+        }
+
         // Ensure the state is advanced to an epoch boundary.
         let slots_per_epoch = E::slots_per_epoch();
         if weak_subj_state.slot() % slots_per_epoch != 0 {
@@ -558,7 +573,7 @@ where
             .map_err(|e| format!("Error writing frozen block roots: {e:?}"))?;
         store
             .cold_db
-            .do_atomically(block_root_batch)
+            .put_batch(DBColumnCold::BlockRoots, block_root_batch)
             .map_err(|e| format!("Error writing frozen block roots: {e:?}"))?;
         debug!(
             from = %weak_subj_block.slot(),
@@ -1152,7 +1167,7 @@ impl<E, THotStore, TColdStore>
     BeaconChainBuilder<Witness<TestingSlotClock, E, THotStore, TColdStore>>
 where
     THotStore: ItemStore<E> + 'static,
-    TColdStore: ItemStore<E> + 'static,
+    TColdStore: ColdStore<E> + 'static,
     E: EthSpec + 'static,
 {
     /// Sets the `BeaconChain` slot clock to `TestingSlotClock`.

beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs

@@ -791,7 +791,10 @@ mod test {
     use fork_choice::PayloadVerificationStatus;
     use logging::create_test_tracing_subscriber;
     use state_processing::ConsensusContext;
-    use store::{HotColdDB, ItemStore, StoreConfig, database::interface::BeaconNodeBackend};
+    use store::{
+        ColdStore, HotColdDB, ItemStore, StoreConfig,
+        database::interface::{BeaconNodeBackend, ColdBackend},
+    };
     use tempfile::{TempDir, tempdir};
     use tracing::info;
     use types::MinimalEthSpec;
@@ -802,7 +805,7 @@ mod test {
     fn get_store_with_spec<E: EthSpec>(
         db_path: &TempDir,
         spec: Arc<ChainSpec>,
-    ) -> Arc<HotColdDB<E, BeaconNodeBackend<E>, BeaconNodeBackend<E>>> {
+    ) -> Arc<HotColdDB<E, BeaconNodeBackend<E>, ColdBackend<E>>> {
         let hot_path = db_path.path().join("hot_db");
         let cold_path = db_path.path().join("cold_db");
         let blobs_path = db_path.path().join("blobs_db");
@@ -861,7 +864,7 @@ mod test {
     where
         E: EthSpec,
         Hot: ItemStore<E>,
-        Cold: ItemStore<E>,
+        Cold: ColdStore<E>,
     {
         let chain = &harness.chain;
         let head = chain.head_snapshot();
@@ -947,7 +950,7 @@ mod test {
         E: EthSpec,
         T: BeaconChainTypes<
                 HotStore = BeaconNodeBackend<E>,
-                ColdStore = BeaconNodeBackend<E>,
+                ColdStore = ColdBackend<E>,
                 EthSpec = E,
             >,
     {

beacon_node/beacon_chain/src/historical_blocks.rs

@@ -10,7 +10,10 @@ use std::borrow::Cow;
 use std::iter;
 use std::time::Duration;
 use store::metadata::DataColumnInfo;
-use store::{AnchorInfo, BlobInfo, DBColumn, Error as StoreError, KeyValueStore, KeyValueStoreOp};
+use store::{
+    AnchorInfo, BlobInfo, ColdStore, DBColumnCold, Error as StoreError, KeyValueStore,
+    KeyValueStoreOp,
+};
 use strum::IntoStaticStr;
 use tracing::{debug, debug_span, instrument};
 use types::{Hash256, Slot};
@@ -108,7 +111,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         let mut new_oldest_data_column_slot = data_column_info.oldest_data_column_slot;
 
         let mut blob_batch = Vec::<KeyValueStoreOp>::new();
-        let mut cold_batch = Vec::with_capacity(blocks_to_import.len());
+        let mut cold_batch: Vec<(Slot, Vec<u8>)> = Vec::with_capacity(blocks_to_import.len());
         let mut hot_batch = Vec::with_capacity(blocks_to_import.len());
         let mut signed_blocks = Vec::with_capacity(blocks_to_import.len());
 
@@ -174,11 +177,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             // Store block roots, including at all skip slots in the freezer DB.
             for slot in (block.slot().as_u64()..prev_block_slot.as_u64()).rev() {
                 debug!(%slot, ?block_root, "Storing frozen block to root mapping");
-                cold_batch.push(KeyValueStoreOp::PutKeyValue(
-                    DBColumn::BeaconBlockRoots,
-                    slot.to_be_bytes().to_vec(),
-                    block_root.as_slice().to_vec(),
-                ));
+                cold_batch.push((Slot::new(slot), block_root.as_slice().to_vec()));
             }
 
             prev_block_slot = block.slot();
@@ -191,11 +190,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             if expected_block_root == self.genesis_block_root {
                 let genesis_slot = self.spec.genesis_slot;
                 for slot in genesis_slot.as_u64()..prev_block_slot.as_u64() {
-                    cold_batch.push(KeyValueStoreOp::PutKeyValue(
-                        DBColumn::BeaconBlockRoots,
-                        slot.to_be_bytes().to_vec(),
-                        self.genesis_block_root.as_slice().to_vec(),
-                    ));
+                    cold_batch.push((Slot::new(slot), self.genesis_block_root.as_slice().to_vec()));
                 }
                 prev_block_slot = genesis_slot;
                 expected_block_root = Hash256::zero();
@@ -261,7 +256,9 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         }
         {
             let _span = debug_span!("backfill_write_cold_db").entered();
-            self.store.cold_db.do_atomically(cold_batch)?;
+            self.store
+                .cold_db
+                .put_batch(DBColumnCold::BlockRoots, cold_batch)?;
         }
 
         let mut anchor_and_blob_batch = Vec::with_capacity(3);

beacon_node/beacon_chain/src/migrate.rs

@@ -7,7 +7,7 @@ use std::sync::{Arc, mpsc};
 use std::thread;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use store::hot_cold_store::{HotColdDBError, migrate_database};
-use store::{Error, ItemStore, Split, StoreOp};
+use store::{ColdStore, Error, ItemStore, Split, StoreOp};
 pub use store::{HotColdDB, MemoryStore};
 use tracing::{debug, error, info, warn};
 use types::{BeaconState, BeaconStateHash, Checkpoint, Epoch, EthSpec, Hash256, Slot};
@@ -30,7 +30,7 @@ pub const DEFAULT_EPOCHS_PER_MIGRATION: u64 = 1;
 
 /// The background migrator runs a thread to perform pruning and migrate state from the hot
 /// to the cold database.
-pub struct BackgroundMigrator<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> {
+pub struct BackgroundMigrator<E: EthSpec, Hot: ItemStore<E>, Cold: ColdStore<E>> {
     db: Arc<HotColdDB<E, Hot, Cold>>,
     /// Record of when the last migration ran, for enforcing `epochs_per_migration`.
     prev_migration: Arc<Mutex<PrevMigration>>,
@@ -135,7 +135,7 @@ pub struct FinalizationNotification {
     pub prev_migration: Arc<Mutex<PrevMigration>>,
 }
 
-impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Hot, Cold> {
+impl<E: EthSpec, Hot: ItemStore<E>, Cold: ColdStore<E>> BackgroundMigrator<E, Hot, Cold> {
     /// Create a new `BackgroundMigrator` and spawn its thread if necessary.
     pub fn new(db: Arc<HotColdDB<E, Hot, Cold>>, config: MigratorConfig) -> Self {
         // Estimate last migration run from DB split slot.

beacon_node/beacon_chain/src/persisted_custody.rs

@@ -1,15 +1,15 @@
 use crate::custody_context::CustodyContextSsz;
 use ssz::{Decode, Encode};
 use std::sync::Arc;
-use store::{DBColumn, Error as StoreError, HotColdDB, ItemStore, StoreItem};
+use store::{ColdStore, DBColumn, Error as StoreError, HotColdDB, ItemStore, StoreItem};
 use types::{EthSpec, Hash256};
 
 /// 32-byte key for accessing the `CustodyContext`. All zero because `CustodyContext` has its own column.
 pub const CUSTODY_DB_KEY: Hash256 = Hash256::ZERO;
 
 pub struct PersistedCustody(pub CustodyContextSsz);
 
-pub fn load_custody_context<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>>(
+pub fn load_custody_context<E: EthSpec, Hot: ItemStore<E>, Cold: ColdStore<E>>(
     store: Arc<HotColdDB<E, Hot, Cold>>,
 ) -> Option<CustodyContextSsz> {
     let res: Result<Option<PersistedCustody>, _> =
@@ -22,7 +22,7 @@ pub fn load_custody_context<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>>(
 }
 
 /// Attempt to persist the custody context object to `self.store`.
-pub fn persist_custody_context<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>>(
+pub fn persist_custody_context<E: EthSpec, Hot: ItemStore<E>, Cold: ColdStore<E>>(
     store: Arc<HotColdDB<E, Hot, Cold>>,
     custody_context: CustodyContextSsz,
 ) -> Result<(), store::Error> {

beacon_node/beacon_chain/src/test_utils.rs

@@ -65,8 +65,8 @@ use std::str::FromStr;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, LazyLock};
 use std::time::Duration;
-use store::database::interface::BeaconNodeBackend;
-use store::{HotColdDB, ItemStore, MemoryStore, config::StoreConfig};
+use store::database::interface::{BeaconNodeBackend, ColdBackend};
+use store::{ColdStore, HotColdDB, ItemStore, MemoryStore, config::StoreConfig};
 use task_executor::TaskExecutor;
 use task_executor::{ShutdownReason, test_utils::TestRuntime};
 use tracing::debug;
@@ -124,7 +124,7 @@ pub fn get_kzg(spec: &ChainSpec) -> Arc<Kzg> {
 pub type BaseHarnessType<E, THotStore, TColdStore> =
     Witness<TestingSlotClock, E, THotStore, TColdStore>;
 
-pub type DiskHarnessType<E> = BaseHarnessType<E, BeaconNodeBackend<E>, BeaconNodeBackend<E>>;
+pub type DiskHarnessType<E> = BaseHarnessType<E, BeaconNodeBackend<E>, ColdBackend<E>>;
 pub type EphemeralHarnessType<E> = BaseHarnessType<E, MemoryStore<E>, MemoryStore<E>>;
 
 pub type BoxedMutator<E, Hot, Cold> = Box<
@@ -350,7 +350,7 @@ impl<E: EthSpec> Builder<DiskHarnessType<E>> {
     /// Disk store, start from genesis.
     pub fn fresh_disk_store(
         mut self,
-        store: Arc<HotColdDB<E, BeaconNodeBackend<E>, BeaconNodeBackend<E>>>,
+        store: Arc<HotColdDB<E, BeaconNodeBackend<E>, ColdBackend<E>>>,
     ) -> Self {
         let validator_keypairs = self
             .validator_keypairs
@@ -384,7 +384,7 @@ impl<E: EthSpec> Builder<DiskHarnessType<E>> {
     /// Disk store, resume.
     pub fn resumed_disk_store(
         mut self,
-        store: Arc<HotColdDB<E, BeaconNodeBackend<E>, BeaconNodeBackend<E>>>,
+        store: Arc<HotColdDB<E, BeaconNodeBackend<E>, ColdBackend<E>>>,
     ) -> Self {
         let mutator = move |builder: BeaconChainBuilder<_>| {
             builder
@@ -400,7 +400,7 @@ impl<E, Hot, Cold> Builder<BaseHarnessType<E, Hot, Cold>>
 where
     E: EthSpec,
     Hot: ItemStore<E>,
-    Cold: ItemStore<E>,
+    Cold: ColdStore<E>,
 {
     pub fn new(eth_spec_instance: E) -> Self {
         let runtime = TestRuntime::default();
@@ -761,7 +761,7 @@ impl<E, Hot, Cold> BeaconChainHarness<BaseHarnessType<E, Hot, Cold>>
 where
     E: EthSpec,
     Hot: ItemStore<E>,
-    Cold: ItemStore<E>,
+    Cold: ColdStore<E>,
 {
     pub fn builder(eth_spec_instance: E) -> Builder<BaseHarnessType<E, Hot, Cold>> {
         create_test_tracing_subscriber();

beacon_node/beacon_chain/tests/op_verification.rs

@@ -15,7 +15,7 @@ use state_processing::per_block_processing::errors::{
 };
 use std::sync::{Arc, LazyLock};
 use store::StoreConfig;
-use store::database::interface::BeaconNodeBackend;
+use store::database::interface::{BeaconNodeBackend, ColdBackend};
 use tempfile::{TempDir, tempdir};
 use types::*;
 
@@ -27,7 +27,7 @@ static KEYPAIRS: LazyLock<Vec<Keypair>> =
 
 type E = MinimalEthSpec;
 type TestHarness = BeaconChainHarness<DiskHarnessType<E>>;
-type HotColdDB = store::HotColdDB<E, BeaconNodeBackend<E>, BeaconNodeBackend<E>>;
+type HotColdDB = store::HotColdDB<E, BeaconNodeBackend<E>, ColdBackend<E>>;
 
 fn get_store(db_path: &TempDir) -> Arc<HotColdDB> {
     let spec = Arc::new(test_spec::<E>());