feat: make store begin_read() callers return Result

bin/ethlambda/src/main.rs

@@ -671,7 +671,7 @@ async fn fetch_initial_state(
     // Checkpoint sync path
 
     // Prefer resuming from a fresh on-disk state to avoid re-downloading what we already have.
-    if let Some(store) = Store::from_db_state(backend.clone(), genesis.genesis_time) {
+    if let Ok(Some(store)) = Store::from_db_state(backend.clone(), genesis.genesis_time) {
         let now_ms = SystemTime::UNIX_EPOCH
             .elapsed()
             .expect("already past the unix epoch")

crates/blockchain/src/coverage.rs

@@ -130,7 +130,7 @@ pub(crate) fn emit_post_block_coverage(
     // the head is normally the block proposed at `reporting_slot + 1`, which
     // carries this round's votes; filter by `data.slot` so we count the same
     // cohort even if the head is at a different slot.
-    if let Some(block) = store.get_block(&store.head()) {
+    if let Ok(Some(block)) = store.get_block(&store.head().expect("head exists")) {
         for att in block.body.attestations.iter() {
             if att.data.slot == reporting_slot {
                 cov_add(&mut block_v, &mut block_s, &att.aggregation_bits);

crates/blockchain/src/lib.rs

@@ -140,7 +140,10 @@ impl BlockChain {
 
         metrics::set_is_aggregator(aggregator.is_enabled());
         metrics::set_node_sync_status(metrics::SyncStatus::Idle);
-        let genesis_time = store.config().genesis_time;
+        let genesis_time = store
+            .config()
+            .expect("failed to load config: config missing or database error")
+            .genesis_time;
         let mut key_manager = key_manager::KeyManager::new(validator_keys);
 
         // Catch XMSS keys up to the current slot before the first tick
@@ -256,7 +259,7 @@ pub struct BlockChainServer {
 
 impl BlockChainServer {
     async fn on_tick(&mut self, timestamp_ms: u64, ctx: &Context<Self>) {
-        let genesis_time_ms = self.store.config().genesis_time * 1000;
+        let genesis_time_ms = self.store.config().expect("config exists").genesis_time * 1000;
 
         // Calculate current slot and interval from milliseconds
         let time_since_genesis_ms = timestamp_ms.saturating_sub(genesis_time_ms);
@@ -270,7 +273,7 @@ impl BlockChainServer {
         // inside VMs, so a tick scheduled for the next interval boundary can fire
         // while the wall clock still reads the previous interval.
         let tick_interval = time_since_genesis_ms / MILLISECONDS_PER_INTERVAL;
-        let store_time = self.store.time();
+        let store_time = self.store.time().expect("store time exists");
 
         if store_time > 0 && tick_interval <= store_time {
             debug!(
@@ -476,7 +479,7 @@ impl BlockChainServer {
         };
 
         let session_id = slot;
-        let genesis_time_ms = self.store.config().genesis_time * 1000;
+        let genesis_time_ms = self.store.config().expect("config exists").genesis_time * 1000;
         let t2_ms = genesis_time_ms + slot * MILLISECONDS_PER_SLOT + 2 * MILLISECONDS_PER_INTERVAL;
         // Interval-2 boundary as a wall-clock instant; the worker holds each
         // produced aggregate until this before sending it back, so nothing
@@ -545,7 +548,7 @@ impl BlockChainServer {
         // Only fire inside the early-aggregation window
         // `[T2 - EARLY_AGGREGATION_WINDOW, T2)`, where T2 is the current
         // slot's interval-2 boundary; the slot is derived from the wall clock.
-        let genesis_time_ms = self.store.config().genesis_time * 1000;
+        let genesis_time_ms = self.store.config().expect("config exists").genesis_time * 1000;
         let Some(ms_since_genesis) = unix_now_ms().checked_sub(genesis_time_ms) else {
             return;
         };
@@ -675,7 +678,7 @@ impl BlockChainServer {
     async fn propose_block(&mut self, slot: u64, validator_id: u64) {
         info!(%slot, %validator_id, "We are the proposer for this slot");
 
-        let genesis_time_ms = self.store.config().genesis_time * 1000;
+        let genesis_time_ms = self.store.config().expect("config exists").genesis_time * 1000;
         let slot_start_ms = genesis_time_ms + slot * MILLISECONDS_PER_SLOT;
 
         // Build the block. `produce_block_with_signatures` advances the store to
@@ -863,8 +866,18 @@ impl BlockChainServer {
     fn process_block(&mut self, signed_block: SignedBlock) -> Result<(), StoreError> {
         store::on_block(&mut self.store, signed_block)?;
         metrics::update_head_slot(self.store.head_slot());
-        metrics::update_latest_justified_slot(self.store.latest_justified().slot);
-        metrics::update_latest_finalized_slot(self.store.latest_finalized().slot);
+        let latest_justified_slot = self
+            .store
+            .latest_justified()
+            .expect("Error: Latest justified checkpoint does not exist")
+            .slot;
+        metrics::update_latest_justified_slot(latest_justified_slot);
+        let latest_finalized_slot = self
+            .store
+            .latest_finalized()
+            .expect("Error: Latest finalized checkpoint does not exist")
+            .slot;
+        metrics::update_latest_finalized_slot(latest_finalized_slot);
         metrics::update_validators_count(self.key_manager.validator_ids().len() as u64);
 
         for table in ALL_TABLES {
@@ -888,7 +901,9 @@ impl BlockChainServer {
         // Prune old states and blocks AFTER the entire cascade completes.
         // Running this mid-cascade would delete states that pending children
         // still need, causing re-processing loops when fallback pruning is active.
-        self.store.prune_old_data();
+        self.store
+            .prune_old_data()
+            .expect("DB pruning should succeed");
     }
 
     /// Try to process a single block. If its parent state is missing, store it
@@ -908,7 +923,12 @@ impl BlockChainServer {
         // already part of the canonical chain and cannot affect fork choice.
         // Discard any pending children: since we won't process this block,
         // children referencing it as parent would remain stuck indefinitely.
-        if slot <= self.store.latest_finalized().slot {
+        let latest_finalized_slot = self
+            .store
+            .latest_finalized()
+            .expect("Error: Latest finalized checkpoint does not exist")
+            .slot;
+        if slot <= latest_finalized_slot {
             self.discard_pending_subtree(block_root);
             return;
         }
@@ -920,7 +940,7 @@ impl BlockChainServer {
         // Catching this early also avoids persisting bogus future blocks to
         // RocksDB and triggering BlocksByRoot fan-out for fabricated parents.
         let block_start_interval = slot.saturating_mul(INTERVALS_PER_SLOT);
-        let store_time = self.store.time();
+        let store_time = self.store.time().expect("store time exists");
         if block_start_interval > store_time + GOSSIP_DISPARITY_INTERVALS {
             warn!(
                 %slot,
@@ -935,7 +955,11 @@ impl BlockChainServer {
         }
 
         // Check if parent state exists before attempting to process
-        if !self.store.has_state(&parent_root) {
+        if !self
+            .store
+            .has_state(&parent_root)
+            .expect("DB read should succeed")
+        {
             info!(%slot, %parent_root, %block_root, "Block parent missing, storing as pending");
 
             // Resolve the actual missing ancestor by walking the chain. A stale entry
@@ -963,14 +987,23 @@ impl BlockChainServer {
             // session, the actual missing block is further up the chain.
             // Note: this loop always terminates — blocks reference parents by hash,
             // so a cycle would require a hash collision.
-            while let Some(header) = self.store.get_block_header(&missing_root) {
-                if self.store.has_state(&header.parent_root) {
+            while let Some(header) = self
+                .store
+                .get_block_header(&missing_root)
+                .expect("DB read should succeed")
+            {
+                if self
+                    .store
+                    .has_state(&header.parent_root)
+                    .expect("DB read should succeed")
+                {
                     // Parent state available — enqueue for processing, cascade
                     // handles the rest via the outer loop.
                     let block = self
                         .store
                         .get_signed_block(&missing_root)
-                        .expect("header and parent state exist, so the full signed block must too");
+                        .expect("header and parent state exist, so the full signed block must too")
+                        .unwrap();
                     queue.push_back(block);
                     return;
                 }
@@ -1082,7 +1115,7 @@ impl BlockChainServer {
             self.pending_block_parents.remove(&block_root);
 
             // Load block data from DB
-            let Some(child_block) = self.store.get_signed_block(&block_root) else {
+            let Ok(Some(child_block)) = self.store.get_signed_block(&block_root) else {
                 warn!(
                     block_root = %ShortRoot(&block_root.0),
                     "Pending block missing from DB, skipping"
@@ -1127,7 +1160,11 @@ impl BlockChainServer {
 
     fn update_sync_status(&mut self, current_slot: u64) {
         let head_slot = self.store.head_slot();
-        let max_seen_slot = self.store.max_live_chain_slot().unwrap_or(head_slot);
+        let max_seen_slot = self
+            .store
+            .max_live_chain_slot()
+            .expect("max live chain slot exists")
+            .unwrap_or(head_slot);
         let status = self
             .sync_status
             .update(current_slot, head_slot, max_seen_slot);
@@ -1152,7 +1189,7 @@ impl BlockChainServer {
         let now_ms = unix_now_ms();
         self.on_tick(now_ms, ctx).await;
 
-        let genesis_time_ms = self.store.config().genesis_time * 1000;
+        let genesis_time_ms = self.store.config().expect("Config exists").genesis_time * 1000;
         let remaining_at_entry = ms_until_next_interval(now_ms, genesis_time_ms);
         let now_after_tick = unix_now_ms();
         let elapsed = now_after_tick.saturating_sub(now_ms);
@@ -1223,7 +1260,7 @@ impl Handler<NewAttestation> for BlockChainServer {
         // Early aggregation only advances the current slot's group counts, so a
         // late- or future-slot attestation can never cross the threshold; skip
         // the check unless this attestation is for the store's current slot.
-        let current_slot = self.store.time() / INTERVALS_PER_SLOT;
+        let current_slot = self.store.time().expect("store time exists") / INTERVALS_PER_SLOT;
         if msg.attestation.data.slot == current_slot {
             self.maybe_start_early_aggregation(ctx).await;
         }

crates/blockchain/src/reaggregate.rs

@@ -61,7 +61,7 @@ pub fn reaggregate_from_block(
     // The multi-message aggregate proof was built against the parent state's
     // validator set. Without it we cannot resolve the pubkey layout the SNARK
     // was bound to.
-    let Some(parent_state) = store.get_state(&block.parent_root) else {
+    let Ok(Some(parent_state)) = store.get_state(&block.parent_root) else {
         debug!(
             block_root = %ethlambda_types::ShortRoot(&block.hash_tree_root().0),
             "Skipping reaggregation: parent state missing"
@@ -239,7 +239,10 @@ struct Candidate {
 /// capped at [`MAX_REAGGREGATIONS_PER_BLOCK`] so an attacker-shaped block
 /// cannot blow past the slot budget.
 fn select_candidates(store: &Store, attestations: &[AggregatedAttestation]) -> Vec<Candidate> {
-    let justified_slot = store.latest_justified().slot;
+    let justified_slot = store
+        .latest_justified()
+        .expect("latest justified checkpoint exists")
+        .slot;
     let mut candidates: Vec<Candidate> = Vec::new();
     for (idx, att) in attestations.iter().enumerate() {
         if att.data.target.slot <= justified_slot {
@@ -315,7 +318,7 @@ mod tests {
         // Justified at slot 5; an attestation with target.slot = 5 must be skipped.
         store
             .update_checkpoints(ethlambda_storage::ForkCheckpoints::new(
-                store.head(),
+                store.head().expect("head exists"),
                 Some(Checkpoint {
                     root: H256::ZERO,
                     slot: 5,