test(capture): specify coherent snapshot view ownership

crates/capture-authority/src/lib.rs

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

crates/capture-authority/src/native_snapshot.rs

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

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> {