feat(doctor): rebuild-state — regenerate the state store from the local archive

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,25 @@ 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.
+    ///
+    /// The result therefore *aliases* the store it came from: both share one
+    /// `Arc<Database>`. 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<Self> {
+        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<Self, ArchiveError> {
         Ok(Self::Redb(dolos_redb3::archive::ArchiveStore::in_memory(
@@ -678,7 +707,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,13 +717,16 @@ impl ArchiveStoreBackend {
 
 pub enum ArchiveWriterBackend {
     Redb(Box<<dolos_redb3::archive::ArchiveStore as CoreArchiveStore>::Writer>),
+    /// Delegates `write_log` and `commit`; discards `apply` and `undo`.
+    LogsOnly(Box<<dolos_redb3::archive::ArchiveStore as CoreArchiveStore>::Writer>),
     NoOp(NoOpArchiveWriter),
 }
 
 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,21 +738,22 @@ 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),
         }
     }
 
     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 +839,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 +851,7 @@ impl CoreArchiveStore for ArchiveStoreBackend {
         keys: &[&LogKey],
     ) -> Result<Vec<Option<EntityValue>>, 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 +862,7 @@ impl CoreArchiveStore for ArchiveStoreBackend {
         range: Range<LogKey>,
     ) -> Result<Self::LogIter, ArchiveError> {
         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 +872,7 @@ impl CoreArchiveStore for ArchiveStoreBackend {
 
     fn get_block_by_slot(&self, slot: &BlockSlot) -> Result<Option<BlockBody>, 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 +883,7 @@ impl CoreArchiveStore for ArchiveStoreBackend {
         to: Option<BlockSlot>,
     ) -> Result<Self::BlockIter<'a>, 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 +893,30 @@ impl CoreArchiveStore for ArchiveStoreBackend {
 
     fn find_intersect(&self, intersect: &[ChainPoint]) -> Result<Option<ChainPoint>, 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<Option<(BlockSlot, BlockBody)>, 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<u64>) -> Result<bool, ArchiveError> {
         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),
         }
     }

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")
         }

src/bin/dolos/data/export.rs

@@ -160,8 +160,11 @@ pub fn run(
 
     // prepare_archive requires direct redb access
     match &mut stores.archive {
+        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::Redb(_) | ArchiveStoreBackend::LogsOnly(_) => {}
         ArchiveStoreBackend::NoOp(_) => {
             bail!("export command is not available for noop archive backend")
         }

src/bin/dolos/data/prune_chain.rs

@@ -38,6 +38,9 @@ pub fn run(config: &RootConfig, args: &Args) -> miette::Result<()> {
 
     // Compaction requires direct redb access
     match &mut stores.archive {
+        ArchiveStoreBackend::LogsOnly(_) => {
+            bail!("chain compaction needs exclusive access to the archive database")
+        }
         ArchiveStoreBackend::Redb(s) => {
             let db = s.db_mut();
 

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)?,