Fix bug in block root storage (sigp#4663)

Fix a bug in the storage of the linear block roots array in the freezer DB. Previously this array was always written as part of state storage (or block backfill). With state pruning enabled by sigp#4610, these states were no longer being written and as a result neither were the block roots.

The impact is quite low, we would just log an error when trying to forwards-iterate the block roots, which for validating nodes only happens when they try to look up blocks for peers:

> Aug 25 03:42:36.980 ERRO Missing chunk in forwards iterator      chunk index: 49726, service: freezer_db

Any node checkpoint synced off `unstable` is affected and has a corrupt database. If you see the log above, you need to re-sync with the fix. Nodes that haven't checkpoint synced recently should _not_ be corrupted, even if they ran the buggy version.

- Use a `ChunkWriter` to write the block roots when states are not being stored.
- Tweak the usage of `get_latest_restore_point` so that it doesn't return a nonsense value when state pruning is enabled.
- Tweak the guarantee on the block roots array so that block roots are assumed available up to the split slot (exclusive). This is a bit nicer than relying on anything to do with the latest restore point, which is a nonsensical concept when there aren't any restore points.

I'm looking forward to deleting the chunked vector code for good when we merge tree-states :grin:

Author: michaelsproul

Cargo.lock

@@ -485,6 +485,21 @@ where
         let (_, updated_builder) = self.set_genesis_state(genesis_state)?;
         self = updated_builder;
 
+        // Fill in the linear block roots between the checkpoint block's slot and the aligned
+        // state's slot. All slots less than the block's slot will be handled by block backfill,
+        // while states greater or equal to the checkpoint state will be handled by `migrate_db`.
+        let block_root_batch = store
+            .store_frozen_block_root_at_skip_slots(
+                weak_subj_block.slot(),
+                weak_subj_state.slot(),
+                weak_subj_block_root,
+            )
+            .map_err(|e| format!("Error writing frozen block roots: {e:?}"))?;
+        store
+            .cold_db
+            .do_atomically(block_root_batch)
+            .map_err(|e| format!("Error writing frozen block roots: {e:?}"))?;
+
         // Write the state and block non-atomically, it doesn't matter if they're forgotten
         // about on a crash restart.
         store

beacon_node/beacon_chain/src/builder.rs

@@ -417,7 +417,7 @@ async fn forwards_iter_block_and_state_roots_until() {
 
     // The last restore point slot is the point at which the hybrid forwards iterator behaviour
     // changes.
-    let last_restore_point_slot = store.get_latest_restore_point_slot();
+    let last_restore_point_slot = store.get_latest_restore_point_slot().unwrap();
     assert!(last_restore_point_slot > 0);
 
     let chain = &harness.chain;

beacon_node/beacon_chain/tests/store_tests.rs

@@ -30,16 +30,16 @@ where
     /// Create a new iterator which can yield elements from `start_vindex` up to the last
     /// index stored by the restore point at `last_restore_point_slot`.
     ///
-    /// The `last_restore_point` slot should be the slot of a recent restore point as obtained from
-    /// `HotColdDB::get_latest_restore_point_slot`. We pass it as a parameter so that the caller can
+    /// The `freezer_upper_limit` slot should be the slot of a recent restore point as obtained from
+    /// `Root::freezer_upper_limit`. We pass it as a parameter so that the caller can
     /// maintain a stable view of the database (see `HybridForwardsBlockRootsIterator`).
     pub fn new(
         store: &'a HotColdDB<E, Hot, Cold>,
         start_vindex: usize,
-        last_restore_point_slot: Slot,
+        freezer_upper_limit: Slot,
         spec: &ChainSpec,
     ) -> Self {
-        let (_, end_vindex) = F::start_and_end_vindex(last_restore_point_slot, spec);
+        let (_, end_vindex) = F::start_and_end_vindex(freezer_upper_limit, spec);
 
         // Set the next chunk to the one containing `start_vindex`.
         let next_cindex = start_vindex / F::chunk_size();

beacon_node/store/src/chunked_iter.rs

@@ -19,6 +19,14 @@ pub trait Root<E: EthSpec>: Field<E, Value = Hash256> {
         end_state: BeaconState<E>,
         end_root: Hash256,
     ) -> Result<SimpleForwardsIterator>;
+
+    /// The first slot for which this field is *no longer* stored in the freezer database.
+    ///
+    /// If `None`, then this field is not stored in the freezer database at all due to pruning
+    /// configuration.
+    fn freezer_upper_limit<Hot: ItemStore<E>, Cold: ItemStore<E>>(
+        store: &HotColdDB<E, Hot, Cold>,
+    ) -> Option<Slot>;
 }
 
 impl<E: EthSpec> Root<E> for BlockRoots {
@@ -39,6 +47,13 @@ impl<E: EthSpec> Root<E> for BlockRoots {
         )?;
         Ok(SimpleForwardsIterator { values })
     }
+
+    fn freezer_upper_limit<Hot: ItemStore<E>, Cold: ItemStore<E>>(
+        store: &HotColdDB<E, Hot, Cold>,
+    ) -> Option<Slot> {
+        // Block roots are stored for all slots up to the split slot (exclusive).
+        Some(store.get_split_slot())
+    }
 }
 
 impl<E: EthSpec> Root<E> for StateRoots {
@@ -59,6 +74,15 @@ impl<E: EthSpec> Root<E> for StateRoots {
         )?;
         Ok(SimpleForwardsIterator { values })
     }
+
+    fn freezer_upper_limit<Hot: ItemStore<E>, Cold: ItemStore<E>>(
+        store: &HotColdDB<E, Hot, Cold>,
+    ) -> Option<Slot> {
+        // State roots are stored for all slots up to the latest restore point (exclusive).
+        // There may not be a latest restore point if state pruning is enabled, in which
+        // case this function will return `None`.
+        store.get_latest_restore_point_slot()
+    }
 }
 
 /// Forwards root iterator that makes use of a flat field table in the freezer DB.
@@ -118,6 +142,7 @@ impl Iterator for SimpleForwardsIterator {
 pub enum HybridForwardsIterator<'a, E: EthSpec, F: Root<E>, Hot: ItemStore<E>, Cold: ItemStore<E>> {
     PreFinalization {
         iter: Box<FrozenForwardsIterator<'a, E, F, Hot, Cold>>,
+        end_slot: Option<Slot>,
         /// Data required by the `PostFinalization` iterator when we get to it.
         continuation_data: Option<Box<(BeaconState<E>, Hash256)>>,
     },
@@ -129,6 +154,7 @@ pub enum HybridForwardsIterator<'a, E: EthSpec, F: Root<E>, Hot: ItemStore<E>, C
     PostFinalization {
         iter: SimpleForwardsIterator,
     },
+    Finished,
 }
 
 impl<'a, E: EthSpec, F: Root<E>, Hot: ItemStore<E>, Cold: ItemStore<E>>
@@ -138,8 +164,8 @@ impl<'a, E: EthSpec, F: Root<E>, Hot: ItemStore<E>, Cold: ItemStore<E>>
     ///
     /// The `get_state` closure should return a beacon state and final block/state root to backtrack
     /// from in the case where the iterated range does not lie entirely within the frozen portion of
-    /// the database. If an `end_slot` is provided and it is before the database's latest restore
-    /// point slot then the `get_state` closure will not be called at all.
+    /// the database. If an `end_slot` is provided and it is before the database's freezer upper
+    /// limit for the field then the `get_state` closure will not be called at all.
     ///
     /// It is OK for `get_state` to hold a lock while this function is evaluated, as the returned
     /// iterator is as lazy as possible and won't do any work apart from calling `get_state`.
@@ -155,27 +181,30 @@ impl<'a, E: EthSpec, F: Root<E>, Hot: ItemStore<E>, Cold: ItemStore<E>>
     ) -> Result<Self> {
         use HybridForwardsIterator::*;
 
-        let latest_restore_point_slot = store.get_latest_restore_point_slot();
+        // First slot at which this field is *not* available in the freezer. i.e. all slots less
+        // than this slot have their data available in the freezer.
+        let freezer_upper_limit = F::freezer_upper_limit(store).unwrap_or(Slot::new(0));
 
-        let result = if start_slot < latest_restore_point_slot {
+        let result = if start_slot < freezer_upper_limit {
             let iter = Box::new(FrozenForwardsIterator::new(
                 store,
                 start_slot,
-                latest_restore_point_slot,
+                freezer_upper_limit,
                 spec,
             ));
 
             // No continuation data is needed if the forwards iterator plans to halt before
             // `end_slot`. If it tries to continue further a `NoContinuationData` error will be
             // returned.
             let continuation_data =
-                if end_slot.map_or(false, |end_slot| end_slot < latest_restore_point_slot) {
+                if end_slot.map_or(false, |end_slot| end_slot < freezer_upper_limit) {
                     None
                 } else {
                     Some(Box::new(get_state()?))
                 };
             PreFinalization {
                 iter,
+                end_slot,
                 continuation_data,
             }
         } else {
@@ -195,6 +224,7 @@ impl<'a, E: EthSpec, F: Root<E>, Hot: ItemStore<E>, Cold: ItemStore<E>>
         match self {
             PreFinalization {
                 iter,
+                end_slot,
                 continuation_data,
             } => {
                 match iter.next() {
@@ -203,10 +233,17 @@ impl<'a, E: EthSpec, F: Root<E>, Hot: ItemStore<E>, Cold: ItemStore<E>>
                     // to a post-finalization iterator beginning from the last slot
                     // of the pre iterator.
                     None => {
+                        // If the iterator has an end slot (inclusive) which has already been
+                        // covered by the (exclusive) frozen forwards iterator, then we're done!
+                        let iter_end_slot = Slot::from(iter.inner.end_vindex);
+                        if end_slot.map_or(false, |end_slot| iter_end_slot == end_slot + 1) {
+                            *self = Finished;
+                            return Ok(None);
+                        }
+
                         let continuation_data = continuation_data.take();
                         let store = iter.inner.store;
-                        let start_slot = Slot::from(iter.inner.end_vindex);
-
+                        let start_slot = iter_end_slot;
                         *self = PostFinalizationLazy {
                             continuation_data,
                             store,
@@ -230,6 +267,7 @@ impl<'a, E: EthSpec, F: Root<E>, Hot: ItemStore<E>, Cold: ItemStore<E>>
                 self.do_next()
             }
             PostFinalization { iter } => iter.next().transpose(),
+            Finished => Ok(None),
         }
     }
 }

beacon_node/store/src/forwards_iter.rs

@@ -45,3 +45,4 @@ network = { path = "../beacon_node/network" }
 testcontainers = { git = "https://github.com/testcontainers/testcontainers-rs/", rev = "0f2c9851" }
 unused_port = { path = "../common/unused_port" }
 task_executor = { path = "../common/task_executor" }
+logging = { path = "../common/logging" }

watch/Cargo.toml

@@ -7,12 +7,21 @@ use beacon_chain::{
 };
 use eth2::{types::BlockId, BeaconNodeHttpClient, SensitiveUrl, Timeouts};
 use http_api::test_utils::{create_api_server, ApiServer};
+use log::error;
+use logging::test_logger;
 use network::NetworkReceivers;
-
 use rand::distributions::Alphanumeric;
 use rand::{thread_rng, Rng};
+use std::collections::HashMap;
+use std::env;
+use std::net::SocketAddr;
+use std::time::Duration;
+use testcontainers::{clients::Cli, core::WaitFor, Image, RunnableImage};
 use tokio::sync::oneshot;
+use tokio::{runtime, task::JoinHandle};
+use tokio_postgres::{config::Config as PostgresConfig, Client, NoTls};
 use types::{Hash256, MainnetEthSpec, Slot};
+use unused_port::unused_tcp4_port;
 use url::Url;
 use watch::{
     client::WatchHttpClient,
@@ -22,17 +31,6 @@ use watch::{
     updater::{handler::*, run_updater, Config as UpdaterConfig, WatchSpec},
 };
 
-use log::error;
-use std::collections::HashMap;
-use std::env;
-use std::net::SocketAddr;
-use std::time::Duration;
-use tokio::{runtime, task::JoinHandle};
-use tokio_postgres::{config::Config as PostgresConfig, Client, NoTls};
-use unused_port::unused_tcp4_port;
-
-use testcontainers::{clients::Cli, core::WaitFor, Image, RunnableImage};
-
 #[derive(Debug)]
 pub struct Postgres(HashMap<String, String>);
 
@@ -132,6 +130,7 @@ impl TesterBuilder {
                 reconstruct_historic_states: true,
                 ..ChainConfig::default()
             })
+            .logger(test_logger())
             .deterministic_keypairs(VALIDATOR_COUNT)
             .fresh_ephemeral_store()
             .build();