diff --git a/crates/cardano/src/genesis/mod.rs b/crates/cardano/src/genesis/mod.rs index 65467d75e..0f93715ca 100644 --- a/crates/cardano/src/genesis/mod.rs +++ b/crates/cardano/src/genesis/mod.rs @@ -167,5 +167,14 @@ pub fn execute( staking::bootstrap::(state, genesis)?; + // Mark genesis state as applied by setting the cursor to Origin. This + // lets crash recovery distinguish "genesis done, no blocks yet" (cursor + // at Origin) from "genesis never ran" (no cursor), so a restart doesn't + // re-run genesis and reset a WAL that already holds blocks. Set last so + // a crash mid-genesis leaves no cursor and genesis re-runs cleanly. + let writer = state.start_writer()?; + writer.set_cursor(ChainPoint::Origin)?; + writer.commit()?; + Ok(()) } diff --git a/crates/core/src/bootstrap.rs b/crates/core/src/bootstrap.rs index 7f038ec17..bb04f5a5b 100644 --- a/crates/core/src/bootstrap.rs +++ b/crates/core/src/bootstrap.rs @@ -8,7 +8,7 @@ use tracing::{error, info, warn}; use crate::{ sync::drain_pending_work, ArchiveStore, ArchiveWriter as _, ChainLogic, ChainPoint, Domain, - DomainError, IndexStore, IndexWriter as _, StateStore, WalStore, + DomainError, EntityMap, IndexStore, IndexWriter as _, StateStore, StateWriter as _, WalStore, }; /// Extension trait for domain bootstrapping operations. @@ -54,9 +54,10 @@ impl BootstrapExt for D { /// Check that WAL is consistent with the state store. /// -/// WAL at or ahead of state is normal — the existing `catch_up_stores` handles -/// replaying WAL entries to bring other stores up. State ahead of WAL is an -/// error that requires explicit repair via `dolos doctor reset-wal`. +/// WAL at or ahead of state is normal — `catch_up_stores` replays WAL entries +/// to bring every other store (state included) up to the WAL tip. State ahead +/// of WAL is an error that requires explicit repair via `dolos doctor +/// reset-wal`. fn check_wal_in_sync_with_state(domain: &D) -> Result<(), DomainError> { let wal = domain.wal().find_tip()?.map(|(point, _)| point); let state = domain.state().read_cursor()?; @@ -95,8 +96,13 @@ fn check_wal_in_sync_with_state(domain: &D) -> Result<(), DomainError }); } (Some(ref wal_tip), None) if wal_tip == &ChainPoint::Origin => { - // WAL at Origin with no state cursor — fresh node after genesis. - info!(%wal_tip, "WAL at origin, no state cursor (post-genesis)"); + // WAL reset to Origin but no state cursor — a crash mid-genesis + // (genesis `commit_wal` resets the WAL to Origin before + // `commit_state` writes the cursor). A completed genesis leaves + // the cursor at Origin, so it takes the `(Some, Some)` arm above; + // this arm only fires when genesis was interrupted. Genesis + // re-runs cleanly on the next block. + info!(%wal_tip, "WAL at origin, no state cursor (interrupted genesis)"); } (Some(ref wal_tip), None) => { error!(%wal_tip, "WAL exists but no state found"); @@ -138,34 +144,161 @@ fn check_archive_in_sync_with_state(domain: &D) -> Result<(), DomainE Ok(()) } -/// Catch up archive and index stores by replaying WAL entries. +/// How to react when a store fails to reach the catch-up target. +enum CatchUpSeverity { + /// Consensus-critical (state): a residual gap is unrecoverable, so abort + /// the boot with `InconsistentState`. + Fatal, + /// Not consensus-critical (archive, indexes): a residual gap is logged + /// but the node still boots, matching `check_archive_in_sync_with_state`. + Lenient, +} + +/// Verify a store reached the catch-up `target` after replay. +/// +/// `reached` is the store's position after catch-up. Comparison is by slot: a +/// `Slot`-only cursor at the target's slot (e.g. post-ESTART) counts as +/// reached even though the points differ. A store can fall short when the WAL +/// no longer holds the entries needed to replay (e.g. after `reset_to`); +/// `severity` decides whether that aborts the boot or is only flagged. +fn verify_caught_up( + store: &str, + reached: Option, + target: &ChainPoint, + severity: CatchUpSeverity, +) -> Result<(), DomainError> { + if reached + .as_ref() + .is_some_and(|point| point.slot() >= target.slot()) + { + return Ok(()); + } + + match severity { + CatchUpSeverity::Fatal => { + error!(?reached, %target, store, "catch-up could not reach the WAL target"); + Err(DomainError::InconsistentState { + wal: Some(target.clone()), + state: reached, + }) + } + CatchUpSeverity::Lenient => { + error!(?reached, %target, store, "store still behind WAL target after catch-up"); + Ok(()) + } + } +} + +/// Catch up state, archive and index stores by replaying WAL entries. /// -/// After `check_wal_in_sync_with_state`, the WAL is at or ahead of the state. -/// If archive or index stores are behind (e.g., crash between state commit -/// and archive/index commit), this function replays the missing WAL entries -/// to bring them back in sync. +/// The WAL commits first in the work-unit lifecycle, so after a crash it is +/// the most advanced store. Every other store reconciles forward to the WAL +/// tip: state first (covering a crash between `commit_wal` and +/// `commit_state`), then archive and indexes (covering a crash between the +/// state commit and the archive/index commits). fn catch_up_stores(domain: &D) -> Result<(), DomainError> { - let state_cursor = match domain.state().read_cursor()? { + let target = match domain.wal().find_tip()? { // nothing to catch up None => return Ok(()), - // Origin means no blocks have been processed yet — archive and indexes - // are correctly empty, so there is nothing to replay. - Some(ChainPoint::Origin) => return Ok(()), - Some(cursor) => cursor, + // Origin means no blocks have been processed yet — state, archive and + // indexes are correctly empty, so there is nothing to replay. + Some((ChainPoint::Origin, _)) => return Ok(()), + Some((point, _)) => point, }; - catch_up_archive(domain, &state_cursor)?; - catch_up_indexes(domain, &state_cursor)?; + catch_up_state(domain, &target)?; + catch_up_archive(domain, &target)?; + catch_up_indexes(domain, &target)?; Ok(()) } +/// Catch up the state store by replaying WAL entries. +/// +/// A crash between `commit_wal` and `commit_state` leaves the WAL holding +/// blocks whose effects never reached the state store. Only roll work units +/// write WAL entries, and each entry fully captures its state mutation +/// (entity deltas + block + resolved inputs), so forward-replaying them here +/// is lossless. Boundary work units never write the WAL, so they can't leave +/// the WAL ahead of state; recovering a crash *during* a boundary is a +/// separate concern (#1018). +fn catch_up_state(domain: &D, target: &ChainPoint) -> Result<(), DomainError> { + let state_cursor = domain.state().read_cursor()?; + + if state_cursor.as_ref() == Some(target) { + return Ok(()); + } + + // Origin (or no cursor) means nothing has been applied yet — replay the + // whole WAL. + let state_slot = match &state_cursor { + None | Some(ChainPoint::Origin) => None, + Some(point) => Some(point.slot()), + }; + + // Find the WAL start point from the state cursor + let start = match state_slot { + Some(slot) => domain.wal().locate_point(slot)?, + None => None, + }; + + let logs = domain.wal().iter_logs(start, Some(target.clone()))?; + + let mut count = 0u64; + + for (point, mut log) in logs { + // Skip entries at or before the current state cursor + if Some(point.slot()) <= state_slot { + continue; + } + + // Skip synthetic entries (from reset_to) — they carry no effects + if log.block.is_empty() { + continue; + } + + let writer = domain.state().start_writer()?; + + // Forward mirror of the rollback loop in `sync.rs`: load each entity + // at its pre-block value, apply the deltas in block order, persist. + let mut entities = EntityMap::default(); + crate::state::apply_delta_chunk::(&mut entities, domain.state(), &mut log.delta)?; + crate::state::save_entities::(&writer, &entities)?; + + let catchup = D::Chain::compute_catchup(&log.block, &log.inputs, point.clone())?; + + writer.apply_utxoset(&catchup.utxo_delta)?; + + writer.set_cursor(point.clone())?; + + // Commit per entry so a later block touching the same entity reloads + // the value this block just wrote. + writer.commit()?; + + count += 1; + } + + if count > 0 { + info!(count, "state caught up from WAL"); + } + + // A WAL that claims a tip the state can't reach (e.g. entries wiped by + // `reset_to`) is unrecoverable — fail loudly instead of leaving a silent + // gap the upstream intersection would then skip past. + verify_caught_up( + "state", + domain.state().read_cursor()?, + target, + CatchUpSeverity::Fatal, + ) +} + /// Catch up archive store by replaying WAL blocks. -fn catch_up_archive(domain: &D, state_cursor: &ChainPoint) -> Result<(), DomainError> { +fn catch_up_archive(domain: &D, target: &ChainPoint) -> Result<(), DomainError> { let archive_tip = domain.archive().get_tip()?.map(|(slot, _)| slot); - let state_slot = state_cursor.slot(); + let target_slot = target.slot(); - if archive_tip == Some(state_slot) { + if archive_tip == Some(target_slot) { return Ok(()); } @@ -176,9 +309,7 @@ fn catch_up_archive(domain: &D, state_cursor: &ChainPoint) -> Result< None => None, }; - let blocks = domain - .wal() - .iter_blocks(start, Some(state_cursor.clone()))?; + let blocks = domain.wal().iter_blocks(start, Some(target.clone()))?; let writer = domain.archive().start_writer()?; let mut count = 0u64; @@ -189,6 +320,11 @@ fn catch_up_archive(domain: &D, state_cursor: &ChainPoint) -> Result< continue; } + // Skip synthetic entries (from reset_to) — they carry no block + if block.is_empty() { + continue; + } + writer.apply(&point, &block)?; count += 1; } @@ -198,14 +334,16 @@ fn catch_up_archive(domain: &D, state_cursor: &ChainPoint) -> Result< info!(count, "archive caught up from WAL"); } - Ok(()) + let archive_tip = domain.archive().get_tip()?.map(|(slot, _)| ChainPoint::Slot(slot)); + + verify_caught_up("archive", archive_tip, target, CatchUpSeverity::Lenient) } /// Catch up index store by replaying WAL log entries. -fn catch_up_indexes(domain: &D, state_cursor: &ChainPoint) -> Result<(), DomainError> { +fn catch_up_indexes(domain: &D, target: &ChainPoint) -> Result<(), DomainError> { let index_cursor = domain.indexes().cursor()?; - if index_cursor.as_ref() == Some(state_cursor) { + if index_cursor.as_ref() == Some(target) { return Ok(()); } @@ -217,7 +355,7 @@ fn catch_up_indexes(domain: &D, state_cursor: &ChainPoint) -> Result< None => None, }; - let logs = domain.wal().iter_logs(start, Some(state_cursor.clone()))?; + let logs = domain.wal().iter_logs(start, Some(target.clone()))?; let writer = domain.indexes().start_writer()?; let mut count = 0u64; @@ -228,6 +366,11 @@ fn catch_up_indexes(domain: &D, state_cursor: &ChainPoint) -> Result< continue; } + // Skip synthetic entries (from reset_to) — they carry no effects + if log.block.is_empty() { + continue; + } + let catchup = D::Chain::compute_catchup(&log.block, &log.inputs, point)?; writer.apply(&catchup.index_delta)?; @@ -239,7 +382,12 @@ fn catch_up_indexes(domain: &D, state_cursor: &ChainPoint) -> Result< info!(count, "indexes caught up from WAL"); } - Ok(()) + verify_caught_up( + "indexes", + domain.indexes().cursor()?, + target, + CatchUpSeverity::Lenient, + ) } #[cfg(test)] diff --git a/crates/core/src/state.rs b/crates/core/src/state.rs index a3979683a..207490da0 100644 --- a/crates/core/src/state.rs +++ b/crates/core/src/state.rs @@ -424,3 +424,91 @@ pub fn load_entity_chunk( Ok(loaded) } + +/// Load into `entities` any entity referenced by `deltas` that isn't already +/// tracked. Entities already in the map keep their in-memory value, so the +/// map doubles as a read-your-own-writes cache for callers that replay +/// multiple delta chunks before committing. +fn load_missing_entities( + entities: &mut EntityMap, + store: &D::State, + deltas: &[D::EntityDelta], +) -> Result<(), StateError> { + let missing: Vec = deltas + .iter() + .map(|delta| delta.key()) + .filter(|key| !entities.contains_key(key)) + .collect(); + + if missing.is_empty() { + return Ok(()); + } + + let loaded = load_entity_chunk::(missing.as_slice(), store)?; + entities.extend(loaded); + + Ok(()) +} + +/// Replay a chunk of deltas forward over the tracked entities, loading any +/// entity not already in the map from the store. +/// +/// This is the same delta application that runs in-memory during normal +/// sync before `commit_state` persists the results; it's shared here so +/// crash-recovery WAL replay (bootstrap catch-up) uses the exact same logic. +pub fn apply_delta_chunk( + entities: &mut EntityMap, + store: &D::State, + deltas: &mut [D::EntityDelta], +) -> Result<(), StateError> { + load_missing_entities::(entities, store, deltas)?; + + for delta in deltas.iter_mut() { + let entity = entities + .get_mut(&delta.key()) + .expect("entity loaded by load_missing_entities"); + + delta.apply(entity); + } + + Ok(()) +} + +/// Counterpart of [`apply_delta_chunk`]: undo a chunk of deltas over the +/// tracked entities. +/// +/// Deltas are undone in reverse application order. Each delta's `prev_*` +/// captures the state immediately before its own apply, so multiple deltas +/// keyed to the same entity must be reversed last-first to correctly walk +/// back through the apply chain. +pub fn undo_delta_chunk( + entities: &mut EntityMap, + store: &D::State, + deltas: &[D::EntityDelta], +) -> Result<(), StateError> { + load_missing_entities::(entities, store, deltas)?; + + for delta in deltas.iter().rev() { + let entity = entities + .get_mut(&delta.key()) + .expect("entity loaded by load_missing_entities"); + + delta.undo(entity); + } + + Ok(()) +} + +/// Persist every tracked entity through the writer: `Some` upserts the +/// record, `None` deletes it. +pub fn save_entities( + writer: &::Writer, + entities: &EntityMap, +) -> Result<(), StateError> { + for (key, entity) in entities.iter() { + let NsKey(ns, key) = key; + writer.save_entity_typed(ns, key, entity.as_ref())?; + } + + Ok(()) +} diff --git a/crates/core/src/sync.rs b/crates/core/src/sync.rs index 28c55dde5..780dc675e 100644 --- a/crates/core/src/sync.rs +++ b/crates/core/src/sync.rs @@ -13,7 +13,7 @@ use std::sync::Arc; use tracing::{debug, info, instrument, warn}; use crate::{ - ArchiveStore as _, BlockSlot, ChainLogic, ChainPoint, Domain, DomainError, EntityDelta as _, + ArchiveStore as _, BlockSlot, ChainLogic, ChainPoint, Domain, DomainError, EntityMap, IndexStore as _, IndexWriter as _, MempoolStore, RawBlock, StateStore, StateWriter as _, TipEvent, WalStore, WorkUnit, }; @@ -70,7 +70,12 @@ impl SyncExt for D { let writer = self.state().start_writer()?; let index_writer = self.indexes().start_writer()?; - for (point, mut log) in undo_blocks.rev() { + // Entities accumulate across all undone entries so consecutive blocks + // touching the same entity unwind from the in-memory value instead of + // re-reading the not-yet-committed store. + let mut entities: EntityMap = EntityMap::default(); + + for (point, log) in undo_blocks.rev() { if point == *to { // Final cursor update - build an empty delta with just the cursor let empty_delta = crate::IndexDelta { @@ -82,26 +87,7 @@ impl SyncExt for D { break; } - let entities = log - .delta - .iter() - .map(|delta| delta.key()) - .collect::>(); - - let mut entities = - crate::state::load_entity_chunk::(entities.as_slice(), self.state())?; - - for (key, entity) in entities.iter_mut() { - // Undo deltas in reverse application order. Each delta's `prev_*` - // captures the state immediately before its own apply, so multiple - // deltas keyed to the same entity must be reversed last-first to - // correctly walk back through the apply chain. - for delta in log.delta.iter_mut().rev() { - if delta.key() == *key { - delta.undo(entity); - } - } - } + crate::state::undo_delta_chunk::(&mut entities, self.state(), &log.delta)?; let block = Arc::new(log.block); @@ -130,6 +116,9 @@ impl SyncExt for D { info!(%point, "block undone"); } + // Persist the undone entities (`Some` upserts, `None` deletes). + crate::state::save_entities::(&writer, &entities)?; + writer.commit()?; index_writer.commit()?; diff --git a/tests/bootstrap.rs b/tests/bootstrap.rs index d67e0085f..1764b51de 100644 --- a/tests/bootstrap.rs +++ b/tests/bootstrap.rs @@ -4,36 +4,54 @@ //! lifecycle with partial commits (WAL + state only), then verifies that //! `bootstrap()` recovers archive and index stores from WAL replay. +use std::str::FromStr as _; use std::sync::Arc; use dolos_core::{ sync::SyncExt as _, BootstrapExt, ChainLogic, ChainPoint, Domain, IndexStore, StateStore, - StateWriter, WalStore, WorkUnit, + StateWriter, TxoRef, WalStore, WorkUnit, }; use dolos_testing::{ synthetic::{build_synthetic_blocks, SyntheticBlockConfig}, toy_domain::ToyDomain, }; -/// Helper: feed blocks into a domain with partial work-unit execution. +/// Which commit phases to run when feeding blocks, simulating a crash at +/// each inter-store boundary of the work-unit lifecycle +/// (commit_wal → commit_state → commit_archive → commit_indexes). +/// `commit_wal` always runs; commit_indexes and finalize never do. /// -/// Runs load → compute → commit_wal → commit_state but **skips** -/// commit_archive and commit_indexes, simulating a crash between -/// the state commit and the archive/index commits. -fn feed_blocks_partial(domain: &ToyDomain, blocks: &[dolos_core::RawBlock]) { +/// Out of scope here: crashes *inside* a phase (mid-shard) and crashes +/// during epoch-boundary work units (RUPD/EWRAP/ESTART), which don't write +/// the WAL — see #1018. +#[derive(Clone, Copy)] +enum CrashAfter { + /// Run commit_wal only — models a crash between `commit_wal` and + /// `commit_state`. + Wal, + /// Run commit_wal + commit_state — models a crash between the state + /// commit and the archive commit. + State, + /// Run commit_wal + commit_state + commit_archive — models a crash + /// between the archive commit and the index commit. + Archive, +} + +/// Helper: feed blocks into a domain with partial work-unit execution. +fn feed_blocks_partial(domain: &ToyDomain, blocks: &[dolos_core::RawBlock], crash_after: CrashAfter) { let mut chain = domain.write_chain(); for block in blocks { if !chain.can_receive_block() { - drain_partial(&mut chain, domain); + drain_partial(&mut chain, domain, crash_after); } chain.receive_block(block.clone()).unwrap(); } - drain_partial(&mut chain, domain); + drain_partial(&mut chain, domain, crash_after); } -fn drain_partial(chain: &mut dolos_cardano::CardanoLogic, domain: &ToyDomain) { +fn drain_partial(chain: &mut dolos_cardano::CardanoLogic, domain: &ToyDomain, crash_after: CrashAfter) { while let Some(mut work) = ::pop_work::(chain, domain) { @@ -44,10 +62,15 @@ fn drain_partial(chain: &mut dolos_cardano::CardanoLogic, domain: &ToyDomain) { WorkUnit::::load(&mut work, domain, shard).unwrap(); WorkUnit::::compute(&mut work, shard).unwrap(); WorkUnit::::commit_wal(&mut work, domain, shard).unwrap(); - WorkUnit::::commit_state(&mut work, domain, shard).unwrap(); - // Intentionally skip commit_archive and commit_indexes — and - // intentionally skip finalize() to model "crash after state - // commit", which is what the recovery test below exercises. + if matches!(crash_after, CrashAfter::State | CrashAfter::Archive) { + WorkUnit::::commit_state(&mut work, domain, shard).unwrap(); + } + if matches!(crash_after, CrashAfter::Archive) { + WorkUnit::::commit_archive(&mut work, domain, shard).unwrap(); + } + // Intentionally skip commit_indexes — and intentionally skip + // finalize() to model a crash mid-lifecycle, which is what the + // recovery tests below exercise. } } } @@ -67,7 +90,7 @@ fn test_catchup_recovers_archive_and_indexes() { let baseline_index = domain.indexes().cursor().unwrap(); // Feed synthetic blocks with partial execution (skip archive + indexes). - feed_blocks_partial(&domain, &blocks); + feed_blocks_partial(&domain, &blocks, CrashAfter::State); // State should have advanced. let state_cursor = domain.state().read_cursor().unwrap().unwrap(); @@ -114,6 +137,171 @@ fn test_catchup_recovers_archive_and_indexes() { ); } +/// A crash between `commit_wal` and `commit_state` leaves the WAL ahead of +/// every other store. Bootstrap must replay the WAL entries into state (and +/// then archive/indexes) instead of leaving state behind — otherwise the +/// upstream intersection resumes from the WAL tip and the skipped blocks' +/// effects are silently lost. +#[test] +fn test_catchup_recovers_state_from_wal() { + let cfg = SyntheticBlockConfig::default(); + let (blocks, vectors, cardano_config) = build_synthetic_blocks(cfg); + + let genesis = Arc::new(dolos_cardano::include::devnet::load()); + let domain = ToyDomain::new_with_genesis_and_config(genesis, cardano_config, None, None); + + let baseline_state = domain.state().read_cursor().unwrap(); + + // Feed blocks committing the WAL only. + feed_blocks_partial(&domain, &blocks, CrashAfter::Wal); + + // WAL advanced; state stayed behind. + let (wal_tip, _) = domain.wal().find_tip().unwrap().unwrap(); + let state_cursor = domain.state().read_cursor().unwrap(); + assert_eq!(state_cursor, baseline_state, "state should not have advanced"); + assert_ne!( + Some(&wal_tip), + baseline_state.as_ref(), + "WAL should have advanced" + ); + + // --- Run bootstrap (which calls catch_up_stores internally) --- + domain.bootstrap().unwrap(); + + // Every store must converge to the WAL tip. + let state_cursor_after = domain.state().read_cursor().unwrap(); + assert_eq!( + state_cursor_after.as_ref(), + Some(&wal_tip), + "state cursor should be at the WAL tip after catch-up" + ); + + let archive_tip_after = domain.archive().get_tip().unwrap().map(|(s, _)| s); + assert_eq!( + archive_tip_after, + Some(wal_tip.slot()), + "archive tip should be at the WAL tip after catch-up" + ); + + let index_cursor_after = domain.indexes().cursor().unwrap(); + assert_eq!( + index_cursor_after.as_ref(), + Some(&wal_tip), + "index cursor should be at the WAL tip after catch-up" + ); + + // The replayed blocks' UTxO effects must be visible in state. Use the + // last tx of the last block — nothing after it can consume its output. + let last_tx = vectors.blocks.last().unwrap().tx_hashes.last().unwrap(); + let txo = TxoRef::from_str(&format!("{last_tx}#0")).unwrap(); + let utxos = domain.state().get_utxos(vec![txo]).unwrap(); + assert_eq!( + utxos.len(), + 1, + "utxo produced by replayed block should be queryable from state" + ); +} + +/// Crash-recovery matrix: state, archive and indexes each at a different +/// point behind the WAL tip. Bootstrap must converge all of them to the +/// WAL tip. +#[test] +fn test_catchup_converges_all_stores_to_wal_tip() { + let cfg = SyntheticBlockConfig::default(); + let (blocks, _vectors, cardano_config) = build_synthetic_blocks(cfg); + assert!( + blocks.len() >= 2, + "synthetic config must produce at least 2 blocks to stagger the stores" + ); + + let genesis = Arc::new(dolos_cardano::include::devnet::load()); + let domain = ToyDomain::new_with_genesis_and_config(genesis, cardano_config, None, None); + + // First batch: WAL + state commit (archive/index stay at baseline). + feed_blocks_partial(&domain, &blocks[..1], CrashAfter::State); + + // Second batch: WAL only (state stays at the first batch). + feed_blocks_partial(&domain, &blocks[1..], CrashAfter::Wal); + + let (wal_tip, _) = domain.wal().find_tip().unwrap().unwrap(); + let state_mid = domain.state().read_cursor().unwrap(); + assert_ne!( + state_mid.as_ref(), + Some(&wal_tip), + "state should be behind the WAL tip" + ); + + domain.bootstrap().unwrap(); + + assert_eq!( + domain.state().read_cursor().unwrap().as_ref(), + Some(&wal_tip), + "state cursor should be at the WAL tip after catch-up" + ); + assert_eq!( + domain.archive().get_tip().unwrap().map(|(s, _)| s), + Some(wal_tip.slot()), + "archive tip should be at the WAL tip after catch-up" + ); + assert_eq!( + domain.indexes().cursor().unwrap().as_ref(), + Some(&wal_tip), + "index cursor should be at the WAL tip after catch-up" + ); +} + +/// Crash between `commit_archive` and `commit_indexes`: WAL, state and +/// archive are all at the tip, only indexes lag. Bootstrap must catch +/// indexes up while leaving the already-current stores untouched. +#[test] +fn test_catchup_recovers_indexes_when_archive_ahead() { + let cfg = SyntheticBlockConfig::default(); + let (blocks, vectors, cardano_config) = build_synthetic_blocks(cfg); + + let genesis = Arc::new(dolos_cardano::include::devnet::load()); + let domain = ToyDomain::new_with_genesis_and_config(genesis, cardano_config, None, None); + + let baseline_index = domain.indexes().cursor().unwrap(); + + // Feed blocks committing everything except indexes. + feed_blocks_partial(&domain, &blocks, CrashAfter::Archive); + + let (wal_tip, _) = domain.wal().find_tip().unwrap().unwrap(); + assert_eq!( + domain.state().read_cursor().unwrap().as_ref(), + Some(&wal_tip), + "state should be at the WAL tip" + ); + assert_eq!( + domain.archive().get_tip().unwrap().map(|(s, _)| s), + Some(wal_tip.slot()), + "archive should be at the WAL tip" + ); + assert_eq!( + domain.indexes().cursor().unwrap(), + baseline_index, + "indexes should not have advanced" + ); + + domain.bootstrap().unwrap(); + + assert_eq!( + domain.indexes().cursor().unwrap().as_ref(), + Some(&wal_tip), + "index cursor should be at the WAL tip after catch-up" + ); + + // Verify index content came through the replay. + let tx_hash_hex = &vectors.blocks[0].tx_hashes[0]; + let tx_hash_bytes = hex::decode(tx_hash_hex).unwrap(); + let slot = domain.indexes().slot_by_tx_hash(&tx_hash_bytes).unwrap(); + assert!( + slot.is_some(), + "tx hash {} should be found in index after catch-up", + tx_hash_hex + ); +} + /// Regression: rolling back through WAL entries that came out of the full sync /// lifecycle must not panic. /// @@ -130,6 +318,11 @@ fn test_catchup_recovers_archive_and_indexes() { /// prior point. With the lifecycle correctly ordered, the WAL rows carry their /// `prev_*` data, undo executes cleanly, and the cursor lands on the rollback /// target. +/// +/// It also verifies entity *persistence*: the accounts namespace must be +/// byte-identical to its snapshot at the rollback target. Before the fix, +/// rollback undid entities in memory but never saved them, leaving entity +/// state reflecting the undone blocks. #[test] fn test_rollback_after_full_sync_lifecycle() { let cfg = SyntheticBlockConfig::default(); @@ -151,10 +344,15 @@ fn test_rollback_after_full_sync_lifecycle() { ChainPoint::Specific(block.slot(), block.hash()) }; - // Feed every block through the live sync lifecycle. `roll_forward` uses + // Feed blocks through the live sync lifecycle. `roll_forward` uses // `run_lifecycle` with `include_wal=true`, so this exercises the same path - // as the live sync pipeline. - for block in &blocks { + // as the live sync pipeline. Feed the first block alone so we can capture + // the entity state that rollback is expected to restore. + domain.roll_forward(blocks[0].clone()).unwrap(); + + let accounts_at_target = snapshot_namespace(&domain, "accounts"); + + for block in &blocks[1..] { domain.roll_forward(block.clone()).unwrap(); } @@ -165,6 +363,14 @@ fn test_rollback_after_full_sync_lifecycle() { "tip should be past the rollback target", ); + // Guard: the blocks past the target must actually touch account state, + // otherwise the restoration assertion below is vacuous. + assert_ne!( + snapshot_namespace(&domain, "accounts"), + accounts_at_target, + "blocks past the rollback target should modify account entities", + ); + // Roll back. Without the fix, this panics inside delta.undo() because the // WAL-deserialized deltas have prev_*=None. domain.rollback(&rollback_target).unwrap(); @@ -175,6 +381,27 @@ fn test_rollback_after_full_sync_lifecycle() { Some(&rollback_target), "state cursor should be at the rollback target after rollback", ); + + // Undone entities must be persisted, restoring the exact state at the + // rollback target. + assert_eq!( + snapshot_namespace(&domain, "accounts"), + accounts_at_target, + "account entities should be restored to their state at the rollback target", + ); +} + +/// Collect all raw (key, value) pairs in a state namespace. +fn snapshot_namespace( + domain: &ToyDomain, + ns: dolos_core::Namespace, +) -> Vec<(dolos_core::EntityKey, dolos_core::EntityValue)> { + domain + .state() + .iter_entities(ns, dolos_core::EntityKey::full_range()) + .unwrap() + .map(|x| x.unwrap()) + .collect() } #[test]