feat(capture): stream snapshot rows with a permanent failure fence

Cargo.lock

@@ -9808,8 +9808,10 @@ dependencies = [
  "redb",
  "serde",
  "serde_json",
+ "sha2 0.11.0",
  "tempfile",
  "thiserror 2.0.21",
+ "tokio",
  "url",
 ]
 

crates/capture-authority/Cargo.toml

@@ -10,8 +10,10 @@ description = "Durable local storage for capture authority"
 [dependencies]
 openraft = { workspace = true }
 redb = { workspace = true }
+sha2 = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 thiserror = { workspace = true }
+tokio = { workspace = true }
 url = { workspace = true }
 
 [dev-dependencies]

crates/capture-authority/src/lib.rs

@@ -11,6 +11,7 @@ mod committed;
 mod created;
 mod node;
 mod raft_types;
+pub mod snapshot;
 pub use created::{
     CaptureBinding, CreatedApplyResult, CreatedEvent, CreatedOperation, CreatedResult, CreatedVersion, DecideCreated,
     MAX_OBJECT_KEY_LENGTH, ObjectIdentity, PreparedIdentity,
@@ -23,6 +24,135 @@ pub use raft_types::{
 mod application_tests;
 mod truncate;
 
+/// OpenRaft log reader backed by exclusive local authority storage.
+#[derive(Clone)]
+pub struct NativeLogReader {
+    store: std::sync::Arc<std::sync::Mutex<Store>>,
+}
+
+impl NativeLogReader {
+    pub fn new(store: Store) -> Self {
+        Self {
+            store: std::sync::Arc::new(std::sync::Mutex::new(store)),
+        }
+    }
+}
+
+/// Async native writes and readers sharing one durable Store.
+#[derive(Clone)]
+pub struct NativeLogWriter {
+    store: std::sync::Arc<std::sync::Mutex<Store>>,
+    ordering: std::sync::Arc<tokio::sync::Mutex<()>>,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum NativeWriteError {
+    #[error(transparent)]
+    Store(#[from] StoreError),
+    #[error("native log writer lock poisoned")]
+    Poisoned,
+    #[error("native log writer worker failed")]
+    Worker(#[source] tokio::task::JoinError),
+}
+
+impl NativeLogWriter {
+    pub fn new(store: Store) -> Self {
+        Self {
+            store: std::sync::Arc::new(std::sync::Mutex::new(store)),
+            ordering: std::sync::Arc::new(tokio::sync::Mutex::new(())),
+        }
+    }
+
+    pub fn reader(&self) -> NativeLogReader {
+        NativeLogReader {
+            store: self.store.clone(),
+        }
+    }
+
+    async fn write<F>(&self, write: F) -> Result<(), NativeWriteError>
+    where
+        F: FnOnce(&mut Store) -> Result<(), StoreError> + Send + 'static,
+    {
+        // Lock order is ordering then Store. The worker owns the ordering guard
+        // until commit completes, even when its awaiting caller is cancelled.
+        let ordering = self.ordering.clone().lock_owned().await;
+        let store = self.store.clone();
+        tokio::task::spawn_blocking(move || {
+            let _ordering = ordering;
+            let mut store = store.lock().map_err(|_| NativeWriteError::Poisoned)?;
+            write(&mut store).map_err(NativeWriteError::Store)
+        })
+        .await
+        .map_err(NativeWriteError::Worker)?
+    }
+
+    pub async fn save_vote(&self, vote: openraft::Vote<u64>) -> Result<(), NativeWriteError> {
+        self.write(move |store| store.save_vote(vote.into())).await
+    }
+
+    pub async fn read_vote(&self) -> Result<Option<openraft::Vote<u64>>, NativeWriteError> {
+        let store = self.store.clone();
+        tokio::task::spawn_blocking(move || {
+            let store = store.lock().map_err(|_| NativeWriteError::Poisoned)?;
+            store
+                .read_vote()
+                .map(|vote| vote.map(Into::into))
+                .map_err(NativeWriteError::Store)
+        })
+        .await
+        .map_err(NativeWriteError::Worker)?
+    }
+
+    pub async fn append_native(&self, entries: Vec<CaptureRaftEntry>) -> Result<(), NativeWriteError> {
+        self.write(move |store| store.append_native(&entries)).await
+    }
+
+    pub async fn truncate_suffix(&self, from: u64) -> Result<(), NativeWriteError> {
+        self.write(move |store| store.truncate_suffix(from)).await
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+enum NativeReaderError {
+    #[error(transparent)]
+    Store(#[from] StoreError),
+    #[error("native log reader lock poisoned")]
+    Poisoned,
+    #[error("native log reader worker failed")]
+    Worker,
+}
+
+fn native_read_error(error: NativeReaderError) -> openraft::StorageError<u64> {
+    openraft::StorageError::IO {
+        source: openraft::StorageIOError::read_logs(openraft::AnyError::new(&error)),
+    }
+}
+
+async fn native_read_worker<F>(
+    store: std::sync::Arc<std::sync::Mutex<Store>>,
+    read: F,
+) -> Result<Vec<CaptureRaftEntry>, openraft::StorageError<u64>>
+where
+    F: FnOnce(&Store) -> Result<Vec<CaptureRaftEntry>, StoreError> + Send + 'static,
+{
+    tokio::task::spawn_blocking(move || {
+        let store = store.lock().map_err(|_| native_read_error(NativeReaderError::Poisoned))?;
+        read(&store).map_err(|error| native_read_error(NativeReaderError::Store(error)))
+    })
+    .await
+    .map_err(|_| native_read_error(NativeReaderError::Worker))?
+}
+
+impl openraft::RaftLogReader<CaptureTypeConfig> for NativeLogReader {
+    async fn try_get_log_entries<RB: std::ops::RangeBounds<u64> + Clone + std::fmt::Debug + openraft::OptionalSend>(
+        &mut self,
+        range: RB,
+    ) -> Result<Vec<CaptureRaftEntry>, openraft::StorageError<u64>> {
+        let bounds = (range.start_bound().cloned(), range.end_bound().cloned());
+        native_read_worker(self.store.clone(), move |store| store.read_native_range(bounds)).await
+    }
+}
+
 const META: TableDefinition<&str, &[u8]> = TableDefinition::new("meta");
 const VOTE: TableDefinition<u8, &[u8]> = TableDefinition::new("vote");
 const LOGS: TableDefinition<u64, &[u8]> = TableDefinition::new("logs");
@@ -459,6 +589,294 @@ mod tests {
             node_id: 0,
         }
     }
+    fn u19_runtime() -> tokio::runtime::Runtime {
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("reader runtime")
+    }
+
+    fn u19_assert_io(error: openraft::StorageError<u64>) {
+        match error {
+            openraft::StorageError::IO { source } => {
+                assert!(source.to_string().starts_with("when Read Logs:"), "U19_READER_ERROR: {source}");
+            }
+            other => panic!("U19_READER_ERROR: unexpected error {other:?}"),
+        }
+    }
+
+    #[test]
+    fn u20_vote_durable_roundtrip() {
+        let (_dir, backend, store) = u17a_store();
+        let writer = NativeLogWriter::new(store);
+        let expected = openraft::Vote::from(vote());
+        u19_runtime().block_on(async {
+            writer.save_vote(expected).await.expect("U20_VOTE_DURABLE save");
+            assert_eq!(writer.read_vote().await.expect("read vote"), Some(expected), "U20_VOTE_DURABLE");
+        });
+        drop(writer);
+        backend.cut();
+        let reopened = Store::from_database(CrashBackend::new(&backend.path).database(), identity()).expect("reopen vote");
+        assert_eq!(reopened.read_vote().expect("durable vote"), Some(vote()), "U20_VOTE_DURABLE");
+    }
+
+    #[test]
+    fn u20_native_append_reader_visibility() {
+        use openraft::RaftLogReader;
+        let (_dir, backend, store) = u17a_store();
+        let writer = NativeLogWriter::new(store);
+        let entries = vec![u17a_blank(0), u17a_membership(1).0, u17a_created(2, u17a_command())];
+        let mut reader = writer.reader();
+        u19_runtime().block_on(async {
+            assert!(writer.append_native(entries.clone()).await.is_ok(), "U20_WRITE_SUCCESS");
+            assert_eq!(reader.try_get_log_entries(..).await.expect("shared reader"), entries);
+        });
+        drop(reader);
+        drop(writer);
+        backend.cut();
+        let reopened = Store::from_database(CrashBackend::new(&backend.path).database(), identity()).expect("reopen entries");
+        assert_eq!(reopened.read_native_range(..).expect("durable entries"), entries);
+    }
+
+    #[test]
+    fn u20_native_truncate_replacement() {
+        use openraft::RaftLogReader;
+        let (_dir, backend, mut store) = u17a_store();
+        let original = vec![u17a_blank(0), u17a_blank(1), u17a_blank(2)];
+        store.append_native(&original).expect("committed fixture entries");
+        store.save_committed(u17a_id(0)).expect("committed prefix");
+        let writer = NativeLogWriter::new(store);
+        let replacement = u17a_created(1, u17a_command());
+        let expected = vec![original[0].clone(), replacement.clone()];
+        u19_runtime().block_on(async {
+            assert!(matches!(
+                writer.truncate_suffix(0).await,
+                Err(NativeWriteError::Store(StoreError::CommittedLogProtected))
+            ));
+            writer.truncate_suffix(1).await.expect("truncate suffix");
+            writer.append_native(vec![replacement]).await.expect("replace suffix");
+            assert_eq!(writer.reader().try_get_log_entries(..).await.expect("replacement reader"), expected);
+        });
+        drop(writer);
+        backend.cut();
+        let reopened = Store::from_database(CrashBackend::new(&backend.path).database(), identity()).expect("reopen replacement");
+        assert_eq!(reopened.read_native_range(..).expect("durable replacement"), expected);
+    }
+
+    #[test]
+    fn u20_worker_thread() {
+        let (_dir, backend, store) = u17a_store();
+        let writer = NativeLogWriter::new(store);
+        backend.media.lock().expect("media").sync_threads.clear();
+        let executor = std::thread::current().id();
+        u19_runtime().block_on(writer.save_vote(vote().into())).expect("worker vote");
+        let media = backend.media.lock().expect("media");
+        assert!(!media.sync_threads.is_empty(), "U20_WORKER_THREAD no real sync");
+        assert!(media.sync_threads.iter().all(|thread| *thread != executor), "U20_WORKER_THREAD");
+    }
+
+    #[test]
+    fn u20_cancelled_write_stays_serialized() {
+        let (_dir, backend, store) = u17a_store();
+        let writer = NativeLogWriter::new(store);
+        let (entered, resume) = backend.pause_sync(1, false);
+        u19_runtime().block_on(async {
+            let first_writer = writer.clone();
+            let first = tokio::spawn(async move { first_writer.save_vote(vote().into()).await });
+            tokio::task::spawn_blocking(move || entered.wait())
+                .await
+                .expect("first sync entered");
+            first.abort();
+            assert!(first.await.expect_err("cancel first await").is_cancelled());
+            let held = writer.ordering.try_lock().is_err();
+            let second_writer = writer.clone();
+            let second = tokio::spawn(async move { second_writer.append_native(vec![u17a_blank(0)]).await });
+            tokio::task::yield_now().await;
+            let waiting = !second.is_finished();
+            tokio::task::spawn_blocking(move || resume.wait())
+                .await
+                .expect("release sync");
+            second.await.expect("second task").expect("second commit");
+            assert!(held && waiting, "U20_WRITE_ORDER cancelled caller released ordering");
+            assert_eq!(writer.read_vote().await.expect("first committed vote"), Some(vote().into()));
+        });
+    }
+
+    #[test]
+    fn u20_failed_commit_disables_mutations() {
+        let (_dir, backend, store) = u17a_store();
+        let writer = NativeLogWriter::new(store);
+        backend.arm(1, false, false);
+        u19_runtime().block_on(async {
+            assert!(
+                matches!(
+                    writer.save_vote(vote().into()).await,
+                    Err(NativeWriteError::Store(StoreError::CommitIndeterminate(_)))
+                ),
+                "U20_WRITE_FAILURE"
+            );
+            assert!(matches!(
+                writer.save_vote(vote().into()).await,
+                Err(NativeWriteError::Store(StoreError::WriteDisabled))
+            ));
+            assert!(matches!(
+                writer.append_native(vec![u17a_blank(0)]).await,
+                Err(NativeWriteError::Store(StoreError::WriteDisabled))
+            ));
+            assert!(matches!(
+                writer.truncate_suffix(0).await,
+                Err(NativeWriteError::Store(StoreError::WriteDisabled))
+            ));
+        });
+    }
+
+    #[test]
+    fn u20_worker_poison_errors() {
+        let (_dir, _backend, store) = u17a_store();
+        let writer = NativeLogWriter::new(store);
+        u19_runtime().block_on(async {
+            assert!(matches!(
+                writer.write(|_| panic!("injected worker panic")).await,
+                Err(NativeWriteError::Worker(_))
+            ));
+            assert!(matches!(writer.save_vote(vote().into()).await, Err(NativeWriteError::Poisoned)));
+            assert!(matches!(writer.append_native(vec![]).await, Err(NativeWriteError::Poisoned)));
+            assert!(matches!(writer.truncate_suffix(0).await, Err(NativeWriteError::Poisoned)));
+            assert!(matches!(writer.read_vote().await, Err(NativeWriteError::Poisoned)));
+        });
+    }
+
+    #[test]
+    fn u19_reader_roundtrip() {
+        use openraft::RaftLogReader;
+        let (_dir, _backend, mut store) = u17a_store();
+        let entries = vec![u17a_blank(0), u17a_membership(1).0, u17a_created(2, u17a_command())];
+        store.append_native(&entries).expect("append reader fixture");
+        let mut reader = NativeLogReader::new(store);
+        let read = u19_runtime().block_on(reader.try_get_log_entries(..)).expect("trait read");
+        assert_eq!(read, entries, "U19_TRAIT_READ_SUCCESS");
+    }
+
+    #[test]
+    fn u19_reader_bounds_reopen() {
+        use openraft::RaftLogReader;
+        use std::ops::Bound::{Excluded, Included, Unbounded};
+        let (dir, backend, mut store) = u17a_store();
+        let entries = vec![u17a_blank(0), u17a_membership(1).0, u17a_created(2, u17a_command())];
+        store.append_native(&entries).expect("append range fixture");
+        drop(store);
+        drop(backend);
+        let store = Store::open(dir.path().join("db"), identity()).expect("reopen reader file");
+        let mut reader = NativeLogReader::new(store);
+        u19_runtime().block_on(async {
+            assert_eq!(
+                reader.try_get_log_entries(1..2).await.expect("bounded"),
+                entries[1..2],
+                "U19_READER_RANGE"
+            );
+            assert_eq!(reader.try_get_log_entries(..=1).await.expect("inclusive"), entries[..2]);
+            assert_eq!(
+                reader
+                    .try_get_log_entries((Excluded(0), Included(2)))
+                    .await
+                    .expect("excluded"),
+                entries[1..]
+            );
+            assert_eq!(reader.try_get_log_entries(2..).await.expect("unbounded end"), entries[2..]);
+            assert_eq!(
+                reader
+                    .try_get_log_entries((Unbounded, Included(u64::MAX)))
+                    .await
+                    .expect("max"),
+                entries
+            );
+            for range in [
+                (Included(2), Excluded(2)),
+                (Included(2), Included(1)),
+                (Excluded(u64::MAX), Unbounded),
+            ] {
+                assert!(reader.try_get_log_entries(range).await.expect("empty bounds").is_empty());
+            }
+            let start = 1;
+            let end = 2;
+            assert_eq!(
+                reader
+                    .try_get_log_entries((Included(&start), Excluded(&end)))
+                    .await
+                    .expect("borrowed bounds"),
+                entries[1..2]
+            );
+        });
+    }
+
+    #[test]
+    fn u19_reader_legacy_corrupt() {
+        use openraft::RaftLogReader;
+        let (_legacy_dir, _legacy_backend, legacy) = u17a_bound();
+        let mut legacy = NativeLogReader::new(legacy);
+        let runtime = u19_runtime();
+        u19_assert_io(
+            runtime
+                .block_on(legacy.try_get_log_entries(0..0))
+                .expect_err("U19_READER_ERROR: legacy empty range"),
+        );
+        let (_dir, _backend, mut store) = u17a_store();
+        store.append_native(&[u17a_blank(0)]).expect("append corruption fixture");
+        let tx = store.database.begin_write().expect("corruption transaction");
+        {
+            let mut logs = tx.open_table(LOGS).expect("logs");
+            logs.insert(0, b"corrupt".as_slice()).expect("corrupt row");
+        }
+        tx.commit().expect("persist corruption");
+        let mut reader = NativeLogReader::new(store);
+        u19_assert_io(
+            runtime
+                .block_on(reader.try_get_log_entries(..))
+                .expect_err("U19_READER_ERROR: corrupt row"),
+        );
+    }
+
+    #[test]
+    fn u19_reader_worker_thread() {
+        use openraft::RaftLogReader;
+        let (_dir, _backend, mut store) = u17a_store();
+        let entries = vec![u17a_blank(0)];
+        store.append_native(&entries).expect("append worker fixture");
+        let mut reader = NativeLogReader::new(store);
+        let runtime = u19_runtime();
+        let async_thread = std::thread::current().id();
+        let (sender, receiver) = mpsc::channel();
+        runtime.block_on(async {
+            let read = native_read_worker(reader.store.clone(), move |store| {
+                sender.send(std::thread::current().id()).expect("send worker thread");
+                store.read_native_range(..)
+            })
+            .await
+            .expect("worker read");
+            assert_eq!(read, entries);
+            assert_ne!(receiver.recv().expect("worker thread"), async_thread, "U19_BLOCKING_THREAD");
+            assert_eq!(reader.try_get_log_entries(..).await.expect("trait read"), entries);
+        });
+    }
+
+    #[test]
+    fn u19_reader_worker_errors() {
+        use openraft::RaftLogReader;
+        let (_dir, _backend, store) = u17a_store();
+        let mut reader = NativeLogReader::new(store);
+        let runtime = u19_runtime();
+        let error = runtime
+            .block_on(native_read_worker(reader.store.clone(), |_| panic!("test worker panic")))
+            .expect_err("worker panic must fail");
+        assert!(error.to_string().contains("native log reader worker failed"));
+        u19_assert_io(error);
+        let error = runtime
+            .block_on(reader.try_get_log_entries(..))
+            .expect_err("poison must fail");
+        assert!(error.to_string().contains("native log reader lock poisoned"));
+        u19_assert_io(error);
+    }
+
     fn u17a_id(index: u64) -> LogId {
         LogId {
             term: 11,
@@ -1371,6 +1789,7 @@ mod tests {
     }
     #[derive(Debug)]
     struct Media {
+        sync_threads: Vec<std::thread::ThreadId>,
         volatile: Vec<u8>,
         dead: bool,
         syncs: usize,
@@ -1392,6 +1811,7 @@ mod tests {
             Self {
                 path: path.to_owned(),
                 media: Arc::new(Mutex::new(Media {
+                    sync_threads: Vec::new(),
                     volatile: bytes,
                     dead: false,
                     syncs: 0,
@@ -1469,6 +1889,7 @@ mod tests {
                 return Err(io::Error::other("power lost"));
             }
             m.syncs += 1;
+            m.sync_threads.push(std::thread::current().id());
             let fault = m.fault.filter(|(n, _, _)| *n == m.syncs);
             let pause = if m.pause.as_ref().is_some_and(|(n, _, _, _)| *n == m.syncs) {
                 m.pause.take()