diff --git a/CHANGELOG.md b/CHANGELOG.md index 8308484c00e..f3dd66991e6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -35,6 +35,13 @@ and this project adheres to [Semantic Versioning](https://semver.org). ### Fixed +- Zebra can now audit the Zakura header store when the state database opens and + self-repair any incoherence (broken linkage, hash↔height index mismatches, + gaps with stranded rows above them, stale rows at committed heights) by + truncating the Zakura column families to the last coherent height in bounded + batches, then letting header sync re-download the truncated suffix. Operators + can opt in with `state.repair_zakura_header_store_on_startup = true`; each + repair emits a warning and the `state.zakura.header_store.incoherent` metric. - Zakura header-store consensus reads now verify store invariants as they walk: the difficulty-context walk checks that every stored header is the block its hash row names and links to the row below it, and the anchor diff --git a/zebra-state/src/config.rs b/zebra-state/src/config.rs index 9cf5a4a0a80..ca443ee8e61 100644 --- a/zebra-state/src/config.rs +++ b/zebra-state/src/config.rs @@ -102,6 +102,15 @@ pub struct Config { /// Zebra's last non-finalized state before it shut down. pub should_backup_non_finalized_state: bool, + /// Whether to audit and repair the Zakura header store when the state database opens. + /// + /// Set to `false` by default. When this is `true`, writable state opens scan + /// the full Zakura header-store frontier and delete any incoherent suffix or + /// stale committed-height rows so header sync can re-download them. This can + /// be expensive while syncing from genesis, because the header frontier can + /// contain millions of rows above the finalized tip. + pub repair_zakura_header_store_on_startup: bool, + /// Whether committed full blocks should seed the Zakura header-only store. /// /// This is enabled by zebrad only for the Zakura v2 path. It is skipped in @@ -427,6 +436,7 @@ impl Default for Config { cache_dir: default_cache_dir(), ephemeral: false, should_backup_non_finalized_state: true, + repair_zakura_header_store_on_startup: false, enable_zakura_header_seed_from_committed_blocks: false, checkpoint_sync: true, vct_fast_sync: true, @@ -455,11 +465,20 @@ mod tests { Config::default().vct_fast_sync, "VCT fast sync is enabled by default when checkpoint sync and embedded frontiers are available" ); + assert!( + !Config::default().repair_zakura_header_store_on_startup, + "Zakura header-store startup repair is opt-in because it scans the full header frontier" + ); let archive: Config = toml::from_str(r#"storage_mode = "archive""#) .expect("archive storage mode deserializes from a string"); assert_eq!(archive.storage_mode, StorageMode::Archive); + let repair_enabled: Config = + toml::from_str(r#"repair_zakura_header_store_on_startup = true"#) + .expect("startup repair config deserializes from a bool"); + assert!(repair_enabled.repair_zakura_header_store_on_startup); + let pruned: Config = toml::from_str(r#"storage_mode = "pruned""#) .expect("pruned storage mode deserializes from a string"); assert_eq!( diff --git a/zebra-state/src/service/finalized_state/zebra_db.rs b/zebra-state/src/service/finalized_state/zebra_db.rs index e7ba5325b2f..7c36fd2b24c 100644 --- a/zebra-state/src/service/finalized_state/zebra_db.rs +++ b/zebra-state/src/service/finalized_state/zebra_db.rs @@ -175,6 +175,16 @@ impl ZebraDb { ) } + // Optionally audit the zakura header store's on-disk invariants and + // truncate any incoherent suffix. This can scan a large header frontier + // while syncing from genesis, so operators opt in when they need a + // startup repair. Read-only instances cannot repair; their reads surface + // any corruption as explicit `StoreIncoherentError`s instead. + if !read_only && config.repair_zakura_header_store_on_startup { + db.audit_and_repair_zakura_header_store() + .expect("startup header-store repair write failed: RocksDB is unavailable"); + } + db.spawn_format_change(format_change); db diff --git a/zebra-state/src/service/finalized_state/zebra_db/block.rs b/zebra-state/src/service/finalized_state/zebra_db/block.rs index 5dab23a1272..a3451c5b864 100644 --- a/zebra-state/src/service/finalized_state/zebra_db/block.rs +++ b/zebra-state/src/service/finalized_state/zebra_db/block.rs @@ -57,6 +57,8 @@ use crate::{ #[cfg(feature = "indexer")] use crate::request::Spend; +mod startup_audit; + #[cfg(test)] mod tests; diff --git a/zebra-state/src/service/finalized_state/zebra_db/block/startup_audit.rs b/zebra-state/src/service/finalized_state/zebra_db/block/startup_audit.rs new file mode 100644 index 00000000000..4bc262b51b4 --- /dev/null +++ b/zebra-state/src/service/finalized_state/zebra_db/block/startup_audit.rs @@ -0,0 +1,571 @@ +//! Startup audit and self-repair for the zakura header store. +//! +//! The zakura header store is five height-indexed column families acting as a +//! replicated view of the canonical header chain above the finalized tip. Its +//! invariants (hash↔height bijection, parent linkage anchored at the finalized +//! tip, no gaps or stranded rows below the header tip) are enforced by the +//! writers, but a store corrupted by an earlier binary stays corrupted on disk +//! and wedges header sync: consensus reads surface the damage as +//! [`StoreIncoherentError`](crate::error::StoreIncoherentError) and refuse to +//! feed stale rows into validation, so the node can no longer make progress +//! past the poisoned window. +//! +//! This module runs the store audit once at [`ZebraDb`] startup and repairs +//! any violation by truncating the zakura column families to the last +//! coherent height in bounded batches. Headers are re-fetchable — header sync +//! re-downloads the truncated suffix — so correctness beats preserved rows: +//! any residual write-path bug in this class becomes a self-healing, +//! observable transient instead of a permanent on-disk wedge. +//! +//! The audit cost is `O(header frontier)`: the zakura column families only +//! hold rows above the finalized tip (plus any stale rows this audit exists +//! to remove), and the verified commitment-roots history below the tip is +//! never scanned. + +use std::{fmt::Debug, ops::Bound::*, sync::Arc}; + +use zebra_chain::block::{self, Height}; + +use super::{ + AdvertisedBodySize, ZAKURA_HEADER_BODY_SIZE_BY_HEIGHT, ZAKURA_HEADER_BY_HEIGHT, + ZAKURA_HEADER_HASH_BY_HEIGHT, ZAKURA_HEADER_HEIGHT_BY_HASH, +}; +use crate::service::finalized_state::{ + disk_db::{DiskWriteBatch, ReadDisk, WriteDisk}, + disk_format::{shielded::CommitmentRootsByHeight, FromDisk, IntoDisk}, + zebra_db::ZebraDb, + COMMITMENT_ROOTS_BY_HEIGHT, +}; + +/// How many violations are included in the repair log line. The full list can +/// be as long as the stranded suffix; the first few identify the fault shape. +const LOGGED_VIOLATIONS: usize = 8; + +/// Maximum number of rows the startup audit materializes from one column +/// family at a time. Roughly 1.5MB with 1.5 KB headers. +const AUDIT_BATCH_ROWS: usize = 100_000; + +/// A single header-store invariant violation found by the startup audit. +/// +/// Every violation is repaired by deleting rows; the variants exist for +/// logging and for test assertions on the fault shape. The fields are +/// diagnostic payload rendered through `Debug` in the repair warning. +#[allow(dead_code)] +#[derive(Clone, Debug)] +pub(crate) enum ZakuraStoreViolation { + /// A zakura row at a height with a committed block (`contains_height`). + /// + /// Committed heights have authoritative full-block rows, and the zakura + /// row at a height is trimmed when its body commits, so a surviving row + /// here is stale (the pre-guard re-delivery bug shape). The predicate is + /// the consensus `hash_by_height` row — which pruning retains — not body + /// presence, so stale rows at pruned heights are found too. + StaleRowAtCommittedHeight { + /// The column family holding the stale row. + cf: &'static str, + /// The committed height of the stale row. + height: Height, + }, + + /// A `hash_by_height` row at `height` has no `header_by_height` row. + MissingHeaderRow { + /// The height with a hash row but no header row. + height: Height, + }, + + /// The header stored at `height` is not the block its hash row names. + HeaderHashMismatch { + /// The height of the divergent rows. + height: Height, + /// The hash the height→hash index names. + indexed: block::Hash, + /// The stored header's actual hash. + computed: block::Hash, + }, + + /// The header at `height` does not link to the stored row below it. + BrokenLinkage { + /// The height of the header whose parent link failed to resolve. + height: Height, + /// The parent hash the header claims (`previous_block_hash`). + expected_parent: block::Hash, + /// The hash actually stored at `height - 1`. + actual_below: block::Hash, + }, + + /// The `height_by_hash` index is missing or wrong for the hash stored at + /// `height` (a forward bijection failure). + WrongHeightByHash { + /// The height whose hash failed the round-trip. + height: Height, + /// The hash stored at `height`. + hash: block::Hash, + /// The height the hash→height index reports, if any. + indexed: Option, + }, + + /// A row above the last coherent height (a stranded suffix: rows above a + /// gap, above a broken link, or above the linked chain's tip). + RowAboveLastCoherent { + /// The column family holding the stranded row. + cf: &'static str, + /// The stranded height. + height: Height, + }, + + /// A `height_by_hash` entry whose target height stores a different hash + /// (or no row at all): a stranded reverse-index row. + OrphanHeightByHash { + /// The hash of the stranded entry. + hash: block::Hash, + /// The height the entry points at. + points_at: Height, + }, +} + +/// The outcome of a startup repair: what was found and what was deleted. +/// +/// The fields are diagnostic payload: read by test assertions and rendered +/// through `Debug`. +#[allow(dead_code)] +#[derive(Debug)] +pub(crate) struct ZakuraStoreRepair { + /// The last height that passed every audit check, and its hash. + /// + /// Rows above it were truncated. `None` means the store had zakura rows + /// but no finalized tip to anchor them at, so all rows were removed. + pub last_coherent: Option<(Height, block::Hash)>, + + /// The number of rows deleted across all five column families. + pub deleted_rows: usize, + + /// Every violation found, in audit order. + pub violations: Vec, +} + +impl ZebraDb { + /// Audits the zakura header store invariants and repairs any violation by + /// deleting the offending rows in bounded batches. + /// + /// Checks, over the whole zakura store (which only holds the header + /// frontier above the finalized tip, so this is cheap): + /// + /// - **bijection**: `hash_by_height` ↔ `height_by_hash` are mutually + /// inverse, and every stored header is the block its hash row names; + /// - **linkage**: rows chain by `previous_block_hash` from the finalized + /// tip hash upward; + /// - **tip integrity**: no rows in any zakura column family above the + /// last linked height, no gaps below it, and no zakura rows at + /// committed heights. + /// + /// On violation, emits the `state.zakura.header_store.incoherent` metric + /// and a warning, then truncates the zakura column families to the last + /// coherent height (and removes stale rows below the finalized tip and + /// orphaned reverse-index entries). Header sync re-downloads the + /// truncated suffix. A store that would fail the linkage-verified + /// consensus reads ([`StoreIncoherentError`](crate::error::StoreIncoherentError)) + /// is always repaired by this audit, because the audit checks are a + /// superset of the read-path checks over the same rows. + /// + /// Returns `Ok(None)` if the store is coherent (nothing is written), or + /// the repair summary after a successful repair write. + /// + /// Verified commitment roots at committed heights are never touched: the + /// roots column family is only scanned above the finalized tip. Each scan + /// is batched so a large header frontier does not have to fit in memory at + /// startup. + pub(crate) fn audit_and_repair_zakura_header_store( + &self, + ) -> Result, rocksdb::Error> { + // Databases opened through `ZebraDb::new` without the zakura column + // families have no header store to audit. + let ( + Some(header_cf), + Some(hash_cf), + Some(height_by_hash_cf), + Some(body_size_cf), + Some(roots_cf), + ) = ( + self.db.cf_handle(ZAKURA_HEADER_BY_HEIGHT), + self.db.cf_handle(ZAKURA_HEADER_HASH_BY_HEIGHT), + self.db.cf_handle(ZAKURA_HEADER_HEIGHT_BY_HASH), + self.db.cf_handle(ZAKURA_HEADER_BODY_SIZE_BY_HEIGHT), + self.db.cf_handle(COMMITMENT_ROOTS_BY_HEIGHT), + ) + else { + return Ok(None); + }; + + let finalized_tip = self.tip(); + + // The roots column family also holds verified rows at committed + // heights (written by body commits, kept through pruning); those are + // not zakura frontier rows and are never audited or repaired. Only + // rows above the finalized tip are provisional header-sync data. + let provisional_roots_start = + finalized_tip.and_then(|(tip_height, _)| tip_height.next().ok()); + let provisional_roots_empty = match (finalized_tip, provisional_roots_start) { + // The finalized tip is at the maximum height: no frontier can + // exist above it. + (Some(_), None) => true, + (Some(_), Some(start)) => self + .db + .zs_forward_range_iter::<_, Height, CommitmentRootsByHeight, _>(&roots_cf, start..) + .next() + .is_none(), + // No finalized tip: every roots row is provisional. + (None, _) => self.db.zs_is_empty(&roots_cf), + }; + + if self.db.zs_is_empty(&header_cf) + && self.db.zs_is_empty(&hash_cf) + && self.db.zs_is_empty(&height_by_hash_cf) + && self.db.zs_is_empty(&body_size_cf) + && provisional_roots_empty + { + // Log the pass so operators can verify the audit ran on this boot. + tracing::info!( + ?finalized_tip, + frontier_rows = 0, + "zakura header store passed its startup coherence audit (empty frontier)" + ); + return Ok(None); + } + + let mut violations = Vec::new(); + + // Walk the linked chain upward from the finalized tip, verifying at + // each height that the hash and header rows agree, the header links + // to the row below, and the reverse index round-trips. The walk stops + // at the first missing hash row (the candidate chain tip) or the + // first violation; everything it passed is the coherent prefix. + let last_coherent = finalized_tip.map(|(anchor_height, anchor_hash)| { + let (mut last_height, mut last_hash) = (anchor_height, anchor_hash); + + while let Ok(height) = last_height.next() { + let Some(hash) = self.db.zs_get(&hash_cf, &height) else { + break; + }; + let Some(header): Option> = self.db.zs_get(&header_cf, &height) + else { + violations.push(ZakuraStoreViolation::MissingHeaderRow { height }); + break; + }; + + let computed = block::Hash::from(&*header); + if computed != hash { + violations.push(ZakuraStoreViolation::HeaderHashMismatch { + height, + indexed: hash, + computed, + }); + break; + } + + if header.previous_block_hash != last_hash { + violations.push(ZakuraStoreViolation::BrokenLinkage { + height, + expected_parent: header.previous_block_hash, + actual_below: last_hash, + }); + break; + } + + let indexed = self.db.zs_get(&height_by_hash_cf, &hash); + if indexed != Some(height) { + violations.push(ZakuraStoreViolation::WrongHeightByHash { + height, + hash, + indexed, + }); + break; + } + + (last_height, last_hash) = (height, hash); + } + + (last_height, last_hash) + }); + + // Committed heights have authoritative full-block rows: the consensus + // `hash_by_height` rows are contiguous from genesis to the finalized + // tip and retained by pruning, so `height <= finalized tip` is + // exactly `contains_height` — the same predicate as the writers' + // committed-height insert gate. (A body-presence predicate would miss + // stale rows at pruned heights.) + let committed = + |height: Height| finalized_tip.is_some_and(|(tip_height, _)| height <= tip_height); + // The coherent frontier window: strictly above the finalized tip, at + // or below the last height the linkage walk verified. + let in_window = |height: Height| { + !committed(height) + && last_coherent.is_some_and(|(last_height, _)| height <= last_height) + }; + + let mut batch = DiskWriteBatch::new(); + let mut deleted_rows = 0; + let mut pending_deletes = 0; + + // Reverse-index entries survive only when their target height is + // inside the window and stores exactly their hash. This removes the + // reverse rows of every deleted hash row, plus entries orphaned by + // earlier overwrites that never cleaned up the displaced hash. + let mut start_after_hash = None; + loop { + let heights_by_hash = + hash_keyed_batch::(&self.db, &height_by_hash_cf, start_after_hash); + let Some(&(last_hash, _)) = heights_by_hash.last() else { + break; + }; + start_after_hash = Some(last_hash); + + for &(hash, points_at) in &heights_by_hash { + let target_matches = + self.db.zs_get::<_, _, block::Hash>(&hash_cf, &points_at) == Some(hash); + if in_window(points_at) && target_matches { + continue; + } + + // Entries whose forward row is deleted above are repair fallout, + // not separate faults; only a live-but-disagreeing target is a + // distinct violation shape worth reporting. + if target_matches { + violations.push(if committed(points_at) { + ZakuraStoreViolation::StaleRowAtCommittedHeight { + cf: ZAKURA_HEADER_HEIGHT_BY_HASH, + height: points_at, + } + } else { + ZakuraStoreViolation::RowAboveLastCoherent { + cf: ZAKURA_HEADER_HEIGHT_BY_HASH, + height: points_at, + } + }); + } else { + violations.push(ZakuraStoreViolation::OrphanHeightByHash { hash, points_at }); + } + + queue_repair_delete( + &self.db, + &mut batch, + &mut pending_deletes, + &height_by_hash_cf, + hash, + )?; + deleted_rows += 1; + } + } + + // Height-keyed zakura rows survive only inside the coherent window. + audit_height_keyed_rows::>( + &self.db, + &header_cf, + ZAKURA_HEADER_BY_HEIGHT, + &committed, + &in_window, + &mut violations, + &mut batch, + &mut pending_deletes, + &mut deleted_rows, + None, + )?; + let frontier_rows = audit_height_keyed_rows::( + &self.db, + &hash_cf, + ZAKURA_HEADER_HASH_BY_HEIGHT, + &committed, + &in_window, + &mut violations, + &mut batch, + &mut pending_deletes, + &mut deleted_rows, + None, + )?; + audit_height_keyed_rows::( + &self.db, + &body_size_cf, + ZAKURA_HEADER_BODY_SIZE_BY_HEIGHT, + &committed, + &in_window, + &mut violations, + &mut batch, + &mut pending_deletes, + &mut deleted_rows, + None, + )?; + + // Provisional roots above the window are part of the stranded suffix. + if let Some(start) = match (finalized_tip, provisional_roots_start) { + (Some(_), None) => None, + (Some(_), Some(start)) => Some(start), + (None, _) => Some(Height(0)), + } { + audit_height_keyed_rows::( + &self.db, + &roots_cf, + COMMITMENT_ROOTS_BY_HEIGHT, + &committed, + &in_window, + &mut violations, + &mut batch, + &mut pending_deletes, + &mut deleted_rows, + Some(start), + )?; + } + + if violations.is_empty() { + // Log the pass so operators can verify the audit ran on this boot + // (the fleet soak requires observing it finding nothing). + tracing::info!( + ?finalized_tip, + ?last_coherent, + frontier_rows, + "zakura header store passed its startup coherence audit" + ); + return Ok(None); + } + + metrics::counter!("state.zakura.header_store.incoherent").increment(1); + tracing::warn!( + ?finalized_tip, + ?last_coherent, + deleted_rows, + violation_count = violations.len(), + first_violations = ?&violations[..violations.len().min(LOGGED_VIOLATIONS)], + "zakura header store failed its startup coherence audit; \ + truncating to the last coherent height so header sync re-downloads the rest" + ); + + flush_repair_batch(&self.db, &mut batch, &mut pending_deletes)?; + + Ok(Some(ZakuraStoreRepair { + last_coherent, + deleted_rows, + violations, + })) + } +} + +fn height_keyed_batch( + db: &crate::service::finalized_state::disk_db::DiskDb, + cf: &rocksdb::ColumnFamilyRef<'_>, + start: Height, +) -> Vec +where + V: FromDisk, +{ + db.zs_forward_range_iter::<_, Height, V, _>(cf, start..) + .map(|(height, _)| height) + .take(AUDIT_BATCH_ROWS) + .collect() +} + +fn hash_keyed_batch( + db: &crate::service::finalized_state::disk_db::DiskDb, + cf: &rocksdb::ColumnFamilyRef<'_>, + start_after: Option, +) -> Vec<(block::Hash, V)> +where + V: FromDisk, +{ + match start_after { + Some(start_after) => db + .zs_forward_range_iter::<_, block::Hash, V, _>(cf, (Excluded(start_after), Unbounded)) + .take(AUDIT_BATCH_ROWS) + .collect(), + None => db + .zs_forward_range_iter::<_, block::Hash, V, _>(cf, ..) + .take(AUDIT_BATCH_ROWS) + .collect(), + } +} + +#[allow(clippy::too_many_arguments)] +fn audit_height_keyed_rows( + db: &crate::service::finalized_state::disk_db::DiskDb, + cf: &rocksdb::ColumnFamilyRef<'_>, + cf_name: &'static str, + committed: &impl Fn(Height) -> bool, + in_window: &impl Fn(Height) -> bool, + violations: &mut Vec, + batch: &mut DiskWriteBatch, + pending_deletes: &mut usize, + deleted_rows: &mut usize, + start: Option, +) -> Result +where + V: FromDisk, +{ + let mut next_start = start.unwrap_or(Height(0)); + let mut scanned_rows = 0; + + loop { + let heights = height_keyed_batch::(db, cf, next_start); + let Some(&last_height) = heights.last() else { + break; + }; + scanned_rows += heights.len(); + next_start = match last_height.next() { + Ok(next_height) => next_height, + Err(_) => break, + }; + + for height in heights { + if in_window(height) { + continue; + } + + violations.push(if committed(height) { + ZakuraStoreViolation::StaleRowAtCommittedHeight { + cf: cf_name, + height, + } + } else { + ZakuraStoreViolation::RowAboveLastCoherent { + cf: cf_name, + height, + } + }); + + queue_repair_delete(db, batch, pending_deletes, cf, height)?; + *deleted_rows += 1; + } + } + + Ok(scanned_rows) +} + +fn queue_repair_delete( + db: &crate::service::finalized_state::disk_db::DiskDb, + batch: &mut DiskWriteBatch, + pending_deletes: &mut usize, + cf: &rocksdb::ColumnFamilyRef<'_>, + key: K, +) -> Result<(), rocksdb::Error> +where + K: IntoDisk + Debug, +{ + batch.zs_delete(cf, key); + *pending_deletes += 1; + + if *pending_deletes >= AUDIT_BATCH_ROWS { + flush_repair_batch(db, batch, pending_deletes)?; + } + + Ok(()) +} + +fn flush_repair_batch( + db: &crate::service::finalized_state::disk_db::DiskDb, + batch: &mut DiskWriteBatch, + pending_deletes: &mut usize, +) -> Result<(), rocksdb::Error> { + if *pending_deletes == 0 { + return Ok(()); + } + + db.write(std::mem::take(batch))?; + *pending_deletes = 0; + Ok(()) +} diff --git a/zebra-state/src/service/finalized_state/zebra_db/block/tests/header_store_coherence.rs b/zebra-state/src/service/finalized_state/zebra_db/block/tests/header_store_coherence.rs index ef8cea5a0a1..669225c0930 100644 --- a/zebra-state/src/service/finalized_state/zebra_db/block/tests/header_store_coherence.rs +++ b/zebra-state/src/service/finalized_state/zebra_db/block/tests/header_store_coherence.rs @@ -34,3 +34,4 @@ mod oracle; mod prop; mod reads; mod scenarios; +mod startup_audit; diff --git a/zebra-state/src/service/finalized_state/zebra_db/block/tests/header_store_coherence/startup_audit.rs b/zebra-state/src/service/finalized_state/zebra_db/block/tests/header_store_coherence/startup_audit.rs new file mode 100644 index 00000000000..95e1e8b03c5 --- /dev/null +++ b/zebra-state/src/service/finalized_state/zebra_db/block/tests/header_store_coherence/startup_audit.rs @@ -0,0 +1,570 @@ +//! Startup-audit self-repair: a hand-corrupted zakura header store heals when +//! the database is reopened. +//! +//! The normal write path refuses to create incoherent stores, so these tests +//! corrupt the column families directly and assert that: +//! +//! - the startup audit finds the violation and truncates the store to the +//! last coherent height in all five column families, leaving a store the +//! full coherence audit passes and header sync can re-download onto; +//! - the repair happens on the real startup path (`ZebraDb::new` at reopen) +//! and persists: a second reopen finds a coherent store and changes nothing; +//! - repair is minimal where truncation is not needed (an orphaned +//! reverse-index row, stale rows at committed heights); +//! - a store with broken linkage-verified reads passes them after the heal; and +//! - the `state.zakura.header_store.incoherent` metric is emitted exactly +//! when a repair runs. + +use std::{ + collections::HashMap, + sync::{Arc, Mutex}, +}; + +use zebra_chain::block::Height; + +use super::super::super::startup_audit::ZakuraStoreViolation; +use super::super::super::{ + ZAKURA_HEADER_BY_HEIGHT, ZAKURA_HEADER_HASH_BY_HEIGHT, ZAKURA_HEADER_HEIGHT_BY_HASH, +}; +use super::super::common::{ + commit_header_range, persistent_config, persistent_state, state_with_genesis_config, +}; +use super::{ + audit::{audit_store, dump_store, StoreDump}, + fabricate::{Universe, BRANCH_A, FORK_HEIGHT}, +}; +use crate::{ + error::{CommitHeaderRangeError, StoreIncoherentError}, + service::finalized_state::{ + disk_db::{DiskWriteBatch, WriteDisk}, + ZebraDb, + }, + Config, +}; + +/// A store holding genesis plus the trunk up to the fork height, built through +/// the production write path, on the given (persistent or ephemeral) config. +fn trunk_state(universe: &Universe, config: Config) -> ZebraDb { + let state = state_with_genesis_config(&universe.network, universe.genesis.clone(), config); + let trunk_headers: Vec<_> = universe.trunk[..FORK_HEIGHT as usize] + .iter() + .map(|fab| fab.header.clone()) + .collect(); + commit_header_range(&state, universe.genesis.hash(), &trunk_headers); + state +} + +/// Closes `state` and reopens the same database, running the startup audit. +fn reopen(state: ZebraDb, config: &Config, universe: &Universe) -> ZebraDb { + let mut state = state; + state.shutdown(true); + drop(state); + persistent_state(config, &universe.network) +} + +fn startup_repair_config(cache_dir: &std::path::Path) -> Config { + Config { + repair_zakura_header_store_on_startup: true, + ..persistent_config(cache_dir) + } +} + +/// The expected store after truncating everything above `last` (the original +/// coherent dump with all five column families filtered to `..= last`). +fn truncated(dump: &StoreDump, last: Height) -> StoreDump { + StoreDump { + headers: dump + .headers + .iter() + .filter(|(height, _)| **height <= last) + .map(|(height, header)| (*height, header.clone())) + .collect(), + hashes: dump + .hashes + .iter() + .filter(|(height, _)| **height <= last) + .map(|(height, hash)| (*height, *hash)) + .collect(), + heights_by_hash: dump + .heights_by_hash + .iter() + .filter(|(_, points_at)| *points_at <= last) + .copied() + .collect(), + body_sizes: dump + .body_sizes + .iter() + .filter(|(height, _)| **height <= last) + .map(|(height, size)| (*height, *size)) + .collect(), + roots: dump + .roots + .iter() + .filter(|(height, _)| **height <= last) + .map(|(height, roots)| (*height, *roots)) + .collect(), + } +} + +fn assert_clean(state: &ZebraDb) { + let violations = audit_store(state); + assert!( + violations.is_empty(), + "expected a coherent store after repair: {violations:?}" + ); +} + +/// A coherent store is untouched: the direct audit reports nothing, and a +/// reopen through the real startup path changes no rows. +#[test] +fn startup_audit_is_noop_on_coherent_store() { + let _init_guard = zebra_test::init(); + let universe = Universe::new(); + let tempdir = tempfile::tempdir().expect("test tempdir is available"); + let config = startup_repair_config(tempdir.path()); + let state = trunk_state(&universe, config.clone()); + let original = dump_store(&state); + + let repair = state + .audit_and_repair_zakura_header_store() + .expect("audit reads and writes succeed"); + assert!( + repair.is_none(), + "no repair expected on a coherent store: {repair:?}" + ); + assert_eq!(dump_store(&state), original); + + let state = reopen(state, &config, &universe); + assert_eq!( + dump_store(&state), + original, + "a reopen must not change a coherent store" + ); +} + +/// Broken linkage mid-frontier (a self-consistent foreign row spliced into the +/// height index — the incident-shaped poison): the audit truncates to the row +/// below the splice, the heal persists across a further reopen, and header +/// sync re-downloads the truncated suffix back to the original store. +#[test] +fn startup_heals_broken_linkage_then_resync_converges() { + let _init_guard = zebra_test::init(); + let universe = Universe::new(); + let tempdir = tempfile::tempdir().expect("test tempdir is available"); + let config = startup_repair_config(tempdir.path()); + let state = trunk_state(&universe, config.clone()); + let original = dump_store(&state); + + // Replace the header and hash rows at height 20 with branch-A's fourth + // row: internally consistent, but it does not link to trunk@19. + let foreign = &universe.branches[BRANCH_A].headers[3]; + let header_cf = state.db.cf_handle(ZAKURA_HEADER_BY_HEIGHT).unwrap(); + let hash_cf = state.db.cf_handle(ZAKURA_HEADER_HASH_BY_HEIGHT).unwrap(); + let mut batch = DiskWriteBatch::new(); + batch.zs_insert(&header_cf, Height(20), foreign.header.clone()); + batch.zs_insert(&hash_cf, Height(20), foreign.hash); + state.db.write(batch).expect("raw insert writes"); + + // The corruption is visible to the Pillar-2 reads before the heal. + let error = state + .recent_header_context(Height(30)) + .expect_err("the walk crosses the corrupted row"); + assert!(matches!(error, StoreIncoherentError::BrokenLinkage { .. })); + + let repair = state + .audit_and_repair_zakura_header_store() + .expect("audit reads and writes succeed") + .expect("the corrupted store is repaired"); + assert_eq!( + repair.last_coherent, + Some((Height(19), universe.trunk_at(19).hash)) + ); + assert!( + repair.violations.iter().any(|violation| matches!( + violation, + ZakuraStoreViolation::BrokenLinkage { height, actual_below, .. } + if *height == Height(20) && *actual_below == universe.trunk_at(19).hash + )), + "expected BrokenLinkage at the splice: {:?}", + repair.violations + ); + + // Everything above the last coherent height is gone, in all five CFs. + assert_eq!(dump_store(&state), truncated(&original, Height(19))); + assert_clean(&state); + state + .recent_header_context(Height(19)) + .expect("the healed store walks cleanly"); + + // The heal persists: a reopen through the startup path changes nothing. + let state = reopen(state, &config, &universe); + assert_eq!(dump_store(&state), truncated(&original, Height(19))); + + // Header sync re-downloads the truncated suffix and converges back onto + // the original chain. + let redelivered: Vec<_> = universe.trunk[19..FORK_HEIGHT as usize] + .iter() + .map(|fab| fab.header.clone()) + .collect(); + commit_header_range(&state, universe.trunk_at(19).hash, &redelivered); + assert_eq!(dump_store(&state), original); + assert_clean(&state); +} + +/// A gap mid-frontier strands every row above it: reopening alone (no direct +/// audit call) heals the store, proving the `ZebraDb::new` hook fires. +#[test] +fn startup_heals_gap_and_stranded_rows_on_reopen() { + let _init_guard = zebra_test::init(); + let universe = Universe::new(); + let tempdir = tempfile::tempdir().expect("test tempdir is available"); + let config = startup_repair_config(tempdir.path()); + let state = trunk_state(&universe, config.clone()); + let original = dump_store(&state); + + let header_cf = state.db.cf_handle(ZAKURA_HEADER_BY_HEIGHT).unwrap(); + let hash_cf = state.db.cf_handle(ZAKURA_HEADER_HASH_BY_HEIGHT).unwrap(); + let mut batch = DiskWriteBatch::new(); + batch.zs_delete(&header_cf, Height(15)); + batch.zs_delete(&hash_cf, Height(15)); + state.db.write(batch).expect("raw delete writes"); + + let state = reopen(state, &config, &universe); + + // Rows 16..=50 were stranded above the gap; the audit truncated to 14. + assert_eq!(dump_store(&state), truncated(&original, Height(14))); + assert_clean(&state); + assert_eq!( + state.best_header_tip(), + Some((Height(14), universe.trunk_at(14).hash)) + ); +} + +/// Reopening with the default config does not run the potentially expensive +/// startup scan, leaving any repair for an explicit audit or an opt-in restart. +#[test] +fn startup_repair_is_disabled_by_default() { + let _init_guard = zebra_test::init(); + let universe = Universe::new(); + let tempdir = tempfile::tempdir().expect("test tempdir is available"); + let config = persistent_config(tempdir.path()); + let state = trunk_state(&universe, config.clone()); + let original = dump_store(&state); + + let header_cf = state.db.cf_handle(ZAKURA_HEADER_BY_HEIGHT).unwrap(); + let hash_cf = state.db.cf_handle(ZAKURA_HEADER_HASH_BY_HEIGHT).unwrap(); + let mut batch = DiskWriteBatch::new(); + batch.zs_delete(&header_cf, Height(15)); + batch.zs_delete(&hash_cf, Height(15)); + state.db.write(batch).expect("raw delete writes"); + + let state = reopen(state, &config, &universe); + + assert_ne!( + dump_store(&state), + truncated(&original, Height(14)), + "default reopen must not repair the corrupted store" + ); +} + +/// A missing reverse-index row breaks the hash↔height bijection: the audit +/// truncates to the row below it. +#[test] +fn startup_heals_missing_reverse_index_row() { + let _init_guard = zebra_test::init(); + let universe = Universe::new(); + let state = trunk_state(&universe, Config::ephemeral()); + let original = dump_store(&state); + + let height_by_hash_cf = state.db.cf_handle(ZAKURA_HEADER_HEIGHT_BY_HASH).unwrap(); + let mut batch = DiskWriteBatch::new(); + batch.zs_delete(&height_by_hash_cf, universe.trunk_at(10).hash); + state.db.write(batch).expect("raw delete writes"); + + let repair = state + .audit_and_repair_zakura_header_store() + .expect("audit reads and writes succeed") + .expect("the corrupted store is repaired"); + assert_eq!( + repair.last_coherent, + Some((Height(9), universe.trunk_at(9).hash)) + ); + assert!( + repair.violations.iter().any(|violation| matches!( + violation, + ZakuraStoreViolation::WrongHeightByHash { height, indexed: None, .. } + if *height == Height(10) + )), + "expected WrongHeightByHash at the deleted entry: {:?}", + repair.violations + ); + assert_eq!(dump_store(&state), truncated(&original, Height(9))); + assert_clean(&state); +} + +/// An orphaned reverse-index row (a stranded `height_by_hash` entry from an +/// overwrite that never cleaned up the displaced hash) is removed on its own: +/// the linked chain is coherent, so nothing is truncated. +#[test] +fn startup_removes_orphan_reverse_row_without_truncating() { + let _init_guard = zebra_test::init(); + let universe = Universe::new(); + let state = trunk_state(&universe, Config::ephemeral()); + let original = dump_store(&state); + + let height_by_hash_cf = state.db.cf_handle(ZAKURA_HEADER_HEIGHT_BY_HASH).unwrap(); + let stray_hash = universe.branches[BRANCH_A].headers[0].hash; + let mut batch = DiskWriteBatch::new(); + batch.zs_insert(&height_by_hash_cf, stray_hash, Height(12)); + state.db.write(batch).expect("raw insert writes"); + + let repair = state + .audit_and_repair_zakura_header_store() + .expect("audit reads and writes succeed") + .expect("the corrupted store is repaired"); + assert_eq!( + repair.last_coherent, + Some((Height(FORK_HEIGHT), universe.trunk_at(FORK_HEIGHT).hash)), + "an orphan reverse row must not shorten the coherent chain" + ); + assert_eq!(repair.deleted_rows, 1); + assert!( + repair.violations.iter().any(|violation| matches!( + violation, + ZakuraStoreViolation::OrphanHeightByHash { hash, points_at } + if *hash == stray_hash && *points_at == Height(12) + )), + "expected OrphanHeightByHash for the stray entry: {:?}", + repair.violations + ); + assert_eq!(dump_store(&state), original); + assert_clean(&state); +} + +/// Stale zakura rows at a committed height (the pre-guard re-delivery shape: +/// rows the release trim never touches again) are removed without disturbing +/// the frontier above. +#[test] +fn startup_removes_stale_rows_at_committed_heights() { + let _init_guard = zebra_test::init(); + let universe = Universe::new(); + let state = trunk_state(&universe, Config::ephemeral()); + let original = dump_store(&state); + + // Genesis is the only committed block; plant zakura rows at its height. + let header_cf = state.db.cf_handle(ZAKURA_HEADER_BY_HEIGHT).unwrap(); + let hash_cf = state.db.cf_handle(ZAKURA_HEADER_HASH_BY_HEIGHT).unwrap(); + let height_by_hash_cf = state.db.cf_handle(ZAKURA_HEADER_HEIGHT_BY_HASH).unwrap(); + let genesis_hash = universe.genesis.hash(); + let mut batch = DiskWriteBatch::new(); + batch.zs_insert(&header_cf, Height(0), universe.genesis.header.clone()); + batch.zs_insert(&hash_cf, Height(0), genesis_hash); + batch.zs_insert(&height_by_hash_cf, genesis_hash, Height(0)); + state.db.write(batch).expect("raw insert writes"); + + let repair = state + .audit_and_repair_zakura_header_store() + .expect("audit reads and writes succeed") + .expect("the corrupted store is repaired"); + assert_eq!( + repair.last_coherent, + Some((Height(FORK_HEIGHT), universe.trunk_at(FORK_HEIGHT).hash)), + "stale committed-height rows must not shorten the frontier" + ); + assert_eq!(repair.deleted_rows, 3); + assert!( + repair.violations.iter().all(|violation| matches!( + violation, + ZakuraStoreViolation::StaleRowAtCommittedHeight { height, .. } + if *height == Height(0) + )), + "expected only StaleRowAtCommittedHeight at genesis: {:?}", + repair.violations + ); + assert_eq!(dump_store(&state), original); + assert_clean(&state); +} + +/// The reads.rs anchor-corruption shape (a hash row overwritten with a foreign +/// hash) makes the Pillar-2 range writer reject with `StoreIncoherent`; after +/// the heal the same delivery commits. This pins the Pillar-2 interaction: any +/// store those reads refuse is healed by this audit. +#[test] +fn startup_heal_unblocks_linkage_verified_reads() { + let _init_guard = zebra_test::init(); + let universe = Universe::new(); + let state = trunk_state(&universe, Config::ephemeral()); + let original = dump_store(&state); + + // The hash row at height 30 now names a different block, while the header + // row and `height_by_hash` still claim the trunk. + let hash_cf = state.db.cf_handle(ZAKURA_HEADER_HASH_BY_HEIGHT).unwrap(); + let stray_hash = universe.branches[BRANCH_A].headers[0].hash; + let mut batch = DiskWriteBatch::new(); + batch.zs_insert(&hash_cf, Height(30), stray_hash); + state.db.write(batch).expect("raw insert writes"); + + // A re-delivered range anchored at trunk@30 is rejected as a local + // storage fault before the heal. + let anchor = universe.trunk_at(30).hash; + let headers: Vec<_> = universe.trunk[30..35] + .iter() + .map(|fab| fab.header.clone()) + .collect(); + let body_sizes = vec![0; headers.len()]; + let mut batch = DiskWriteBatch::new(); + let error = batch + .prepare_header_range_batch(&state, anchor, &headers, &body_sizes) + .expect_err("the anchor round-trip fails on the corrupted index"); + assert!(matches!( + error, + CommitHeaderRangeError::StoreIncoherent(StoreIncoherentError::BijectionMismatch { .. }) + )); + + let repair = state + .audit_and_repair_zakura_header_store() + .expect("audit reads and writes succeed") + .expect("the corrupted store is repaired"); + assert_eq!( + repair.last_coherent, + Some((Height(29), universe.trunk_at(29).hash)) + ); + assert!( + repair.violations.iter().any(|violation| matches!( + violation, + ZakuraStoreViolation::HeaderHashMismatch { height, indexed, .. } + if *height == Height(30) && *indexed == stray_hash + )), + "expected HeaderHashMismatch at the overwritten hash row: {:?}", + repair.violations + ); + assert_eq!(dump_store(&state), truncated(&original, Height(29))); + assert_clean(&state); + + // The same fork of history now commits, anchored at the healed tip. + let redelivered: Vec<_> = universe.trunk[29..FORK_HEIGHT as usize] + .iter() + .map(|fab| fab.header.clone()) + .collect(); + commit_header_range(&state, universe.trunk_at(29).hash, &redelivered); + assert_eq!(dump_store(&state), original); + assert_clean(&state); +} + +/// A minimal local metrics recorder capturing counter increments by name. +#[derive(Clone, Default)] +struct CounterCapture(Arc>>); + +impl CounterCapture { + fn get(&self, name: &str) -> u64 { + self.0 + .lock() + .expect("counter capture lock is never poisoned") + .get(name) + .copied() + .unwrap_or(0) + } +} + +struct CaptureHandle { + name: String, + store: Arc>>, +} + +impl metrics::CounterFn for CaptureHandle { + fn increment(&self, value: u64) { + *self + .store + .lock() + .expect("counter capture lock is never poisoned") + .entry(self.name.clone()) + .or_insert(0) += value; + } + + fn absolute(&self, value: u64) { + self.store + .lock() + .expect("counter capture lock is never poisoned") + .insert(self.name.clone(), value); + } +} + +impl metrics::Recorder for CounterCapture { + fn describe_counter( + &self, + _: metrics::KeyName, + _: Option, + _: metrics::SharedString, + ) { + } + + fn describe_gauge( + &self, + _: metrics::KeyName, + _: Option, + _: metrics::SharedString, + ) { + } + + fn describe_histogram( + &self, + _: metrics::KeyName, + _: Option, + _: metrics::SharedString, + ) { + } + + fn register_counter(&self, key: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Counter { + metrics::Counter::from_arc(Arc::new(CaptureHandle { + name: key.name().to_string(), + store: self.0.clone(), + })) + } + + fn register_gauge(&self, _: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Gauge { + metrics::Gauge::noop() + } + + fn register_histogram( + &self, + _: &metrics::Key, + _: &metrics::Metadata<'_>, + ) -> metrics::Histogram { + metrics::Histogram::noop() + } +} + +/// The `state.zakura.header_store.incoherent` metric fires exactly when a +/// startup repair runs: once for the healing reopen, and not at all for the +/// clean reopen after it. +#[test] +fn startup_repair_emits_incoherent_metric() { + let _init_guard = zebra_test::init(); + let universe = Universe::new(); + let tempdir = tempfile::tempdir().expect("test tempdir is available"); + let config = startup_repair_config(tempdir.path()); + let state = trunk_state(&universe, config.clone()); + + let header_cf = state.db.cf_handle(ZAKURA_HEADER_BY_HEIGHT).unwrap(); + let hash_cf = state.db.cf_handle(ZAKURA_HEADER_HASH_BY_HEIGHT).unwrap(); + let mut batch = DiskWriteBatch::new(); + batch.zs_delete(&header_cf, Height(15)); + batch.zs_delete(&hash_cf, Height(15)); + state.db.write(batch).expect("raw delete writes"); + + const METRIC: &str = "state.zakura.header_store.incoherent"; + + let healing = CounterCapture::default(); + let state = metrics::with_local_recorder(&healing, || reopen(state, &config, &universe)); + assert_eq!( + healing.get(METRIC), + 1, + "the healing reopen emits the metric" + ); + assert_clean(&state); + + let clean = CounterCapture::default(); + let state = metrics::with_local_recorder(&clean, || reopen(state, &config, &universe)); + assert_eq!(clean.get(METRIC), 0, "a clean reopen emits nothing"); + assert_clean(&state); +}