diff --git a/Cargo.lock b/Cargo.lock index 75c15ba781..27d7b655e3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9051,6 +9051,7 @@ dependencies = [ "http_api", "hyper", "log", + "logging", "network", "r2d2", "rand 0.7.3", diff --git a/beacon_node/beacon_chain/src/builder.rs b/beacon_node/beacon_chain/src/builder.rs index bb629d6624..54739f2b8a 100644 --- a/beacon_node/beacon_chain/src/builder.rs +++ b/beacon_node/beacon_chain/src/builder.rs @@ -481,6 +481,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 diff --git a/beacon_node/beacon_chain/tests/store_tests.rs b/beacon_node/beacon_chain/tests/store_tests.rs index bf68657b4f..ab54af42c7 100644 --- a/beacon_node/beacon_chain/tests/store_tests.rs +++ b/beacon_node/beacon_chain/tests/store_tests.rs @@ -421,7 +421,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; diff --git a/beacon_node/store/src/chunked_iter.rs b/beacon_node/store/src/chunked_iter.rs index 8ef0b6d201..b3322b5225 100644 --- a/beacon_node/store/src/chunked_iter.rs +++ b/beacon_node/store/src/chunked_iter.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, 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(); diff --git a/beacon_node/store/src/forwards_iter.rs b/beacon_node/store/src/forwards_iter.rs index 353be6bf05..125b73a458 100644 --- a/beacon_node/store/src/forwards_iter.rs +++ b/beacon_node/store/src/forwards_iter.rs @@ -19,6 +19,14 @@ pub trait Root: Field { end_state: BeaconState, end_root: Hash256, ) -> Result; + + /// 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, Cold: ItemStore>( + store: &HotColdDB, + ) -> Option; } impl Root for BlockRoots { @@ -39,6 +47,13 @@ impl Root for BlockRoots { )?; Ok(SimpleForwardsIterator { values }) } + + fn freezer_upper_limit, Cold: ItemStore>( + store: &HotColdDB, + ) -> Option { + // Block roots are stored for all slots up to the split slot (exclusive). + Some(store.get_split_slot()) + } } impl Root for StateRoots { @@ -59,6 +74,15 @@ impl Root for StateRoots { )?; Ok(SimpleForwardsIterator { values }) } + + fn freezer_upper_limit, Cold: ItemStore>( + store: &HotColdDB, + ) -> Option { + // 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, Hot: ItemStore, Cold: ItemStore> { PreFinalization { iter: Box>, + end_slot: Option, /// Data required by the `PostFinalization` iterator when we get to it. continuation_data: Option, Hash256)>>, }, @@ -129,6 +154,7 @@ pub enum HybridForwardsIterator<'a, E: EthSpec, F: Root, Hot: ItemStore, C PostFinalization { iter: SimpleForwardsIterator, }, + Finished, } impl<'a, E: EthSpec, F: Root, Hot: ItemStore, Cold: ItemStore> @@ -138,8 +164,8 @@ impl<'a, E: EthSpec, F: Root, Hot: ItemStore, Cold: ItemStore> /// /// 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,13 +181,15 @@ impl<'a, E: EthSpec, F: Root, Hot: ItemStore, Cold: ItemStore> ) -> Result { 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, )); @@ -169,13 +197,14 @@ impl<'a, E: EthSpec, F: Root, Hot: ItemStore, Cold: ItemStore> // `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, Hot: ItemStore, Cold: ItemStore> match self { PreFinalization { iter, + end_slot, continuation_data, } => { match iter.next() { @@ -203,10 +233,17 @@ impl<'a, E: EthSpec, F: Root, Hot: ItemStore, Cold: ItemStore> // 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, Hot: ItemStore, Cold: ItemStore> self.do_next() } PostFinalization { iter } => iter.next().transpose(), + Finished => Ok(None), } } } diff --git a/beacon_node/store/src/hot_cold_store.rs b/beacon_node/store/src/hot_cold_store.rs index 47839ed3e9..87f8e0ffc3 100644 --- a/beacon_node/store/src/hot_cold_store.rs +++ b/beacon_node/store/src/hot_cold_store.rs @@ -18,7 +18,7 @@ use crate::metadata::{ }; use crate::metrics; use crate::{ - get_key_for_col, DBColumn, DatabaseBlock, Error, ItemStore, KeyValueStoreOp, + get_key_for_col, ChunkWriter, DBColumn, DatabaseBlock, Error, ItemStore, KeyValueStoreOp, PartialBeaconState, StoreItem, StoreOp, }; use itertools::process_results; @@ -963,6 +963,9 @@ impl, Cold: ItemStore> HotColdDB ops.push(op); // 2. Store updated vector entries. + // Block roots need to be written here as well as by the `ChunkWriter` in `migrate_db` + // because states may require older block roots, and the writer only stores block roots + // between the previous split point and the new split point. let db = &self.cold_db; store_updated_vector(BlockRoots, db, state, &self.spec, ops)?; store_updated_vector(StateRoots, db, state, &self.spec, ops)?; @@ -1243,10 +1246,21 @@ impl, Cold: ItemStore> HotColdDB }; } - /// Fetch the slot of the most recently stored restore point. - pub fn get_latest_restore_point_slot(&self) -> Slot { - (self.get_split_slot() - 1) / self.config.slots_per_restore_point - * self.config.slots_per_restore_point + /// Fetch the slot of the most recently stored restore point (if any). + pub fn get_latest_restore_point_slot(&self) -> Option { + let split_slot = self.get_split_slot(); + let anchor = self.get_anchor_info(); + + // There are no restore points stored if the state upper limit lies in the hot database. + // It hasn't been reached yet, and may never be. + if anchor.map_or(false, |a| a.state_upper_limit >= split_slot) { + None + } else { + Some( + (split_slot - 1) / self.config.slots_per_restore_point + * self.config.slots_per_restore_point, + ) + } } /// Load the database schema version from disk. @@ -1585,6 +1599,25 @@ impl, Cold: ItemStore> HotColdDB ) } + /// Update the linear array of frozen block roots with the block root for several skipped slots. + /// + /// Write the block root at all slots from `start_slot` (inclusive) to `end_slot` (exclusive). + pub fn store_frozen_block_root_at_skip_slots( + &self, + start_slot: Slot, + end_slot: Slot, + block_root: Hash256, + ) -> Result, Error> { + let mut ops = vec![]; + let mut block_root_writer = + ChunkWriter::::new(&self.cold_db, start_slot.as_usize())?; + for slot in start_slot.as_usize()..end_slot.as_usize() { + block_root_writer.set(slot, block_root, &mut ops)?; + } + block_root_writer.write(&mut ops)?; + Ok(ops) + } + /// Try to prune all execution payloads, returning early if there is no need to prune. pub fn try_prune_execution_payloads(&self, force: bool) -> Result<(), Error> { let split = self.get_split_info(); @@ -1725,7 +1758,14 @@ pub fn migrate_database, Cold: ItemStore>( return Err(HotColdDBError::FreezeSlotUnaligned(finalized_state.slot()).into()); } - let mut hot_db_ops: Vec> = Vec::new(); + let mut hot_db_ops = vec![]; + let mut cold_db_ops = vec![]; + + // Chunk writer for the linear block roots in the freezer DB. + // Start at the new upper limit because we iterate backwards. + let new_frozen_block_root_upper_limit = finalized_state.slot().as_usize().saturating_sub(1); + let mut block_root_writer = + ChunkWriter::::new(&store.cold_db, new_frozen_block_root_upper_limit)?; // 1. Copy all of the states between the new finalized state and the split slot, from the hot DB // to the cold DB. Delete the execution payloads of these now-finalized blocks. @@ -1750,6 +1790,9 @@ pub fn migrate_database, Cold: ItemStore>( // Delete the old summary, and the full state if we lie on an epoch boundary. hot_db_ops.push(StoreOp::DeleteState(state_root, Some(slot))); + // Store the block root for this slot in the linear array of frozen block roots. + block_root_writer.set(slot.as_usize(), block_root, &mut cold_db_ops)?; + // Do not try to store states if a restore point is yet to be stored, or will never be // stored (see `STATE_UPPER_LIMIT_NO_RETAIN`). Make an exception for the genesis state // which always needs to be copied from the hot DB to the freezer and should not be deleted. @@ -1759,29 +1802,34 @@ pub fn migrate_database, Cold: ItemStore>( .map_or(false, |anchor| slot < anchor.state_upper_limit) { debug!(store.log, "Pruning finalized state"; "slot" => slot); + continue; } - let mut cold_db_ops: Vec = Vec::new(); - - if slot % store.config.slots_per_restore_point == 0 { - let state: BeaconState = get_full_state(&store.hot_db, &state_root, &store.spec)? - .ok_or(HotColdDBError::MissingStateToFreeze(state_root))?; - - store.store_cold_state(&state_root, &state, &mut cold_db_ops)?; - } - // Store a pointer from this state root to its slot, so we can later reconstruct states // from their state root alone. let cold_state_summary = ColdStateSummary { slot }; let op = cold_state_summary.as_kv_store_op(state_root); cold_db_ops.push(op); - // There are data dependencies between calls to `store_cold_state()` that prevent us from - // doing one big call to `store.cold_db.do_atomically()` at end of the loop. - store.cold_db.do_atomically(cold_db_ops)?; + if slot % store.config.slots_per_restore_point == 0 { + let state: BeaconState = get_full_state(&store.hot_db, &state_root, &store.spec)? + .ok_or(HotColdDBError::MissingStateToFreeze(state_root))?; + + store.store_cold_state(&state_root, &state, &mut cold_db_ops)?; + + // Commit the batch of cold DB ops whenever a full state is written. Each state stored + // may read the linear fields of previous states stored. + store + .cold_db + .do_atomically(std::mem::take(&mut cold_db_ops))?; + } } + // Finish writing the block roots and commit the remaining cold DB ops. + block_root_writer.write(&mut cold_db_ops)?; + store.cold_db.do_atomically(cold_db_ops)?; + // Warning: Critical section. We have to take care not to put any of the two databases in an // inconsistent state if the OS process dies at any point during the freezeing // procedure. diff --git a/watch/Cargo.toml b/watch/Cargo.toml index b29df14439..3dc3b7c190 100644 --- a/watch/Cargo.toml +++ b/watch/Cargo.toml @@ -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" } diff --git a/watch/tests/tests.rs b/watch/tests/tests.rs index a54386eb75..dc0b8af6e3 100644 --- a/watch/tests/tests.rs +++ b/watch/tests/tests.rs @@ -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); @@ -132,6 +130,7 @@ impl TesterBuilder { reconstruct_historic_states: true, ..ChainConfig::default() }) + .logger(test_logger()) .deterministic_keypairs(VALIDATOR_COUNT) .fresh_ephemeral_store() .build();