feat(capture): atomically migrate legacy Created history

crates/capture-authority/src/application.rs

@@ -98,7 +98,7 @@ fn decode_apply(b: &[u8]) -> Result<CreatedApplyResult, StoreError> {
     d.finish()?;
     Ok(r)
 }
-fn read_binding(meta: &impl ReadableTable<&'static str, &'static [u8]>) -> Result<CaptureBinding, StoreError> {
+pub(crate) fn read_binding(meta: &impl ReadableTable<&'static str, &'static [u8]>) -> Result<CaptureBinding, StoreError> {
     let b = meta.get("capture_binding")?.ok_or(StoreError::BindingNotPinned)?;
     let mut d = Decoder::new(b.value());
     let binding = d.binding()?;
@@ -298,11 +298,16 @@ fn entry_command(entry: &crate::LogEntry, native: bool) -> Result<DecideCreated,
 }
 
 pub(crate) fn native_application_state(tx: &ReadTransaction) -> Result<(Option<LogId>, crate::CaptureMembership), StoreError> {
-    let meta = tx.open_table(META)?;
+    native_application_metadata(&tx.open_table(META)?)
+}
+
+fn native_application_metadata(
+    meta: &impl ReadableTable<&'static str, &'static [u8]>,
+) -> Result<(Option<LogId>, crate::CaptureMembership), StoreError> {
     if number(required(meta.get("schema")?)?.value())? != 4 {
         return Err(StoreError::UnsupportedSchema);
     }
-    let last = read_applied(&meta)?;
+    let last = read_applied(meta)?;
     let membership = crate::CaptureMembership::decode(required(meta.get("native_membership")?)?.value())
         .map_err(|_| StoreError::CorruptRecord)?;
     if membership
@@ -315,6 +320,63 @@ pub(crate) fn native_application_state(tx: &ReadTransaction) -> Result<(Option<L
 }
 
 impl Store {
+    pub fn migrate_legacy_created_to_native(&mut self) -> Result<(), StoreError> {
+        let tx = self.write_transaction()?;
+        {
+            let mut meta = tx.open_table(META)?;
+            let schema = number(required(meta.get("schema")?)?.value())?;
+            if !matches!(schema, 3 | 4) {
+                return Err(StoreError::UnsupportedSchema);
+            }
+            let mut logs = tx.open_table(LOGS)?;
+            let heads = tx.open_table(HEADS)?;
+            let versions = tx.open_table(VERSIONS)?;
+            let events = tx.open_table(EVENTS)?;
+            let operations = tx.open_table(OPERATIONS)?;
+            let mut results = tx.open_table(APPLY_RESULTS)?;
+            validate_application_tables(&meta, &logs, &heads, &versions, &events, &operations, &results)?;
+            if schema == 4 {
+                return Ok(());
+            }
+            if meta.get("native_membership")?.is_some() {
+                return Err(StoreError::CorruptRecord);
+            }
+            // Validate the entire retained suffix before replacing any record.
+            let mut native_logs = Vec::new();
+            for row in logs.iter()? {
+                let (index, bytes) = row?;
+                let entry = decode_log(bytes.value())?;
+                let command = DecideCreated::decode(&entry.payload)?;
+                if entry.id.index != index.value() || command.encode()? != entry.payload {
+                    return Err(StoreError::CorruptRecord);
+                }
+                let native = crate::CaptureRaftEntry {
+                    log_id: entry.id.into(),
+                    payload: openraft::EntryPayload::Normal(crate::CaptureCommand::try_created(command)?),
+                };
+                native_logs.push((index.value(), encode_native_entry(&native)?));
+            }
+            let mut native_results = Vec::new();
+            for row in results.iter()? {
+                let (index, bytes) = row?;
+                let response = crate::CaptureResponse::created(decode_apply(bytes.value())?)?;
+                native_results.push((index.value(), encode_response(&response)?));
+            }
+            for (index, bytes) in native_logs {
+                logs.insert(index, bytes.as_slice())?;
+            }
+            for (index, bytes) in native_results {
+                results.insert(index, bytes.as_slice())?;
+            }
+            let membership = crate::CaptureMembership::new(None, vec![], BTreeMap::new())?;
+            meta.insert("native_membership", membership.encode()?.as_slice())?;
+            meta.insert("schema", 4u64.to_be_bytes().as_slice())?;
+            // Replay native decisions inside the same durable writer boundary.
+            validate_application_tables(&meta, &logs, &heads, &versions, &events, &operations, &results)?;
+        }
+        self.commit(tx)
+    }
+
     /// Explicitly opt an empty, bound store into native entry persistence.
     pub fn enable_native_empty(&mut self) -> Result<(), StoreError> {
         let tx = self.write_transaction()?;
@@ -626,23 +688,40 @@ impl Store {
 
 pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreError> {
     let meta = tx.open_table(META)?;
+    if !matches!(number(required(meta.get("schema")?)?.value())?, 3 | 4) {
+        return Ok(());
+    }
+    validate_application_tables(
+        &tx.open_table(META)?,
+        &tx.open_table(LOGS)?,
+        &tx.open_table(HEADS)?,
+        &tx.open_table(VERSIONS)?,
+        &tx.open_table(EVENTS)?,
+        &tx.open_table(OPERATIONS)?,
+        &tx.open_table(APPLY_RESULTS)?,
+    )
+}
+
+fn validate_application_tables(
+    meta: &impl ReadableTable<&'static str, &'static [u8]>,
+    logs: &impl ReadableTable<u64, &'static [u8]>,
+    heads: &impl ReadableTable<&'static [u8], &'static [u8]>,
+    versions: &impl ReadableTable<&'static [u8], &'static [u8]>,
+    events: &impl ReadableTable<&'static [u8], &'static [u8]>,
+    operations: &impl ReadableTable<&'static [u8], &'static [u8]>,
+    results: &impl ReadableTable<u64, &'static [u8]>,
+) -> Result<(), StoreError> {
     let schema = number(required(meta.get("schema")?)?.value())?;
     let native = schema == 4;
     if !matches!(schema, 3 | 4) {
         return Ok(());
     }
-    let binding = read_binding(&meta).map_err(|error| match error {
+    let binding = read_binding(meta).map_err(|error| match error {
         StoreError::BindingNotPinned => StoreError::CorruptRecord,
         other => other,
     })?;
-    let last = read_applied(&meta)?;
-    let logs = tx.open_table(LOGS)?;
-    let committed = validate_committed(&meta, &logs)?;
-    let heads = tx.open_table(HEADS)?;
-    let versions = tx.open_table(VERSIONS)?;
-    let events = tx.open_table(EVENTS)?;
-    let operations = tx.open_table(OPERATIONS)?;
-    let results = tx.open_table(APPLY_RESULTS)?;
+    let last = read_applied(meta)?;
+    let committed = validate_committed(meta, logs)?;
     if last.is_some_and(|last| committed.is_none_or(|c| last.index > c.index)) {
         return Err(StoreError::CorruptRecord);
     }
@@ -741,7 +820,7 @@ pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreErro
         return Err(StoreError::CorruptRecord);
     }
     if native {
-        if native_application_state(tx)?.1 != expected_membership
+        if native_application_metadata(meta)?.1 != expected_membership
             || heads.len()? != u64::try_from(replay_heads.len()).map_err(|_| StoreError::CorruptRecord)?
             || versions.len()? != u64::try_from(replay_versions.len()).map_err(|_| StoreError::CorruptRecord)?
         {

crates/capture-authority/src/lib.rs

@@ -9,7 +9,9 @@ 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::{
@@ -63,6 +65,13 @@ impl NativeLogWriter {
         }
     }
 
+    pub async fn snapshot_view(&self) -> Result<NativeSnapshotView, NativeSnapshotError> {
+        let store = self.store.clone();
+        tokio::task::spawn_blocking(move || NativeSnapshotView::acquire(store))
+            .await
+            .map_err(NativeSnapshotError::Worker)?
+    }
+
     pub fn reader(&self) -> NativeLogReader {
         NativeLogReader {
             store: self.store.clone(),
@@ -613,6 +622,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},
@@ -1226,6 +1236,294 @@ mod tests {
         u19_assert_io(error);
     }
 
+    fn u17b_snapshot(store: &Store) -> Vec<Vec<(Vec<u8>, Vec<u8>)>> {
+        u17b_snapshot_database(&store.database)
+    }
+    fn u17b_snapshot_database(database: &Database) -> Vec<Vec<(Vec<u8>, Vec<u8>)>> {
+        use crate::application::{APPLY_RESULTS, EVENTS, HEADS, OPERATIONS, VERSIONS};
+        let tx = database.begin_read().expect("snapshot transaction");
+        let meta = tx.open_table(META).expect("snapshot metadata");
+        let mut snapshot = vec![
+            meta.iter()
+                .expect("metadata iterator")
+                .map(|row| {
+                    let (key, value) = row.expect("metadata row");
+                    (key.value().as_bytes().to_vec(), value.value().to_vec())
+                })
+                .collect(),
+        ];
+        for table in [LOGS, APPLY_RESULTS] {
+            snapshot.push(
+                tx.open_table(table)
+                    .expect("snapshot indexed table")
+                    .iter()
+                    .expect("indexed iterator")
+                    .map(|row| {
+                        let (key, value) = row.expect("indexed row");
+                        (key.value().to_be_bytes().to_vec(), value.value().to_vec())
+                    })
+                    .collect(),
+            );
+        }
+        for table in [HEADS, VERSIONS, EVENTS, OPERATIONS] {
+            snapshot.push(
+                tx.open_table(table)
+                    .expect("snapshot business table")
+                    .iter()
+                    .expect("business iterator")
+                    .map(|row| {
+                        let (key, value) = row.expect("business row");
+                        (key.value().to_vec(), value.value().to_vec())
+                    })
+                    .collect(),
+            );
+        }
+        snapshot
+    }
+    fn u17b_history() -> (tempfile::TempDir, CrashBackend, Store, Vec<DecideCreated>) {
+        let (dir, backend, mut store) = u17a_bound();
+        let first = u17a_command();
+        let mut rejected = first.clone();
+        rejected.operation_id = [10; 16];
+        rejected.prepared.version_id = [11; 16];
+        let mut second = first.clone();
+        second.operation_id = [12; 16];
+        second.prepared.version_id = [13; 16];
+        second.expected_head = Some(first.prepared.version_id);
+        second.prepared.content_length = 8193;
+        let mut suffix = second.clone();
+        suffix.operation_id = [14; 16];
+        suffix.prepared.version_id = [15; 16];
+        suffix.expected_head = Some(second.prepared.version_id);
+        let commands = vec![first.clone(), first, rejected, second, suffix];
+        let entries = commands
+            .iter()
+            .enumerate()
+            .map(|(index, command)| LogEntry {
+                id: u17a_id(u64::try_from(index).expect("fixture index")),
+                payload: command.encode().expect("canonical legacy Created"),
+            })
+            .collect::<Vec<_>>();
+        store.append(&entries).expect("legacy history and suffix");
+        store.save_committed(u17a_id(4)).expect("committed suffix");
+        for index in 0..4 {
+            store.apply_next_created(u17a_id(index)).expect("legacy history apply");
+        }
+        assert_eq!(committed_schema(&store), 3);
+        (dir, backend, store, commands)
+    }
+    #[test]
+    fn u17b_native_migration_populated_suffix() {
+        let (_dir, backend, mut store, commands) = u17b_history();
+        let before = u17b_snapshot(&store);
+        let results = (0..4)
+            .map(|index| {
+                store
+                    .created_apply_result(index)
+                    .expect("legacy result")
+                    .expect("applied result")
+            })
+            .collect::<Vec<_>>();
+        assert!(
+            store.migrate_legacy_created_to_native().is_ok(),
+            "U17B_MIGRATION_SUCCESS canonical populated schema3 must migrate"
+        );
+        assert_eq!(committed_schema(&store), 4);
+        assert_eq!(&u17b_snapshot(&store)[3..], &before[3..], "business bytes preserved");
+        assert_eq!(store.capture_binding().expect("binding"), Some(u17a_binding()));
+        assert_eq!(store.log_state().expect("tail").last, Some(u17a_id(4)));
+        assert_eq!(store.read_committed().expect("committed"), Some(u17a_id(4)));
+        assert_eq!(
+            store.native_application_state().expect("application pair"),
+            (
+                Some(u17a_id(3)),
+                CaptureMembership::new(None, vec![], std::collections::BTreeMap::new()).expect("empty membership")
+            )
+        );
+        assert_eq!(
+            store.read_native_entries(0, 5).expect("all entries"),
+            commands
+                .iter()
+                .enumerate()
+                .map(|(index, command)| u17a_created(u64::try_from(index).expect("index"), command.clone()))
+                .collect::<Vec<_>>()
+        );
+        for result in results {
+            assert_eq!(
+                store.created_apply_result(result.log_id.index).expect("migrated result"),
+                Some(result.clone())
+            );
+            assert_eq!(
+                store
+                    .native_apply_response(result.log_id.index)
+                    .expect("native result")
+                    .expect("response")
+                    .created_result(),
+                Some(&result.result)
+            );
+        }
+        assert!(store.native_apply_response(4).expect("suffix response").is_none());
+        assert_eq!(
+            store
+                .created_operation(commands[0].operation_id)
+                .expect("first operation")
+                .expect("operation")
+                .result,
+            CreatedResult::Created {
+                version_id: commands[0].prepared.version_id,
+                event_id: commands[0].operation_id,
+                decision_log_id: u17a_id(0)
+            }
+        );
+        drop(store);
+        let mut store = Store::open(&backend.path, identity()).expect("migrated reopen");
+        assert_eq!(&u17b_snapshot(&store)[3..], &before[3..]);
+        let response = store.apply_next_native(u17a_id(4)).expect("real suffix application");
+        assert_eq!(
+            response.created_result(),
+            Some(&CreatedResult::Created {
+                version_id: commands[4].prepared.version_id,
+                event_id: commands[4].operation_id,
+                decision_log_id: u17a_id(4)
+            })
+        );
+        assert_eq!(
+            store.created_head(&commands[4].object).expect("suffix head"),
+            Some(commands[4].prepared.version_id)
+        );
+        assert_eq!(store.last_applied().expect("suffix frontier"), Some(u17a_id(4)));
+    }
+    #[test]
+    fn u17b_native_migration_opaque_refusal() {
+        let canonical = u17a_command().encode().expect("canonical command");
+        let mut noncanonical = canonical.clone();
+        noncanonical.push(0);
+        for payload in [
+            b"opaque payload".to_vec(),
+            canonical[..canonical.len() - 1].to_vec(),
+            noncanonical,
+        ] {
+            let (_dir, backend, mut store) = u17a_bound();
+            store
+                .append(&[LogEntry { id: u17a_id(0), payload }])
+                .expect("invalid legacy suffix append");
+            let before = u17b_snapshot(&store);
+            assert!(store.migrate_legacy_created_to_native().is_err(), "invalid payload must refuse");
+            assert_eq!(u17b_snapshot(&store), before, "refusal changes no table bytes");
+            drop(store);
+            let store = Store::open(&backend.path, identity()).expect("refused suffix reopen");
+            assert_eq!(u17b_snapshot(&store), before);
+            assert_eq!(committed_schema(&store), 3);
+        }
+    }
+    #[test]
+    fn u17b_native_migration_empty_idempotent() {
+        let (_dir, backend, store) = u17a_bound();
+        drop(store);
+        let mut store = Store::from_database(backend.database(), identity()).expect("open legacy without upgrade");
+        assert_eq!(committed_schema(&store), 3);
+        store.migrate_legacy_created_to_native().expect("empty migration");
+        assert_eq!(committed_schema(&store), 4);
+        let before = u17b_snapshot(&store);
+        backend.media.lock().expect("reset migration syncs").syncs = 0;
+        // Reopen through the instrumented backend before checking the no-commit path.
+        drop(store);
+        let mut store = Store::from_database(backend.database(), identity()).expect("native reopen");
+        backend.media.lock().expect("reset reopen syncs").syncs = 0;
+        store.migrate_legacy_created_to_native().expect("native repeat");
+        assert_eq!(backend.media.lock().expect("repeat syncs").syncs, 0, "repeat must not commit");
+        assert_eq!(u17b_snapshot(&store), before);
+        store.writable = false;
+        assert_eq!(
+            store
+                .migrate_legacy_created_to_native()
+                .expect_err("disabled even on schema4")
+                .code(),
+            "write_disabled"
+        );
+    }
+    #[test]
+    fn u17b_native_migration_ack_and_cut() {
+        let (_dir, backend, mut store, _) = u17b_history();
+        let old = u17b_snapshot(&store);
+        backend.media.lock().expect("reset migration syncs").syncs = 0;
+        store.migrate_legacy_created_to_native().expect("migration ACK");
+        let reached = backend.media.lock().expect("actual migration syncs").syncs;
+        assert!(reached > 0, "migration must reach durability IO");
+        let new = u17b_snapshot(&store);
+        backend.cut();
+        drop(store);
+        assert_eq!(u17b_snapshot(&Store::open(&backend.path, identity()).expect("ACK recovery")), new);
+        for sync in 1..=reached {
+            for after in [false, true] {
+                let (_dir, backend, mut store, _) = u17b_history();
+                assert_eq!(u17b_snapshot(&store), old);
+                let (entered, resume) = backend.pause_sync(sync, after);
+                let writer = std::thread::spawn(move || {
+                    let result = store.migrate_legacy_created_to_native();
+                    (store, result)
+                });
+                entered.wait();
+                backend.cut();
+                resume.wait();
+                let (store, result) = writer.join().expect("migration cut writer");
+                drop(store);
+                // A logical open refusal must not hide a physically durable mixed migration.
+                if let Ok(database) = Database::open(&backend.path) {
+                    let actual = u17b_snapshot_database(&database);
+                    assert!(actual == old || actual == new, "U17B_ATOMIC_SCHEMA_MIGRATION");
+                }
+                if let Some(store) = u17a_cut_recovery(&backend) {
+                    let actual = u17b_snapshot(&store);
+                    assert!(
+                        actual == old || actual == new,
+                        "migration cut must never mix schema/log/result/business tables"
+                    );
+                    if result.is_ok() {
+                        assert_eq!(actual, new, "ACK cannot recover old state");
+                    }
+                } else {
+                    assert!(result.is_err(), "ACK recovery must succeed");
+                }
+            }
+        }
+        let (_dir, backend, mut store, _) = u17b_history();
+        backend.arm(1, false, false);
+        assert_eq!(
+            store.migrate_legacy_created_to_native().expect_err("uncertain commit").code(),
+            "commit_indeterminate"
+        );
+        assert_eq!(
+            store
+                .migrate_legacy_created_to_native()
+                .expect_err("migration disabled")
+                .code(),
+            "write_disabled"
+        );
+        assert_eq!(store.append_native(&[]).expect_err("native append disabled").code(), "write_disabled");
+        assert_eq!(store.append(&[]).expect_err("legacy append disabled").code(), "write_disabled");
+        assert_eq!(
+            store.apply_next_native(u17a_id(4)).expect_err("native apply disabled").code(),
+            "write_disabled"
+        );
+        assert_eq!(
+            store
+                .apply_next_created(u17a_id(4))
+                .expect_err("legacy apply disabled")
+                .code(),
+            "write_disabled"
+        );
+        assert_eq!(
+            store
+                .pin_capture_binding(u17a_binding())
+                .expect_err("binding disabled")
+                .code(),
+            "write_disabled"
+        );
+        assert_eq!(store.save_committed(u17a_id(4)).expect_err("committed disabled").code(), "write_disabled");
+        assert_eq!(store.save_vote(vote()).expect_err("vote disabled").code(), "write_disabled");
+        assert_eq!(store.truncate_suffix(5).expect_err("truncate disabled").code(), "write_disabled");
+    }
     fn u17a_id(index: u64) -> LogId {
         LogId {
             term: 11,
@@ -1301,6 +1599,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();

crates/capture-authority/src/native_snapshot.rs

@@ -0,0 +1,95 @@
+//! Owned consistent native application snapshot view.
+use crate::{Store, StoreError, application, snapshot};
+use redb::{ReadTransaction, ReadableDatabase, ReadableTable, ReadableTableMetadata};
+use std::{
+    io::Write,
+    sync::{Arc, Mutex},
+};
+
+#[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),
+}
+
+pub struct NativeSnapshotView {
+    // Drop the transaction before the owner so the Database stays open throughout the view.
+    transaction: ReadTransaction,
+    header: snapshot::Header,
+    counts: [u64; 4],
+    _owner: Arc<Mutex<Store>>,
+}
+impl NativeSnapshotView {
+    pub(crate) fn acquire(owner: Arc<Mutex<Store>>) -> Result<Self, NativeSnapshotError> {
+        let transaction = {
+            let store = owner.lock().map_err(|_| NativeSnapshotError::Poisoned)?;
+            if !store.writable {
+                return Err(StoreError::WriteDisabled.into());
+            }
+            store.database.begin_read().map_err(StoreError::from)?
+        };
+        let (applied, membership) = application::native_application_state(&transaction)?;
+        let header = {
+            let meta = transaction.open_table(crate::META).map_err(StoreError::from)?;
+            let identity = meta
+                .get("identity")
+                .map_err(StoreError::from)?
+                .ok_or(StoreError::CorruptRecord)?;
+            snapshot::Header {
+                cluster_id: crate::decode_identity(identity.value())?.cluster_id,
+                applied,
+                membership,
+                binding: application::read_binding(&meta)?,
+            }
+        };
+        let mut counts = [0; 4];
+        for (index, definition) in [
+            application::HEADS,
+            application::VERSIONS,
+            application::EVENTS,
+            application::OPERATIONS,
+        ]
+        .into_iter()
+        .enumerate()
+        {
+            counts[index] = transaction
+                .open_table(definition)
+                .map_err(StoreError::from)?
+                .len()
+                .map_err(StoreError::from)?;
+        }
+        Ok(Self {
+            transaction,
+            header,
+            counts,
+            _owner: owner,
+        })
+    }
+
+    pub fn stream<W: Write>(self, writer: W, limits: snapshot::Limits) -> Result<W, NativeSnapshotError> {
+        let mut encoder = snapshot::Encoder::new(writer, &self.header, self.counts, limits)?;
+        for (definition, section) in [
+            (application::HEADS, snapshot::Section::Heads),
+            (application::VERSIONS, snapshot::Section::Versions),
+            (application::EVENTS, snapshot::Section::Events),
+            (application::OPERATIONS, snapshot::Section::Operations),
+        ] {
+            let table = self.transaction.open_table(definition).map_err(StoreError::from)?;
+            for row in table.iter().map_err(StoreError::from)? {
+                let (key, value) = row.map_err(StoreError::from)?;
+                encoder.row(snapshot::Row {
+                    section,
+                    key: key.value(),
+                    value: value.value(),
+                })?;
+            }
+        }
+        Ok(encoder.finish()?)
+    }
+}