feat(capture): apply committed native entries on the ordered worker

crates/capture-authority/src/lib.rs

@@ -69,9 +69,10 @@ impl NativeLogWriter {
         }
     }
 
-    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 +111,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)]
@@ -605,6 +658,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 +1085,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)));

crates/capture-authority/src/snapshot.rs

@@ -167,6 +167,130 @@ fn row_lengths(k: u64, v: u64, limits: Limits) -> Result<u64, SnapshotError> {
     limit(v <= limits.value_bytes, "value bytes")?;
     k.checked_add(v).ok_or(SnapshotError::Limit("record bytes"))
 }
+/// Incremental encoder for isolated snapshot output. Completion verifies framing,
+/// not persistence or installation. A row error permanently fences the writer.
+pub struct Encoder<W: Write> {
+    out: Output<W>,
+    counts: [u64; 4],
+    section: usize,
+    written: u64,
+    previous: Option<Vec<u8>>,
+    limits: Limits,
+    failed: bool,
+}
+impl<W: Write> Encoder<W> {
+    pub fn new(writer: W, header: &Header, counts: [u64; 4], limits: Limits) -> Result<Self, SnapshotError> {
+        header_valid(header)?;
+        let count = counts
+            .iter()
+            .try_fold(0u64, |n, c| n.checked_add(*c).ok_or(SnapshotError::Limit("records")))?;
+        limit(count <= limits.records, "records")?;
+        format(header.applied.is_some() || count == 0, "rows without applied cut")?;
+        membership_size(header, limits)?;
+        let membership = header.membership.encode().map_err(|_| SnapshotError::Format("membership"))?;
+        let membership_len = u64::try_from(membership.len()).map_err(|_| SnapshotError::Limit("membership"))?;
+        limit(membership_len <= limits.membership_bytes, "membership")?;
+        let mut out = Output {
+            io: writer,
+            hash: Sha256::new(),
+            used: 0,
+            max: limits.total_bytes,
+        };
+        out.bytes(MAGIC)?;
+        out.bytes(&header.cluster_id)?;
+        match header.applied {
+            None => out.bytes(&[0])?,
+            Some(id) => {
+                out.bytes(&[1])?;
+                for n in [id.term, id.leader_node, id.index] {
+                    out.number(n)?;
+                }
+            }
+        }
+        out.number(membership_len)?;
+        out.bytes(&membership)?;
+        out.number(header.binding.revision)?;
+        out.bytes(&header.binding.binding_id)?;
+        out.bytes(&header.binding.target_id)?;
+        out.bytes(&[1])?;
+        out.number(counts[0])?;
+        let mut encoder = Self {
+            out,
+            counts,
+            section: 0,
+            written: 0,
+            previous: None,
+            limits,
+            failed: false,
+        };
+        encoder.advance_sections()?;
+        Ok(encoder)
+    }
+
+    fn advance_sections(&mut self) -> Result<(), SnapshotError> {
+        while self.section < SECTIONS.len() && self.written == self.counts[self.section] {
+            self.section += 1;
+            self.written = 0;
+            self.previous = None;
+            if self.section < SECTIONS.len() {
+                self.out
+                    .bytes(&[u8::try_from(self.section + 1).map_err(|_| SnapshotError::Format("section"))?])?;
+                self.out.number(self.counts[self.section])?;
+            }
+        }
+        Ok(())
+    }
+
+    pub fn row(&mut self, row: Row<'_>) -> Result<(), SnapshotError> {
+        format(!self.failed, "encoder failed")?;
+        let result = self.write_row(row);
+        if result.is_err() {
+            self.failed = true;
+        }
+        result
+    }
+
+    fn write_row(&mut self, row: Row<'_>) -> Result<(), SnapshotError> {
+        format(self.section < SECTIONS.len(), "record count")?;
+        format(row.section == SECTIONS[self.section], "section order")?;
+        let k = u64::try_from(row.key.len()).map_err(|_| SnapshotError::Limit("key bytes"))?;
+        let v = u64::try_from(row.value.len()).map_err(|_| SnapshotError::Limit("value bytes"))?;
+        let length = row_lengths(k, v, self.limits)?
+            .checked_add(16)
+            .ok_or(SnapshotError::Limit("record bytes"))?;
+        // Reserve the still-unwritten section framing as well as the footer.
+        let framing = u64::try_from(SECTIONS.len() - self.section - 1)
+            .map_err(|_| SnapshotError::Limit("total bytes"))?
+            .checked_mul(9)
+            .ok_or(SnapshotError::Limit("total bytes"))?;
+        limit(
+            self.out
+                .used
+                .checked_add(length)
+                .and_then(|n| n.checked_add(framing))
+                .and_then(|n| n.checked_add(32))
+                .is_some_and(|n| n <= self.out.max),
+            "total bytes",
+        )?;
+        format(self.previous.as_ref().is_none_or(|p| p.as_slice() < row.key), "key order")?;
+        self.out.number(k)?;
+        self.out.number(v)?;
+        self.out.bytes(row.key)?;
+        self.out.bytes(row.value)?;
+        self.previous = Some(row.key.to_vec());
+        self.written += 1;
+        self.advance_sections()
+    }
+
+    pub fn finish(self) -> Result<W, SnapshotError> {
+        format(!self.failed, "encoder failed")?;
+        format(self.section == SECTIONS.len(), "record count")?;
+        let mut out = self.out;
+        out.io.write_all(&out.hash.finalize())?;
+        Ok(out.io)
+    }
+}
+
 pub fn encode<'a>(
     writer: impl Write,
     header: &Header,
@@ -174,68 +298,11 @@ pub fn encode<'a>(
     rows: impl IntoIterator<Item = Row<'a>>,
     limits: Limits,
 ) -> Result<(), SnapshotError> {
-    header_valid(header)?;
-    let count = counts
-        .iter()
-        .try_fold(0u64, |n, c| n.checked_add(*c).ok_or(SnapshotError::Limit("records")))?;
-    limit(count <= limits.records, "records")?;
-    format(header.applied.is_some() || count == 0, "rows without applied cut")?;
-    membership_size(header, limits)?;
-    let membership = header.membership.encode().map_err(|_| SnapshotError::Format("membership"))?;
-    let membership_len = u64::try_from(membership.len()).map_err(|_| SnapshotError::Limit("membership"))?;
-    limit(membership_len <= limits.membership_bytes, "membership")?;
-    let mut out = Output {
-        io: writer,
-        hash: Sha256::new(),
-        used: 0,
-        max: limits.total_bytes,
-    };
-    out.bytes(MAGIC)?;
-    out.bytes(&header.cluster_id)?;
-    match header.applied {
-        None => out.bytes(&[0])?,
-        Some(id) => {
-            out.bytes(&[1])?;
-            for n in [id.term, id.leader_node, id.index] {
-                out.number(n)?;
-            }
-        }
+    let mut encoder = Encoder::new(writer, header, counts, limits)?;
+    for row in rows {
+        encoder.row(row)?;
     }
-    out.number(membership_len)?;
-    out.bytes(&membership)?;
-    out.number(header.binding.revision)?;
-    out.bytes(&header.binding.binding_id)?;
-    out.bytes(&header.binding.target_id)?;
-    let mut rows = rows.into_iter();
-    for (index, section) in SECTIONS.into_iter().enumerate() {
-        out.bytes(&[u8::try_from(index + 1).map_err(|_| SnapshotError::Format("section"))?])?;
-        out.number(counts[index])?;
-        let mut previous: Option<Vec<u8>> = None;
-        for _ in 0..counts[index] {
-            let row = rows.next().ok_or(SnapshotError::Format("record count"))?;
-            format(row.section == section, "section order")?;
-            let k = u64::try_from(row.key.len()).map_err(|_| SnapshotError::Limit("key bytes"))?;
-            let v = u64::try_from(row.value.len()).map_err(|_| SnapshotError::Limit("value bytes"))?;
-            let length = row_lengths(k, v, limits)?
-                .checked_add(16)
-                .ok_or(SnapshotError::Limit("record bytes"))?;
-            limit(
-                out.used
-                    .checked_add(length)
-                    .and_then(|n| n.checked_add(32))
-                    .is_some_and(|n| n <= out.max),
-                "total bytes",
-            )?;
-            format(previous.as_ref().is_none_or(|p| p.as_slice() < row.key), "key order")?;
-            out.number(k)?;
-            out.number(v)?;
-            out.bytes(row.key)?;
-            out.bytes(row.value)?;
-            previous = Some(row.key.to_vec());
-        }
-    }
-    format(rows.next().is_none(), "record count")?;
-    out.io.write_all(&out.hash.finalize())?;
+    encoder.finish()?;
     Ok(())
 }
 pub fn decode(reader: impl Read, sink: &mut impl StagingSink, limits: Limits) -> Result<Header, SnapshotError> {
@@ -403,6 +470,257 @@ mod tests {
         assert!(result.is_ok(), "U22_CODEC_ROUNDTRIP: {result:?}");
         b
     }
+    #[test]
+    fn u23_redb_guard_rows() -> Result<(), Box<dyn std::error::Error>> {
+        use redb::{ReadableDatabase, ReadableTable, TableDefinition};
+        const TABLE: TableDefinition<&str, &str> = TableDefinition::new("u23_rows");
+        let dir = tempfile::tempdir()?;
+        let db = redb::Database::create(dir.path().join("rows.redb"))?;
+        let fixture = [("a", "head"), ("b", "next")];
+        let write = db.begin_write()?;
+        {
+            let mut table = write.open_table(TABLE)?;
+            for (key, value) in fixture {
+                table.insert(key, value)?;
+            }
+        }
+        write.commit()?;
+        let read = db.begin_read()?;
+        let table = read.open_table(TABLE)?;
+        let mut bytes = Vec::new();
+        let result = Encoder::new(&mut bytes, &header(), [2, 0, 0, 0], limits());
+        assert!(result.is_ok(), "U23_STREAM_GUARD: constructor failed");
+        let mut encoder = result?;
+        for row in table.iter()? {
+            let (key, value) = row?;
+            encoder.row(Row {
+                section: Section::Heads,
+                key: key.value().as_bytes(),
+                value: value.value().as_bytes(),
+            })?;
+        }
+        encoder.finish()?;
+        let mut expected = Vec::new();
+        encode(
+            &mut expected,
+            &header(),
+            [2, 0, 0, 0],
+            fixture.into_iter().map(|(key, value)| Row {
+                section: Section::Heads,
+                key: key.as_bytes(),
+                value: value.as_bytes(),
+            }),
+            limits(),
+        )?;
+        assert_eq!(bytes, expected, "U23_STREAM_GUARD: wire changed");
+        let mut sink = Sink::default();
+        assert_eq!(decode(bytes.as_slice(), &mut sink, limits())?, header());
+        assert_eq!(sink.0.len(), fixture.len());
+        for ((section, key, value), (expected_key, expected_value)) in sink.0.iter().zip(fixture) {
+            assert_eq!(*section, Section::Heads);
+            assert_eq!(key, expected_key.as_bytes());
+            assert_eq!(value, expected_value.as_bytes());
+        }
+        Ok(())
+    }
+
+    #[test]
+    fn u23_failure_fence() {
+        use std::{cell::Cell, rc::Rc};
+        struct Recovering {
+            fail: Rc<Cell<bool>>,
+            writes: Rc<Cell<usize>>,
+        }
+        impl Write for Recovering {
+            fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
+                self.writes.set(self.writes.get() + 1);
+                if self.fail.replace(false) {
+                    Err(io::Error::other("once"))
+                } else {
+                    Ok(bytes.len())
+                }
+            }
+            fn flush(&mut self) -> io::Result<()> {
+                Ok(())
+            }
+        }
+        let fail = Rc::new(Cell::new(false));
+        let writes = Rc::new(Cell::new(0));
+        let mut encoder = Encoder::new(
+            Recovering {
+                fail: fail.clone(),
+                writes: writes.clone(),
+            },
+            &header(),
+            [1, 0, 0, 0],
+            limits(),
+        )
+        .expect("U23_FAILURE_FENCE: construct");
+        fail.set(true);
+        assert!(matches!(
+            encoder.row(Row {
+                section: Section::Heads,
+                key: b"a",
+                value: b"v"
+            }),
+            Err(SnapshotError::Io(_))
+        ));
+        let after_failure = writes.get();
+        assert!(
+            encoder
+                .row(Row {
+                    section: Section::Heads,
+                    key: b"a",
+                    value: b"v"
+                })
+                .is_err(),
+            "U23_FAILURE_FENCE: recovered after IO error"
+        );
+        assert!(encoder.finish().is_err(), "U23_FAILURE_FENCE: finish after IO error");
+        assert_eq!(writes.get(), after_failure, "U23_FAILURE_FENCE: wrote after error");
+        for invalid in [
+            Row {
+                section: Section::Versions,
+                key: b"a",
+                value: b"v",
+            },
+            Row {
+                section: Section::Heads,
+                key: &[0; 33],
+                value: b"v",
+            },
+        ] {
+            let mut output = Vec::new();
+            let mut encoder = Encoder::new(&mut output, &header(), [1, 0, 0, 0], limits()).expect("construct validation encoder");
+            assert!(encoder.row(invalid).is_err());
+            assert!(
+                encoder
+                    .row(Row {
+                        section: Section::Heads,
+                        key: b"a",
+                        value: b"v"
+                    })
+                    .is_err(),
+                "U23_FAILURE_FENCE: validation retry"
+            );
+            assert!(encoder.finish().is_err(), "U23_FAILURE_FENCE: validation finish");
+        }
+        for key in [b"b".as_slice(), b"a".as_slice()] {
+            let mut output = Vec::new();
+            let mut encoder = Encoder::new(&mut output, &header(), [2, 0, 0, 0], limits()).expect("construct ordered encoder");
+            encoder
+                .row(Row {
+                    section: Section::Heads,
+                    key: b"b",
+                    value: b"v",
+                })
+                .expect("first row");
+            assert!(
+                encoder
+                    .row(Row {
+                        section: Section::Heads,
+                        key,
+                        value: b"v"
+                    })
+                    .is_err()
+            );
+            assert!(encoder.finish().is_err(), "U23_FAILURE_FENCE: key order finish");
+        }
+    }
+
+    #[test]
+    fn u23_finish_counts() {
+        for counts in [[1, 0, 0, 0], [0, 1, 0, 0], [0, 0, 0, 1]] {
+            let mut output = Vec::new();
+            let encoder = Encoder::new(&mut output, &header(), counts, limits()).expect("construct incomplete encoder");
+            assert!(encoder.finish().is_err(), "U23_FINISH_COUNT: incomplete accepted");
+            assert!(
+                decode(output.as_slice(), &mut Sink::default(), limits()).is_err(),
+                "U23_FINISH_COUNT: incomplete footer"
+            );
+        }
+        for counts in [[0, 0, 0, 0], [0, 0, 0, 1]] {
+            let mut output = Vec::new();
+            let mut encoder = Encoder::new(&mut output, &header(), counts, limits()).expect("construct empty-section encoder");
+            if counts[3] == 1 {
+                encoder
+                    .row(Row {
+                        section: Section::Operations,
+                        key: b"o",
+                        value: b"v",
+                    })
+                    .expect("skip empty sections");
+            }
+            encoder.finish().expect("U23_FINISH_COUNT: exact count");
+            decode(output.as_slice(), &mut Sink::default(), limits()).expect("valid finished snapshot");
+        }
+        let mut output = Vec::new();
+        let mut encoder = Encoder::new(&mut output, &header(), [1, 0, 0, 0], limits()).expect("construct extra-row encoder");
+        encoder
+            .row(Row {
+                section: Section::Heads,
+                key: b"a",
+                value: b"v",
+            })
+            .expect("declared row");
+        assert!(
+            encoder
+                .row(Row {
+                    section: Section::Heads,
+                    key: b"b",
+                    value: b"v"
+                })
+                .is_err(),
+            "U23_FINISH_COUNT: extra row"
+        );
+        assert!(encoder.finish().is_err(), "U23_FINISH_COUNT: extra row finish");
+    }
+
+    #[test]
+    fn u23_row_limits() {
+        use std::{cell::Cell, rc::Rc};
+        struct Counting(Rc<Cell<usize>>);
+        impl Write for Counting {
+            fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
+                self.0.set(self.0.get() + bytes.len());
+                Ok(bytes.len())
+            }
+            fn flush(&mut self) -> io::Result<()> {
+                Ok(())
+            }
+        }
+        let empty = {
+            let mut bytes = Vec::new();
+            encode(&mut bytes, &header(), [0; 4], [], limits()).expect("empty framing length");
+            u64::try_from(bytes.len()).expect("length")
+        };
+        for (bounded, key, value, label) in [
+            (limits(), &[0; 33][..], b"v".as_slice(), "key bytes"),
+            (limits(), b"a".as_slice(), &[0; 65][..], "value bytes"),
+            (
+                Limits {
+                    total_bytes: empty + 17,
+                    ..limits()
+                },
+                b"a".as_slice(),
+                b"v".as_slice(),
+                "total bytes",
+            ),
+        ] {
+            let written = Rc::new(Cell::new(0));
+            let mut encoder =
+                Encoder::new(Counting(written.clone()), &header(), [1, 0, 0, 0], bounded).expect("construct bounded encoder");
+            let before = written.get();
+            assert!(
+                matches!(encoder.row(Row { section: Section::Heads, key, value }), Err(SnapshotError::Limit(name)) if name == label),
+                "U23_ROW_LIMIT: {label}"
+            );
+            assert_eq!(written.get(), before, "U23_ROW_LIMIT: wrote rejected row");
+            assert!(encoder.finish().is_err(), "U23_ROW_LIMIT: failed finish");
+            assert_eq!(written.get(), before, "U23_ROW_LIMIT: wrote footer");
+        }
+    }
+
     // Independent wire fixture: magic, version, hash algorithm, cluster, optional
     // applied ID, length-prefixed complete membership, binding, four sections.
     fn raw(h: &Header, entries: &[(u8, Vec<(Vec<u8>, Vec<u8>)>)]) -> Vec<u8> {

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.