feat(capture): add bounded streaming snapshot codec

Cargo.lock

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

crates/capture-authority/Cargo.toml

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

crates/capture-authority/src/lib.rs

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

crates/capture-authority/src/snapshot.rs

@@ -0,0 +1,704 @@
+//! Streaming checkpoint framing. Decoding targets isolated staging only: a failure
+//! can leave partial rows. Success verifies framing and digest, not business
+//! semantics; callers must validate semantics before promotion to live storage.
+use crate::{CaptureBinding, CaptureMembership, LogId};
+use sha2::{Digest, Sha256};
+use std::io::{Read, Write};
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct Header {
+    pub cluster_id: [u8; 16],
+    pub applied: Option<LogId>,
+    pub membership: CaptureMembership,
+    pub binding: CaptureBinding,
+}
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Section {
+    Heads,
+    Versions,
+    Events,
+    Operations,
+}
+#[derive(Clone, Copy, Debug)]
+pub struct Limits {
+    pub total_bytes: u64,
+    pub membership_bytes: u64,
+    pub key_bytes: u64,
+    pub value_bytes: u64,
+    pub records: u64,
+}
+#[derive(Debug, thiserror::Error)]
+pub enum SnapshotError {
+    #[error("invalid snapshot framing: {0}")]
+    Format(&'static str),
+    #[error("snapshot limit exceeded: {0}")]
+    Limit(&'static str),
+    #[error("snapshot digest mismatch")]
+    Digest,
+    #[error("snapshot IO failed")]
+    Io(#[from] std::io::Error),
+}
+/// Row values remain opaque. Operation values will contain the canonical decision
+/// LogId followed by the existing canonical operation bytes, validated by callers.
+pub struct Row<'a> {
+    pub section: Section,
+    pub key: &'a [u8],
+    pub value: &'a [u8],
+}
+/// Implementations must write exclusively to isolated staging, never live tables.
+pub trait StagingSink {
+    fn row(&mut self, section: Section, key: &[u8], value: &[u8]) -> std::io::Result<()>;
+}
+const MAGIC: &[u8; 6] = b"CSNP\x01\x01";
+const SECTIONS: [Section; 4] = [Section::Heads, Section::Versions, Section::Events, Section::Operations];
+
+fn limit(ok: bool, name: &'static str) -> Result<(), SnapshotError> {
+    if ok { Ok(()) } else { Err(SnapshotError::Limit(name)) }
+}
+fn format(ok: bool, name: &'static str) -> Result<(), SnapshotError> {
+    if ok { Ok(()) } else { Err(SnapshotError::Format(name)) }
+}
+fn header_valid(h: &Header) -> Result<(), SnapshotError> {
+    format(h.cluster_id != [0; 16], "zero cluster ID")?;
+    match (h.applied, *h.membership.log_id()) {
+        (None, _) => format(
+            h.membership.log_id().is_none() && h.membership.configs().is_empty() && h.membership.nodes().is_empty(),
+            "membership without applied cut",
+        ),
+        (Some(applied), Some(member)) => {
+            format(member.index < applied.index || member == applied, "membership beyond applied cut")
+        }
+        _ => Ok(()),
+    }
+}
+fn membership_size(h: &Header, limits: Limits) -> Result<(), SnapshotError> {
+    // Preflight the existing codec's wire sizes before its output allocation.
+    let mut size = 6u64 + if h.membership.log_id().is_some() { 24 } else { 0 } + 8;
+    for group in h.membership.configs() {
+        let voters = u64::try_from(group.len()).map_err(|_| SnapshotError::Limit("membership"))?;
+        size = size
+            .checked_add(4)
+            .and_then(|n| voters.checked_mul(8).and_then(|v| n.checked_add(v)))
+            .ok_or(SnapshotError::Limit("membership"))?;
+        limit(size <= limits.membership_bytes, "membership")?;
+    }
+    for node in h.membership.nodes().values() {
+        // CNOD v1: five-byte prefix, four-byte URI length, raw URI bytes.
+        let length = u64::try_from(node.rpc_uri().len())
+            .map_err(|_| SnapshotError::Limit("membership"))?
+            .checked_add(9)
+            .ok_or(SnapshotError::Limit("membership"))?;
+        size = size
+            .checked_add(12)
+            .and_then(|n| n.checked_add(length))
+            .ok_or(SnapshotError::Limit("membership"))?;
+        limit(size <= limits.membership_bytes, "membership")?;
+    }
+    limit(size <= limits.membership_bytes, "membership")?;
+    let header_bytes = 6u64 + 16 + 1 + if h.applied.is_some() { 24 } else { 0 } + 8 + 40 + 36 + 32;
+    limit(size.checked_add(header_bytes).is_some_and(|n| n <= limits.total_bytes), "total bytes")
+}
+struct Output<W> {
+    io: W,
+    hash: Sha256,
+    used: u64,
+    max: u64,
+}
+impl<W: Write> Output<W> {
+    fn bytes(&mut self, bytes: &[u8]) -> Result<(), SnapshotError> {
+        let len = u64::try_from(bytes.len()).map_err(|_| SnapshotError::Limit("total bytes"))?;
+        let next = self.used.checked_add(len).ok_or(SnapshotError::Limit("total bytes"))?;
+        limit(next.checked_add(32).is_some_and(|n| n <= self.max), "total bytes")?;
+        self.io.write_all(bytes)?;
+        self.hash.update(bytes);
+        self.used = next;
+        Ok(())
+    }
+    fn number(&mut self, n: u64) -> Result<(), SnapshotError> {
+        self.bytes(&n.to_be_bytes())
+    }
+}
+struct Input<R> {
+    io: R,
+    hash: Sha256,
+    used: u64,
+    max: u64,
+}
+impl<R: Read> Input<R> {
+    fn reserve(&self, len: u64) -> Result<(), SnapshotError> {
+        limit(
+            self.used
+                .checked_add(len)
+                .and_then(|n| n.checked_add(32))
+                .is_some_and(|n| n <= self.max),
+            "total bytes",
+        )
+    }
+    fn bytes(&mut self, bytes: &mut [u8]) -> Result<(), SnapshotError> {
+        let len = u64::try_from(bytes.len()).map_err(|_| SnapshotError::Limit("total bytes"))?;
+        self.reserve(len)?;
+        self.io.read_exact(bytes)?;
+        self.hash.update(&*bytes);
+        self.used += len;
+        Ok(())
+    }
+    fn byte(&mut self) -> Result<u8, SnapshotError> {
+        let mut b = [0];
+        self.bytes(&mut b)?;
+        Ok(b[0])
+    }
+    fn number(&mut self) -> Result<u64, SnapshotError> {
+        let mut b = [0; 8];
+        self.bytes(&mut b)?;
+        Ok(u64::from_be_bytes(b))
+    }
+    fn vector(&mut self, len: u64) -> Result<Vec<u8>, SnapshotError> {
+        self.reserve(len)?;
+        let len = usize::try_from(len).map_err(|_| SnapshotError::Limit("allocation"))?;
+        let mut bytes = Vec::new();
+        bytes.try_reserve_exact(len).map_err(|_| SnapshotError::Limit("allocation"))?;
+        bytes.resize(len, 0);
+        self.bytes(&mut bytes)?;
+        Ok(bytes)
+    }
+}
+fn row_lengths(k: u64, v: u64, limits: Limits) -> Result<u64, SnapshotError> {
+    limit(k <= limits.key_bytes, "key bytes")?;
+    limit(v <= limits.value_bytes, "value bytes")?;
+    k.checked_add(v).ok_or(SnapshotError::Limit("record bytes"))
+}
+pub fn encode<'a>(
+    writer: impl Write,
+    header: &Header,
+    counts: [u64; 4],
+    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)?;
+            }
+        }
+    }
+    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())?;
+    Ok(())
+}
+pub fn decode(reader: impl Read, sink: &mut impl StagingSink, limits: Limits) -> Result<Header, SnapshotError> {
+    let mut input = Input {
+        io: reader,
+        hash: Sha256::new(),
+        used: 0,
+        max: limits.total_bytes,
+    };
+    let mut magic = [0; 6];
+    input.bytes(&mut magic)?;
+    format(&magic == MAGIC, "magic/version/algorithm")?;
+    let mut cluster_id = [0; 16];
+    input.bytes(&mut cluster_id)?;
+    let applied = match input.byte()? {
+        0 => None,
+        1 => Some(LogId {
+            term: input.number()?,
+            leader_node: input.number()?,
+            index: input.number()?,
+        }),
+        _ => return Err(SnapshotError::Format("applied tag")),
+    };
+    let m = input.number()?;
+    limit(m <= limits.membership_bytes, "membership")?;
+    let membership = CaptureMembership::decode(&input.vector(m)?).map_err(|_| SnapshotError::Format("membership"))?;
+    let revision = input.number()?;
+    let mut binding_id = [0; 16];
+    input.bytes(&mut binding_id)?;
+    let mut target_id = [0; 16];
+    input.bytes(&mut target_id)?;
+    let header = Header {
+        cluster_id,
+        applied,
+        membership,
+        binding: CaptureBinding {
+            revision,
+            binding_id,
+            target_id,
+        },
+    };
+    header_valid(&header)?;
+    let mut records = 0u64;
+    for (index, section) in SECTIONS.into_iter().enumerate() {
+        let tag = input.byte()?;
+        format(usize::from(tag) == index + 1, "section order")?;
+        let count = input.number()?;
+        records = records.checked_add(count).ok_or(SnapshotError::Limit("records"))?;
+        limit(records <= limits.records, "records")?;
+        format(header.applied.is_some() || count == 0, "rows without applied cut")?;
+        input.reserve(count.checked_mul(16).ok_or(SnapshotError::Limit("record bytes"))?)?;
+        let mut previous: Option<Vec<u8>> = None;
+        for _ in 0..count {
+            let k = input.number()?;
+            let v = input.number()?;
+            let len = row_lengths(k, v, limits)?;
+            input.reserve(len)?;
+            let key = input.vector(k)?;
+            let value = input.vector(v)?;
+            format(previous.as_ref().is_none_or(|p| p.as_slice() < key.as_slice()), "key order")?;
+            sink.row(section, &key, &value)?;
+            previous = Some(key);
+        }
+    }
+    let mut footer = [0; 32];
+    input.io.read_exact(&mut footer)?;
+    if input.hash.finalize().as_slice() != footer {
+        return Err(SnapshotError::Digest);
+    }
+    let mut extra = [0];
+    loop {
+        match input.io.read(&mut extra) {
+            Ok(0) => break,
+            Ok(_) => return Err(SnapshotError::Format("trailing bytes")),
+            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
+            Err(e) => return Err(SnapshotError::Io(e)),
+        }
+    }
+    Ok(header)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use sha2::{Digest, Sha256};
+    use std::{
+        collections::BTreeMap,
+        io,
+        sync::{
+            Arc,
+            atomic::{AtomicUsize, Ordering},
+        },
+    };
+    fn limits() -> Limits {
+        Limits {
+            total_bytes: 4096,
+            membership_bytes: 512,
+            key_bytes: 32,
+            value_bytes: 64,
+            records: 16,
+        }
+    }
+    fn header() -> Header {
+        Header {
+            cluster_id: [1; 16],
+            applied: Some(LogId {
+                term: 2,
+                leader_node: 7,
+                index: 9,
+            }),
+            membership: CaptureMembership::new(None, vec![], BTreeMap::new()).expect("empty membership"),
+            binding: CaptureBinding {
+                revision: 3,
+                binding_id: [4; 16],
+                target_id: [5; 16],
+            },
+        }
+    }
+    const SECTIONS: [Section; 4] = [Section::Heads, Section::Versions, Section::Events, Section::Operations];
+    fn rows() -> Vec<Row<'static>> {
+        vec![
+            Row {
+                section: Section::Heads,
+                key: b"a",
+                value: b"head",
+            },
+            Row {
+                section: Section::Heads,
+                key: b"b",
+                value: b"next",
+            },
+            Row {
+                section: Section::Versions,
+                key: b"v",
+                value: b"version",
+            },
+            Row {
+                section: Section::Operations,
+                key: b"o",
+                value: b"opaque-operation",
+            },
+        ]
+    }
+    #[derive(Default)]
+    struct Sink(Vec<(Section, Vec<u8>, Vec<u8>)>);
+    impl StagingSink for Sink {
+        fn row(&mut self, s: Section, k: &[u8], v: &[u8]) -> io::Result<()> {
+            self.0.push((s, k.to_vec(), v.to_vec()));
+            Ok(())
+        }
+    }
+    // No Seek implementation; every read is deliberately short.
+    struct Chunked<'a>(&'a [u8]);
+    impl Read for Chunked<'_> {
+        fn read(&mut self, b: &mut [u8]) -> io::Result<usize> {
+            let n = b.len().min(3).min(self.0.len());
+            b[..n].copy_from_slice(&self.0[..n]);
+            self.0 = &self.0[n..];
+            Ok(n)
+        }
+    }
+    fn encoded() -> Vec<u8> {
+        let mut b = vec![];
+        let result = encode(&mut b, &header(), [2, 1, 0, 1], rows(), limits());
+        assert!(result.is_ok(), "U22_CODEC_ROUNDTRIP: {result:?}");
+        b
+    }
+    // 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> {
+        let mut b = b"CSNP\x01\x01".to_vec();
+        b.extend(h.cluster_id);
+        match h.applied {
+            None => b.push(0),
+            Some(id) => {
+                b.push(1);
+                for n in [id.term, id.leader_node, id.index] {
+                    b.extend(n.to_be_bytes());
+                }
+            }
+        }
+        let m = h.membership.encode().expect("membership fixture");
+        b.extend(u64::try_from(m.len()).expect("fixture length").to_be_bytes());
+        b.extend(m);
+        b.extend(h.binding.revision.to_be_bytes());
+        b.extend(h.binding.binding_id);
+        b.extend(h.binding.target_id);
+        for (tag, rows) in entries {
+            b.push(*tag);
+            b.extend(u64::try_from(rows.len()).expect("fixture count").to_be_bytes());
+            for (k, v) in rows {
+                b.extend(u64::try_from(k.len()).expect("key length").to_be_bytes());
+                b.extend(u64::try_from(v.len()).expect("value length").to_be_bytes());
+                b.extend(k);
+                b.extend(v);
+            }
+        }
+        resign(&mut b);
+        b
+    }
+    fn empty_entries() -> Vec<(u8, Vec<(Vec<u8>, Vec<u8>)>)> {
+        (1..=4).map(|t| (t, vec![])).collect()
+    }
+    fn resign_payload(b: &mut Vec<u8>) {
+        b.truncate(b.len() - 32);
+        resign(b);
+    }
+    fn resign(b: &mut Vec<u8>) {
+        let d = Sha256::digest(&*b);
+        b.extend_from_slice(&d);
+    }
+    fn first_section_offset(h: &Header) -> usize {
+        6 + 16 + 1 + if h.applied.is_some() { 24 } else { 0 } + 8 + h.membership.encode().expect("membership").len() + 40
+    }
+    fn rejected(b: &[u8]) {
+        assert!(decode(Chunked(b), &mut Sink::default(), limits()).is_err(), "invalid framing accepted");
+    }
+    #[test]
+    fn u22_roundtrip_stream() {
+        let b = encoded();
+        let mut sink = Sink::default();
+        let h = decode(Chunked(&b), &mut sink, limits()).expect("U22_CODEC_ROUNDTRIP decode");
+        assert_eq!(h, header(), "U22_CODEC_ROUNDTRIP header");
+        let expected: Vec<_> = rows()
+            .into_iter()
+            .map(|r| (r.section, r.key.to_vec(), r.value.to_vec()))
+            .collect();
+        assert_eq!(sink.0, expected, "U22_CODEC_ROUNDTRIP rows");
+        let entries = vec![
+            (1, vec![(b"a".to_vec(), b"head".to_vec()), (b"b".to_vec(), b"next".to_vec())]),
+            (2, vec![(b"v".to_vec(), b"version".to_vec())]),
+            (3, vec![]),
+            (4, vec![(b"o".to_vec(), b"opaque-operation".to_vec())]),
+        ];
+        assert_eq!(b, raw(&header(), &entries), "deterministic wire");
+    }
+    #[test]
+    fn u22_tampered_digest() {
+        let mut b = raw(&header(), &empty_entries());
+        b[6] ^= 1;
+        let result = decode(Chunked(&b), &mut Sink::default(), limits());
+        assert!(matches!(result, Err(SnapshotError::Digest)), "U22_DIGEST: {result:?}");
+        let mut b = raw(&header(), &empty_entries());
+        let end = b.len() - 1;
+        b[end] ^= 1;
+        assert!(
+            matches!(decode(Chunked(&b), &mut Sink::default(), limits()), Err(SnapshotError::Digest)),
+            "U22_DIGEST footer"
+        );
+    }
+    #[test]
+    fn u22_strict_framing() {
+        let good = raw(&header(), &empty_entries());
+        for offset in [0, 4, 5, 22] {
+            let mut b = good.clone();
+            b[offset] = 255;
+            resign_payload(&mut b);
+            rejected(&b);
+        }
+        for end in [0, 5, 22, good.len() - 1] {
+            rejected(&good[..end]);
+        }
+        let mut b = good.clone();
+        b.push(0);
+        rejected(&b);
+        let mut b = good.clone();
+        let o = first_section_offset(&header());
+        b[o + 1..o + 9].copy_from_slice(&1u64.to_be_bytes());
+        resign_payload(&mut b);
+        rejected(&b);
+        let mut e = empty_entries();
+        e.pop();
+        rejected(&raw(&header(), &e));
+        let mut e = empty_entries();
+        e[2].0 = 9;
+        rejected(&raw(&header(), &e));
+    }
+    #[test]
+    fn u22_bounded_lengths() {
+        let mut e = empty_entries();
+        e[0].1.push((vec![1; 33], vec![2]));
+        let b = raw(&header(), &e);
+        let mut sink = Sink::default();
+        let result = decode(Chunked(&b), &mut sink, limits());
+        assert!(matches!(result, Err(SnapshotError::Limit(_))), "U22_LIMIT: {result:?}");
+        assert!(sink.0.is_empty(), "U22_LIMIT before staging");
+        let o = first_section_offset(&header());
+        for (offset, length) in [(o + 9, u64::MAX), (o + 17, u64::MAX), (o + 1, u64::MAX), (47, u64::MAX)] {
+            let mut b = b.clone();
+            b[offset..offset + 8].copy_from_slice(&length.to_be_bytes());
+            resign_payload(&mut b);
+            let result = decode(Chunked(&b), &mut Sink::default(), limits());
+            assert!(matches!(result, Err(SnapshotError::Limit(_))), "U22_LIMIT overflow {offset}: {result:?}");
+        }
+        let good = raw(&header(), &empty_entries());
+        let mut l = limits();
+        l.total_bytes = 16;
+        assert!(matches!(decode(Chunked(&good), &mut Sink::default(), l), Err(SnapshotError::Limit(_))));
+        let mut out = vec![];
+        let mut l = limits();
+        l.key_bytes = 0;
+        assert!(matches!(
+            encode(&mut out, &header(), [2, 1, 0, 1], rows(), l),
+            Err(SnapshotError::Limit(_))
+        ));
+    }
+    #[test]
+    fn u22_ordered_sections() {
+        for keys in [[b"b", b"a"], [b"a", b"a"]] {
+            let mut e = empty_entries();
+            e[0].1 = keys.into_iter().map(|k| (k.to_vec(), vec![1])).collect();
+            let b = raw(&header(), &e);
+            let result = decode(Chunked(&b), &mut Sink::default(), limits());
+            assert!(matches!(result, Err(SnapshotError::Format(_))), "U22_ORDER: {result:?}");
+        }
+        let mut e = empty_entries();
+        e.swap(0, 1);
+        rejected(&raw(&header(), &e));
+        let mut out = vec![];
+        assert!(encode(&mut out, &header(), [1, 0, 0, 0], [], limits()).is_err(), "count mismatch");
+        let rs = [
+            Row {
+                section: SECTIONS[1],
+                key: b"a",
+                value: b"b",
+            },
+            Row {
+                section: SECTIONS[0],
+                key: b"a",
+                value: b"b",
+            },
+        ];
+        assert!(encode(&mut out, &header(), [1, 1, 0, 0], rs, limits()).is_err(), "section regression");
+    }
+    struct Broken;
+    impl Read for Broken {
+        fn read(&mut self, _: &mut [u8]) -> io::Result<usize> {
+            Err(io::Error::other("read fault"))
+        }
+    }
+    impl Write for Broken {
+        fn write(&mut self, _: &[u8]) -> io::Result<usize> {
+            Err(io::Error::other("write fault"))
+        }
+        fn flush(&mut self) -> io::Result<()> {
+            Ok(())
+        }
+    }
+    impl StagingSink for Broken {
+        fn row(&mut self, _: Section, _: &[u8], _: &[u8]) -> io::Result<()> {
+            Err(io::Error::other("sink fault"))
+        }
+    }
+    #[test]
+    fn u22_io_failures() {
+        let r = decode(Broken, &mut Sink::default(), limits());
+        assert!(matches!(r, Err(SnapshotError::Io(_))));
+        let e = encode(Broken, &header(), [0; 4], [], limits());
+        assert!(matches!(e, Err(SnapshotError::Io(_))));
+        let mut e = empty_entries();
+        e[0].1.push((vec![1], vec![2]));
+        let r = decode(Chunked(&raw(&header(), &e)), &mut Broken, limits());
+        assert!(matches!(r, Err(SnapshotError::Io(_))));
+        use std::error::Error;
+        let err = SnapshotError::Io(io::Error::other("cause"));
+        assert!(err.source().is_some());
+    }
+    #[test]
+    fn u22_header_pair() {
+        let mut h = header();
+        h.cluster_id = [0; 16];
+        rejected(&raw(&h, &empty_entries()));
+        let mut h = header();
+        h.applied = None;
+        let b = raw(&h, &empty_entries());
+        assert_eq!(decode(Chunked(&b), &mut Sink::default(), limits()).expect("empty checkpoint"), h);
+        let mut e = empty_entries();
+        e[0].1.push((vec![1], vec![2]));
+        rejected(&raw(&h, &e));
+        for id in [
+            LogId {
+                term: 2,
+                leader_node: 7,
+                index: 10,
+            },
+            LogId {
+                term: 3,
+                leader_node: 7,
+                index: 9,
+            },
+        ] {
+            let mut h = header();
+            h.membership = CaptureMembership::new(Some(id), vec![], BTreeMap::new()).expect("membership ID");
+            rejected(&raw(&h, &empty_entries()));
+            h.applied = None;
+            rejected(&raw(&h, &empty_entries()));
+        }
+        let mut joint = header();
+        let nodes = (1..=4)
+            .map(|id| {
+                (
+                    id,
+                    crate::CaptureNode::new(format!("http://127.0.0.1:{}", 9000 + id)).expect("joint node"),
+                )
+            })
+            .collect();
+        joint.membership =
+            CaptureMembership::new(joint.applied, vec![[1, 2].into_iter().collect(), [2, 3].into_iter().collect()], nodes)
+                .expect("joint membership and learner");
+        let mut wire = vec![];
+        encode(&mut wire, &joint, [0; 4], [], limits()).expect("encode complete joint membership");
+        assert_eq!(
+            decode(Chunked(&wire), &mut Sink::default(), limits()).expect("decode complete joint membership"),
+            joint
+        );
+        let mut h = header();
+        h.membership = CaptureMembership::new(h.applied, vec![], BTreeMap::new()).expect("equal cut");
+        assert_eq!(
+            decode(Chunked(&raw(&h, &empty_entries())), &mut Sink::default(), limits()).expect("matching ID"),
+            h
+        );
+    }
+    #[test]
+    fn u22_incremental_sink() {
+        struct ProgressRead<'a> {
+            bytes: &'a [u8],
+            read: Arc<AtomicUsize>,
+        }
+        impl Read for ProgressRead<'_> {
+            fn read(&mut self, b: &mut [u8]) -> io::Result<usize> {
+                let n = b.len().min(1).min(self.bytes.len());
+                b[..n].copy_from_slice(&self.bytes[..n]);
+                self.bytes = &self.bytes[n..];
+                self.read.fetch_add(n, Ordering::SeqCst);
+                Ok(n)
+            }
+        }
+        struct ProgressSink {
+            read: Arc<AtomicUsize>,
+            end: usize,
+            rows: usize,
+        }
+        impl StagingSink for ProgressSink {
+            fn row(&mut self, _: Section, _: &[u8], _: &[u8]) -> io::Result<()> {
+                assert!(self.read.load(Ordering::SeqCst) <= self.end, "sink must precede footer");
+                self.rows += 1;
+                Ok(())
+            }
+        }
+        let b = encoded();
+        let read = Arc::new(AtomicUsize::new(0));
+        let mut sink = ProgressSink {
+            read: read.clone(),
+            end: b.len() - 32,
+            rows: 0,
+        };
+        decode(ProgressRead { bytes: &b, read }, &mut sink, limits()).expect("incremental decode");
+        assert_eq!(sink.rows, 4);
+        let mut b = b;
+        let last = b.len() - 1;
+        b[last] ^= 1;
+        let mut sink = Sink::default();
+        assert!(matches!(decode(Chunked(&b), &mut sink, limits()), Err(SnapshotError::Digest)));
+        assert_eq!(sink.0.len(), 4, "failure leaves isolated partial staging");
+    }
+}