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()?;
@@ -138,34 +139,127 @@ fn check_archive_in_sync_with_state<D: Domain>(domain: &D) -> Result<(), DomainE
     Ok(())
 }
 
-/// Catch up archive and index stores by replaying WAL entries.
+/// 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");
+    }
+
+    // Post-condition: the replay must actually reach the target. A WAL that
+    // claims a tip the state can't reach (e.g. entries wiped by `reset_to`)
+    // is unrecoverable here — fail loudly instead of leaving a silent gap.
+    // Compared by slot: a Slot-only cursor at the target's slot (post-ESTART)
+    // is already at the target even though the points differ.
+    let cursor = domain.state().read_cursor()?;
+
+    let reached = cursor
+        .as_ref()
+        .is_some_and(|cursor| cursor.slot() >= target.slot());
+
+    if !reached {
+        error!(?cursor, %target, "state catch-up could not reach the WAL target");
+        return Err(DomainError::InconsistentState {
+            wal: Some(target.clone()),
+            state: cursor,
+        });
+    }
 
     Ok(())
 }
 
 /// 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 +270,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 +281,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 +295,27 @@ fn catch_up_archive<D: Domain>(domain: &D, state_cursor: &ChainPoint) -> Result<
         info!(count, "archive caught up from WAL");
     }
 
+    // Archive can legitimately remain behind when the WAL no longer holds the
+    // missing blocks (e.g. after `reset_to`). Flag it instead of failing —
+    // archive completeness is not consensus-critical, matching the lenient
+    // handling in `check_archive_in_sync_with_state`.
+    let archive_tip = domain.archive().get_tip()?.map(|(slot, _)| slot);
+
+    if archive_tip < Some(target_slot) {
+        error!(
+            ?archive_tip,
+            target_slot, "archive still behind WAL target after catch-up"
+        );
+    }
+
     Ok(())
 }
 
 /// 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 +327,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 +338,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,6 +354,15 @@ fn catch_up_indexes<D: Domain>(domain: &D, state_cursor: &ChainPoint) -> Result<
         info!(count, "indexes caught up from WAL");
     }
 
+    // Same lenient handling as archive: flag a residual lag instead of
+    // failing the boot.
+    let index_cursor = domain.indexes().cursor()?;
+    let index_slot = index_cursor.as_ref().map(|p| p.slot());
+
+    if index_slot < Some(target.slot()) {
+        error!(?index_cursor, %target, "indexes still behind WAL target after catch-up");
+    }
+
     Ok(())
 }
 

crates/core/src/state.rs

@@ -424,3 +424,91 @@ pub fn load_entity_chunk<D: Domain>(
 
     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<D: Domain>(
+    entities: &mut EntityMap<D::Entity>,
+    store: &D::State,
+    deltas: &[D::EntityDelta],
+) -> Result<(), StateError> {
+    let missing: Vec<NsKey> = deltas
+        .iter()
+        .map(|delta| delta.key())
+        .filter(|key| !entities.contains_key(key))
+        .collect();
+
+    if missing.is_empty() {
+        return Ok(());
+    }
+
+    let loaded = load_entity_chunk::<D>(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<D: Domain>(
+    entities: &mut EntityMap<D::Entity>,
+    store: &D::State,
+    deltas: &mut [D::EntityDelta],
+) -> Result<(), StateError> {
+    load_missing_entities::<D>(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<D: Domain>(
+    entities: &mut EntityMap<D::Entity>,
+    store: &D::State,
+    deltas: &[D::EntityDelta],
+) -> Result<(), StateError> {
+    load_missing_entities::<D>(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<D: Domain>(
+    writer: &<D::State as StateStore>::Writer,
+    entities: &EntityMap<D::Entity>,
+) -> Result<(), StateError> {
+    for (key, entity) in entities.iter() {
+        let NsKey(ns, key) = key;
+        writer.save_entity_typed(ns, key, entity.as_ref())?;
+    }
+
+    Ok(())
+}