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,15 +65,23 @@ 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(),
         }
     }
 
-    async fn write<F>(&self, write: F) -> Result<(), NativeWriteError>
+    async fn write<T, F>(&self, write: F) -> Result<T, NativeWriteError>
     where
-        F: FnOnce(&mut Store) -> Result<(), StoreError> + Send + 'static,
+        T: Send + 'static,
+        F: FnOnce(&mut Store) -> Result<T, 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.
@@ -110,6 +120,58 @@ impl NativeLogWriter {
     pub async fn truncate_suffix(&self, from: u64) -> Result<(), NativeWriteError> {
         self.write(move |store| store.truncate_suffix(from)).await
     }
+
+    /// Apply entries already committed by Raft consensus. Only the trusted Raft
+    /// apply path may call this; local persistence does not prove quorum.
+    pub async fn apply_committed(&self, entries: Vec<CaptureRaftEntry>) -> Result<Vec<CaptureResponse>, NativeWriteError> {
+        self.write(move |store| {
+            if !store.writable {
+                return Err(StoreError::WriteDisabled);
+            }
+            let (last_applied, _) = store.native_application_state()?;
+            if entries.is_empty() {
+                return Ok(Vec::new());
+            }
+            for entry in &entries {
+                if store.read_native_entry(entry.log_id.index)?.as_ref() != Some(entry) {
+                    return Err(StoreError::InvalidCommittedLog);
+                }
+            }
+            for pair in entries.windows(2) {
+                if pair[0].log_id.index.checked_add(1) != Some(pair[1].log_id.index) {
+                    return Err(StoreError::InvalidApplySequence);
+                }
+            }
+            let first = entries[0].log_id.index;
+            match last_applied {
+                None if first != 0 => return Err(StoreError::InvalidApplySequence),
+                Some(last) if first > last.index && last.index.checked_add(1) != Some(first) => {
+                    return Err(StoreError::InvalidApplySequence);
+                }
+                _ => {}
+            }
+            // Validate the entire request before either durable marker advances.
+            let end = entries[entries.len() - 1].log_id;
+            if store.read_committed()?.is_none_or(|committed| committed.index < end.index) {
+                store.save_committed(end.into())?;
+            }
+            entries
+                .into_iter()
+                .map(|entry| store.apply_next_native(entry.log_id.into()))
+                .collect()
+        })
+        .await
+    }
+
+    pub async fn applied_state(&self) -> Result<(Option<LogId>, CaptureMembership), NativeWriteError> {
+        self.write(|store| {
+            if !store.writable {
+                return Err(StoreError::WriteDisabled);
+            }
+            store.native_application_state()
+        })
+        .await
+    }
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -560,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},
@@ -605,6 +668,302 @@ mod tests {
         }
     }
 
+    fn u21_fixture() -> (tempfile::TempDir, CrashBackend, NativeLogWriter, Vec<CaptureRaftEntry>, CaptureMembership) {
+        let (dir, backend, mut store) = u17a_store();
+        let (entry, membership) = u17a_membership(1);
+        let entries = vec![u17a_blank(0), entry, u17a_created(2, u17a_command())];
+        store.append_native(&entries).expect("persist native batch");
+        (dir, backend, NativeLogWriter::new(store), entries, membership)
+    }
+
+    #[test]
+    fn u21_apply_native_batch() {
+        let (_dir, backend, writer, entries, membership) = u21_fixture();
+        let result = u19_runtime().block_on(writer.apply_committed(entries));
+        assert!(result.is_ok(), "U21_APPLY_SUCCESS: {result:?}");
+        let responses = result.expect("batch responses");
+        let expected = {
+            let store = writer.store.lock().expect("store");
+            (0..3)
+                .map(|i| {
+                    store
+                        .native_apply_response(i)
+                        .expect("receipt")
+                        .expect("U21_APPLY_DURABLE persisted receipt")
+                })
+                .collect::<Vec<_>>()
+        };
+        assert_eq!(responses, expected, "U21_APPLY_DURABLE");
+        assert_eq!(responses[0], CaptureResponse::blank(u17a_id(0)).expect("Blank"));
+        assert_eq!(responses[1], CaptureResponse::membership(u17a_id(1)).expect("Membership"));
+        {
+            let store = writer.store.lock().expect("store");
+            assert_eq!(
+                responses[2],
+                CaptureResponse::created(store.created_apply_result(2).expect("Created").expect("durable Created"))
+                    .expect("Created response")
+            );
+            assert!(
+                store
+                    .created_operation(u17a_command().operation_id)
+                    .expect("operation")
+                    .is_some(),
+                "U21_APPLY_DURABLE"
+            );
+        }
+        assert_eq!(
+            u19_runtime().block_on(writer.applied_state()).expect("checkpoint"),
+            (Some(u17a_id(2)), membership.clone())
+        );
+        backend.cut();
+        drop(writer);
+        let store = Store::open(&backend.path, identity()).expect("cut reopen");
+        assert_eq!(store.native_application_state().expect("reopened pair"), (Some(u17a_id(2)), membership));
+        for (i, response) in expected.iter().enumerate() {
+            assert_eq!(
+                store
+                    .native_apply_response(u64::try_from(i).expect("index"))
+                    .expect("reopened receipt")
+                    .as_ref(),
+                Some(response)
+            );
+        }
+    }
+
+    #[test]
+    fn u21_apply_batch_rejects_mismatch() {
+        let (_dir, backend, writer, entries, _) = u21_fixture();
+        let old = u19_runtime().block_on(writer.applied_state()).expect("old pair");
+        let mut payload = entries.clone();
+        payload[1].payload = openraft::EntryPayload::Blank;
+        let mut id = entries.clone();
+        id[1].log_id = LogId { term: 12, ..u17a_id(1) }.into();
+        let mut command = u17a_command();
+        command.prepared.content_length += 1;
+        let mut changed = entries.clone();
+        changed[2] = u17a_created(2, command);
+        let syncs = backend.media.lock().expect("media").syncs;
+        for invalid in [
+            payload,
+            id,
+            changed,
+            vec![entries[0].clone(), entries[2].clone()],
+            vec![entries[1].clone(), entries[0].clone()],
+            vec![entries[0].clone(), entries[0].clone()],
+            vec![entries[2].clone()],
+        ] {
+            assert!(u19_runtime().block_on(writer.apply_committed(invalid)).is_err(), "U21_ENTRY_MATCH");
+            assert_eq!(
+                u19_runtime().block_on(writer.applied_state()).expect("unchanged pair"),
+                old,
+                "U21_ENTRY_MATCH"
+            );
+            let store = writer.store.lock().expect("store");
+            assert_eq!(store.read_committed().expect("commit unchanged"), None, "U21_ENTRY_MATCH");
+            assert_eq!(
+                store
+                    .created_operation(u17a_command().operation_id)
+                    .expect("business unchanged"),
+                None,
+                "U21_ENTRY_MATCH"
+            );
+        }
+        assert_eq!(backend.media.lock().expect("media").syncs, syncs);
+    }
+
+    #[test]
+    fn u21_apply_replay_prefix() {
+        let (_dir, backend, writer, entries, membership) = u21_fixture();
+        let runtime = u19_runtime();
+        let prefix = runtime
+            .block_on(writer.apply_committed(entries[..2].to_vec()))
+            .expect("prefix");
+        let responses = runtime
+            .block_on(writer.apply_committed(entries.clone()))
+            .expect("prefix replay and suffix");
+        assert_eq!(responses[..2], prefix);
+        let before = {
+            let store = writer.store.lock().expect("store");
+            (
+                store.created_operation(u17a_command().operation_id).expect("operation"),
+                store.created_event(u17a_command().operation_id).expect("event"),
+                store
+                    .created_version(&u17a_command().object, u17a_command().prepared.version_id)
+                    .expect("version"),
+            )
+        };
+        let syncs = backend.media.lock().expect("media").syncs;
+        assert_eq!(
+            runtime
+                .block_on(writer.apply_committed(entries.clone()))
+                .expect("full replay"),
+            responses
+        );
+        assert_eq!(
+            runtime
+                .block_on(writer.apply_committed(entries[..2].to_vec()))
+                .expect("short replay"),
+            prefix
+        );
+        let store = writer.store.lock().expect("store");
+        assert_eq!(store.read_committed().expect("no regression"), Some(u17a_id(2)));
+        assert_eq!(store.native_application_state().expect("pair"), (Some(u17a_id(2)), membership));
+        assert_eq!(
+            (
+                store.created_operation(u17a_command().operation_id).expect("operation"),
+                store.created_event(u17a_command().operation_id).expect("event"),
+                store
+                    .created_version(&u17a_command().object, u17a_command().prepared.version_id)
+                    .expect("version")
+            ),
+            before
+        );
+        assert_eq!(backend.media.lock().expect("media").syncs, syncs);
+    }
+
+    #[test]
+    fn u21_apply_empty_failure_fence() {
+        let (_dir, backend, writer, _, _) = u21_fixture();
+        let syncs = backend.media.lock().expect("media").syncs;
+        assert_eq!(u19_runtime().block_on(writer.apply_committed(vec![])).expect("empty"), vec![]);
+        assert_eq!(backend.media.lock().expect("media").syncs, syncs);
+        let (_dir, _backend, store) = media_store();
+        let legacy = NativeLogWriter::new(store);
+        assert!(matches!(
+            u19_runtime().block_on(legacy.apply_committed(vec![])),
+            Err(NativeWriteError::Store(StoreError::UnsupportedSchema))
+        ));
+        legacy.store.lock().expect("store").writable = false;
+        assert!(matches!(
+            u19_runtime().block_on(legacy.apply_committed(vec![])),
+            Err(NativeWriteError::Store(StoreError::WriteDisabled))
+        ));
+    }
+
+    #[test]
+    fn u21_apply_cut_and_retry() {
+        let runtime = u19_runtime();
+        let (_dir, backend, writer, entries, _) = u21_fixture();
+        backend.media.lock().expect("media").syncs = 0;
+        let expected = runtime
+            .block_on(writer.apply_committed(entries))
+            .expect("measure real successful apply");
+        let syncs = backend.media.lock().expect("media").syncs;
+        assert!(syncs > 0, "real apply syncs");
+        for sync in 1..=syncs {
+            for after in [false, true] {
+                let (_dir, backend, writer, entries, membership) = u21_fixture();
+                backend.arm(sync, after, false);
+                assert!(
+                    matches!(
+                        runtime.block_on(writer.apply_committed(entries.clone())),
+                        Err(NativeWriteError::Store(StoreError::CommitIndeterminate(_)))
+                    ),
+                    "cut must not ACK {sync}/{after}"
+                );
+                assert_eq!(backend.media.lock().expect("media").syncs, sync, "actual boundary reached");
+                assert!(
+                    matches!(
+                        runtime.block_on(writer.applied_state()),
+                        Err(NativeWriteError::Store(StoreError::WriteDisabled))
+                    ),
+                    "U21_CHECKPOINT_FENCE"
+                );
+                backend.cut();
+                drop(writer);
+                let store = Store::open(&backend.path, identity()).expect("reopen interrupted apply");
+                let pair = store.native_application_state().expect("recovered pair");
+                if let Some(last) = pair.0 {
+                    assert!(last.index <= 2);
+                    assert_eq!(
+                        pair.1,
+                        if last.index >= 1 {
+                            membership.clone()
+                        } else {
+                            CaptureMembership::new(None, Vec::new(), std::collections::BTreeMap::new()).expect("empty membership")
+                        }
+                    );
+                    for i in 0..=last.index {
+                        assert_eq!(
+                            store.native_apply_response(i).expect("recovered receipt"),
+                            Some(expected[usize::try_from(i).expect("index")].clone())
+                        );
+                    }
+                } else {
+                    assert_eq!(
+                        pair.1,
+                        CaptureMembership::new(None, Vec::new(), std::collections::BTreeMap::new()).expect("empty membership")
+                    );
+                }
+                let retry = NativeLogWriter::new(store);
+                assert_eq!(
+                    runtime
+                        .block_on(retry.apply_committed(entries))
+                        .expect("retry complete batch"),
+                    expected
+                );
+                assert_eq!(
+                    runtime.block_on(retry.applied_state()).expect("retry checkpoint"),
+                    (Some(u17a_id(2)), membership)
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn u21_apply_worker_and_order() {
+        let (_dir, backend, writer, entries, _) = u21_fixture();
+        backend.media.lock().expect("media").sync_threads.clear();
+        let executor = std::thread::current().id();
+        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.apply_committed(entries).await });
+            tokio::task::spawn_blocking(move || entered.wait())
+                .await
+                .expect("apply entered sync");
+            first.abort();
+            assert!(first.await.expect_err("cancelled await").is_cancelled());
+            assert!(writer.ordering.try_lock().is_err(), "cancel retains ordering guard");
+            let second_writer = writer.clone();
+            let second = tokio::spawn(async move { second_writer.save_vote(vote().into()).await });
+            tokio::task::yield_now().await;
+            assert!(!second.is_finished(), "later write waits");
+            tokio::task::spawn_blocking(move || resume.wait())
+                .await
+                .expect("release sync");
+            second.await.expect("later task").expect("later durable write");
+            assert_eq!(writer.applied_state().await.expect("completed apply").0, Some(u17a_id(2)));
+        });
+        let media = backend.media.lock().expect("media");
+        assert!(!media.sync_threads.is_empty());
+        assert!(media.sync_threads.iter().all(|thread| *thread != executor));
+    }
+
+    #[test]
+    fn u21_checkpoint_read_coherent() {
+        let (_dir, backend, writer, entries, membership) = u21_fixture();
+        let old = u19_runtime().block_on(writer.applied_state()).expect("old pair");
+        let (entered, resume) = backend.pause_sync(1, false);
+        u19_runtime().block_on(async {
+            let apply_writer = writer.clone();
+            let apply = tokio::spawn(async move { apply_writer.apply_committed(entries).await });
+            tokio::task::spawn_blocking(move || entered.wait())
+                .await
+                .expect("apply entered");
+            let read_writer = writer.clone();
+            let read = tokio::spawn(async move { read_writer.applied_state().await });
+            tokio::task::yield_now().await;
+            assert!(!read.is_finished(), "checkpoint waits for store transaction");
+            tokio::task::spawn_blocking(move || resume.wait())
+                .await
+                .expect("release apply");
+            apply.await.expect("apply task").expect("apply");
+            let pair = read.await.expect("read task").expect("read pair");
+            assert!(pair == old || pair == (Some(u17a_id(2)), membership), "full coherent pair");
+        });
+    }
+
     #[test]
     fn u20_vote_durable_roundtrip() {
         let (_dir, backend, store) = u17a_store();
@@ -736,7 +1095,7 @@ mod tests {
         let writer = NativeLogWriter::new(store);
         u19_runtime().block_on(async {
             assert!(matches!(
-                writer.write(|_| panic!("injected worker panic")).await,
+                writer.write::<(), _>(|_| panic!("injected worker panic")).await,
                 Err(NativeWriteError::Worker(_))
             ));
             assert!(matches!(writer.save_vote(vote().into()).await, Err(NativeWriteError::Poisoned)));
@@ -877,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,
@@ -952,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()?)
+    }
+}

docs/architecture/README.md

@@ -32,6 +32,7 @@ Required headings and strings in these files are asserted by `scripts/check_arch
 
 | Document | Use this when |
 |---|---|
+| [capture-checkpoint-contract.md](capture-checkpoint-contract.md) | changing Capture checkpoint capture, provenance, installation, recovery, or log cleanup |
 | [local-descriptor-cache.md](local-descriptor-cache.md) | changing descriptor cache keys/open modes or invalidation after local mutations |
 | [erasure-coding.md](erasure-coding.md) | changing anything under `crates/ecstore/src/erasure/`, `crates/filemeta/`, `crates/ecstore/src/set_disk/`, storage-class or layout code, or any decode, quorum, or heal boundary (normative spec) |
 | [placement-repair-invariants.md](placement-repair-invariants.md) | changing anything that resolves an object to a pool, set, or disk, or that admits scanner or heal work |

docs/architecture/capture-checkpoint-contract.md

@@ -0,0 +1,211 @@
+**Use this when:** changing Capture checkpoint capture, provenance, installation, recovery, or log cleanup.
+
+# Capture checkpoint contract
+
+This document defines the requirements for a recoverable Capture checkpoint.
+It is a design contract, not evidence that production capture, installation,
+purge, quorum integration, or S3 behavior implements these requirements.
+
+## Existing evidence and the missing boundary
+
+In `crates/capture-authority/src/application.rs`, `apply_created` inserts one
+canonical command and result on the first decision for an operation ID. An
+identical command returns that stored result; a different command returns
+`OperationConflict` without replacing the canonical record. The first result
+remains authoritative even when later state would produce a different result.
+
+`CreatedOperation` in `crates/capture-authority/src/created.rs` contains command
+bytes and a result. Only `CreatedResult::Created` carries `decision_log_id`.
+Negative results carry no first-decision ID. Heads, versions, events, and
+operations therefore cannot reconstruct the source of every first decision.
+For example, a stored `HeadMismatch` records an observed head, but does not
+prove that the head existed at the original decision cut. Later retries do
+not supply the missing historical evidence.
+
+`validate_application` in `crates/capture-authority/src/application.rs` derives
+first operation IDs from retained log/application records and, for native
+schema 4, replays decisions from the beginning using in-memory maps and sets.
+That intact-prefix validator is evidence for existing behavior; it is not a
+checkpoint-baseline validator or a bounded recovery algorithm.
+
+The `Row` comment and `StagingSink` in
+`crates/capture-authority/src/snapshot.rs` reserve operation provenance for
+caller validation and require isolated staging. Rows remain opaque to the
+framing decoder. A framing digest detects byte corruption; it does not establish
+semantic consistency, historical truth, or authority to install a checkpoint.
+
+## Canonical operation provenance
+
+Checkpoint-capable storage must use an explicitly versioned operation envelope
+containing the full first-decision `LogId` and the existing canonical operation
+bytes, including their stored result. Preserve those bytes and result; do not
+recompute a negative result against checkpoint-time heads or binding state.
+The envelope version is a storage capability boundary, distinct from the
+snapshot framing version. The existing CSNP v1 framing and header must not be
+silently extended. Header capability metadata requires a declared framing
+revision or an authenticated, versioned manifest whose consistency with the
+captured header is validated. Distinct storage and framing versions do not
+permit silently changing the v1 format.
+
+The first insertion must atomically record the envelope, application response,
+applied position, and any successful head/version/event changes in the same
+write transaction. Its first-decision ID must identify that insertion's log
+entry. An identical-command retry returns the original result and leaves both
+bytes and first-decision ID unchanged. A conflicting-command retry records its
+own application response as `OperationConflict`; it cannot overwrite the
+canonical envelope or claim to be the first decision.
+
+```mermaid
+flowchart TD
+    A[Committed command and operation ID] --> B{Canonical operation exists?}
+    B -->|No| C[Evaluate once at this log ID]
+    C --> D[Atomically persist first ID, original bytes and result]
+    B -->|Yes| E{Same canonical command bytes?}
+    E -->|Yes| F[Return original result; preserve first ID]
+    E -->|No| G[Return OperationConflict; preserve canonical record]
+```
+
+## Compatibility and promotion
+
+The existing schema 3 to schema 4 migration remains unchanged. Adding provenance
+requires a separately declared storage capability with explicit readers, writers,
+upgrade, and downgrade rules. A schema 4 label alone must not imply provenance
+support. Unknown envelope versions and stores missing required provenance must
+be rejected by checkpoint installation and purge authorization.
+
+An old schema 4 store may backfill first IDs only from its complete, intact,
+validated prefix: replay and validate the original command/results, identify the
+first canonical insertion for each operation, then atomically persist the
+capability and envelopes. A retry's ID or a successful event's ID cannot stand
+in for missing negative provenance. If the prefix has already been removed or
+cannot be validated, refuse promotion and cleanup; do not invent IDs. Readers
+without the new capability must refuse the upgraded format rather than silently
+interpret it as the old operation encoding. Retaining all history forever is
+not the checkpoint design: validated promotion supplies a recoverable baseline
+that permits bounded suffix retention.
+
+## Capture cut and resource ownership
+
+Capture must read the header, heads, versions, events, and operations through
+one owned read transaction. The view must retain the owning store/database
+handle for the entire transaction, cursor, and output lifetime; an owned
+transaction type alone does not prove database lifetime safety. Keep the store
+`Arc` alive so dropping another handle cannot invalidate the view. These are
+ownership requirements, not a claim about unverified library behavior.
+
+The header must describe that same committed, applied cut with its full `LogId`,
+full effective membership including its decision position and node metadata,
+cluster identity, binding identity/revision, and supported storage/envelope
+capabilities. Membership and binding must come from the captured transaction,
+not from a later live read. Never capture beyond the durable committed position.
+A four-table view without this cut and metadata is not independently installable.
+
+Emit rows incrementally in deterministic section/key order. Bound row size,
+section and total bytes, counts, temporary disk consumption, and concurrent
+capture admission. Use checked arithmetic. Do not collect the whole snapshot
+in RAM. Cancellation, disk exhaustion, or limit failure must leave the live
+store unchanged and release task-owned staging resources.
+
+## Isolated validation and trust
+
+Decode only into isolated disk staging. Check framing, limits, section order,
+unique keys, complete header metadata, capability versions, cluster/binding
+compatibility, canonical command/result encoding, envelope IDs at or before the
+cut, and all head/version/event/operation relationships. A successful operation
+must agree with its version, event, and first decision. Negative operations must
+have no success event or side effect attributable to that decision.
+
+These checks still cannot prove the historical predicate behind a negative
+result. Neither a digest nor a first ID proves that a `HeadMismatch`,
+`BindingMismatch`, or `VersionConflict` was justified when first decided.
+Remote installation therefore requires an explicitly authenticated and
+authorized Raft authority for the claimed cluster, cut, and membership, or a
+complete historical proof validated from a trusted baseline. Reject input that
+has neither. Authentication and authorization must precede publication and
+purge; a caller's assertion of trust is insufficient. This document does not
+claim that this remote Raft authentication boundary is implemented.
+
+Replay validation must use disk-backed staging/indexes and bounded buffers,
+including duplicate-operation and version/head tracking. Putting the input on
+disk while retaining replay maps proportional to history does not meet this
+requirement. The existing `validate_application` RAM replay cannot simply be
+reused to claim bounded checkpoint installation.
+
+## Atomic publication and recovery
+
+After isolated validation, publication must have one durable atomic point that
+selects the complete checkpoint generation, its application state, cut,
+membership, binding, capability, and recovery metadata. A pointer switch is
+acceptable only when all referenced bytes are already durable and recovery
+cannot observe a partial generation. Serialize publication with application so
+no committed suffix write is lost between validation and the switch.
+
+Preserve local node identity and persisted vote/hard state. Remote checkpoint
+metadata cannot overwrite either. Validate cluster identity, binding, full cut
+identity, and membership before replacing state. Reject rollback behind the
+local durable application/checkpoint floor; an equal index with a different
+term/leader identity is a conflict, not an equivalent cut. Retain valid local
+suffix and commit state consistently with the installed baseline, rather than
+resetting them from remote metadata.
+
+On reopen, select only a fully published generation and verify its recoverable
+baseline before replay. The suffix starts exactly at cut.index + 1, with checked
+index arithmetic; the complete cut ID must agree wherever the boundary is
+retained. Validate contiguous suffix indices, entry identities, committed and
+applied bounds, typed application responses, canonical retry behavior, and
+membership transitions relative to checkpoint state. Reject gaps, conflicting
+boundary IDs, stale membership, or a suffix belonging to another cluster or
+binding. Recovery must begin from checkpoint heads and canonical operations,
+not from empty state or the assumption that the retained log starts at zero.
+
+Purge is authorized only after the published checkpoint can be reopened and
+validated as a recovery baseline. Delete only entries/results covered by that
+exact cut, retaining everything needed for the continuous suffix and local
+Raft durability. Re-check the selected generation and cut immediately before
+cleanup. Persist a recoverable purge boundary so interrupted cleanup is
+idempotent. Never delete the only evidence before provenance promotion or
+checkpoint durability succeeds.
+
+```mermaid
+flowchart TD
+    A[One owned transaction and store lifetime] --> B[Stream captured cut to isolated disk staging]
+    B --> C[Validate framing, semantics, provenance and authorized source]
+    C --> D[Make complete generation durable]
+    D --> E[Single atomic publication point]
+    E --> F[Reopen baseline and validate continuous suffix]
+    F --> G[Authorize purge only through that exact cut]
+    G --> H[Idempotent cleanup; preserve recoverable baseline and suffix]
+    B -->|Failure| I[Discard staging; live generation remains usable]
+    C -->|Failure| I
+    D -->|Before publication crash| I
+```
+
+A crash before publication leaves the previous generation recoverable. A crash
+after publication must recover the new complete baseline, including when purge
+has not begun or is only partly complete. Orphan staging is never selected as
+live state. Old generations may be reclaimed only after no active view needs
+them and the selected generation remains recoverable.
+
+## Behavioral verification requirements
+
+The following are required behavior checks, not reports of tests run or gates
+passed. Existing symbols above anchor the semantics; future checkpoint paths
+must demonstrate the additional requirements directly.
+
+| Behavior | Required observation |
+|---|---|
+| First negative result | Original command/result and first full ID survive capture, install, and reopen; later state does not re-evaluate it. |
+| Identical and conflicting retries | `apply_created` semantics hold across the checkpoint boundary: original result/ID retained; conflict cannot overwrite the canonical record. |
+| Old schema 4 promotion | An intact prefix validates and supplies first IDs; missing/corrupt prefix refuses promotion and purge; schema 3 to 4 behavior stays unchanged. |
+| Snapshot cut under concurrent writes | Header, membership, binding, and all rows describe one transaction cut; later writes remain in the suffix. |
+| View lifetime | Dropping other store handles during row emission cannot invalidate the retained transaction or database. |
+| Framing versus authority | Valid digest with forged negative history is insufficient; unauthenticated or unauthorized remote input is rejected. |
+| Cross-record corruption | Altered command, first ID, event, version, head, membership, or binding fails isolated validation before publication. |
+| Recovery boundary | Exact cut plus continuous suffix recovers; gaps, conflicting full IDs, cross-cluster input, and rollback are rejected. |
+| Local Raft state | Installation preserves local identity, vote/hard state, and compatible durable suffix/commit information. |
+| Resource exhaustion | Oversize rows, arithmetic overflow, disk exhaustion, cancellation, and large histories fail safely within declared memory/disk bounds. |
+| Crash publication | Crashes before and after the atomic point select a complete old or new generation, never a mixture. |
+| Interrupted purge | Reopen succeeds before cleanup is authorized; each interruption resumes without deleting the recoverable baseline or required suffix. |
+
+Documentation path and whitespace guards only validate document hygiene. They
+cannot establish these production behaviors or the remote trust boundary.