feat(capture): add explicit schema-bound checkpoint capabilities

crates/capture-authority/src/application.rs

@@ -103,6 +103,10 @@ fn decode_provenance(b: &[u8]) -> Result<CreatedOperationProvenance, StoreError>
     })
 }
 
+pub(crate) fn validate_provenance_encoding(bytes: &[u8]) -> Result<(), StoreError> {
+    decode_provenance(bytes).map(|_| ())
+}
+
 fn decode_stored_operation(b: &[u8], schema: u64) -> Result<CreatedOperation, StoreError> {
     if matches!(schema, 5 | 6) {
         Ok(decode_provenance(b)?.operation)

crates/capture-authority/src/lib.rs

@@ -22,6 +22,7 @@ pub use node::{CaptureNode, CaptureNodeError};
 pub use raft_types::{
     CaptureCommand, CaptureMembership, CaptureRaftEntry, CaptureResponse, CaptureResponseKind, CaptureTypeConfig, MembershipError,
 };
+pub use snapshot::{CapabilityArtifact, CapabilityManifest};
 #[cfg(test)]
 mod application_tests;
 mod truncate;
@@ -2231,6 +2232,407 @@ mod tests {
             membership,
         )
     }
+    #[derive(Default)]
+    struct U27aSink(Vec<(snapshot::Section, Vec<u8>, Vec<u8>)>);
+    impl snapshot::StagingSink for U27aSink {
+        fn row(&mut self, section: snapshot::Section, key: &[u8], value: &[u8]) -> io::Result<()> {
+            self.0.push((section, key.to_vec(), value.to_vec()));
+            Ok(())
+        }
+    }
+
+    fn u27a_manifest(payload: &[u8]) -> snapshot::CapabilityManifest {
+        use sha2::{Digest, Sha256};
+        let mut bytes = [0; 56];
+        bytes[..4].copy_from_slice(b"CCAP");
+        bytes[4] = 1;
+        bytes[5] = 1;
+        bytes[6..14].copy_from_slice(&6u64.to_be_bytes());
+        bytes[14] = 1;
+        bytes[15] = 1;
+        bytes[16..24].copy_from_slice(&u64::try_from(payload.len()).expect("small test payload length").to_be_bytes());
+        bytes[24..].copy_from_slice(&Sha256::digest(payload));
+        snapshot::CapabilityManifest { bytes }
+    }
+
+    fn u27a_fixture() -> (tempfile::TempDir, CrashBackend, Store, Vec<DecideCreated>) {
+        let (dir, backend, mut store) = u17a_store();
+        provenance_promote(&mut store);
+        let commands = provenance_commands();
+        let mut entries = vec![u17a_blank(0), u17a_membership(1).0];
+        for (index, command) in commands.iter().enumerate() {
+            entries.push(u17a_created(u64::try_from(index).expect("fixture index") + 2, command.clone()));
+        }
+        u24_apply(&mut store, &entries);
+        assert_eq!(committed_schema(&store), 6, "schema6 fixture setup");
+        (dir, backend, store, commands)
+    }
+
+    fn u27a_rows(store: &Store) -> Vec<(Vec<u8>, Vec<u8>)> {
+        let tx = store.database.begin_read().expect("operation fixture transaction");
+        let table = tx.open_table(application::OPERATIONS).expect("operation fixture table");
+        let rows = table
+            .iter()
+            .expect("operation fixture cursor")
+            .map(|row| {
+                let (key, value) = row.expect("operation fixture row");
+                (key.value().to_vec(), value.value().to_vec())
+            })
+            .collect();
+        rows
+    }
+
+    fn u27a_artifact(writer: &NativeLogWriter) -> snapshot::CapabilityArtifact<Vec<u8>> {
+        let view = u19_runtime()
+            .block_on(writer.snapshot_view())
+            .expect("capability fixture view");
+        let result = view.stream_capability_artifact(Vec::new(), u24_limits());
+        assert_eq!(result.is_ok(), true, "U27A_CAPABILITY_ENABLED");
+        result.expect("asserted capability production")
+    }
+
+    fn u27a_decode(
+        artifact: snapshot::CapabilityArtifact<Vec<u8>>,
+    ) -> Result<(snapshot::Header, U27aSink), snapshot::SnapshotError> {
+        let mut sink = U27aSink::default();
+        let header = snapshot::decode_capability_artifact(
+            snapshot::CapabilityArtifact {
+                manifest: artifact.manifest,
+                payload: io::Cursor::new(artifact.payload),
+            },
+            &mut sink,
+            u24_limits(),
+        )?;
+        Ok((header, sink))
+    }
+
+    #[test]
+    fn u27a_checkpoint_schema6_copr_stream() {
+        let (_dir, _backend, store, commands) = u27a_fixture();
+        let expected = u27a_rows(&store);
+        let writer = NativeLogWriter::new(store);
+        let artifact = u27a_artifact(&writer);
+        assert_eq!(
+            artifact.manifest.encode(),
+            u27a_manifest(&artifact.payload).encode(),
+            "U27A_PAYLOAD_BINDING"
+        );
+        let decoded = u27a_decode(artifact);
+        assert_eq!(decoded.is_ok(), true, "U27A_COPR_BYTES");
+        let (header, sink) = decoded.expect("asserted capability consumer");
+        assert_eq!(header.applied, Some(u17a_id(10)), "U27A_COPR_BYTES");
+        assert_eq!(header.membership.log_id(), &Some(u17a_id(1)), "U27A_COPR_BYTES");
+        let operations: Vec<_> = sink
+            .0
+            .into_iter()
+            .filter(|row| row.0 == snapshot::Section::Operations)
+            .map(|(_, key, value)| (key, value))
+            .collect();
+        assert_eq!(operations, expected, "U27A_COPR_BYTES");
+        assert_eq!(operations.len(), 4, "U27A_COPR_BYTES");
+        for (command_index, first_index) in [(0, 2), (3, 5), (4, 6), (5, 7)] {
+            let command = &commands[command_index];
+            let value = &operations
+                .iter()
+                .find(|(key, _)| key.as_slice() == command.operation_id.as_slice())
+                .expect("canonical operation")
+                .1;
+            assert_eq!(&value[..5], b"COPR\x01", "U27A_COPR_BYTES");
+            assert_eq!(&value[5..29], encode_id(u17a_id(first_index)), "U27A_FIRST_ID");
+            let stored = writer
+                .store
+                .lock()
+                .expect("fixture store")
+                .created_operation_provenance(command.operation_id)
+                .expect("provenance read")
+                .expect("operation");
+            assert_eq!(stored.first_decision_log_id, u17a_id(first_index), "U27A_FIRST_ID");
+            assert_eq!(
+                stored.operation.canonical_command,
+                command.encode().expect("original command"),
+                "U27A_COPR_BYTES"
+            );
+        }
+        let store = writer.store.lock().expect("conflict receipt store");
+        assert_eq!(
+            store
+                .native_apply_response(4)
+                .expect("receipt")
+                .expect("response")
+                .created_result(),
+            Some(&CreatedResult::OperationConflict),
+            "U27A_FIRST_ID"
+        );
+    }
+
+    #[test]
+    fn u27a_checkpoint_schema4_legacy_unchanged() {
+        let (_dir, _backend, mut store) = u17a_store();
+        u24_apply(&mut store, &[u17a_blank(0), u17a_membership(1).0, u17a_created(2, u17a_command())]);
+        let before = u17b_snapshot(&store);
+        // Fixed bytes exported by the accepted pre-capability binary for this schema4 fixture.
+        const LEGACY_PAYLOAD_HEX: &str = concat!(
+            "43534e5001010707070707070707070707070707070701000000000000000b00000000000000070000000000000002000000000000010e434d454d01",
+            "01000000000000000b000000000000000700000000000000010000000200000002000000000000000700000000000000080000000200000000000000",
+            "08000000000000000900000004000000000000000700000024434e4f44010000001b68747470733a2f2f706565722d372e6578616d706c653a373030",
+            "37000000000000000800000023434e4f44010000001a687474703a2f2f706565722d382e6578616d706c653a37303038000000000000000900000024",
+            "434e4f44010000001b68747470733a2f2f706565722d392e6578616d706c653a37303039000000000000000a00000025434e4f44010000001c687474",
+            "70733a2f2f6c6561726e65722e6578616d706c653a373031300000000000000009060606060606060606060606060606060707070707070707070707",
+            "070707070701000000000000000100000000000000210000000000000010020202020202020202020202020202020000000d6e61746976652d6f626a",
+            "656374030303030303030303030303030303030200000000000000010000000000000031000000000000006802020202020202020202020202020202",
+            "0000000d6e61746976652d6f626a65637403030303030303030303030303030303030303030303030303030303030303030404040404040404040404",
+            "040404040405050505050505050505050505050505050505050505050505050505050505050000000000001001000000000000000b00000000000000",
+            "0700000000000000020000000000000009030000000000000001000000000000001000000000000000a9010101010101010101010101010101010202",
+            "02020202020202020202020202020000000d6e61746976652d6f626a6563740303030303030303030303030303030304040404040404040404040404",
+            "04040405050505050505050505050505050505050505050505050505050505050505050000000000001001000000000000000b000000000000000700",
+            "000000000000020000000000000009060606060606060606060606060606060707070707070707070707070707070704000000000000000100000000",
+            "0000001000000000000000c8010101010101010101010101010101010000000000000087435254440101010101010101010101010101010101020202",
+            "020202020202020202020202020000000d6e61746976652d6f626a656374000303030303030303030303030303030304040404040404040404040404",
+            "040404050505050505050505050505050505050505050505050505050505050505050500000000000010010000000000000009000303030303030303",
+            "030303030303030301010101010101010101010101010101000000000000000b000000000000000700000000000000024f43d9e18a25d702afdacde7",
+            "e9b00f87177c40b11232952e61ddb2aa18411bce",
+        );
+        let original: Vec<u8> = LEGACY_PAYLOAD_HEX
+            .as_bytes()
+            .chunks_exact(2)
+            .map(|pair| {
+                let digits = std::str::from_utf8(pair).expect("fixed legacy hex");
+                u8::from_str_radix(digits, 16).expect("fixed legacy byte")
+            })
+            .collect();
+        let writer = NativeLogWriter::new(store);
+        let view = u19_runtime().block_on(writer.snapshot_view()).expect("legacy view");
+        let payload = view.stream(Vec::new(), u24_limits()).expect("original stream");
+        assert_eq!(payload, original, "U27A_V1_OPAQUE");
+        assert_eq!(&payload[..6], b"CSNP\x01\x01", "U27A_V1_OPAQUE");
+        let view = u19_runtime()
+            .block_on(writer.snapshot_view())
+            .expect("legacy capability view");
+        let result = view.stream_capability_artifact(Vec::new(), u24_limits());
+        assert_eq!(
+            matches!(result, Err(NativeSnapshotError::Store(StoreError::UnsupportedSchema))),
+            true,
+            "U27A_SCHEMA_GATE"
+        );
+        let store = writer.store.lock().expect("legacy unchanged store");
+        assert_eq!(committed_schema(&store), 4, "U27A_SCHEMA_GATE");
+        assert_eq!(u17b_snapshot(&store), before, "U27A_SCHEMA_GATE");
+    }
+
+    #[test]
+    fn u27a_checkpoint_codec_mismatch_rejected() {
+        let (_dir, _backend, mut store) = u17a_store();
+        u24_apply(&mut store, &[u17a_created(0, u17a_command())]);
+        let writer = NativeLogWriter::new(store);
+        let old = u19_runtime()
+            .block_on(writer.snapshot_view())
+            .expect("legacy codec view")
+            .stream(Vec::new(), u24_limits())
+            .expect("legacy codec payload");
+        assert_eq!(
+            u27a_decode(snapshot::CapabilityArtifact {
+                manifest: u27a_manifest(&old),
+                payload: old
+            })
+            .is_err(),
+            true,
+            "U27A_CODEC_DISPATCH"
+        );
+        let (_dir6, _backend6, store6, _) = u27a_fixture();
+        let writer6 = NativeLogWriter::new(store6);
+        let artifact = u27a_artifact(&writer6);
+        for (offset, wrong) in [(6, 5), (14, 0), (14, 2), (15, 0), (15, 2)] {
+            let mut manifest = artifact.manifest.clone();
+            if offset == 6 {
+                manifest.bytes[6..14].copy_from_slice(&u64::from(wrong).to_be_bytes());
+            } else {
+                manifest.bytes[offset] = wrong;
+            }
+            assert_eq!(
+                u27a_decode(snapshot::CapabilityArtifact {
+                    manifest,
+                    payload: artifact.payload.clone()
+                })
+                .is_err(),
+                true,
+                "U27A_CODEC_DISPATCH"
+            );
+        }
+    }
+
+    #[test]
+    fn u27a_checkpoint_manifest_binding_rejected() {
+        struct ShortWriter {
+            bytes: Vec<u8>,
+            interrupted: bool,
+        }
+        impl Write for ShortWriter {
+            fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
+                if !self.interrupted {
+                    self.interrupted = true;
+                    return Err(io::ErrorKind::Interrupted.into());
+                }
+                let length = bytes.len().min(3);
+                self.bytes.extend_from_slice(&bytes[..length]);
+                Ok(length)
+            }
+            fn flush(&mut self) -> io::Result<()> {
+                Ok(())
+            }
+        }
+        let (_dir, _backend, store, _) = u27a_fixture();
+        let writer = NativeLogWriter::new(store);
+        let view = u19_runtime().block_on(writer.snapshot_view()).expect("short output view");
+        let result = view.stream_capability_artifact(
+            ShortWriter {
+                bytes: Vec::new(),
+                interrupted: false,
+            },
+            u24_limits(),
+        );
+        assert_eq!(result.is_ok(), true, "U27A_PAYLOAD_BINDING");
+        let artifact = result.expect("asserted short write producer");
+        assert_eq!(
+            artifact.manifest.encode(),
+            u27a_manifest(&artifact.payload.bytes).encode(),
+            "U27A_PAYLOAD_BINDING"
+        );
+        let encoded = artifact.manifest.encode();
+        assert_eq!(snapshot::CapabilityManifest::decode(&encoded).is_ok(), true, "U27A_PAYLOAD_BINDING");
+        for size in [0, 55, 57] {
+            let mut raw = encoded.to_vec();
+            raw.resize(size, 0);
+            assert_eq!(snapshot::CapabilityManifest::decode(&raw).is_err(), true, "U27A_PAYLOAD_BINDING");
+        }
+        for offset in [0, 4, 5, 14, 15] {
+            let mut raw = encoded;
+            raw[offset] = 255;
+            assert_eq!(snapshot::CapabilityManifest::decode(&raw).is_err(), true, "U27A_PAYLOAD_BINDING");
+        }
+        for offset in [16, 23, 24, 55] {
+            let mut manifest = artifact.manifest.clone();
+            manifest.bytes[offset] ^= 1;
+            assert_eq!(
+                u27a_decode(snapshot::CapabilityArtifact {
+                    manifest,
+                    payload: artifact.payload.bytes.clone()
+                })
+                .is_err(),
+                true,
+                "U27A_PAYLOAD_BINDING"
+            );
+        }
+        let mut corrupt = artifact.payload.bytes.clone();
+        *corrupt.last_mut().expect("footer byte") ^= 1;
+        assert_eq!(
+            u27a_decode(snapshot::CapabilityArtifact {
+                manifest: artifact.manifest.clone(),
+                payload: corrupt
+            })
+            .is_err(),
+            true,
+            "U27A_PAYLOAD_BINDING"
+        );
+        let mut extra = artifact.payload.bytes.clone();
+        extra.push(0);
+        assert_eq!(
+            u27a_decode(snapshot::CapabilityArtifact {
+                manifest: artifact.manifest,
+                payload: extra
+            })
+            .is_err(),
+            true,
+            "U27A_PAYLOAD_BINDING"
+        );
+    }
+
+    #[test]
+    fn u27a_checkpoint_negative_bytes_first_id() {
+        let (_dir, _backend, mut store, commands) = u27a_fixture();
+        let before = u27a_rows(&store);
+        let mut later = commands[0].clone();
+        later.operation_id = [20; 16];
+        later.prepared.version_id = [21; 16];
+        later.expected_head = Some(commands[0].prepared.version_id);
+        u24_apply(
+            &mut store,
+            &[
+                u17a_created(11, later),
+                u17a_created(12, commands[4].clone()),
+                u17a_created(13, commands[2].clone()),
+            ],
+        );
+        assert_eq!(
+            store.created_head(&commands[0].object).expect("changed head"),
+            Some([21; 16]),
+            "U27A_FIRST_ID"
+        );
+        let after = u27a_rows(&store);
+        for command_index in [3, 4, 5] {
+            let key = commands[command_index].operation_id.to_vec();
+            let old = before.iter().find(|row| row.0 == key).expect("first negative bytes");
+            assert_eq!(after.iter().find(|row| row.0 == key), Some(old), "U27A_FIRST_ID");
+        }
+        assert_eq!(
+            store
+                .native_apply_response(13)
+                .expect("conflict response")
+                .expect("receipt")
+                .created_result(),
+            Some(&CreatedResult::OperationConflict),
+            "U27A_FIRST_ID"
+        );
+        let writer = NativeLogWriter::new(store);
+        let (_, sink) = u27a_decode(u27a_artifact(&writer)).expect("capability preserves negatives");
+        let captured: Vec<_> = sink
+            .0
+            .into_iter()
+            .filter(|row| row.0 == snapshot::Section::Operations)
+            .map(|(_, key, value)| (key, value))
+            .collect();
+        assert_eq!(captured, after, "U27A_FIRST_ID");
+    }
+
+    #[test]
+    fn u27a_checkpoint_v1_framing_is_opaque() {
+        let (_dir, _backend, mut store) = u17a_store();
+        u24_apply(&mut store, &[u17a_blank(0)]);
+        let tx = store.database.begin_read().expect("opaque header transaction");
+        let (applied, membership) = application::native_application_state(&tx).expect("opaque application pair");
+        let header = snapshot::Header {
+            cluster_id: identity().cluster_id,
+            applied,
+            membership,
+            binding: u17a_binding(),
+        };
+        let mut payload = Vec::new();
+        snapshot::encode(
+            &mut payload,
+            &header,
+            [0, 0, 0, 1],
+            [snapshot::Row {
+                section: snapshot::Section::Operations,
+                key: b"o",
+                value: b"opaque-operation",
+            }],
+            u24_limits(),
+        )
+        .expect("original opaque encode");
+        let mut sink = U27aSink::default();
+        let original = snapshot::decode(payload.as_slice(), &mut sink, u24_limits());
+        assert_eq!(original.is_ok(), true, "U27A_V1_OPAQUE");
+        assert_eq!(sink.0[0].2, b"opaque-operation", "U27A_V1_OPAQUE");
+        assert_eq!(
+            u27a_decode(snapshot::CapabilityArtifact {
+                manifest: u27a_manifest(&payload),
+                payload
+            })
+            .is_err(),
+            true,
+            "U27A_V1_OPAQUE"
+        );
+    }
+
     fn u24_limits() -> snapshot::Limits {
         snapshot::Limits {
             total_bytes: 1_000_000,

crates/capture-authority/src/native_snapshot.rs

@@ -23,6 +23,7 @@ pub struct NativeSnapshotView {
     transaction: ReadTransaction,
     header: snapshot::Header,
     counts: [u64; 4],
+    schema: u64,
     _owner: Arc<Mutex<Store>>,
 }
 impl NativeSnapshotView {
@@ -35,18 +36,27 @@ impl NativeSnapshotView {
             store.database.begin_read().map_err(StoreError::from)?
         };
         let (applied, membership) = application::native_application_state(&transaction)?;
-        let header = {
+        let (header, schema) = {
             let meta = transaction.open_table(crate::META).map_err(StoreError::from)?;
             let identity = meta
                 .get("identity")
                 .map_err(StoreError::from)?
                 .ok_or(StoreError::CorruptRecord)?;
-            snapshot::Header {
-                cluster_id: crate::decode_identity(identity.value())?.cluster_id,
-                applied,
-                membership,
-                binding: application::read_binding(&meta)?,
-            }
+            let schema = crate::number(
+                meta.get("schema")
+                    .map_err(StoreError::from)?
+                    .ok_or(StoreError::CorruptRecord)?
+                    .value(),
+            )?;
+            (
+                snapshot::Header {
+                    cluster_id: crate::decode_identity(identity.value())?.cluster_id,
+                    applied,
+                    membership,
+                    binding: application::read_binding(&meta)?,
+                },
+                schema,
+            )
         };
         let mut counts = [0; 4];
         for (index, definition) in [
@@ -68,10 +78,23 @@ impl NativeSnapshotView {
             transaction,
             header,
             counts,
+            schema,
             _owner: owner,
         })
     }
 
+    pub fn stream_capability_artifact<W: Write>(
+        self,
+        writer: W,
+        limits: snapshot::Limits,
+    ) -> Result<snapshot::CapabilityArtifact<W>, NativeSnapshotError> {
+        if self.schema != 6 {
+            return Err(StoreError::UnsupportedSchema.into());
+        }
+        let output = snapshot::CapabilityOutput::new(writer, limits.total_bytes);
+        Ok(self.stream(output, limits)?.finish())
+    }
+
     pub fn stream<W: Write>(self, writer: W, limits: snapshot::Limits) -> Result<W, NativeSnapshotError> {
         let mut encoder = snapshot::Encoder::new(writer, &self.header, self.counts, limits)?;
         for (definition, section) in [

crates/capture-authority/src/snapshot.rs

@@ -38,6 +38,161 @@ pub enum SnapshotError {
     #[error("snapshot IO failed")]
     Io(#[from] std::io::Error),
 }
+/// Fixed capability declaration. Consumers validate these untrusted bytes before decoding.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct CapabilityManifest {
+    pub bytes: [u8; 56],
+}
+impl CapabilityManifest {
+    pub fn encode(&self) -> [u8; 56] {
+        self.bytes
+    }
+
+    pub fn decode(bytes: &[u8]) -> Result<Self, SnapshotError> {
+        format(bytes.len() == 56, "capability length")?;
+        format(&bytes[..6] == b"CCAP\x01\x01", "capability version/framing")?;
+        format(bytes[6..14] == 6u64.to_be_bytes(), "capability schema")?;
+        format(bytes[14..16] == [1, 1], "operation codec")?;
+        let mut encoded = [0; 56];
+        encoded.copy_from_slice(bytes);
+        Ok(Self { bytes: encoded })
+    }
+
+    fn payload_length(&self) -> u64 {
+        let mut bytes = [0; 8];
+        bytes.copy_from_slice(&self.bytes[16..24]);
+        u64::from_be_bytes(bytes)
+    }
+}
+
+/// Manifest and payload form one explicit input. This does not grant installation authority.
+pub struct CapabilityArtifact<T> {
+    pub manifest: CapabilityManifest,
+    pub payload: T,
+}
+
+pub(crate) struct CapabilityOutput<W> {
+    io: W,
+    hash: Sha256,
+    used: u64,
+    max: u64,
+}
+impl<W: Write> CapabilityOutput<W> {
+    pub(crate) fn new(io: W, max: u64) -> Self {
+        Self {
+            io,
+            hash: Sha256::new(),
+            used: 0,
+            max,
+        }
+    }
+
+    pub(crate) fn finish(self) -> CapabilityArtifact<W> {
+        let mut bytes = [0; 56];
+        bytes[..6].copy_from_slice(b"CCAP\x01\x01");
+        bytes[6..14].copy_from_slice(&6u64.to_be_bytes());
+        bytes[14..16].copy_from_slice(&[1, 1]);
+        bytes[16..24].copy_from_slice(&self.used.to_be_bytes());
+        bytes[24..].copy_from_slice(&self.hash.finalize());
+        CapabilityArtifact {
+            manifest: CapabilityManifest { bytes },
+            payload: self.io,
+        }
+    }
+}
+impl<W: Write> Write for CapabilityOutput<W> {
+    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
+        let requested = u64::try_from(bytes.len()).map_err(std::io::Error::other)?;
+        if self.used.checked_add(requested).is_none_or(|n| n > self.max) {
+            return Err(std::io::Error::other("capability payload limit"));
+        }
+        let written = self.io.write(bytes)?;
+        let part = bytes
+            .get(..written)
+            .ok_or_else(|| std::io::Error::other("invalid writer count"))?;
+        let count = u64::try_from(written).map_err(std::io::Error::other)?;
+        self.used = self
+            .used
+            .checked_add(count)
+            .ok_or_else(|| std::io::Error::other("capability length overflow"))?;
+        self.hash.update(part);
+        Ok(written)
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        self.io.flush()
+    }
+}
+
+struct CapabilityInput<R> {
+    io: R,
+    hash: Sha256,
+    used: u64,
+    max: u64,
+}
+impl<R: Read> Read for CapabilityInput<R> {
+    fn read(&mut self, bytes: &mut [u8]) -> std::io::Result<usize> {
+        // At the boundary, permit only the existing decoder's one-byte EOF probe.
+        let remaining = self
+            .max
+            .checked_sub(self.used)
+            .ok_or_else(|| std::io::Error::other("capability length overflow"))?;
+        let requested = u64::try_from(bytes.len()).map_err(std::io::Error::other)?;
+        let allowed = requested.min(if remaining == 0 { 1 } else { remaining });
+        let allowed = usize::try_from(allowed).map_err(std::io::Error::other)?;
+        let read = self.io.read(&mut bytes[..allowed])?;
+        let part = bytes
+            .get(..read)
+            .ok_or_else(|| std::io::Error::other("invalid reader count"))?;
+        let count = u64::try_from(read).map_err(std::io::Error::other)?;
+        let used = self
+            .used
+            .checked_add(count)
+            .ok_or_else(|| std::io::Error::other("capability length overflow"))?;
+        if used > self.max {
+            return Err(std::io::Error::other("capability payload limit"));
+        }
+        self.used = used;
+        self.hash.update(part);
+        Ok(read)
+    }
+}
+
+struct CapabilitySink<'a, S>(&'a mut S);
+impl<S: StagingSink> StagingSink for CapabilitySink<'_, S> {
+    fn row(&mut self, section: Section, key: &[u8], value: &[u8]) -> std::io::Result<()> {
+        if section == Section::Operations {
+            crate::application::validate_provenance_encoding(value).map_err(std::io::Error::other)?;
+        }
+        self.0.row(section, key, value)
+    }
+}
+
+/// Validates framing, declared capability and operation encoding into isolated staging.
+/// Failure may leave partial staging rows, which the caller must discard.
+/// Success does not prove cross-table consistency, historical truth or source authority.
+pub fn decode_capability_artifact(
+    artifact: CapabilityArtifact<impl Read>,
+    sink: &mut impl StagingSink,
+    limits: Limits,
+) -> Result<Header, SnapshotError> {
+    let manifest = CapabilityManifest::decode(&artifact.manifest.bytes)?;
+    let length = manifest.payload_length();
+    limit(length <= limits.total_bytes, "capability payload")?;
+    let mut reader = CapabilityInput {
+        io: artifact.payload,
+        hash: Sha256::new(),
+        used: 0,
+        max: length,
+    };
+    let header = decode(&mut reader, &mut CapabilitySink(sink), limits)?;
+    format(reader.used == length, "capability payload length")?;
+    if reader.hash.finalize().as_slice() != &manifest.bytes[24..] {
+        return Err(SnapshotError::Digest);
+    }
+    Ok(header)
+}
+
 /// 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> {

crates/ecstore/src/api/mod.rs

@@ -372,15 +372,6 @@ pub mod config {
             with_server_config_write_lock,
         };
 
-        /// Observe persisted config bytes and their reader-associated metadata within a byte limit.
-        pub async fn read_config_observation(
-            api: std::sync::Arc<crate::store::ECStore>,
-            file: &str,
-            max_bytes: usize,
-        ) -> crate::error::Result<(Vec<u8>, crate::object_api::ObjectInfo)> {
-            crate::config::com::read_config_observation(api, file, max_bytes).await
-        }
-
         /// Read config bytes and metadata without treating an empty payload as absent.
         pub async fn read_config_preserve_empty_with_metadata_opts(
             api: std::sync::Arc<crate::store::ECStore>,

crates/ecstore/src/config/com.rs

@@ -449,22 +449,6 @@ where
     Ok(data)
 }
 
-pub(crate) async fn read_config_observation<S>(api: Arc<S>, file: &str, max_bytes: usize) -> Result<(Vec<u8>, ObjectInfo)>
-where
-    S: EcstoreObjectIO,
-{
-    let opts = ObjectOptions {
-        suppress_read_repair: true,
-        no_lock: false,
-        skip_verify_bitrot: false,
-        ..Default::default()
-    };
-    crate::object_api::without_get_object_body_cache_hook(read_config_limited_preserve_empty_with_metadata_opts(
-        api, file, &opts, max_bytes,
-    ))
-    .await
-}
-
 pub(crate) async fn read_config_limited_preserve_empty<S>(api: Arc<S>, file: &str, max_bytes: usize) -> Result<Vec<u8>>
 where
     S: EcstoreObjectIO,
@@ -5441,166 +5425,6 @@ mod tests {
         assert_eq!(store.write_calls.load(Ordering::SeqCst), 0);
     }
 
-    struct ObservationStream {
-        inner: Cursor<Vec<u8>>,
-        consumed: Arc<AtomicUsize>,
-        fail: bool,
-    }
-
-    impl AsyncRead for ObservationStream {
-        fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
-            if self.fail {
-                return Poll::Ready(Err(std::io::Error::other("observation stream failed")));
-            }
-            let before = buf.filled().len();
-            let result = Pin::new(&mut self.inner).poll_read(cx, buf);
-            self.consumed.fetch_add(buf.filled().len() - before, Ordering::SeqCst);
-            result
-        }
-    }
-
-    struct ObservationStore {
-        body: Vec<u8>,
-        declared_size: i64,
-        error: Option<Error>,
-        stream_error: bool,
-        consumed: Arc<AtomicUsize>,
-        options: std::sync::Mutex<Option<(bool, bool, bool)>>,
-        writes: AtomicUsize,
-    }
-
-    impl Debug for ObservationStore {
-        fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-            f.debug_struct("ObservationStore").finish()
-        }
-    }
-
-    impl ObservationStore {
-        fn new(body: Vec<u8>, declared_size: i64) -> Self {
-            Self {
-                body,
-                declared_size,
-                error: None,
-                stream_error: false,
-                consumed: Arc::new(AtomicUsize::new(0)),
-                options: std::sync::Mutex::new(None),
-                writes: AtomicUsize::new(0),
-            }
-        }
-    }
-
-    #[async_trait::async_trait]
-    impl crate::storage_api_contracts::object::ObjectIO for ObservationStore {
-        type Error = Error;
-        type RangeSpec = HTTPRangeSpec;
-        type HeaderMap = HeaderMap;
-        type ObjectOptions = ObjectOptions;
-        type ObjectInfo = ObjectInfo;
-        type GetObjectReader = GetObjectReader;
-        type PutObjectReader = PutObjReader;
-
-        async fn get_object_reader(
-            &self,
-            bucket: &str,
-            object: &str,
-            range: Option<HTTPRangeSpec>,
-            _headers: HeaderMap,
-            opts: &ObjectOptions,
-        ) -> Result<GetObjectReader> {
-            assert_eq!(bucket, RUSTFS_META_BUCKET);
-            assert_eq!(object, "config/observation.bin");
-            assert!(range.is_none());
-            *self.options.lock().expect("observation options") =
-                Some((opts.suppress_read_repair, opts.no_lock, opts.skip_verify_bitrot));
-            if let Some(error) = &self.error {
-                return Err(error.clone());
-            }
-            Ok(GetObjectReader {
-                stream: Box::new(ObservationStream {
-                    inner: Cursor::new(self.body.clone()),
-                    consumed: self.consumed.clone(),
-                    fail: self.stream_error,
-                }),
-                object_info: ObjectInfo {
-                    size: self.declared_size,
-                    etag: Some("observation-etag".to_string()),
-                    ..Default::default()
-                },
-                buffered_body: None,
-                body_source: Default::default(),
-            })
-        }
-
-        async fn put_object(
-            &self,
-            _bucket: &str,
-            _object: &str,
-            _data: &mut PutObjReader,
-            _opts: &ObjectOptions,
-        ) -> Result<ObjectInfo> {
-            self.writes.fetch_add(1, Ordering::SeqCst);
-            Err(Error::other("observation must not write"))
-        }
-    }
-
-    #[tokio::test]
-    async fn observation_preserves_options_and_errors() {
-        for body in [Vec::new(), b"raw config".to_vec()] {
-            let store = Arc::new(ObservationStore::new(body.clone(), i64::try_from(body.len()).expect("fixture size")));
-            let result = super::read_config_observation(store.clone(), "config/observation.bin", 32).await;
-            assert!(result.is_ok(), "observation must preserve empty and nonempty objects: {result:?}");
-            let (observed, info) = result.expect("successful observation");
-            assert_eq!(observed, body);
-            assert_eq!(info.etag.as_deref(), Some("observation-etag"));
-            assert_eq!(
-                *store.options.lock().expect("options"),
-                Some((true, false, false)),
-                "I05_REPAIR_SUPPRESSED"
-            );
-            assert_eq!(store.writes.load(Ordering::SeqCst), 0);
-        }
-        for (source, expected) in [
-            (Error::FileNotFound, Error::ConfigNotFound),
-            (
-                Error::ObjectNotFound(RUSTFS_META_BUCKET.to_string(), "config/observation.bin".to_string()),
-                Error::ConfigNotFound,
-            ),
-            (Error::ErasureReadQuorum, Error::ErasureReadQuorum),
-        ] {
-            let mut fixture = ObservationStore::new(Vec::new(), 0);
-            fixture.error = Some(source);
-            let store = Arc::new(fixture);
-            let result = super::read_config_observation(store.clone(), "config/observation.bin", 32).await;
-            assert_eq!(result.expect_err("read error must propagate"), expected);
-            assert_eq!(store.writes.load(Ordering::SeqCst), 0);
-        }
-        let mut fixture = ObservationStore::new(Vec::new(), 0);
-        fixture.stream_error = true;
-        let store = Arc::new(fixture);
-        let result = super::read_config_observation(store.clone(), "config/observation.bin", 32).await;
-        let error = result.expect_err("stream error must propagate");
-        assert!(error.to_string().contains("observation stream failed"));
-        assert_eq!(store.writes.load(Ordering::SeqCst), 0);
-    }
-
-    #[tokio::test]
-    async fn observation_enforces_byte_limit() {
-        let store = Arc::new(ObservationStore::new(vec![7; 64], 0));
-        let result = super::read_config_observation(store.clone(), "config/observation.bin", 8).await;
-        assert_eq!(result.is_err(), true, "I05_BOUND_ENFORCED");
-        assert_eq!(
-            result.expect_err("bounded observation error"),
-            Error::CorruptedFormat,
-            "I05_BOUND_ENFORCED"
-        );
-        assert_eq!(store.consumed.load(Ordering::SeqCst), 9, "I05_BOUND_ENFORCED: drain exactly limit + 1");
-        let store = Arc::new(ObservationStore::new(vec![7; 8], 8));
-        let result = super::read_config_observation(store.clone(), "config/observation.bin", 8).await;
-        assert!(result.is_ok(), "exact byte limit must succeed: {result:?}");
-        assert_eq!(result.expect("exact limit").0, vec![7; 8]);
-        assert_eq!(store.consumed.load(Ordering::SeqCst), 8);
-    }
-
     #[tokio::test]
     async fn read_config_preserve_empty_distinguishes_empty_object() {
         let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(Vec::new()), None));

crates/ecstore/src/set_disk/mod.rs

@@ -1095,217 +1095,6 @@ mod prepared_get_object_metadata_tests {
     use http::HeaderMap;
     use tokio::io::AsyncReadExt;
 
-    const OBSERVATION_HOOK_BODY_LEN: usize = 1024 * 1024 + 1;
-
-    struct ObservationBodyHook {
-        object: &'static str,
-        calls: Arc<AtomicU64>,
-        body: bytes::Bytes,
-    }
-
-    #[async_trait::async_trait]
-    impl crate::object_api::GetObjectBodyCacheHook for ObservationBodyHook {
-        async fn lookup(&self, bucket: &str, object: &str, info: &ObjectInfo) -> Option<bytes::Bytes> {
-            if bucket == crate::disk::RUSTFS_META_BUCKET && object == self.object {
-                assert_eq!(info.size, i64::try_from(self.body.len()).expect("hook fixture size"));
-                self.calls.fetch_add(1, Ordering::SeqCst);
-                Some(self.body.clone())
-            } else {
-                None
-            }
-        }
-    }
-
-    struct ObservationHookGuard;
-
-    impl Drop for ObservationHookGuard {
-        fn drop(&mut self) {
-            crate::object_api::unregister_get_object_body_cache_hook();
-        }
-    }
-
-    async fn ordinary_observation_get(set: &SetDisks, object: &str) -> Vec<u8> {
-        let mut reader = set
-            .get_object_reader(crate::disk::RUSTFS_META_BUCKET, object, None, HeaderMap::new(), &ObjectOptions::default())
-            .await
-            .expect("ordinary GET positive control");
-        let mut bytes = Vec::new();
-        reader.stream.read_to_end(&mut bytes).await.expect("ordinary GET body");
-        bytes
-    }
-
-    fn observation_register_hook(object: &'static str, body: Vec<u8>) -> (ObservationHookGuard, Arc<AtomicU64>) {
-        let calls = Arc::new(AtomicU64::new(0));
-        crate::object_api::register_get_object_body_cache_hook(Arc::new(ObservationBodyHook {
-            object,
-            calls: calls.clone(),
-            body: bytes::Bytes::from(body),
-        }));
-        (ObservationHookGuard, calls)
-    }
-
-    #[tokio::test]
-    #[serial_test::serial(body_cache_hook)]
-    async fn observation_real_backend_boundaries() {
-        let object = "config/observation-real.bin";
-        let payload = b"persisted config";
-        let (_dirs, set) = make_local_set_disks(4, 2).await;
-        let mut reader = PutObjReader::from_vec(payload.to_vec());
-        let written = set
-            .put_object(crate::disk::RUSTFS_META_BUCKET, object, &mut reader, &ObjectOptions::default())
-            .await
-            .expect("persist real config fixture");
-        let result = crate::config::com::read_config_observation(set.clone(), object, payload.len()).await;
-        assert_eq!(result.is_ok(), true, "I05_OBSERVATION_SUCCESS: {result:?}");
-        let (bytes, info) = result.expect("real observation success");
-        assert_eq!(bytes, payload);
-        assert_eq!(info.size, written.size);
-        assert_eq!(info.etag, written.etag);
-        assert_eq!(info.data_dir, written.data_dir);
-        assert_eq!(info.name, object);
-        let over = crate::config::com::read_config_observation(set.clone(), object, payload.len() - 1).await;
-        assert_eq!(over.is_err(), true, "I05_BOUND_ENFORCED");
-        assert_eq!(over.expect_err("over limit must fail"), Error::CorruptedFormat, "I05_BOUND_ENFORCED");
-        let absent = crate::config::com::read_config_observation(set.clone(), "config/observation-absent.bin", 32).await;
-        assert_eq!(absent.expect_err("absent config"), Error::ConfigNotFound);
-        let empty_object = "config/observation-empty.bin";
-        let mut empty = PutObjReader::from_vec(Vec::new());
-        set.put_object(crate::disk::RUSTFS_META_BUCKET, empty_object, &mut empty, &ObjectOptions::default())
-            .await
-            .expect("persist empty config");
-        let empty = crate::config::com::read_config_observation(set, empty_object, 0).await;
-        assert!(empty.is_ok(), "existing empty config must succeed: {empty:?}");
-        let (bytes, info) = empty.expect("empty observation");
-        assert!(bytes.is_empty());
-        assert_eq!(info.size, 0);
-    }
-
-    #[tokio::test]
-    #[serial_test::serial(body_cache_hook)]
-    async fn observation_bypasses_registered_body_hook() {
-        let object = "config/observation-hook.bin";
-        let (_dirs, set) = make_local_set_disks(4, 2).await;
-        let mut reader = PutObjReader::from_vec(vec![b'p'; OBSERVATION_HOOK_BODY_LEN]);
-        set.put_object(crate::disk::RUSTFS_META_BUCKET, object, &mut reader, &ObjectOptions::default())
-            .await
-            .expect("persist real config fixture");
-        let (_hook, calls) = observation_register_hook(object, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
-        assert_eq!(ordinary_observation_get(&set, object).await, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
-        assert_eq!(calls.swap(0, Ordering::SeqCst), 1);
-        let result = crate::config::com::read_config_observation(set, object, OBSERVATION_HOOK_BODY_LEN).await;
-        assert_eq!(result.is_ok(), true, "I05_HOOK_BYPASS: {result:?}");
-        assert_eq!(
-            result.expect("observed persisted body").0,
-            vec![b'p'; OBSERVATION_HOOK_BODY_LEN],
-            "I05_HOOK_BYPASS"
-        );
-        assert_eq!(calls.load(Ordering::SeqCst), 0, "I05_HOOK_BYPASS");
-    }
-
-    #[tokio::test]
-    #[serial_test::serial(body_cache_hook)]
-    async fn observation_hook_scope_restores_after_error() {
-        let object = "config/observation-hook-error.bin";
-        let (_dirs, set) = make_local_set_disks(4, 2).await;
-        let mut reader = PutObjReader::from_vec(vec![b'p'; OBSERVATION_HOOK_BODY_LEN]);
-        set.put_object(crate::disk::RUSTFS_META_BUCKET, object, &mut reader, &ObjectOptions::default())
-            .await
-            .expect("persist real config fixture");
-        let (_hook, calls) = observation_register_hook(object, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
-        let result = crate::config::com::read_config_observation(set.clone(), object, OBSERVATION_HOOK_BODY_LEN - 1).await;
-        assert_eq!(result.expect_err("bounded observation error"), Error::CorruptedFormat);
-        assert_eq!(calls.load(Ordering::SeqCst), 0);
-        assert_eq!(ordinary_observation_get(&set, object).await, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
-        assert_eq!(calls.load(Ordering::SeqCst), 1);
-    }
-
-    struct PendingObservationStore {
-        set: Arc<SetDisks>,
-        ready: std::sync::Mutex<Option<tokio::sync::oneshot::Sender<()>>>,
-    }
-
-    impl std::fmt::Debug for PendingObservationStore {
-        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-            f.debug_struct("PendingObservationStore").finish()
-        }
-    }
-
-    #[async_trait::async_trait]
-    impl crate::storage_api_contracts::object::ObjectIO for PendingObservationStore {
-        type Error = Error;
-        type RangeSpec = HTTPRangeSpec;
-        type HeaderMap = HeaderMap;
-        type ObjectOptions = ObjectOptions;
-        type ObjectInfo = ObjectInfo;
-        type GetObjectReader = GetObjectReader;
-        type PutObjectReader = PutObjReader;
-
-        async fn get_object_reader(
-            &self,
-            bucket: &str,
-            object: &str,
-            range: Option<HTTPRangeSpec>,
-            headers: HeaderMap,
-            opts: &ObjectOptions,
-        ) -> Result<GetObjectReader> {
-            let mut reader = self.set.get_object_reader(bucket, object, range, headers, opts).await?;
-            let mut body = Vec::new();
-            reader.stream.read_to_end(&mut body).await.expect("pending producer real GET");
-            assert_eq!(
-                body,
-                vec![b'p'; OBSERVATION_HOOK_BODY_LEN],
-                "I05_HOOK_BYPASS: producer must enter its own scope"
-            );
-            self.ready
-                .lock()
-                .expect("pending signal")
-                .take()
-                .expect("one GET")
-                .send(())
-                .expect("producer entry signal");
-            std::future::pending::<Result<GetObjectReader>>().await
-        }
-
-        async fn put_object(
-            &self,
-            _bucket: &str,
-            _object: &str,
-            _data: &mut PutObjReader,
-            _opts: &ObjectOptions,
-        ) -> Result<ObjectInfo> {
-            assert!(false, "observation must not PUT");
-            Err(Error::other("unexpected observation PUT"))
-        }
-    }
-
-    #[tokio::test]
-    #[serial_test::serial(body_cache_hook)]
-    async fn observation_hook_scope_restores_after_cancel() {
-        let object = "config/observation-hook-cancel.bin";
-        let (_dirs, set) = make_local_set_disks(4, 2).await;
-        let mut reader = PutObjReader::from_vec(vec![b'p'; OBSERVATION_HOOK_BODY_LEN]);
-        set.put_object(crate::disk::RUSTFS_META_BUCKET, object, &mut reader, &ObjectOptions::default())
-            .await
-            .expect("persist real config fixture");
-        let (_hook, calls) = observation_register_hook(object, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
-        assert_eq!(ordinary_observation_get(&set, object).await, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
-        assert_eq!(calls.swap(0, Ordering::SeqCst), 1);
-        let (ready_tx, ready_rx) = tokio::sync::oneshot::channel();
-        let adapter = Arc::new(PendingObservationStore {
-            set: set.clone(),
-            ready: std::sync::Mutex::new(Some(ready_tx)),
-        });
-        let mut scope = Box::pin(crate::config::com::read_config_observation(adapter, object, OBSERVATION_HOOK_BODY_LEN));
-        tokio::select! {
-            result = ready_rx => { result.expect("producer reached pending GET"); }
-            result = &mut scope => { assert!(false, "producer must remain pending until cancelled: {result:?}"); }
-        }
-        assert_eq!(calls.load(Ordering::SeqCst), 0);
-        drop(scope);
-        assert_eq!(ordinary_observation_get(&set, object).await, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
-        assert_eq!(calls.load(Ordering::SeqCst), 1);
-    }
-
     const READ_VERSION_BARRIER_GUARD: std::time::Duration = std::time::Duration::from_secs(10);
 
     fn object_with_initial_data_shards(bucket: &str, prefix: &str) -> String {