test(capture): specify coherent snapshot view ownership

Cargo.lock

@@ -9808,6 +9808,7 @@ dependencies = [
  "redb",
  "serde",
  "serde_json",
+ "sha2 0.11.0",
  "tempfile",
  "thiserror 2.0.21",
  "tokio",

crates/capture-authority/Cargo.toml

@@ -10,6 +10,7 @@ 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 }

crates/capture-authority/src/lib.rs

@@ -9,8 +9,11 @@ use std::{
 mod application;
 mod committed;
 mod created;
+mod native_snapshot;
 mod node;
+pub use native_snapshot::{NativeSnapshotError, NativeSnapshotView};
 mod raft_types;
+pub mod snapshot;
 pub use created::{
     CaptureBinding, CreatedApplyResult, CreatedEvent, CreatedOperation, CreatedResult, CreatedVersion, DecideCreated,
     MAX_OBJECT_KEY_LENGTH, ObjectIdentity, PreparedIdentity,
@@ -37,6 +40,84 @@ impl NativeLogReader {
     }
 }
 
+/// 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 async fn snapshot_view(&self) -> Result<NativeSnapshotView, NativeSnapshotError> {
+        Err(NativeSnapshotError::Unavailable)
+    }
+
+    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)]
@@ -485,6 +566,7 @@ fn decode_log(b: &[u8]) -> Result<LogEntry, StoreError> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use redb::ReadableTableMetadata;
     use std::{
         io::{BufRead, BufReader, Read, Write},
         process::{Command, Stdio},
@@ -530,6 +612,147 @@ mod tests {
         }
     }
 
+    #[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;
@@ -736,6 +959,205 @@ mod tests {
             membership,
         )
     }
+    fn u24_limits() -> snapshot::Limits {
+        snapshot::Limits {
+            total_bytes: 1_000_000,
+            membership_bytes: 100_000,
+            key_bytes: 100_000,
+            value_bytes: 100_000,
+            records: 1000,
+        }
+    }
+    fn u24_apply(store: &mut Store, entries: &[CaptureRaftEntry]) {
+        store.append_native(entries).expect("append snapshot fixture");
+        store
+            .save_committed(entries.last().expect("nonempty fixture").log_id.into())
+            .expect("commit snapshot fixture");
+        for entry in entries {
+            store.apply_next_native(entry.log_id.into()).expect("apply snapshot fixture");
+        }
+    }
+    fn u24_expected(store: &Store) -> Vec<u8> {
+        let tx = store.database.begin_read().expect("fixture read transaction");
+        let (applied, membership) = application::native_application_state(&tx).expect("fixture application head");
+        let header = snapshot::Header {
+            cluster_id: identity().cluster_id,
+            applied,
+            membership,
+            binding: u17a_binding(),
+        };
+        let definitions = [
+            application::HEADS,
+            application::VERSIONS,
+            application::EVENTS,
+            application::OPERATIONS,
+        ];
+        let mut counts = [0; 4];
+        for (i, definition) in definitions.iter().enumerate() {
+            counts[i] = tx
+                .open_table(*definition)
+                .expect("fixture table")
+                .len()
+                .expect("fixture count");
+        }
+        let mut encoder = snapshot::Encoder::new(Vec::new(), &header, counts, u24_limits()).expect("fixture encoder");
+        for (definition, section) in definitions.into_iter().zip([
+            snapshot::Section::Heads,
+            snapshot::Section::Versions,
+            snapshot::Section::Events,
+            snapshot::Section::Operations,
+        ]) {
+            let table = tx.open_table(definition).expect("fixture rows");
+            for row in table.iter().expect("fixture iterator") {
+                let (key, value) = row.expect("fixture row");
+                encoder
+                    .row(snapshot::Row {
+                        section,
+                        key: key.value(),
+                        value: value.value(),
+                    })
+                    .expect("fixture encode row");
+            }
+        }
+        encoder.finish().expect("fixture complete")
+    }
+    #[test]
+    fn u24_coherent_view() {
+        let (_dir, _backend, mut store) = u17a_store();
+        u24_apply(&mut store, &[u17a_blank(0), u17a_membership(1).0, u17a_created(2, u17a_command())]);
+        let expected_a = u24_expected(&store);
+        let writer = NativeLogWriter::new(store);
+        let runtime = u19_runtime();
+        let result = runtime.block_on(writer.snapshot_view());
+        assert!(result.is_ok(), "U24_COHERENT_VIEW: acquisition failed");
+        let view = result.expect("A view");
+        let mut command = u17a_command();
+        command.operation_id = [8; 16];
+        command.object.key = b"second-object".to_vec();
+        command.prepared.version_id = [9; 16];
+        let expected_b = {
+            let mut store = writer.store.lock().expect("B store");
+            u24_apply(&mut store, &[u17a_membership(3).0, u17a_created(4, command)]);
+            u24_expected(&store)
+        };
+        assert_eq!(view.stream(Vec::new(), u24_limits()).expect("A stream"), expected_a, "U24_COHERENT_VIEW");
+        let new_view = runtime.block_on(writer.snapshot_view()).expect("B view");
+        assert_eq!(new_view.stream(Vec::new(), u24_limits()).expect("B stream"), expected_b);
+        assert_ne!(expected_a, expected_b, "B changes all tables and application membership cut");
+    }
+    #[test]
+    fn u24_stream_releases_writer() {
+        struct BlockingOutput {
+            entered: Option<mpsc::Sender<()>>,
+            release: mpsc::Receiver<()>,
+        }
+        impl Write for BlockingOutput {
+            fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
+                if let Some(entered) = self.entered.take() {
+                    entered.send(()).map_err(io::Error::other)?;
+                    self.release.recv_timeout(Duration::from_secs(10)).map_err(io::Error::other)?;
+                }
+                Ok(bytes.len())
+            }
+            fn flush(&mut self) -> io::Result<()> {
+                Ok(())
+            }
+        }
+        let (_dir, _backend, mut store) = u17a_store();
+        u24_apply(&mut store, &[u17a_blank(0)]);
+        let writer = NativeLogWriter::new(store);
+        let view = u19_runtime().block_on(writer.snapshot_view()).expect("stream view");
+        let (entered_tx, entered_rx) = mpsc::channel();
+        let (release_tx, release_rx) = mpsc::channel();
+        let stream = std::thread::spawn(move || {
+            view.stream(
+                BlockingOutput {
+                    entered: Some(entered_tx),
+                    release: release_rx,
+                },
+                u24_limits(),
+            )
+        });
+        let entered = entered_rx.recv_timeout(Duration::from_secs(5));
+        let (done_tx, done_rx) = mpsc::channel();
+        let write = std::thread::spawn(move || {
+            let result = u19_runtime().block_on(writer.write(|store| {
+                u24_apply(store, &[u17a_created(1, u17a_command())]);
+                Ok(())
+            }));
+            let _ = done_tx.send(result.is_ok());
+        });
+        let done = done_rx.recv_timeout(Duration::from_secs(5));
+        let _ = release_tx.send(());
+        let streamed = stream.join().expect("stream joined");
+        write.join().expect("writer joined");
+        assert!(entered.is_ok(), "output entered rendezvous");
+        assert!(matches!(done, Ok(true)), "U24_STREAM_RELEASES_WRITER");
+        assert!(streamed.is_ok(), "stream completed");
+    }
+    #[test]
+    fn u24_view_schema_fence() {
+        let (_dir, _backend, store) = u17a_bound();
+        let writer = NativeLogWriter::new(store);
+        assert!(
+            matches!(
+                u19_runtime().block_on(writer.snapshot_view()),
+                Err(NativeSnapshotError::Store(StoreError::UnsupportedSchema))
+            ),
+            "U24_SCHEMA_FENCE"
+        );
+        assert_eq!(committed_schema(&writer.store.lock().expect("schema unchanged")), 3);
+        let (_dir2, _backend2, mut store) = u17a_store();
+        store.writable = false;
+        let writer = NativeLogWriter::new(store);
+        assert!(
+            matches!(
+                u19_runtime().block_on(writer.snapshot_view()),
+                Err(NativeSnapshotError::Store(StoreError::WriteDisabled))
+            ),
+            "U24_WRITE_FENCE"
+        );
+    }
+    #[test]
+    fn u24_view_lifetime_errors() {
+        struct Broken;
+        impl Write for Broken {
+            fn write(&mut self, _: &[u8]) -> io::Result<usize> {
+                Err(io::Error::new(io::ErrorKind::BrokenPipe, "snapshot test output"))
+            }
+            fn flush(&mut self) -> io::Result<()> {
+                Ok(())
+            }
+        }
+        let (_dir, _backend, mut store) = u17a_store();
+        u24_apply(&mut store, &[u17a_blank(0)]);
+        let expected = u24_expected(&store);
+        let writer = NativeLogWriter::new(store);
+        let view = u19_runtime().block_on(writer.snapshot_view()).expect("owned view");
+        drop(writer);
+        assert_eq!(view.stream(Vec::new(), u24_limits()).expect("view outlives writer"), expected);
+        let (_dir2, _backend2, store) = u17a_store();
+        let writer = NativeLogWriter::new(store);
+        let view = u19_runtime().block_on(writer.snapshot_view()).expect("error view");
+        let result = view.stream(Broken, u24_limits());
+        match result {
+            Err(NativeSnapshotError::Snapshot(snapshot::SnapshotError::Io(error))) => {
+                assert_eq!(error.kind(), io::ErrorKind::BrokenPipe)
+            }
+            _ => panic!("U24_TYPED_IO_ERROR"),
+        }
+        u19_runtime()
+            .block_on(writer.write(|store| {
+                u24_apply(store, &[u17a_blank(0)]);
+                Ok(())
+            }))
+            .expect("write after failed stream");
+        u19_runtime()
+            .block_on(writer.snapshot_view())
+            .expect("view after failed stream")
+            .stream(Vec::new(), u24_limits())
+            .expect("stream after failed output");
+    }
     #[test]
     fn u18_native_read_roundtrip() {
         let (_dir, _backend, mut store) = u17a_store();
@@ -1573,6 +1995,7 @@ mod tests {
     }
     #[derive(Debug)]
     struct Media {
+        sync_threads: Vec<std::thread::ThreadId>,
         volatile: Vec<u8>,
         dead: bool,
         syncs: usize,
@@ -1594,6 +2017,7 @@ mod tests {
             Self {
                 path: path.to_owned(),
                 media: Arc::new(Mutex::new(Media {
+                    sync_threads: Vec::new(),
                     volatile: bytes,
                     dead: false,
                     syncs: 0,
@@ -1671,6 +2095,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()

crates/capture-authority/src/native_snapshot.rs

@@ -0,0 +1,26 @@
+//! Owned consistent native application snapshot view.
+use crate::{StoreError, snapshot};
+use std::io::Write;
+
+#[derive(Debug, thiserror::Error)]
+pub enum NativeSnapshotError {
+    #[error(transparent)]
+    Store(#[from] StoreError),
+    #[error(transparent)]
+    Snapshot(#[from] snapshot::SnapshotError),
+    #[error("native snapshot Store lock poisoned")]
+    Poisoned,
+    #[error("native snapshot worker failed")]
+    Worker(#[source] tokio::task::JoinError),
+    #[error("native snapshot view unavailable")]
+    Unavailable,
+}
+
+pub struct NativeSnapshotView {
+    _private: (),
+}
+impl NativeSnapshotView {
+    pub fn stream<W: Write>(self, _writer: W, _limits: snapshot::Limits) -> Result<W, NativeSnapshotError> {
+        Err(NativeSnapshotError::Unavailable)
+    }
+}