fix(bootstrap): recover state from WAL when WAL is ahead after crash

crates/cardano/src/genesis/mod.rs

@@ -167,5 +167,14 @@ pub fn execute<D: Domain>(
 
     staking::bootstrap::<D>(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(())
 }

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<D: Domain> 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<D: Domain>(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<D: Domain>(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<D: Domain>(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<ChainPoint>,
+    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<D: Domain>(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<D: Domain>(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::<D>(&mut entities, domain.state(), &mut log.delta)?;
+        crate::state::save_entities::<D>(&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<D: Domain>(domain: &D, state_cursor: &ChainPoint) -> Result<(), DomainError> {
+fn catch_up_archive<D: Domain>(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<D: Domain>(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<D: Domain>(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<D: Domain>(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<D: Domain>(domain: &D, state_cursor: &ChainPoint) -> Result<(), DomainError> {
+fn catch_up_indexes<D: Domain>(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<D: Domain>(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<D: Domain>(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<D: Domain>(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)]