diff --git a/bin/stress-test/Cargo.toml b/bin/stress-test/Cargo.toml index eb82ce982..36128cc4e 100644 --- a/bin/stress-test/Cargo.toml +++ b/bin/stress-test/Cargo.toml @@ -29,5 +29,11 @@ rand = { workspace = true } rayon = { workspace = true } tokio = { workspace = true } +[features] +# Renders spans as a timing tree on stdout, showing the per-phase breakdown of e.g. the +# `load-state` benchmark. Opt-in because feature unification would otherwise switch the log +# format of every binary in a workspace-wide build. +tracing-forest = ["miden-node-utils/tracing-forest"] + [dev-dependencies] tempfile = { workspace = true } diff --git a/bin/stress-test/README.md b/bin/stress-test/README.md index d2908f154..3e102e4b2 100644 --- a/bin/stress-test/README.md +++ b/bin/stress-test/README.md @@ -107,13 +107,35 @@ Latency measurements represent pure store processing time without network overhe #### load-state +Measures full store startup (`State::load`) against the seeded data directory. `--load-iterations` (default 3) repeats +the load; the first iteration may pay RocksDB WAL recovery and a cold OS page cache, while later iterations measure a +clean warm restart. + ```text -State loaded in 42.959271667s -Database contains 99961 accounts and 99960 nullifiers +Iteration 0: state loaded in 38.623292ms +Iteration 1: state loaded in 20.376417ms +Iteration 2: state loaded in 17.526916ms +... +Database contains 52 accounts and 50 nullifiers ``` -Account tree loading (~21.3s) and nullifier tree loading (~21.5s) were the primary bottlenecks; MMR loading and database -connection were negligible (<3ms each). +Build with `--features tracing-forest` to render the per-phase breakdown of each load as a timing tree, including the +RocksDB opens (`open_tree_storage`, `open_forest_storage`): + +```text +INFO load [ 36.1ms | 0.00% / 100.00% ] +INFO ┕━ load_with_database_options [ 36.1ms | 0.00% / 100.00% ] +INFO ┝━ load_with_pool_size [ 8.47ms | 23.48% ] +INFO ┝━ load_mmr [ 1.27ms | 3.52% ] +INFO ┝━ open_tree_storage [ 10.3ms | 28.68% ] path: "accounttree" +INFO ┝━ load_account_tree [ 2.19ms | 6.08% ] +INFO ┝━ open_tree_storage [ 6.38ms | 17.70% ] path: "nullifiertree" +INFO ┝━ load_nullifier_tree [ 822µs | 2.28% ] +INFO ┝━ verify_tree_consistency [ 68.0µs | 0.19% ] +INFO ┝━ open_forest_storage [ 5.98ms | 16.60% ] path: "accountstateforest" +INFO ┝━ load_account_state_forest [ 74.4µs | 0.21% ] block.number: 2 +INFO ┕━ verify_account_state_forest_consistency [ 458µs | 1.27% ] +``` #### sync-notes diff --git a/bin/stress-test/src/main.rs b/bin/stress-test/src/main.rs index fe9e5770d..734213aca 100644 --- a/bin/stress-test/src/main.rs +++ b/bin/stress-test/src/main.rs @@ -148,7 +148,12 @@ pub enum Endpoint { #[command(name = "sync-chain-mmr")] SyncChainMmr, #[command(name = "load-state")] - LoadState, + LoadState { + /// Number of times to load the state. The first iteration may pay `RocksDB` WAL recovery + /// and a cold OS page cache; later iterations measure a clean warm restart. + #[arg(long, value_name = "LOAD_ITERATIONS", default_value = "3")] + load_iterations: NonZeroUsize, + }, #[command(name = "get-account")] GetAccount { /// Storage slot name to request with all entries. @@ -228,8 +233,8 @@ async fn main() { Endpoint::SyncChainMmr => { bench_sync_chain_mmr(data_directory, iterations, concurrency).await; }, - Endpoint::LoadState => { - load_state(&data_directory).await; + Endpoint::LoadState { load_iterations } => { + load_state(&data_directory, load_iterations.get()).await; }, Endpoint::GetAccount { storage_map_slot } => { bench_get_account(data_directory, iterations, concurrency, storage_map_slot).await; diff --git a/bin/stress-test/src/store/mod.rs b/bin/stress-test/src/store/mod.rs index c8c078968..01cf4c54b 100644 --- a/bin/stress-test/src/store/mod.rs +++ b/bin/stress-test/src/store/mod.rs @@ -696,12 +696,25 @@ fn transaction_record_to_proto( // LOAD STATE // ================================================================================================ -pub async fn load_state(data_directory: &Path) { - let start = Instant::now(); - // The writer is never started: this bench only measures load time, and dropping the un-started - // state releases the tree storage the writer owns. - let _loaded = State::load(data_directory, StorageOptions::default()).await.unwrap(); - let elapsed = start.elapsed(); +pub async fn load_state(data_directory: &Path, iterations: usize) { + let mut durations = Vec::with_capacity(iterations); + for iteration in 0..iterations { + let start = Instant::now(); + // The writer is never started: this bench only measures load time, and dropping the + // un-started state releases the tree storage the writer owns. + let loaded = State::load(data_directory, StorageOptions::default()).await.unwrap(); + let elapsed = start.elapsed(); + drop(loaded); + + // The first iteration may pay RocksDB WAL recovery and a cold OS page cache; later + // iterations measure a clean warm restart. + println!("Iteration {iteration}: state loaded in {elapsed:?}"); + durations.push(elapsed); + } + + if durations.len() > 1 { + print_summary(&durations); + } // Get database path and run SQL commands to count records let data_directory = @@ -727,6 +740,5 @@ pub async fn load_state(data_directory: &Path) { |output| String::from_utf8_lossy(&output.stdout).trim().to_string(), ); - println!("State loaded in {elapsed:?}"); println!("Database contains {account_count} accounts and {nullifier_count} nullifiers"); } diff --git a/crates/store/src/db/mod.rs b/crates/store/src/db/mod.rs index 2155eeaf9..74971a039 100644 --- a/crates/store/src/db/mod.rs +++ b/crates/store/src/db/mod.rs @@ -398,6 +398,24 @@ impl Db { .await } + /// Returns the block header commitments starting at `from` (inclusive), ordered by block + /// number. + #[miden_instrument( + level = "debug", + target = COMPONENT, + err, + )] + pub async fn select_block_header_commitments_from( + &self, + from: BlockNumber, + ) -> Result> { + self.transact("block header commitments from", move |conn| { + let raw = queries::select_block_header_commitments_from(conn, from)?; + Ok(raw) + }) + .await + } + /// Returns a page of account commitments for tree rebuilding. #[miden_instrument( level = "debug", diff --git a/crates/store/src/db/models/queries/block_headers.rs b/crates/store/src/db/models/queries/block_headers.rs index e147a902c..05404552d 100644 --- a/crates/store/src/db/models/queries/block_headers.rs +++ b/crates/store/src/db/models/queries/block_headers.rs @@ -152,6 +152,35 @@ pub fn select_all_block_header_commitments( Ok(commitments) } +/// Select the commitments of all block headers starting at `from` (inclusive) from the DB using +/// the given [`SqliteConnection`], ordered by block number. +/// +/// # Returns +/// +/// A vector of [`BlockHeaderCommitment`] or an error. +/// +/// # Raw SQL +/// +/// ```sql +/// SELECT commitment +/// FROM block_headers +/// WHERE block_num >= :from +/// ORDER BY block_num ASC +/// ``` +pub fn select_block_header_commitments_from( + conn: &mut SqliteConnection, + from: BlockNumber, +) -> Result, DatabaseError> { + let raw_commitments = + QueryDsl::select(schema::block_headers::table, schema::block_headers::commitment) + .filter(schema::block_headers::block_num.ge(from.to_raw_sql())) + .order(schema::block_headers::block_num.asc()) + .load::>(conn)?; + let commitments = + Result::from_iter(raw_commitments.into_iter().map(BlockHeaderCommitment::from_raw_sql))?; + Ok(commitments) +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] #[repr(transparent)] pub struct BlockHeaderCommitment(pub(crate) Word); diff --git a/crates/store/src/state/chain_mmr_checkpoint.rs b/crates/store/src/state/chain_mmr_checkpoint.rs new file mode 100644 index 000000000..cb1274353 --- /dev/null +++ b/crates/store/src/state/chain_mmr_checkpoint.rs @@ -0,0 +1,370 @@ +//! On-disk checkpoint of the chain MMR. +//! +//! The blockchain MMR is derived entirely from the block header commitments in SQLite, but +//! rebuilding it hashes one merge per block, so the cost of a rebuild grows with chain height. +//! The checkpoint caches the derived structure in a flat file: startup restores it with a single +//! sequential read and only appends the blocks committed since the checkpoint was taken. +//! +//! The file is the MMR's node array stored verbatim: a concatenation of 32-byte nodes with no +//! header or framing. The MMR is append-only, so refreshing the checkpoint appends just the nodes +//! added since the previous refresh, and any prefix of the file covering a whole number of blocks +//! is itself a valid checkpoint of the corresponding chain prefix. Appends are buffered and never +//! synced: a crash may tear or lose the tail, which the read path drops before restoring the +//! remaining prefix. +//! +//! The checkpoint is a pure cache. A stale checkpoint is simply topped up by the loader; a corrupt +//! or divergent checkpoint is discarded in favour of a full rebuild, guarded by the +//! chain-commitment consistency check in the loader. Deleting the file is always safe. + +use std::io::Write as _; +use std::path::{Path, PathBuf}; + +use miden_crypto::merkle::mmr::Mmr; +use miden_node_utils::ErrorReport; +use miden_protocol::Word; +use miden_protocol::block::Blockchain; +use miden_protocol::utils::serde::{ByteWriter, Deserializable, Serializable}; +use tracing::warn; + +use crate::LOG_TARGET; + +/// File name of the chain MMR checkpoint within the data directory. +pub(crate) const CHAIN_MMR_CHECKPOINT_FILENAME: &str = "chainmmr.bin"; + +/// Serialized size of one MMR node. +const NODE_SIZE: usize = Word::SERIALIZED_SIZE; + +/// Handle to the chain MMR checkpoint file within the data directory. +/// +/// Reads and writes are best-effort: any failure is logged and treated as "no checkpoint", never +/// surfaced as an error, since the database remains the source of truth. +#[derive(Debug, Clone)] +pub(crate) struct ChainMmrCheckpoint { + path: PathBuf, +} + +impl ChainMmrCheckpoint { + pub fn new(data_dir: &Path) -> Self { + Self { + path: data_dir.join(CHAIN_MMR_CHECKPOINT_FILENAME), + } + } + + /// Reads the checkpoint, restoring the longest usable prefix of the file: a torn tail from an + /// interrupted append is dropped, and a checkpoint with more blocks than `chain_length` (e.g. + /// the database was restored from an older backup) is cut back to the database's chain length. + /// Returns `None` if the file is missing, unreadable, undecodable, or holds no complete block. + pub fn read(&self, chain_length: u32) -> Option { + let bytes = fs_err::read(&self.path) + .inspect_err(|err| + if err.kind() != std::io::ErrorKind::NotFound { + warn!(target: LOG_TARGET, err = %err.as_report(), "Failed to read the chain MMR checkpoint; rebuilding from the database"); + } + ) + .ok()?; + + let num_blocks = largest_chain_prefix(bytes.len() / NODE_SIZE, chain_length); + if num_blocks == 0 { + return None; + } + let num_nodes = num_nodes_in_chain(num_blocks); + // Restoring the valid prefix needs no dedicated logic: the decode below always takes + // exactly `num_nodes` nodes, which in the healthy case is the whole file. This branch only + // reports when tail bytes are actually being dropped. + if num_nodes * NODE_SIZE != bytes.len() { + warn!( + target: LOG_TARGET, + file_bytes = bytes.len(), + restored_blocks = num_blocks, + chain_length, + "Chain MMR checkpoint tail is unusable (torn append or ahead of the database); restoring the valid prefix" + ); + } + + // The node array is an in-memory struct at this point; what is missing is a constructor + // that turns raw nodes into an `Mmr` without re-hashing them (`nodes` is private, and + // `Mmr::try_from_iter` re-hashes every merge). The only non-hashing path in is the + // `Deserializable` impl, so the node bytes are prefixed with the two integers its layout + // expects — the forest's leaf count and the node-array length — and decoded. + // + // TODO: Replace with `Mmr::from_nodes_unchecked` (decode the node bytes into `Word`s and + // construct directly) once the node is on a miden-vm release containing + // 0xMiden/miden-vm#3585, removing the extra copy. The layout coupling is safe meanwhile: + // upstream documents the flat `forest || nodes` encoding as the stable wire format. + let mut serialized = Vec::with_capacity(2 * size_of::() + num_nodes * NODE_SIZE); + serialized.write_usize(num_blocks as usize); + serialized.write_usize(num_nodes); + serialized.extend_from_slice(&bytes[..num_nodes * NODE_SIZE]); + + match Mmr::read_from_bytes(&serialized) { + Ok(mmr) => Some(Blockchain::from_mmr_unchecked(mmr)), + Err(err) => { + warn!(target: LOG_TARGET, err = %err.as_report(), "Failed to decode the chain MMR checkpoint; rebuilding from the database"); + None + }, + } + } + + /// Atomically replaces the checkpoint with the given blockchain. + /// + /// The replacement is atomic against concurrent readers (write to a temporary file, then + /// rename), but like [`Self::append`] it is never fsynced: a crash may lose the replacement, + /// which only costs the next startup a rebuild or a longer top-up. + pub fn write(&self, blockchain: &Blockchain) { + let bytes = node_bytes_from(blockchain, 0); + + let tmp_path = self.path.with_extension("tmp"); + let result = + fs_err::write(&tmp_path, &bytes).and_then(|()| fs_err::rename(&tmp_path, &self.path)); + if let Err(err) = result { + warn!(target: LOG_TARGET, err = %err.as_report(), "Failed to write the chain MMR checkpoint"); + } + } + + /// Appends the nodes added after the first `from_blocks` blocks to the checkpoint file, + /// returning whether the file ends at `blockchain`'s tip as a result. + /// + /// The write goes to the OS page cache and is never fsynced: the checkpoint is a pure cache, + /// so durability is not required, and skipping the sync keeps the append cheap enough to run + /// inline on the block-apply path. A crash may lose or tear the unsynced tail; [`Self::read`] + /// drops it and the loader tops the difference up from the database. + /// + /// The append is only valid if the file ends exactly where the new nodes begin; when it does + /// not (a torn earlier append, an ahead-of-database file that `read` cut back, or a failed + /// earlier write), the file is left untouched and `false` is returned. Healing — rewriting + /// the checkpoint via [`Self::write`] — is deliberately left to callers off the block-apply + /// path, since the full rewrite's I/O grows with chain height. + pub fn append(&self, blockchain: &Blockchain, from_blocks: u32) -> bool { + debug_assert!(from_blocks <= blockchain.num_blocks()); + + let expected_len = (num_nodes_in_chain(from_blocks) * NODE_SIZE) as u64; + let file_len = match fs_err::metadata(&self.path) { + Ok(metadata) => metadata.len(), + Err(err) if err.kind() == std::io::ErrorKind::NotFound => 0, + Err(err) => { + warn!(target: LOG_TARGET, err = %err.as_report(), "Failed to stat the chain MMR checkpoint; skipping the append"); + return false; + }, + }; + if file_len != expected_len { + warn!( + target: LOG_TARGET, + file_len, + expected_len, + "Chain MMR checkpoint does not end at the expected node; skipping the append" + ); + return false; + } + if from_blocks == blockchain.num_blocks() { + return true; + } + + let bytes = node_bytes_from(blockchain, from_blocks); + let result = fs_err::OpenOptions::new() + .append(true) + .create(true) + .open(&self.path) + .and_then(|mut file| file.write_all(&bytes)); + match result { + Ok(()) => true, + Err(err) => { + warn!(target: LOG_TARGET, err = %err.as_report(), "Failed to append to the chain MMR checkpoint"); + false + }, + } + } +} + +/// Number of nodes in the MMR of a chain with `num_blocks` blocks: each block adds its own leaf +/// node plus one parent node per binary-tree merge, totalling `2 * blocks - popcount(blocks)`. +fn num_nodes_in_chain(num_blocks: u32) -> usize { + 2 * num_blocks as usize - num_blocks.count_ones() as usize +} + +/// Returns the largest block count whose MMR node array fits within `num_nodes` complete nodes, +/// capped at `chain_length`. +/// +/// Node counts don't map 1:1 to block counts (see [`num_nodes_in_chain`]): a file cut short by a +/// torn append usually ends between block boundaries, and only the nodes up to the last whole +/// block are usable. This binary-searches for the largest block count whose complete node array +/// is present, capped at the database's `chain_length` for the case where the checkpoint is +/// ahead of the database (e.g. the database was restored from an older backup). +/// [`ChainMmrCheckpoint::read`] then decodes just that prefix of the file and ignores the rest. +fn largest_chain_prefix(num_nodes: usize, chain_length: u32) -> u32 { + // `num_nodes_in_chain` is strictly increasing, so "fits in the file" is a monotone predicate + // and the largest block count satisfying it is found by binary search. `lo` always satisfies + // the predicate; `hi` starts at the answer's cheap upper bounds (`chain_length`, and + // `num_nodes` since every block contributes at least one node). + let mut lo = 0u64; + let mut hi = u64::from(chain_length).min(num_nodes as u64); + while lo < hi { + // Round the probe up: with rounding down, `hi == lo + 1` would probe `lo` and loop forever. + // The u64 arithmetic avoids overflow near `u32::MAX`. + let mid = lo + (hi - lo).div_ceil(2); + if num_nodes_in_chain(mid as u32) <= num_nodes { + lo = mid; + } else { + hi = mid - 1; + } + } + lo as u32 +} + +/// Serializes the MMR nodes appended after the first `from_blocks` blocks. +/// +/// TODO: Serialize just the new nodes via `Mmr::nodes_from(start)` once the node is on a +/// miden-vm release containing 0xMiden/miden-vm#3585. Until then the whole MMR is serialized — +/// its layout ends with the node array — and the new nodes' bytes are taken from the tail. +fn node_bytes_from(blockchain: &Blockchain, from_blocks: u32) -> Vec { + let mmr = blockchain.as_mmr(); + let serialized = mmr.to_bytes(); + let new_bytes = (mmr.forest().num_nodes() - num_nodes_in_chain(from_blocks)) * NODE_SIZE; + serialized[serialized.len() - new_bytes..].to_vec() +} + +#[cfg(test)] +mod tests { + use miden_crypto::merkle::mmr::Mmr; + use miden_protocol::Word; + + use super::*; + + fn chain(blocks: u32) -> Blockchain { + let mmr = Mmr::try_from_iter((0..blocks).map(|i| Word::from([i, 0, 0, 0u32]))) + .expect("test MMR should build"); + Blockchain::from_mmr_unchecked(mmr) + } + + #[test] + fn read_returns_none_when_missing() { + let dir = tempfile::tempdir().expect("temp directory should be created"); + assert!(ChainMmrCheckpoint::new(dir.path()).read(5).is_none()); + } + + #[test] + fn write_read_round_trips() { + let dir = tempfile::tempdir().expect("temp directory should be created"); + let checkpoint = ChainMmrCheckpoint::new(dir.path()); + let blockchain = chain(5); + + checkpoint.write(&blockchain); + + let restored = checkpoint.read(5).expect("checkpoint should round-trip"); + assert_eq!(restored.num_blocks(), 5); + assert_eq!(restored.commitment(), blockchain.commitment()); + } + + #[test] + fn read_truncates_checkpoint_ahead_of_database() { + let dir = tempfile::tempdir().expect("temp directory should be created"); + let checkpoint = ChainMmrCheckpoint::new(dir.path()); + checkpoint.write(&chain(5)); + + // A checkpoint with more blocks than the database is cut back to the database's chain + // length; at or behind the chain length it is restored whole. + let truncated = checkpoint.read(3).expect("prefix should be restored"); + assert_eq!(truncated.num_blocks(), 3); + assert_eq!(truncated.commitment(), chain(3).commitment()); + + let whole = checkpoint.read(10).expect("whole checkpoint should be restored"); + assert_eq!(whole.num_blocks(), 5); + } + + #[test] + fn append_extends_existing_checkpoint() { + let dir = tempfile::tempdir().expect("temp directory should be created"); + let checkpoint = ChainMmrCheckpoint::new(dir.path()); + checkpoint.write(&chain(3)); + + assert!(checkpoint.append(&chain(8), 3)); + + let restored = checkpoint.read(8).expect("appended checkpoint should be restored"); + assert_eq!(restored.num_blocks(), 8); + assert_eq!(restored.commitment(), chain(8).commitment()); + } + + #[test] + fn append_creates_missing_checkpoint() { + let dir = tempfile::tempdir().expect("temp directory should be created"); + let checkpoint = ChainMmrCheckpoint::new(dir.path()); + + assert!(checkpoint.append(&chain(4), 0)); + + let restored = checkpoint.read(4).expect("created checkpoint should be restored"); + assert_eq!(restored.num_blocks(), 4); + assert_eq!(restored.commitment(), chain(4).commitment()); + } + + #[test] + fn append_skips_on_length_mismatch() { + let dir = tempfile::tempdir().expect("temp directory should be created"); + let checkpoint = ChainMmrCheckpoint::new(dir.path()); + checkpoint.write(&chain(5)); + + // The file holds 5 blocks but the append expects it to end at 3; the file is left untouched + // — healing by full rewrite is the caller's call, off the block-apply path. + assert!(!checkpoint.append(&chain(8), 3)); + + let restored = checkpoint.read(8).expect("untouched checkpoint should be restored"); + assert_eq!(restored.num_blocks(), 5); + assert_eq!(restored.commitment(), chain(5).commitment()); + } + + #[test] + fn read_drops_torn_tail() { + let dir = tempfile::tempdir().expect("temp directory should be created"); + let checkpoint = ChainMmrCheckpoint::new(dir.path()); + let blockchain = chain(5); + checkpoint.write(&blockchain); + + // A crash mid-append leaves trailing bytes that don't form complete blocks: a partial node, + // and a whole node that doesn't complete a block on its own. + for torn_tail in [&[0u8; 10][..], &[0u8; NODE_SIZE][..]] { + let mut file = fs_err::OpenOptions::new() + .append(true) + .open(dir.path().join(CHAIN_MMR_CHECKPOINT_FILENAME)) + .expect("checkpoint file should open for appending"); + file.write_all(torn_tail).expect("torn tail should be written"); + drop(file); + + let restored = checkpoint.read(10).expect("valid prefix should be restored"); + assert_eq!(restored.num_blocks(), 5); + assert_eq!(restored.commitment(), blockchain.commitment()); + + checkpoint.write(&blockchain); + } + } + + #[test] + fn largest_chain_prefix_finds_last_whole_block() { + // Every node count between two block boundaries maps back to the lower boundary. + for blocks in 0..=64u32 { + let boundary = num_nodes_in_chain(blocks); + let next_boundary = num_nodes_in_chain(blocks + 1); + for num_nodes in boundary..next_boundary { + assert_eq!(largest_chain_prefix(num_nodes, u32::MAX), blocks); + } + } + } + + #[test] + fn largest_chain_prefix_caps_at_chain_length() { + // A checkpoint ahead of the database (e.g. restored from an older backup) is cut back. + assert_eq!(largest_chain_prefix(num_nodes_in_chain(64), 10), 10); + assert_eq!(largest_chain_prefix(num_nodes_in_chain(64), 0), 0); + // The u64 midpoint arithmetic holds up at the u32 extreme. + assert_eq!(largest_chain_prefix(num_nodes_in_chain(u32::MAX), u32::MAX), u32::MAX); + } + + #[test] + fn read_rejects_undecodable_nodes() { + let dir = tempfile::tempdir().expect("temp directory should be created"); + let checkpoint = ChainMmrCheckpoint::new(dir.path()); + + // 0xFF bytes are not canonical field elements, so decoding fails. + fs_err::write(dir.path().join(CHAIN_MMR_CHECKPOINT_FILENAME), [0xFF; NODE_SIZE]) + .expect("corrupt checkpoint should be written"); + + assert!(checkpoint.read(5).is_none()); + } +} diff --git a/crates/store/src/state/lifecycle.rs b/crates/store/src/state/lifecycle.rs index 572bb71f6..b5b93d6ef 100644 --- a/crates/store/src/state/lifecycle.rs +++ b/crates/store/src/state/lifecycle.rs @@ -9,9 +9,11 @@ use arc_swap::ArcSwap; use miden_node_utils::ErrorReport; use miden_node_utils::clap::StorageOptions; use miden_node_utils::shutdown::CancellationToken; +use miden_node_utils::spawn::spawn_blocking_in_current_span; use miden_node_utils::tracing::miden_instrument; -use miden_protocol::block::BlockNumber; use tokio::sync::{mpsc, watch}; +use tokio::task::JoinHandle; +use tracing::Instrument; use crate::account_state_forest::AccountStateForestBackend; use crate::accounts::AccountTreeWithHistory; @@ -19,6 +21,7 @@ use crate::blocks::BlockStore; use crate::db::Db; use crate::errors::StateInitializationError; use crate::proven_tip::ProvenTipWriter; +use crate::state::chain_mmr_checkpoint::ChainMmrCheckpoint; use crate::state::loader::{ ACCOUNT_STATE_FOREST_STORAGE_DIR, ACCOUNT_TREE_STORAGE_DIR, @@ -42,6 +45,19 @@ use crate::state::{ }; use crate::{COMPONENT, DataDirectory, DatabaseOptions}; +/// Awaits a spawned load task, forwarding its result. +/// +/// The load tasks are never aborted, so a join error is a panic from the task; it is resumed on +/// the caller so panics keep propagating as panics. +async fn join_load_task( + handle: JoinHandle>, +) -> Result { + match handle.await { + Ok(result) => result, + Err(err) => std::panic::resume_unwind(err.into_panic()), + } +} + /// Number of recent committed blocks held in the in-memory cache for replica subscriptions. const BLOCK_CACHE_CAPACITY: NonZeroUsize = NonZeroUsize::new(512).unwrap(); @@ -138,15 +154,22 @@ impl State { ); let database_filepath = data_directory.database_path(); - let mut db = Db::load_with_pool_size( - database_filepath.clone(), - database_options.connection_pool_size, - ) - .await - .map_err(StateInitializationError::DatabaseLoadError)?; + let db = Arc::new( + Db::load_with_pool_size( + database_filepath.clone(), + database_options.connection_pool_size, + ) + .await + .map_err(StateInitializationError::DatabaseLoadError)?, + ); - let blockchain = load_mmr(&mut db).await?; - let latest_block_num = blockchain.chain_tip().unwrap_or(BlockNumber::GENESIS); + // The chain tip drives forest loading and the account tree history below; `load_mmr`'s + // consistency check also pins the chain MMR to this header. + let latest_block_num = db + .select_block_header_by_block_num(None) + .await? + .ok_or(StateInitializationError::GenesisBlockMissing)? + .block_num(); let apply_block_thread_priority = storage_options.apply_block_thread_priority; @@ -159,31 +182,91 @@ impl State { #[cfg(not(feature = "rocksdb"))] let (account_storage_config, nullifier_storage_config, forest_storage_config) = ((), (), ()); - let account_storage = - TreeStorage::create(data_path, &account_storage_config, ACCOUNT_TREE_STORAGE_DIR)?; - let account_tree = account_storage.load_account_tree(&mut db).await?; - let nullifier_storage = - TreeStorage::create(data_path, &nullifier_storage_config, NULLIFIER_TREE_STORAGE_DIR)?; - let nullifier_tree = nullifier_storage.load_nullifier_tree(&mut db).await?; + // The four structures live in independent storages and the database pool supports + // concurrent readers, so open and load them concurrently. Each branch is a spawned task + // because loading has long synchronous sections (RocksDB opens, MMR hashing, SMT top + // reconstruction) that would serialize if polled from a single task. Spawning is eager, so + // all four run from this point; the join below only collects their results. + let chain_mmr_checkpoint = ChainMmrCheckpoint::new(data_path); + + let mmr_task = tokio::spawn( + { + let (db, checkpoint) = (Arc::clone(&db), chain_mmr_checkpoint.clone()); + async move { load_mmr(&db, &checkpoint).await } + } + .in_current_span(), + ); + let account_tree_task = tokio::spawn( + { + let (db, path) = (Arc::clone(&db), data_path.to_path_buf()); + async move { + join_load_task(spawn_blocking_in_current_span(move || { + TreeStorage::create( + &path, + &account_storage_config, + ACCOUNT_TREE_STORAGE_DIR, + ) + })) + .await? + .load_account_tree(&db) + .await + } + } + .in_current_span(), + ); + let nullifier_tree_task = tokio::spawn( + { + let (db, path) = (Arc::clone(&db), data_path.to_path_buf()); + async move { + join_load_task(spawn_blocking_in_current_span(move || { + TreeStorage::create( + &path, + &nullifier_storage_config, + NULLIFIER_TREE_STORAGE_DIR, + ) + })) + .await? + .load_nullifier_tree(&db) + .await + } + } + .in_current_span(), + ); + let forest_task = tokio::spawn( + { + let (db, path) = (Arc::clone(&db), data_path.to_path_buf()); + async move { + let forest = join_load_task(spawn_blocking_in_current_span(move || { + AccountStateForestBackend::create( + &path, + &forest_storage_config, + ACCOUNT_STATE_FOREST_STORAGE_DIR, + ) + })) + .await? + .load_account_state_forest(&db, latest_block_num) + .await?; + verify_account_state_forest_consistency(&forest, &db).await?; + Ok(forest) + } + } + .in_current_span(), + ); + let (blockchain, account_tree, nullifier_tree, forest) = tokio::try_join!( + join_load_task(mmr_task), + join_load_task(account_tree_task), + join_load_task(nullifier_tree_task), + join_load_task(forest_task), + )?; // Verify that tree roots match the expected roots from the database. This catches any // divergence between persistent storage and the database caused by corruption or incomplete // shutdown. - verify_tree_consistency(account_tree.root(), nullifier_tree.root(), &mut db).await?; + verify_tree_consistency(account_tree.root(), nullifier_tree.root(), &db).await?; let account_tree = AccountTreeWithHistory::new(account_tree, latest_block_num); - let forest_backend = AccountStateForestBackend::create( - data_path, - &forest_storage_config, - ACCOUNT_STATE_FOREST_STORAGE_DIR, - )?; - let forest = forest_backend.load_account_state_forest(&mut db, latest_block_num).await?; - verify_account_state_forest_consistency(&forest, &mut db).await?; - - let db = Arc::new(db); - // Initialize the proven tip from the block store. let proven_tip_init = block_store .load_proven_tip() @@ -229,6 +312,7 @@ impl State { nullifier_tree, account_tree, blockchain, + chain_mmr_checkpoint, forest, snapshots_live, apply_block_thread_priority, diff --git a/crates/store/src/state/loader.rs b/crates/store/src/state/loader.rs index a2cdd45de..bdf3dee74 100644 --- a/crates/store/src/state/loader.rs +++ b/crates/store/src/state/loader.rs @@ -32,16 +32,14 @@ use miden_protocol::block::{BlockHeader, BlockNumber, Blockchain}; use miden_protocol::crypto::merkle::smt::MemoryStorage; use miden_protocol::crypto::merkle::smt::{LargeSmt, LargeSmtError, SmtStorage}; use miden_protocol::{Felt, Word}; -#[cfg(feature = "rocksdb")] -use tracing::info; +use tracing::{info, warn}; -use crate::COMPONENT; -#[cfg(feature = "rocksdb")] -use crate::LOG_TARGET; use crate::account_state_forest::AccountStateForest; use crate::db::Db; use crate::db::models::queries::BlockHeaderCommitment; use crate::errors::{DatabaseError, StateInitializationError}; +use crate::state::chain_mmr_checkpoint::ChainMmrCheckpoint; +use crate::{COMPONENT, LOG_TARGET}; // CONSTANTS // ================================================================================================ @@ -130,13 +128,13 @@ pub trait TreeStorageLoader: SmtStorage + Sized { /// Loads an account tree, either from persistent storage or by rebuilding from DB. fn load_account_tree( self, - db: &mut Db, + db: &Db, ) -> impl Future>, StateInitializationError>> + Send; /// Loads a nullifier tree, either from persistent storage or by rebuilding from DB. fn load_nullifier_tree( self, - db: &mut Db, + db: &Db, ) -> impl Future>, StateInitializationError>> + Send; } @@ -162,7 +160,7 @@ pub(crate) trait AccountForestLoader: Backend + Sized { /// Loads the account state forest, either from persistent storage or by rebuilding from DB. fn load_account_state_forest( self, - db: &mut Db, + db: &Db, block_num: BlockNumber, ) -> impl Future, StateInitializationError>> + Send; } @@ -186,7 +184,7 @@ impl TreeStorageLoader for MemoryStorage { )] async fn load_account_tree( self, - db: &mut Db, + db: &Db, ) -> Result>, StateInitializationError> { let mut smt = LargeSmt::with_entries(self, std::iter::empty()) .map_err(account_tree_large_smt_error_to_init_error)?; @@ -230,7 +228,7 @@ impl TreeStorageLoader for MemoryStorage { )] async fn load_nullifier_tree( self, - db: &mut Db, + db: &Db, ) -> Result>, StateInitializationError> { let mut smt = LargeSmt::with_entries(self, std::iter::empty()) .map_err(account_tree_large_smt_error_to_init_error)?; @@ -270,6 +268,13 @@ impl TreeStorageLoader for MemoryStorage { #[cfg(feature = "rocksdb")] impl TreeStorageLoader for RocksDbStorage { type Config = RocksDbOptions; + // Opening RocksDB replays unflushed WAL segments and fills the table cache, which can take + // seconds; the span makes this cost visible in startup traces. + #[miden_instrument( + target = COMPONENT, + name = "open_tree_storage", + fields(path = domain), + )] fn create( data_dir: &Path, storage_options: &Self::Config, @@ -288,7 +293,7 @@ impl TreeStorageLoader for RocksDbStorage { )] async fn load_account_tree( self, - db: &mut Db, + db: &Db, ) -> Result>, StateInitializationError> { // If RocksDB storage has data, load from it directly @@ -342,7 +347,7 @@ impl TreeStorageLoader for RocksDbStorage { )] async fn load_nullifier_tree( self, - db: &mut Db, + db: &Db, ) -> Result>, StateInitializationError> { // If RocksDB storage has data, load from it directly @@ -410,7 +415,7 @@ impl AccountForestLoader for ForestInMemoryBackend { )] async fn load_account_state_forest( self, - db: &mut Db, + db: &Db, block_num: BlockNumber, ) -> Result, StateInitializationError> { let mut forest = AccountStateForest::from_backend(self) @@ -424,6 +429,13 @@ impl AccountForestLoader for ForestInMemoryBackend { impl AccountForestLoader for ForestPersistentBackend { type Config = RocksDbOptions; + // Opening RocksDB replays unflushed WAL segments and fills the table cache, which can take + // seconds; the span makes this cost visible in startup traces. + #[miden_instrument( + target = COMPONENT, + name = "open_forest_storage", + fields(path = domain), + )] fn create( data_dir: &Path, storage_options: &Self::Config, @@ -456,7 +468,7 @@ impl AccountForestLoader for ForestPersistentBackend { )] async fn load_account_state_forest( self, - db: &mut Db, + db: &Db, block_num: BlockNumber, ) -> Result, StateInitializationError> { let mut forest = AccountStateForest::from_backend(self) @@ -487,12 +499,100 @@ pub fn load_smt(storage: S) -> Result, StateInitializ // TREE LOADING FUNCTIONS // ================================================================================================ -/// Loads the blockchain MMR from all block headers in the database. +/// Loads the blockchain MMR, either by restoring the on-disk checkpoint or by rebuilding from all +/// block headers in the database. +/// +/// The checkpoint is a cache of the derived MMR structure; the MMR is append-only, so a stale +/// checkpoint is a valid prefix and is topped up with the block commitments it is missing. A +/// missing, corrupt, or divergent checkpoint falls back to a full rebuild. Either path is verified +/// against the latest header's chain commitment, and the topped-up blocks are appended to the +/// checkpoint whenever it was behind the database. #[miden_instrument( target = COMPONENT, )] -pub async fn load_mmr(db: &mut Db) -> Result { +pub async fn load_mmr( + db: &Db, + checkpoint: &ChainMmrCheckpoint, +) -> Result { let latest_header = db.select_block_header_by_block_num(None).await?; + + // Fast path: restore the checkpoint and top it up. Falls through to a full rebuild if there is + // no usable checkpoint, or if the topped-up result doesn't match the database. + if let Some(chain_mmr) = + try_restore_mmr_from_checkpoint(db, checkpoint, latest_header.as_ref()).await? + { + return Ok(chain_mmr); + } + + rebuild_mmr_from_database(db, checkpoint, latest_header.as_ref()).await +} + +/// Attempts to restore the chain MMR from the on-disk checkpoint. +/// +/// Returns `Ok(None)` — never an error — when the checkpoint can't be used, so the caller falls +/// back to a full rebuild: +/// - there is no genesis block yet (nothing to restore against); +/// - the checkpoint file is missing, unreadable, or holds no usable prefix (see +/// [`ChainMmrCheckpoint::read`]); +/// - the checkpoint, topped up with the blocks committed since it was taken, does not reproduce +/// the latest header's chain commitment (e.g. the file diverged from the database). +async fn try_restore_mmr_from_checkpoint( + db: &Db, + checkpoint: &ChainMmrCheckpoint, + latest_header: Option<&BlockHeader>, +) -> Result, StateInitializationError> { + let Some(header) = latest_header else { + return Ok(None); + }; + + let chain_length = header.block_num().as_u32() + 1; + let Some(mut chain_mmr) = checkpoint.read(chain_length) else { + return Ok(None); + }; + + // The MMR is append-only, so the checkpoint is a valid prefix of the chain; append the + // commitments for the blocks committed after it was taken. + let checkpoint_blocks = chain_mmr.num_blocks(); + let missing_commitments = db + .select_block_header_commitments_from(BlockNumber::from(checkpoint_blocks)) + .await?; + for commitment in missing_commitments { + chain_mmr.push(commitment.word()); + } + + if let Err(err) = verify_chain_mmr_consistency(&chain_mmr, Some(header)) { + warn!( + target: LOG_TARGET, + err = %miden_node_utils::ErrorReport::as_report(&err), + "Chain MMR checkpoint diverged from the database; rebuilding" + ); + return Ok(None); + } + + info!( + target: LOG_TARGET, + checkpoint_blocks, + appended_blocks = chain_length - checkpoint_blocks, + "Loaded chain MMR from checkpoint" + ); + + // Keep the on-disk checkpoint caught up so the next startup has fewer blocks to append. Startup + // is off the block-apply path, so a file the append cannot extend (e.g. one whose torn tail + // `read` skipped over) is healed here with a full rewrite. + if !checkpoint.append(&chain_mmr, checkpoint_blocks) { + checkpoint.write(&chain_mmr); + } + + Ok(Some(chain_mmr)) +} + +/// Rebuilds the chain MMR from every block header commitment in the database, then writes it to the +/// checkpoint so the next startup can restore it instead of rebuilding. +async fn rebuild_mmr_from_database( + db: &Db, + checkpoint: &ChainMmrCheckpoint, + latest_header: Option<&BlockHeader>, +) -> Result { let block_commitments = db.select_all_block_header_commitments().await?; // SAFETY: We assume the loaded MMR is valid and does not have more than u32::MAX entries. @@ -500,7 +600,9 @@ pub async fn load_mmr(db: &mut Db) -> Result, - db: &mut Db, + db: &Db, block_num: BlockNumber, ) -> Result<(), StateInitializationError> { use miden_protocol::account::AccountPatch; @@ -606,7 +708,7 @@ pub async fn rebuild_account_state_forest( pub async fn verify_tree_consistency( account_tree_root: Word, nullifier_tree_root: Word, - db: &mut Db, + db: &Db, ) -> Result<(), StateInitializationError> { // Fetch the latest block header to get the expected roots let latest_header = db.select_block_header_by_block_num(None).await?; @@ -647,9 +749,11 @@ pub async fn verify_tree_consistency( target = COMPONENT, )] pub async fn verify_account_state_forest_consistency( - forest: &AccountStateForest, - db: &mut Db, + forest: &AccountStateForest, + db: &Db, ) -> Result<(), StateInitializationError> { + use rayon::iter::{IntoParallelIterator, ParallelIterator}; + let mut cursor = None; loop { @@ -661,14 +765,15 @@ pub async fn verify_account_state_forest_consistency( break; } - for account in page.accounts { + // Per-account checks are independent, so verify each page in parallel. + page.accounts.into_par_iter().try_for_each(|account| { verify_account_state_forest_record( forest, account.account_id, account.vault_root, &account.storage_header, - )?; - } + ) + })?; cursor = page.next_cursor; if cursor.is_none() { @@ -792,7 +897,7 @@ mod tests { let headers = build_headers(5); let signing_key = SigningKey::new(); - let mut db = crate::db::Db::load(db_path).await.expect("test database should load"); + let db = crate::db::Db::load(db_path).await.expect("test database should load"); db.query("insert corrupted block headers", move |conn| { for header in &headers { @@ -816,7 +921,7 @@ mod tests { .await .expect("test block headers should be inserted"); - let error = load_mmr(&mut db) + let error = load_mmr(&db, &ChainMmrCheckpoint::new(temp_dir.path())) .await .expect_err("startup MMR load should reject inconsistent block headers"); @@ -826,6 +931,90 @@ mod tests { ); } + /// Bootstraps a test database at `db_path` and inserts `count` consistent block headers. + async fn seed_headers(db_path: &Path, count: u32) -> (Vec, crate::db::Db) { + crate::db::bootstrap_database(db_path).expect("test database should bootstrap"); + let headers = build_headers(count); + let db = crate::db::Db::load(db_path.to_path_buf()) + .await + .expect("test database should load"); + + let headers_to_insert = headers.clone(); + let signing_key = SigningKey::new(); + db.query("insert block headers", move |conn| { + for header in &headers_to_insert { + let signatures = miden_protocol::block::BlockSignatures::new(vec![ + signing_key.sign(header.commitment()), + ]) + .expect("one signature is within bounds"); + crate::db::models::queries::insert_block_header(conn, header, &signatures)?; + } + Ok::<_, DatabaseError>(()) + }) + .await + .expect("test block headers should be inserted"); + + (headers, db) + } + + #[tokio::test] + #[miden_node_test_macro::enable_logging] + async fn load_mmr_writes_and_restores_checkpoint() { + let temp_dir = tempfile::tempdir().expect("temp directory should be created"); + let (_headers, db) = seed_headers(&temp_dir.path().join("store.sqlite"), 5).await; + let checkpoint = ChainMmrCheckpoint::new(temp_dir.path()); + + // The first load rebuilds from the database and writes the checkpoint. + let rebuilt = load_mmr(&db, &checkpoint).await.expect("MMR should rebuild from database"); + let written = checkpoint.read(5).expect("checkpoint should have been written"); + assert_eq!(written.num_blocks(), 5); + assert_eq!(written.commitment(), rebuilt.commitment()); + + // The second load restores the checkpoint and yields the same chain. + let restored = + load_mmr(&db, &checkpoint).await.expect("MMR should restore from checkpoint"); + assert_eq!(restored.commitment(), rebuilt.commitment()); + } + + #[tokio::test] + #[miden_node_test_macro::enable_logging] + async fn load_mmr_tops_up_stale_checkpoint() { + let temp_dir = tempfile::tempdir().expect("temp directory should be created"); + let (headers, db) = seed_headers(&temp_dir.path().join("store.sqlite"), 5).await; + let checkpoint = ChainMmrCheckpoint::new(temp_dir.path()); + + // Checkpoint the chain as of block 2; the database is 2 blocks ahead. + let stale_mmr = Mmr::try_from_iter(headers[..3].iter().map(BlockHeader::commitment)) + .expect("test MMR should build"); + checkpoint.write(&Blockchain::from_mmr_unchecked(stale_mmr)); + + let chain = load_mmr(&db, &checkpoint).await.expect("stale checkpoint should be topped up"); + assert_eq!(chain.num_blocks(), 5); + + // The topped-up chain replaces the stale checkpoint. + let refreshed = checkpoint.read(5).expect("refreshed checkpoint should exist"); + assert_eq!(refreshed.num_blocks(), 5); + assert_eq!(refreshed.commitment(), chain.commitment()); + } + + #[tokio::test] + #[miden_node_test_macro::enable_logging] + async fn load_mmr_recovers_from_corrupt_checkpoint() { + use crate::state::chain_mmr_checkpoint::CHAIN_MMR_CHECKPOINT_FILENAME; + + let temp_dir = tempfile::tempdir().expect("temp directory should be created"); + let (_headers, db) = seed_headers(&temp_dir.path().join("store.sqlite"), 5).await; + let checkpoint = ChainMmrCheckpoint::new(temp_dir.path()); + + fs_err::write(temp_dir.path().join(CHAIN_MMR_CHECKPOINT_FILENAME), b"garbage") + .expect("corrupt checkpoint should be written"); + + let chain = load_mmr(&db, &checkpoint) + .await + .expect("corrupt checkpoint should fall back to a database rebuild"); + assert_eq!(chain.num_blocks(), 5); + } + #[test] fn account_state_forest_consistency_detects_storage_map_root_mismatch() { let account_id = AccountId::try_from(ACCOUNT_ID_REGULAR_PUBLIC_ACCOUNT_IMMUTABLE_CODE) diff --git a/crates/store/src/state/mod.rs b/crates/store/src/state/mod.rs index 3a519e36a..609e462dc 100644 --- a/crates/store/src/state/mod.rs +++ b/crates/store/src/state/mod.rs @@ -5,6 +5,7 @@ mod block_lifecycle; mod bootstrap; +mod chain_mmr_checkpoint; mod disk_monitor; mod lifecycle; mod loader; diff --git a/crates/store/src/state/writer/worker.rs b/crates/store/src/state/writer/worker.rs index 515fb42b6..a4c6c2d6d 100644 --- a/crates/store/src/state/writer/worker.rs +++ b/crates/store/src/state/writer/worker.rs @@ -31,6 +31,7 @@ use crate::blocks::BlockStore; use crate::db::{Db, NoteRecord}; use crate::errors::{ApplyBlockError, InvalidBlockError}; use crate::state::block_lifecycle::{BlockLifecycle, lifecycle_events_enabled}; +use crate::state::chain_mmr_checkpoint::ChainMmrCheckpoint; use crate::state::loader::TreeStorage; use crate::state::view::{ PublishedGenerations, @@ -44,6 +45,14 @@ use crate::{COMPONENT, HistoricalError, LOG_TARGET}; // WRITE WORKER // ================================================================================================ +/// Number of new blocks after which the chain MMR checkpoint is refreshed. +/// +/// A refresh appends only the nodes added since the previous one (at most two per block) as a +/// buffered, unsynced write, so it is cheap enough to run inline on the write path. The interval +/// bounds how far the checkpoint can lag after a crash, which startup then tops up from the +/// database. +const CHECKPOINT_INTERVAL_BLOCKS: u32 = 500; + /// Single-task owner of the mutable trees. Processes [`WriteRequest`]s serially. /// /// The writer owns the writable trees directly, so no locks are held at any point: validation and @@ -63,6 +72,11 @@ pub(in crate::state) struct WriteWorker { account_tree: AccountTreeWithHistory, /// The blockchain MMR owned by this writer. blockchain: Blockchain, + /// On-disk checkpoint of the chain MMR, appended to every [`CHECKPOINT_INTERVAL_BLOCKS`] blocks + /// and on shutdown so the next startup only tops up the blocks committed after that. + chain_mmr_checkpoint: ChainMmrCheckpoint, + /// Number of blocks already persisted in the chain MMR checkpoint file. + checkpointed_blocks: u32, /// The mutable account state forest owned by this writer. forest: AccountStateForest, /// Shared counter of live snapshot generations, for observability. @@ -103,6 +117,7 @@ impl WriteWorker { nullifier_tree: NullifierTree>, account_tree: AccountTreeWithHistory, blockchain: Blockchain, + chain_mmr_checkpoint: ChainMmrCheckpoint, forest: AccountStateForest, snapshots_live: Arc, apply_block_thread_priority: bool, @@ -121,6 +136,10 @@ impl WriteWorker { let apply_pool = Arc::new(pool_builder.build().expect("apply_block thread pool should build")); + // The loader refreshed the checkpoint up to the loaded chain tip (best-effort; a failed + // refresh is healed by the length check on the next append). + let checkpointed_blocks = blockchain.num_blocks(); + Self { db, block_store, @@ -131,6 +150,8 @@ impl WriteWorker { nullifier_tree, account_tree, blockchain, + chain_mmr_checkpoint, + checkpointed_blocks, forest, snapshots_live, published_generations, @@ -156,7 +177,35 @@ impl WriteWorker { }; let result = self.write_block(req.signed_block).await; let _ = req.result_tx.send(result); + + if self.blockchain.num_blocks() - self.checkpointed_blocks >= CHECKPOINT_INTERVAL_BLOCKS + { + self.refresh_checkpoint(); + } + } + + // Refresh the on-disk chain MMR checkpoint so the next startup only tops up the blocks + // committed after this point. Shutdown is off the block-apply path, so here — unlike in the + // periodic refresh — a checkpoint the appends could not extend is healed by rewriting it in + // full (via write). + if !self.chain_mmr_checkpoint.append(&self.blockchain, self.checkpointed_blocks) { + self.chain_mmr_checkpoint.write(&self.blockchain); + } + } + + /// Appends the blocks committed since the last refresh to the on-disk chain MMR checkpoint. + /// + /// A checkpoint that cannot be extended is left as is: healing it means rewriting the file in + /// full, whose I/O grows with chain height, so it is deferred to the shutdown refresh (or the + /// next startup). The tracked block count advances regardless, keeping later refreshes on the + /// cheap length-check-and-skip path rather than retrying ever larger appends. + fn refresh_checkpoint(&mut self) { + let num_blocks = self.blockchain.num_blocks(); + if num_blocks == self.checkpointed_blocks { + return; } + self.chain_mmr_checkpoint.append(&self.blockchain, self.checkpointed_blocks); + self.checkpointed_blocks = num_blocks; } /// Validates and commits a signed block to all persistent and in-memory stores.