From 5435be72b862638af448d6265ff3a2d0028fe762 Mon Sep 17 00:00:00 2001 From: Santiago Date: Tue, 18 Aug 2026 08:22:52 -0300 Subject: [PATCH 1/4] feat(doctor): rebuild-state command replays the local archive into a fresh state store MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New `dolos doctor rebuild-state`: regenerates the state store from the instance's own archive through the import lifecycle — no network, no snapshot re-import, no re-writing of the archive or indexes. In-place by default (crash-safe sequence: WAL to origin, wipe, replay, WAL reseed), with --target and --ephemeral for isolated outputs, --stop-epoch to bound an isolated replay, and --rewrite-logs to overwrite the archive's derived log rows through a new write-gated ArchiveStoreBackend::LogsOnly view over the already-open redb store. Co-Authored-By: Claude Fable 5 --- src/adapters/storage.rs | 53 ++- src/bin/dolos/data/cardinality_stats.rs | 2 +- src/bin/dolos/data/export.rs | 8 +- src/bin/dolos/data/prune_chain.rs | 2 +- src/bin/dolos/doctor/mod.rs | 5 + src/bin/dolos/doctor/rebuild_state.rs | 458 ++++++++++++++++++++++++ tests/node/mod.rs | 18 + tests/rebuild_state.rs | 239 +++++++++++++ 8 files changed, 770 insertions(+), 15 deletions(-) create mode 100644 src/bin/dolos/doctor/rebuild_state.rs create mode 100644 tests/rebuild_state.rs diff --git a/src/adapters/storage.rs b/src/adapters/storage.rs index fbb29fad6..b7f32b6d4 100644 --- a/src/adapters/storage.rs +++ b/src/adapters/storage.rs @@ -632,6 +632,16 @@ impl CoreStateStore for StateStoreBackend { #[derive(Clone)] pub enum ArchiveStoreBackend { Redb(dolos_redb3::archive::ArchiveStore), + /// Write-gated view over an already-open redb archive: reads and + /// derived-log writes pass through, block appends and undos are + /// discarded. + /// + /// This exists for replays over an archive that already holds the chain + /// (`dolos doctor rebuild-state --rewrite-logs`): block appends are not + /// idempotent — replaying them would double every segment file — while + /// boundary log keys are slot-derived and identical under replay, so + /// re-written log rows overwrite the originals in place. + LogsOnly(dolos_redb3::archive::ArchiveStore), NoOp(NoOpArchiveStore), } @@ -652,6 +662,19 @@ impl ArchiveStoreBackend { Self::NoOp(NoOpArchiveStore) } + /// Wrap this backend's already-open redb store in a [`Self::LogsOnly`] + /// write gate. + /// + /// Clones the handle out of the open store rather than opening the path + /// again, because redb refuses to open the same file twice. Returns + /// `None` when there is no redb store to wrap. + pub fn logs_only(&self) -> Option { + match self { + Self::Redb(s) | Self::LogsOnly(s) => Some(Self::LogsOnly(s.clone())), + Self::NoOp(_) => None, + } + } + /// Create an in-memory archive store. pub fn in_memory(schema: StateSchema) -> Result { Ok(Self::Redb(dolos_redb3::archive::ArchiveStore::in_memory( @@ -678,7 +701,7 @@ impl ArchiveStoreBackend { pub fn shutdown(&self) -> Result<(), ArchiveError> { match self { - Self::Redb(s) => s + Self::Redb(s) | Self::LogsOnly(s) => s .shutdown() .map_err(|e| ArchiveError::InternalError(e.to_string())), Self::NoOp(s) => s.shutdown(), @@ -688,6 +711,8 @@ impl ArchiveStoreBackend { pub enum ArchiveWriterBackend { Redb(Box<::Writer>), + /// Delegates `write_log` and `commit`; discards `apply` and `undo`. + LogsOnly(Box<::Writer>), NoOp(NoOpArchiveWriter), } @@ -695,6 +720,7 @@ impl CoreArchiveWriter for ArchiveWriterBackend { fn apply(&self, point: &ChainPoint, block: &RawBlock) -> Result<(), ArchiveError> { match self { Self::Redb(w) => w.apply(point, block), + Self::LogsOnly(_) => Ok(()), Self::NoOp(w) => w.apply(point, block), } } @@ -706,7 +732,7 @@ impl CoreArchiveWriter for ArchiveWriterBackend { value: &EntityValue, ) -> Result<(), ArchiveError> { match self { - Self::Redb(w) => w.write_log(ns, key, value), + Self::Redb(w) | Self::LogsOnly(w) => w.write_log(ns, key, value), Self::NoOp(w) => w.write_log(ns, key, value), } } @@ -714,13 +740,14 @@ impl CoreArchiveWriter for ArchiveWriterBackend { fn undo(&self, point: &ChainPoint) -> Result<(), ArchiveError> { match self { Self::Redb(w) => w.undo(point), + Self::LogsOnly(_) => Ok(()), Self::NoOp(w) => w.undo(point), } } fn commit(self) -> Result<(), ArchiveError> { match self { - Self::Redb(w) => (*w).commit(), + Self::Redb(w) | Self::LogsOnly(w) => (*w).commit(), Self::NoOp(w) => w.commit(), } } @@ -806,6 +833,8 @@ impl CoreArchiveStore for ArchiveStoreBackend { match self { Self::Redb(s) => CoreArchiveStore::start_writer(s) .map(|writer| ArchiveWriterBackend::Redb(Box::new(writer))), + Self::LogsOnly(s) => CoreArchiveStore::start_writer(s) + .map(|writer| ArchiveWriterBackend::LogsOnly(Box::new(writer))), Self::NoOp(s) => CoreArchiveStore::start_writer(s).map(ArchiveWriterBackend::NoOp), } } @@ -816,7 +845,7 @@ impl CoreArchiveStore for ArchiveStoreBackend { keys: &[&LogKey], ) -> Result>, ArchiveError> { match self { - Self::Redb(s) => CoreArchiveStore::read_logs(s, ns, keys), + Self::Redb(s) | Self::LogsOnly(s) => CoreArchiveStore::read_logs(s, ns, keys), Self::NoOp(s) => CoreArchiveStore::read_logs(s, ns, keys), } } @@ -827,7 +856,7 @@ impl CoreArchiveStore for ArchiveStoreBackend { range: Range, ) -> Result { match self { - Self::Redb(s) => CoreArchiveStore::iter_logs(s, ns, range) + Self::Redb(s) | Self::LogsOnly(s) => CoreArchiveStore::iter_logs(s, ns, range) .map(|iter| ArchiveLogIterBackend::Redb(Box::new(iter))), Self::NoOp(s) => { CoreArchiveStore::iter_logs(s, ns, range).map(ArchiveLogIterBackend::NoOp) @@ -837,7 +866,7 @@ impl CoreArchiveStore for ArchiveStoreBackend { fn get_block_by_slot(&self, slot: &BlockSlot) -> Result, ArchiveError> { match self { - Self::Redb(s) => CoreArchiveStore::get_block_by_slot(s, slot), + Self::Redb(s) | Self::LogsOnly(s) => CoreArchiveStore::get_block_by_slot(s, slot), Self::NoOp(s) => CoreArchiveStore::get_block_by_slot(s, slot), } } @@ -848,7 +877,7 @@ impl CoreArchiveStore for ArchiveStoreBackend { to: Option, ) -> Result, ArchiveError> { match self { - Self::Redb(s) => CoreArchiveStore::get_range(s, from, to) + Self::Redb(s) | Self::LogsOnly(s) => CoreArchiveStore::get_range(s, from, to) .map(|iter| ArchiveBlockIterBackend::Redb(Box::new(iter))), Self::NoOp(s) => { CoreArchiveStore::get_range(s, from, to).map(ArchiveBlockIterBackend::NoOp) @@ -858,28 +887,30 @@ impl CoreArchiveStore for ArchiveStoreBackend { fn find_intersect(&self, intersect: &[ChainPoint]) -> Result, ArchiveError> { match self { - Self::Redb(s) => CoreArchiveStore::find_intersect(s, intersect), + Self::Redb(s) | Self::LogsOnly(s) => CoreArchiveStore::find_intersect(s, intersect), Self::NoOp(s) => CoreArchiveStore::find_intersect(s, intersect), } } fn get_tip(&self) -> Result, ArchiveError> { match self { - Self::Redb(s) => CoreArchiveStore::get_tip(s), + Self::Redb(s) | Self::LogsOnly(s) => CoreArchiveStore::get_tip(s), Self::NoOp(s) => CoreArchiveStore::get_tip(s), } } fn prune_history(&self, max_slots: u64, max_prune: Option) -> Result { match self { - Self::Redb(s) => CoreArchiveStore::prune_history(s, max_slots, max_prune), + Self::Redb(s) | Self::LogsOnly(s) => { + CoreArchiveStore::prune_history(s, max_slots, max_prune) + } Self::NoOp(s) => CoreArchiveStore::prune_history(s, max_slots, max_prune), } } fn truncate_front(&self, after: &ChainPoint) -> Result<(), ArchiveError> { match self { - Self::Redb(s) => CoreArchiveStore::truncate_front(s, after), + Self::Redb(s) | Self::LogsOnly(s) => CoreArchiveStore::truncate_front(s, after), Self::NoOp(s) => CoreArchiveStore::truncate_front(s, after), } } diff --git a/src/bin/dolos/data/cardinality_stats.rs b/src/bin/dolos/data/cardinality_stats.rs index af6843f29..63b180150 100644 --- a/src/bin/dolos/data/cardinality_stats.rs +++ b/src/bin/dolos/data/cardinality_stats.rs @@ -101,7 +101,7 @@ pub fn run(config: &RootConfig, args: &Args) -> miette::Result<()> { // This command requires direct redb access let archive = match archive { - ArchiveStoreBackend::Redb(s) => s, + ArchiveStoreBackend::Redb(s) | ArchiveStoreBackend::LogsOnly(s) => s, ArchiveStoreBackend::NoOp(_) => { bail!("cardinality-stats command is not available for noop archive backend") } diff --git a/src/bin/dolos/data/export.rs b/src/bin/dolos/data/export.rs index 0a9abe962..ffa9885e0 100644 --- a/src/bin/dolos/data/export.rs +++ b/src/bin/dolos/data/export.rs @@ -160,8 +160,12 @@ pub fn run( // prepare_archive requires direct redb access match &mut stores.archive { - ArchiveStoreBackend::Redb(s) if !args.skip_sanitization => prepare_archive(s, &pb)?, - ArchiveStoreBackend::Redb(_) => {} + ArchiveStoreBackend::Redb(s) | ArchiveStoreBackend::LogsOnly(s) + if !args.skip_sanitization => + { + prepare_archive(s, &pb)? + } + ArchiveStoreBackend::Redb(_) | ArchiveStoreBackend::LogsOnly(_) => {} ArchiveStoreBackend::NoOp(_) => { bail!("export command is not available for noop archive backend") } diff --git a/src/bin/dolos/data/prune_chain.rs b/src/bin/dolos/data/prune_chain.rs index 81d74d5d0..fc1676ef7 100644 --- a/src/bin/dolos/data/prune_chain.rs +++ b/src/bin/dolos/data/prune_chain.rs @@ -38,7 +38,7 @@ pub fn run(config: &RootConfig, args: &Args) -> miette::Result<()> { // Compaction requires direct redb access match &mut stores.archive { - ArchiveStoreBackend::Redb(s) => { + ArchiveStoreBackend::Redb(s) | ArchiveStoreBackend::LogsOnly(s) => { let db = s.db_mut(); while db.compact().into_diagnostic()? { diff --git a/src/bin/dolos/doctor/mod.rs b/src/bin/dolos/doctor/mod.rs index 49d3d8030..3b9ed0e6f 100644 --- a/src/bin/dolos/doctor/mod.rs +++ b/src/bin/dolos/doctor/mod.rs @@ -5,6 +5,7 @@ use crate::feedback::Feedback; mod catchup_stores; mod check; +mod rebuild_state; mod reset_wal; mod rollback; mod update_entity; @@ -21,6 +22,9 @@ pub enum Command { /// catch up store data from WAL records CatchupStores(catchup_stores::Args), + /// rebuild the state store by replaying the local archive + RebuildState(rebuild_state::Args), + // Reset WAL position using state cursor ResetWal(reset_wal::Args), @@ -48,6 +52,7 @@ pub fn run(config: &RootConfig, args: &Args, feedback: &Feedback) -> miette::Res match &args.command { Command::Check(x) => check::run(config, x)?, Command::CatchupStores(x) => catchup_stores::run(config, x, feedback)?, + Command::RebuildState(x) => rebuild_state::run(config, x, feedback)?, Command::ResetWal(x) => reset_wal::run(config, x, feedback)?, Command::WalIntegrity(x) => wal_integrity::run(config, x)?, Command::Rollback(x) => rollback::run(config, x)?, diff --git a/src/bin/dolos/doctor/rebuild_state.rs b/src/bin/dolos/doctor/rebuild_state.rs new file mode 100644 index 000000000..da212acdb --- /dev/null +++ b/src/bin/dolos/doctor/rebuild_state.rs @@ -0,0 +1,458 @@ +//! `dolos doctor rebuild-state` — regenerate the state store from the local +//! archive. +//! +//! The archive holds the full raw chain, and state computation never reads +//! the archive, so a synced instance can rebuild its state store offline by +//! replaying its own archive through the import lifecycle: no network, no +//! snapshot re-import, no re-writing of the archive or indexes. This is the +//! debugging loop for state-math fixes — a full replay re-writes every store +//! to test a change that only touches state. +//! +//! Three output modes: in place (the default — wipe and regenerate the +//! instance's own state store), `--target ` (a fresh state store +//! somewhere else, instance untouched), and `--ephemeral` (an in-memory +//! state store, discarded on exit — a pure validation run). +//! +//! The in-place sequence is ordered for crash safety: the WAL is reset to +//! origin *before* the state store is wiped, and reseeded to the final +//! cursor only after the replay completes. A crash mid-rebuild therefore +//! leaves WAL(origin) behind a partial state cursor, which the next startup +//! refuses loudly (`InconsistentState`); re-running this command recovers. + +use std::io::IsTerminal as _; +use std::path::PathBuf; +use std::sync::Arc; + +use dolos_core::config::{ChainConfig, RootConfig, StateStoreConfig}; +use dolos_core::ImportExt as _; +use indicatif::ProgressBar; +use miette::{bail, Context as _, IntoDiagnostic as _}; +use pallas::ledger::traverse::MultiEraBlock; + +use dolos::adapters::DomainAdapter; +use dolos::prelude::*; +use dolos::storage::{ + ArchiveStoreBackend, IndexStoreBackend, MempoolBackend, StateStoreBackend, WalStoreBackend, +}; + +use crate::feedback::Feedback; + +#[derive(Debug, clap::Args)] +pub struct Args { + /// Rebuild into a fresh state store at this path, leaving the instance's + /// own stores untouched + #[arg(long, conflicts_with = "ephemeral")] + target: Option, + + /// Rebuild into an in-memory state store and discard it (a validation + /// run that writes nothing) + #[arg(long)] + ephemeral: bool, + + /// Also re-write the derived log records the archive carries (StakeLog, + /// reward logs, EpochState); in-place mode only + #[arg(long, conflicts_with_all = ["target", "ephemeral"])] + rewrite_logs: bool, + + /// Stop cleanly once the replay reaches this epoch (requires --target or + /// --ephemeral) + #[arg(long)] + stop_epoch: Option, + + /// Number of blocks to import per chunk + #[arg(long, default_value_t = 500)] + chunk: usize, + + /// Skip the interactive confirmation of the in-place wipe (required when + /// no terminal is attached) + #[arg(long)] + force: bool, +} + +enum Mode { + InPlace, + Target(PathBuf), + Ephemeral, +} + +/// Add the "instance appears to be running" reading to a store-open failure. +/// +/// The backend file locks are the only concurrency guard this command has, so +/// a failure to open a store on a synced instance is most often a daemon (or +/// another dolos command) still holding it. +fn open_store(what: &str, result: Result) -> miette::Result { + result.map_err(|e| { + miette::miette!( + help = "if the error mentions a lock, the instance appears to be running; stop the \ + daemon (or the other dolos command) and re-run", + "opening the {what} store failed: {e}", + ) + }) +} + +/// Refuse an in-place wipe that would take another store with it. +/// +/// The wipe removes `state_path` recursively, so any other configured store +/// path at or under it is a configuration this command must not act on. Paths +/// are compared as the configuration resolves them, which is also how the +/// stores themselves are opened. +fn check_wipe_scope(config: &RootConfig, state_path: &std::path::Path) -> miette::Result<()> { + let mut others: Vec<(&str, PathBuf)> = vec![("storage root", config.storage.path.clone())]; + + if let Some(path) = config.storage.wal_path() { + others.push(("wal", path)); + } + if let Some(path) = config.storage.archive_path() { + others.push(("archive", path)); + } + if let Some(path) = config.storage.index_path() { + others.push(("index", path)); + } + if let Some(path) = config.storage.mempool_path() { + others.push(("mempool", path)); + } + + // Segment files can live outside the archive directory; the backend takes + // `blocks_path` verbatim, so this check does too. + if let dolos_core::config::ArchiveStoreConfig::Redb(cfg) = &config.storage.archive { + if let Some(path) = &cfg.blocks_path { + others.push(("archive segments", path.clone())); + } + } + + for (what, path) in others { + if path.starts_with(state_path) { + bail!( + "refusing the in-place wipe: the {what} path {} sits at or under the state path \ + {}; wiping the state store would take it too", + path.display(), + state_path.display(), + ); + } + } + + Ok(()) +} + +fn confirm_wipe(state_path: &std::path::Path, force: bool) -> miette::Result<()> { + if force { + return Ok(()); + } + + if !std::io::stdin().is_terminal() { + bail!( + "in-place rebuild wipes {} and no terminal is attached to confirm it; pass --force", + state_path.display(), + ); + } + + eprint!( + "about to wipe {} and rebuild it from the local archive; continue? [y/N] ", + state_path.display(), + ); + + let mut answer = String::new(); + std::io::stdin() + .read_line(&mut answer) + .into_diagnostic() + .context("reading confirmation")?; + + if !matches!(answer.trim(), "y" | "Y" | "yes" | "YES") { + bail!("aborted"); + } + + Ok(()) +} + +/// Check the archive is usable as a replay source and return its tip point. +/// +/// Two gates: the first block must sit at the start of the chain (an archive +/// pruned by `max_history` cannot rebuild state from origin, by design), and +/// the tip must decode. Deliberately *not* a strict prev-hash walk from +/// origin: Byron EBBs are overwritten in the slot-keyed blocks table, so that +/// walk reports broken continuity at essentially every Byron epoch on a +/// legitimate mainnet archive. Real continuity is enforced during the replay +/// itself by `check_extension`, which tolerates exactly the EBB hole and +/// aborts on any mid-epoch gap. +fn preflight(archive: &ArchiveStoreBackend) -> miette::Result { + let first = archive + .get_range(None, None) + .into_diagnostic() + .context("iterating archive blocks")? + .next(); + + let Some((first_slot, first_body)) = first else { + bail!("the archive is empty; there is nothing to rebuild state from"); + }; + + let first = MultiEraBlock::decode(&first_body) + .into_diagnostic() + .with_context(|| format!("decoding the archive's first block at slot {first_slot}"))?; + + if first.number() > 1 { + bail!( + help = "instances running a `max_history` window cannot use this command; the replay \ + always starts from origin", + "the archive does not start at the beginning of the chain (first block is #{} at \ + slot {first_slot}); it looks pruned", + first.number(), + ); + } + + let (tip_slot, tip_body) = archive + .get_tip() + .into_diagnostic() + .context("reading archive tip")? + .expect("archive with a first block has a tip"); + + let tip = MultiEraBlock::decode(&tip_body) + .into_diagnostic() + .with_context(|| format!("decoding the archive tip block at slot {tip_slot}"))?; + + Ok(ChainPoint::Specific(tip_slot, tip.hash())) +} + +/// Assemble the rebuild domain: the fresh state store, an empty in-memory WAL +/// (the import lifecycle skips `commit_wal`, so it stays empty), the given +/// archive backend (no-op, or the write-gated view under `--rewrite-logs`), +/// no-op indexes and an ephemeral mempool. Genesis fires automatically on the +/// first imported block; its one index delta is discarded by the no-op index, +/// which is correct — the live indexes already carry it. +fn build_domain( + config: &RootConfig, + state: StateStoreBackend, + archive: ArchiveStoreBackend, + stop_epoch: Option, +) -> miette::Result { + let genesis = Arc::new(crate::common::open_genesis_files(&config.genesis)?); + + let ChainConfig::Cardano(mut chain_config) = config.chain.clone(); + + if stop_epoch.is_some() { + chain_config.stop_epoch = stop_epoch; + } + + let chain = + dolos_cardano::CardanoLogic::initialize::(chain_config, &state, &genesis) + .into_diagnostic() + .context("initializing chain logic against the fresh state")?; + + let wal = WalStoreBackend::in_memory() + .into_diagnostic() + .context("creating the in-memory WAL")?; + + let (tip_broadcast, _) = tokio::sync::broadcast::channel(100); + + Ok(DomainAdapter { + storage_config: Arc::new(config.storage.clone()), + sync_config: Arc::new(config.sync.clone()), + genesis, + chain: Arc::new(std::sync::RwLock::new(chain)), + wal, + state, + archive, + indexes: IndexStoreBackend::noop(), + mempool: MempoolBackend::Ephemeral(dolos_core::builtin::EphemeralMempool::new()), + tip_broadcast, + }) +} + +/// Replay the archive into the rebuild domain in chunks. +/// +/// The range iterator is re-opened per chunk: it holds a redb read +/// transaction, and one held across writes blocks page reclamation for the +/// whole run. Returns whether the replay stopped at `stop_epoch` rather than +/// at the archive tip. +fn replay( + domain: &DomainAdapter, + source: &ArchiveStoreBackend, + chunk_size: usize, + progress: &ProgressBar, +) -> miette::Result { + let mut cursor: Option = None; + + loop { + let blocks: Vec = source + .get_range(cursor.map(|slot| slot + 1), None) + .into_diagnostic() + .context("iterating archive blocks")? + .take(chunk_size) + .map(|(_, body)| Arc::new(body)) + .collect(); + + if blocks.is_empty() { + return Ok(false); + } + + match domain.import_blocks(blocks) { + Ok(last) => { + cursor = Some(last); + progress.set_position(last); + } + Err(DomainError::StopEpochReached) => return Ok(true), + Err(e) => { + return Err(miette::miette!("{e}")).with_context(|| { + format!( + "importing a block chunk after slot {}", + cursor.unwrap_or_default(), + ) + }) + } + } + } +} + +pub fn run(config: &RootConfig, args: &Args, feedback: &Feedback) -> miette::Result<()> { + crate::common::setup_tracing_error_only()?; + + let mode = match (&args.target, args.ephemeral) { + (Some(path), _) => Mode::Target(path.clone()), + (None, true) => Mode::Ephemeral, + (None, false) => Mode::InPlace, + }; + + if args.stop_epoch.is_some() && matches!(mode, Mode::InPlace) { + bail!( + "--stop-epoch needs --target or --ephemeral: a partial in-place state cannot be \ + reconciled with the live stores" + ); + } + + let archive = open_store("archive", crate::common::open_archive_store(config))?; + + let tip = preflight(&archive)?; + + // The domain's archive: writes are discarded, except under --rewrite-logs + // where derived-log writes pass through to the already-open store (redb + // will not open the same file twice, so the open handle is what gets + // wrapped). + let domain_archive = if args.rewrite_logs { + archive + .logs_only() + .ok_or_else(|| miette::miette!("--rewrite-logs needs a persistent redb archive"))? + } else { + ArchiveStoreBackend::noop() + }; + + let (state, wal) = match &mode { + Mode::InPlace => { + let Some(state_path) = config.storage.state_path() else { + bail!( + "the configured state backend is in_memory, so there is no on-disk state \ + store to rebuild in place; use --ephemeral or --target" + ); + }; + + check_wipe_scope(config, &state_path)?; + confirm_wipe(&state_path, args.force)?; + + let wal = open_store("wal", crate::common::open_wal_store(config))?; + + // Reset the WAL *before* the wipe: from here until the final + // reseed, a crash leaves WAL(origin) behind the state cursor, + // which the next startup refuses loudly instead of serving a + // half-rebuilt ledger. Re-running this command recovers. + wal.reset_to(&ChainPoint::Origin) + .into_diagnostic() + .context("resetting the WAL to origin")?; + + if state_path.exists() { + std::fs::remove_dir_all(&state_path) + .into_diagnostic() + .with_context(|| { + format!("wiping the state store at {}", state_path.display()) + })?; + } + + let state = open_store("state", crate::common::open_state_store(config))?; + + (state, Some(wal)) + } + Mode::Target(path) => { + if path.exists() && path.read_dir().into_diagnostic()?.next().is_some() { + bail!( + "--target {} is not empty; pick a fresh directory", + path.display(), + ); + } + + let state = match &config.storage.state { + StateStoreConfig::Fjall(cfg) => StateStoreBackend::open_fjall(path, cfg), + _ => StateStoreBackend::open_fjall( + path, + &dolos_core::config::FjallStateConfig::default(), + ), + } + .into_diagnostic() + .with_context(|| format!("opening the target state store at {}", path.display()))?; + + (state, None) + } + Mode::Ephemeral => { + let state = StateStoreBackend::in_memory() + .into_diagnostic() + .context("creating the in-memory state store")?; + + (state, None) + } + }; + + let domain = build_domain(config, state.clone(), domain_archive, args.stop_epoch)?; + + let progress = feedback.slot_progress_bar(); + progress.set_message("rebuilding state from archive"); + progress.set_length(tip.slot()); + + let replayed = replay(&domain, &archive, args.chunk, &progress); + + // Shut down even when the replay failed: fjall in particular has + // background work to flush before the handle drops. + let shutdown = domain.shutdown(); + + let stopped = replayed?; + shutdown.map_err(|e| miette::miette!("shutting down the rebuild domain: {e}"))?; + + progress.finish_with_message("replay complete"); + + let cursor = state + .read_cursor() + .into_diagnostic() + .context("reading the rebuilt state cursor")?; + + let Some(cursor) = cursor else { + bail!("the rebuilt state has no cursor; the replay produced nothing"); + }; + + if !cursor.is_fully_defined() { + bail!("the rebuilt state cursor {cursor} carries no block hash"); + } + + if !stopped && cursor != tip { + bail!("the rebuilt state cursor {cursor} does not match the archive tip {tip}"); + } + + if let Some(wal) = wal { + wal.reset_to(&cursor) + .into_diagnostic() + .context("reseeding the WAL from the rebuilt state cursor")?; + } + + match &mode { + Mode::InPlace => { + println!("state rebuilt in place; cursor at {cursor}, WAL reseeded"); + println!("run `dolos data check` to verify the rebuilt stores"); + } + Mode::Target(path) => { + println!( + "state rebuilt into {}; cursor at {cursor}; the instance's stores were not \ + touched", + path.display(), + ); + } + Mode::Ephemeral => { + println!("ephemeral rebuild completed; cursor at {cursor}; nothing was written"); + } + } + + Ok(()) +} diff --git a/tests/node/mod.rs b/tests/node/mod.rs index 543551e5d..824d0abb2 100644 --- a/tests/node/mod.rs +++ b/tests/node/mod.rs @@ -105,6 +105,24 @@ impl Node { ..Default::default() }); + // The synthetic chain's initial UTxOs are seeded through the chain + // config's `custom_utxos`, so they are part of this chain's genesis. + // Persist them into `dolos.toml`: a separate process replaying from + // origin (`doctor rebuild-state`) runs genesis from the file and must + // seed the same set. + let config_path = self.config_path(); + let mut document: toml::Value = + toml::from_str(&std::fs::read_to_string(&config_path).unwrap()).unwrap(); + document + .get_mut("chain") + .and_then(|chain| chain.as_table_mut()) + .unwrap() + .insert( + "custom_utxos".to_owned(), + toml::Value::try_from(&chain_config.custom_utxos).unwrap(), + ); + std::fs::write(&config_path, toml::to_string(&document).unwrap()).unwrap(); + let chain = dolos_cardano::CardanoLogic::initialize::( chain_config, &stores.state, diff --git a/tests/rebuild_state.rs b/tests/rebuild_state.rs new file mode 100644 index 000000000..2c09a83bb --- /dev/null +++ b/tests/rebuild_state.rs @@ -0,0 +1,239 @@ +//! `dolos doctor rebuild-state`, end to end against real on-disk stores. +//! +//! The rebuild-equality claim: a chain built through the import lifecycle, +//! its state wiped, rebuilt from nothing but the instance's own archive, +//! produces the same state store — cursor, every entity in every namespace, +//! and the full UTxO set. The comparison is deliberately exhaustive rather +//! than a tip check: a rebuild that landed the right cursor over a diverged +//! ledger must fail here. + +mod node; + +use std::collections::BTreeMap; +use std::path::Path; + +use dolos_core::{ + ArchiveStore as _, ChainPoint, EntityKey, LogKey, StateStore as _, UtxoEntry, WalStore as _, +}; +use node::{assert_ok, Node}; + +/// Everything a state store holds, read through the same backend enums the +/// node runs on. +#[derive(Debug, PartialEq)] +struct StateContents { + cursor: Option, + entities: BTreeMap<&'static str, Vec<(EntityKey, Vec)>>, + utxos: Vec, +} + +impl StateContents { + fn read(state: &dolos::storage::StateStoreBackend) -> Self { + let mut entities = BTreeMap::new(); + + for ns in dolos_cardano::model::build_schema().keys() { + let rows: Vec<_> = state + .iter_entities(ns, EntityKey::full_range()) + .unwrap() + .collect::>() + .unwrap(); + + entities.insert(*ns, rows); + } + + let utxos = state + .iter_utxos() + .unwrap() + .collect::>() + .unwrap(); + + Self { + cursor: state.read_cursor().unwrap(), + entities, + utxos, + } + } +} + +/// The instance seen whole: state, the archive's derived-log rows, the WAL +/// tip, and the archive segment files byte for byte. +#[derive(Debug, PartialEq)] +struct InstanceContents { + state: StateContents, + logs: BTreeMap<&'static str, Vec<(LogKey, Vec)>>, + wal_tip: Option, + segments: BTreeMap>, +} + +impl InstanceContents { + fn read(node: &Node) -> Self { + let stores = + dolos::storage::open_data_stores::(&node.config).unwrap(); + + let mut logs = BTreeMap::new(); + + for ns in dolos_cardano::model::build_schema().keys() { + let rows: Vec<_> = stores + .archive + .iter_logs(ns, LogKey::full_range()) + .unwrap() + .collect::>() + .unwrap(); + + logs.insert(*ns, rows); + } + + let contents = Self { + state: StateContents::read(&stores.state), + logs, + wal_tip: stores.wal.find_tip().unwrap().map(|(point, _)| point), + segments: read_segments(&node.config.storage.archive_path().unwrap()), + }; + + drop(stores); + + contents + } +} + +/// The archive's block segment files, whole. The redb index (`index`) is +/// excluded: opening the store at all advances its transaction counter, so +/// byte-identity is a claim only the segment files can carry. +fn read_segments(archive_dir: &Path) -> BTreeMap> { + let mut out = BTreeMap::new(); + + for entry in std::fs::read_dir(archive_dir).unwrap() { + let entry = entry.unwrap(); + + if !entry.file_type().unwrap().is_file() { + continue; + } + + let name = entry.file_name().to_string_lossy().into_owned(); + + if name == "index" { + continue; + } + + out.insert(name, std::fs::read(entry.path()).unwrap()); + } + + out +} + +fn rebuild(node: &Node, extra: &[&str]) -> std::process::Output { + let mut command = std::process::Command::new(env!("CARGO_BIN_EXE_dolos")); + + command + .arg("--config") + .arg(node.config_path()) + .args(["doctor", "rebuild-state"]) + .args(extra); + + command.output().unwrap() +} + +/// The rebuild-equality claim itself, in place: wipe nothing by hand, let the +/// command run its own sequence, and require the state back byte for byte — +/// with the archive segments untouched and the WAL reseeded at the cursor. +#[test] +fn an_in_place_rebuild_reproduces_the_synced_state() { + let node = Node::new(); + node.sync(); + + let before = InstanceContents::read(&node); + assert!(before.state.cursor.is_some(), "fixture did not sync"); + + let stdout = assert_ok(&rebuild(&node, &["--force"])); + assert!(stdout.contains("rebuilt in place"), "{stdout}"); + + let after = InstanceContents::read(&node); + + assert_eq!(after.state, before.state); + assert_eq!(after.segments, before.segments); + assert_eq!(after.logs, before.logs); + assert_eq!(after.wal_tip, after.state.cursor, "WAL was not reseeded"); +} + +/// `--rewrite-logs` replays with derived-log writes passing through to the +/// live archive. The keys are slot-derived, so a faithful replay overwrites +/// every row with the value it already carries — and the blocks and segment +/// files stay exactly as they were. +#[test] +fn rewrite_logs_reproduces_the_log_rows_and_leaves_blocks_alone() { + let node = Node::new(); + node.sync(); + + let before = InstanceContents::read(&node); + + assert_ok(&rebuild(&node, &["--force", "--rewrite-logs"])); + + let after = InstanceContents::read(&node); + + assert_eq!(after.logs, before.logs); + assert_eq!(after.segments, before.segments); + assert_eq!(after.state, before.state); +} + +/// `--target` writes the rebuilt state somewhere else and `--ephemeral` +/// writes it nowhere; in both runs the instance's own stores stay untouched, +/// WAL included. +#[test] +fn target_and_ephemeral_leave_the_instance_untouched() { + let node = Node::new(); + node.sync(); + + let before = InstanceContents::read(&node); + + let target = node.root.path().join("rebuilt-state"); + assert_ok(&rebuild( + &node, + &["--target", &target.display().to_string()], + )); + + assert_ok(&rebuild(&node, &["--ephemeral"])); + + let after = InstanceContents::read(&node); + assert_eq!(after, before); + + // And the alternate-path rebuild is the same state the instance holds. + let rebuilt = dolos::storage::StateStoreBackend::open_fjall( + &target, + &dolos_core::config::FjallStateConfig::default(), + ) + .unwrap(); + + assert_eq!(StateContents::read(&rebuilt), before.state); +} + +/// The in-place wipe asks first. With no terminal attached there is nobody to +/// ask, so the command refuses before touching anything — `--force` is the +/// non-interactive spelling of yes. +#[test] +fn without_force_a_non_interactive_rebuild_refuses_before_touching_anything() { + let node = Node::new(); + node.sync(); + + let before = InstanceContents::read(&node); + + let output = rebuild(&node, &[]); + assert!(!output.status.success()); + + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("--force"), "{stderr}"); + + assert_eq!(InstanceContents::read(&node), before); +} + +/// `--stop-epoch` in place is refused: a partial in-place state cannot be +/// reconciled with the live stores. +#[test] +fn stop_epoch_requires_an_isolated_output() { + let node = Node::new(); + node.sync(); + + let output = rebuild(&node, &["--force", "--stop-epoch", "1"]); + assert!(!output.status.success()); + + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("--target"), "{stderr}"); +} From f8ca93287a638bfc69bfab6e93b29707a7ab8c26 Mon Sep 17 00:00:00 2001 From: Santiago Date: Tue, 18 Aug 2026 15:36:59 -0300 Subject: [PATCH 2/4] docs(doctor): note that --ephemeral holds the whole ledger in RAM An unbounded --ephemeral rebuild of a preprod instance exhausted a 16 GB workstation: the builtin memory state store keeps every entity and the full UTxO set uncompressed, which is a different order of magnitude from the same state in fjall's LSM tree on disk. Say so where an operator chooses the flag. Co-Authored-By: Claude Opus 5 (1M context) --- src/bin/dolos/doctor/rebuild_state.rs | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/src/bin/dolos/doctor/rebuild_state.rs b/src/bin/dolos/doctor/rebuild_state.rs index da212acdb..f5abd957c 100644 --- a/src/bin/dolos/doctor/rebuild_state.rs +++ b/src/bin/dolos/doctor/rebuild_state.rs @@ -13,6 +13,14 @@ //! somewhere else, instance untouched), and `--ephemeral` (an in-memory //! state store, discarded on exit — a pure validation run). //! +//! `--ephemeral` holds the whole ledger in memory: every entity and the +//! full UTxO set, uncompressed and unspilled. That is a different order of +//! magnitude from the same state on disk, where fjall keeps it compressed +//! in an LSM tree — a public-network state store that occupies a couple of +//! GB on disk does not fit in the RAM of an ordinary workstation. So the +//! mode is for bounded replays (`--stop-epoch`) and small chains; a +//! full-chain validation run on a public network wants `--target` instead. +//! //! The in-place sequence is ordered for crash safety: the WAL is reset to //! origin *before* the state store is wiped, and reseeded to the final //! cursor only after the replay completes. A crash mid-rebuild therefore @@ -45,7 +53,8 @@ pub struct Args { target: Option, /// Rebuild into an in-memory state store and discard it (a validation - /// run that writes nothing) + /// run that writes nothing). Holds the whole ledger in RAM — pair it + /// with --stop-epoch on a public network, or use --target instead #[arg(long)] ephemeral: bool, From 003810c4ec3ea11c4bcbe8a71b43baa03442e7ed Mon Sep 17 00:00:00 2001 From: Santiago Date: Tue, 18 Aug 2026 16:28:21 -0300 Subject: [PATCH 3/4] style: normalize comments across the PR #1222 diff Comment-only sweep to the TxPipe comment standard: removed 2 inline comments (8 lines) that restated policy already carried by docstrings (the module docstring's crash-safety ordering; the LogsOnly/logs_only docstrings at their call site), trimmed 0, kept the remaining comments as-is. Co-Authored-By: Claude Fable 5 --- src/bin/dolos/doctor/rebuild_state.rs | 8 -------- 1 file changed, 8 deletions(-) diff --git a/src/bin/dolos/doctor/rebuild_state.rs b/src/bin/dolos/doctor/rebuild_state.rs index f5abd957c..525575646 100644 --- a/src/bin/dolos/doctor/rebuild_state.rs +++ b/src/bin/dolos/doctor/rebuild_state.rs @@ -331,10 +331,6 @@ pub fn run(config: &RootConfig, args: &Args, feedback: &Feedback) -> miette::Res let tip = preflight(&archive)?; - // The domain's archive: writes are discarded, except under --rewrite-logs - // where derived-log writes pass through to the already-open store (redb - // will not open the same file twice, so the open handle is what gets - // wrapped). let domain_archive = if args.rewrite_logs { archive .logs_only() @@ -357,10 +353,6 @@ pub fn run(config: &RootConfig, args: &Args, feedback: &Feedback) -> miette::Res let wal = open_store("wal", crate::common::open_wal_store(config))?; - // Reset the WAL *before* the wipe: from here until the final - // reseed, a crash leaves WAL(origin) behind the state cursor, - // which the next startup refuses loudly instead of serving a - // half-rebuilt ledger. Re-running this command recovers. wal.reset_to(&ChainPoint::Origin) .into_diagnostic() .context("resetting the WAL to origin")?; From 9e55f5115c800a1aa336be7a28ef868879dc6d02 Mon Sep 17 00:00:00 2001 From: Santiago Date: Tue, 18 Aug 2026 17:25:02 -0300 Subject: [PATCH 4/4] fix(doctor): refuse a zero chunk before the wipe, and keep LogsOnly away from db_mut MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two review findings: - `--chunk 0` imports no blocks, so the replay ended immediately and the command failed on the missing cursor — in place, that is after the WAL reset and the state wipe, so a typo cost the operator their instance. Refuse it up front, with a test that asserts nothing was touched. - `logs_only()` clones the archive handle, so a LogsOnly value always aliases the Arc of the store it came from and can never satisfy `db_mut` (`Arc::get_mut(..).unwrap()`). Folding it into the Redb arm of prune-chain and export made a guaranteed panic look supported; those two now refuse it explicitly, and the aliasing is documented on logs_only. Also swap the archive-tip `expect` for a diagnostic, matching the rest of the command's operator-facing errors. Co-Authored-By: Claude Opus 5 (1M context) --- src/adapters/storage.rs | 6 ++++++ src/bin/dolos/data/export.rs | 7 +++---- src/bin/dolos/data/prune_chain.rs | 5 ++++- src/bin/dolos/doctor/rebuild_state.rs | 10 ++++++++-- tests/rebuild_state.rs | 19 +++++++++++++++++++ 5 files changed, 40 insertions(+), 7 deletions(-) diff --git a/src/adapters/storage.rs b/src/adapters/storage.rs index b7f32b6d4..b27115cba 100644 --- a/src/adapters/storage.rs +++ b/src/adapters/storage.rs @@ -668,6 +668,12 @@ impl ArchiveStoreBackend { /// Clones the handle out of the open store rather than opening the path /// again, because redb refuses to open the same file twice. Returns /// `None` when there is no redb store to wrap. + /// + /// The result therefore *aliases* the store it came from: both share one + /// `Arc`. Anything reaching for exclusive database access — + /// `db_mut`, and so redb compaction — must refuse a `LogsOnly` value + /// rather than treat it as a `Redb` one, because `Arc::get_mut` cannot + /// succeed while the original handle is alive. pub fn logs_only(&self) -> Option { match self { Self::Redb(s) | Self::LogsOnly(s) => Some(Self::LogsOnly(s.clone())), diff --git a/src/bin/dolos/data/export.rs b/src/bin/dolos/data/export.rs index ffa9885e0..59f16b6b0 100644 --- a/src/bin/dolos/data/export.rs +++ b/src/bin/dolos/data/export.rs @@ -160,11 +160,10 @@ pub fn run( // prepare_archive requires direct redb access match &mut stores.archive { - ArchiveStoreBackend::Redb(s) | ArchiveStoreBackend::LogsOnly(s) - if !args.skip_sanitization => - { - prepare_archive(s, &pb)? + ArchiveStoreBackend::LogsOnly(_) if !args.skip_sanitization => { + bail!("archive sanitization needs exclusive access to the archive database") } + ArchiveStoreBackend::Redb(s) if !args.skip_sanitization => prepare_archive(s, &pb)?, ArchiveStoreBackend::Redb(_) | ArchiveStoreBackend::LogsOnly(_) => {} ArchiveStoreBackend::NoOp(_) => { bail!("export command is not available for noop archive backend") diff --git a/src/bin/dolos/data/prune_chain.rs b/src/bin/dolos/data/prune_chain.rs index fc1676ef7..60c7cc0b9 100644 --- a/src/bin/dolos/data/prune_chain.rs +++ b/src/bin/dolos/data/prune_chain.rs @@ -38,7 +38,10 @@ pub fn run(config: &RootConfig, args: &Args) -> miette::Result<()> { // Compaction requires direct redb access match &mut stores.archive { - ArchiveStoreBackend::Redb(s) | ArchiveStoreBackend::LogsOnly(s) => { + ArchiveStoreBackend::LogsOnly(_) => { + bail!("chain compaction needs exclusive access to the archive database") + } + ArchiveStoreBackend::Redb(s) => { let db = s.db_mut(); while db.compact().into_diagnostic()? { diff --git a/src/bin/dolos/doctor/rebuild_state.rs b/src/bin/dolos/doctor/rebuild_state.rs index 525575646..e4213e0d1 100644 --- a/src/bin/dolos/doctor/rebuild_state.rs +++ b/src/bin/dolos/doctor/rebuild_state.rs @@ -208,11 +208,13 @@ fn preflight(archive: &ArchiveStoreBackend) -> miette::Result { ); } - let (tip_slot, tip_body) = archive + let Some((tip_slot, tip_body)) = archive .get_tip() .into_diagnostic() .context("reading archive tip")? - .expect("archive with a first block has a tip"); + else { + bail!("the archive reported a first block but no tip; it changed under this command"); + }; let tip = MultiEraBlock::decode(&tip_body) .into_diagnostic() @@ -327,6 +329,10 @@ pub fn run(config: &RootConfig, args: &Args, feedback: &Feedback) -> miette::Res ); } + if args.chunk == 0 { + bail!("--chunk must be at least 1"); + } + let archive = open_store("archive", crate::common::open_archive_store(config))?; let tip = preflight(&archive)?; diff --git a/tests/rebuild_state.rs b/tests/rebuild_state.rs index 2c09a83bb..d37de393b 100644 --- a/tests/rebuild_state.rs +++ b/tests/rebuild_state.rs @@ -224,6 +224,25 @@ fn without_force_a_non_interactive_rebuild_refuses_before_touching_anything() { assert_eq!(InstanceContents::read(&node), before); } +/// A zero chunk imports no blocks, so the replay would end instantly. In +/// place that lands *after* the WAL reset and the wipe, so the refusal has to +/// come before either — a typo must not cost the operator their instance. +#[test] +fn a_zero_chunk_is_refused_before_touching_anything() { + let node = Node::new(); + node.sync(); + + let before = InstanceContents::read(&node); + + let output = rebuild(&node, &["--force", "--chunk", "0"]); + assert!(!output.status.success()); + + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("--chunk"), "{stderr}"); + + assert_eq!(InstanceContents::read(&node), before); +} + /// `--stop-epoch` in place is refused: a partial in-place state cannot be /// reconciled with the live stores. #[test]