fix(capture): validate checkpoint cut against captured committed log

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,804 @@ 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 u27b_rows(bytes: Vec<u8>, expected: &[Vec<(Vec<u8>, Vec<u8>)>], cut: LogId, member: LogId) {
+        let mut sink = U27aSink::default();
+        let header = snapshot::decode(io::Cursor::new(bytes), &mut sink, u24_limits()).expect("decode captured bytes");
+        assert_eq!(header.applied, Some(cut), "U27B_SAME_TX");
+        assert_eq!(header.membership, u17a_membership(member.index).1, "U27B_SAME_TX");
+        assert_eq!(header.cluster_id, identity().cluster_id, "U27B_SAME_TX");
+        assert_eq!(header.binding, u17a_binding(), "U27B_SAME_TX");
+        for (section, rows) in [
+            snapshot::Section::Heads,
+            snapshot::Section::Versions,
+            snapshot::Section::Events,
+            snapshot::Section::Operations,
+        ]
+        .into_iter()
+        .zip(&expected[3..7])
+        {
+            let actual: Vec<_> = sink
+                .0
+                .iter()
+                .filter(|row| row.0 == section)
+                .map(|row| (row.1.clone(), row.2.clone()))
+                .collect();
+            assert_eq!(&actual, rows, "U27B_SAME_TX");
+        }
+    }
+
+    #[test]
+    fn u27b_checkpoint_same_tx_cut() {
+        let (_dir, _backend, store, _) = u27a_fixture();
+        let before = u17b_snapshot(&store);
+        let writer = NativeLogWriter::new(store);
+        let owner = writer.store.clone();
+        let _hook = native_snapshot::after_capture(move || {
+            let mut store = owner.lock().expect("live corruption owner");
+            let tx = store.write_transaction().expect("live corruption transaction");
+            tx.open_table(META)
+                .expect("live metadata")
+                .remove("committed")
+                .expect("remove live marker");
+            store.commit(tx).expect("durable live corruption");
+            assert!(
+                matches!(store.read_committed(), Err(StoreError::CorruptRecord)),
+                "live transaction sees missing committed"
+            );
+        });
+        let old = NativeSnapshotView::acquire(writer.store.clone());
+        assert_eq!(old.is_ok(), true, "U27B_SAME_TX");
+        u27b_rows(
+            old.expect("asserted old view")
+                .stream(Vec::new(), u24_limits())
+                .expect("old stream"),
+            &before,
+            u17a_id(10),
+            u17a_id(1),
+        );
+        assert!(
+            matches!(
+                NativeSnapshotView::acquire(writer.store.clone()),
+                Err(NativeSnapshotError::Store(StoreError::CorruptRecord))
+            ),
+            "U27B_SAME_TX"
+        );
+        drop(_hook);
+
+        let (_dir2, _backend2, store, _) = u27a_fixture();
+        let before = u17b_snapshot(&store);
+        let writer = NativeLogWriter::new(store);
+        let old = NativeSnapshotView::acquire(writer.store.clone()).expect("normal old view");
+        let after = {
+            let mut store = writer.store.lock().expect("normal writer");
+            let mut command = u17a_command();
+            command.operation_id = [81; 16];
+            command.object.key = b"u27b-next-object".to_vec();
+            command.prepared.version_id = [82; 16];
+            u24_apply(&mut store, &[u17a_membership(11).0, u17a_created(12, command)]);
+            u17b_snapshot(&store)
+        };
+        for i in 3..7 {
+            assert_ne!(before[i], after[i], "U27B_SAME_TX new table changes");
+        }
+        u27b_rows(
+            old.stream(Vec::new(), u24_limits()).expect("normal old stream"),
+            &before,
+            u17a_id(10),
+            u17a_id(1),
+        );
+        u27b_rows(
+            NativeSnapshotView::acquire(writer.store.clone())
+                .expect("normal fresh view")
+                .stream(Vec::new(), u24_limits())
+                .expect("normal fresh stream"),
+            &after,
+            u17a_id(12),
+            u17a_id(11),
+        );
+    }
+
+    #[test]
+    fn u27b_checkpoint_bounds() {
+        for case in 0..4 {
+            let (_dir, _backend, mut store, _) = u27a_fixture();
+            let tx = store.write_transaction().expect("bounds fixture transaction");
+            {
+                let mut meta = tx.open_table(META).expect("bounds metadata");
+                match case {
+                    0 => {
+                        meta.insert("committed", encode_id(u17a_id(9)).as_slice())
+                            .expect("committed below applied");
+                    }
+                    1 => {
+                        let mut applied = vec![1];
+                        applied.extend(encode_id(LogId { term: 12, ..u17a_id(10) }));
+                        meta.insert("last_applied", applied.as_slice())
+                            .expect("applied term mismatch");
+                    }
+                    2 => {
+                        let mut applied = vec![1];
+                        applied.extend(encode_id(LogId {
+                            leader_node: 8,
+                            ..u17a_id(10)
+                        }));
+                        meta.insert("last_applied", applied.as_slice())
+                            .expect("applied leader mismatch");
+                    }
+                    _ => {
+                        meta.remove("committed").expect("missing committed fixture");
+                    }
+                }
+            }
+            store.commit(tx).expect("persist bounds fixture");
+            let expected_applied = match case {
+                1 => LogId { term: 12, ..u17a_id(10) },
+                2 => LogId {
+                    leader_node: 8,
+                    ..u17a_id(10)
+                },
+                _ => u17a_id(10),
+            };
+            assert_eq!(
+                store.native_application_state().expect("valid applied encoding").0,
+                Some(expected_applied),
+                "U27B_FULL_CUT decoded fixture"
+            );
+            let before = u17b_snapshot(&store);
+            let writer = NativeLogWriter::new(store);
+            let result = NativeSnapshotView::acquire(writer.store.clone());
+            assert_eq!(
+                matches!(result, Err(NativeSnapshotError::Store(StoreError::CorruptRecord))),
+                true,
+                "U27B_FULL_CUT case {case}"
+            );
+            assert_eq!(
+                u17b_snapshot(&writer.store.lock().expect("unchanged bounds owner")),
+                before,
+                "U27B_FULL_CUT zero writes"
+            );
+        }
+        let (_dir, _backend, store, _) = u27a_fixture();
+        let writer = NativeLogWriter::new(store);
+        assert!(NativeSnapshotView::acquire(writer.store.clone()).is_ok(), "U27B_FULL_CUT equal full ID");
+        u19_runtime()
+            .block_on(writer.write(|store| {
+                store.append_native(&[u17a_blank(11)])?;
+                store.save_committed(u17a_id(11))
+            }))
+            .expect("advance committed only");
+        assert!(
+            NativeSnapshotView::acquire(writer.store.clone()).is_ok(),
+            "U27B_FULL_CUT applied below committed"
+        );
+
+        for schema6 in [false, true] {
+            let (_dir, _backend, mut store) = u17a_store();
+            if schema6 {
+                provenance_promote(&mut store);
+            }
+            let schema = if schema6 { 6 } else { 4 };
+            let before = u17b_snapshot(&store);
+            let writer = NativeLogWriter::new(store);
+            let bytes = NativeSnapshotView::acquire(writer.store.clone())
+                .expect("legal empty native view")
+                .stream(Vec::new(), u24_limits())
+                .expect("empty bare stream");
+            let mut sink = U27aSink::default();
+            let header = snapshot::decode(io::Cursor::new(bytes), &mut sink, u24_limits()).expect("empty decode");
+            assert_eq!(header.applied, None, "U27B_FULL_CUT empty applied");
+            assert!(
+                header.membership.log_id().is_none()
+                    && header.membership.configs().is_empty()
+                    && header.membership.nodes().is_empty(),
+                "U27B_FULL_CUT empty membership"
+            );
+            assert!(sink.0.is_empty(), "U27B_FULL_CUT empty rows");
+            let capability = NativeSnapshotView::acquire(writer.store.clone())
+                .expect("empty capability view")
+                .stream_capability_artifact(Vec::new(), u24_limits());
+            if schema6 {
+                assert!(capability.is_ok(), "U27B_FULL_CUT explicit schema6");
+            } else {
+                assert!(
+                    matches!(capability, Err(NativeSnapshotError::Store(StoreError::UnsupportedSchema))),
+                    "U27B_FULL_CUT schema4 gate"
+                );
+            }
+            assert_eq!(
+                committed_schema(&writer.store.lock().expect("schema owner")),
+                schema,
+                "U27B_FULL_CUT no promotion"
+            );
+            assert_eq!(
+                u17b_snapshot(&writer.store.lock().expect("empty unchanged owner")),
+                before,
+                "U27B_FULL_CUT zero writes"
+            );
+            u19_runtime()
+                .block_on(writer.write(|store| {
+                    store.append_native(&[u17a_blank(0)])?;
+                    store.save_committed(u17a_id(0))
+                }))
+                .expect("empty applied with committed log");
+            assert!(
+                NativeSnapshotView::acquire(writer.store.clone()).is_ok(),
+                "U27B_FULL_CUT committed unapplied accepted"
+            );
+        }
+        let (_dir, _backend, mut store) = u17a_store();
+        let tx = store.write_transaction().expect("missing applied fixture");
+        tx.open_table(META)
+            .expect("empty metadata")
+            .remove("last_applied")
+            .expect("remove empty applied encoding");
+        store.commit(tx).expect("commit missing applied");
+        let writer = NativeLogWriter::new(store);
+        assert!(
+            matches!(
+                NativeSnapshotView::acquire(writer.store.clone()),
+                Err(NativeSnapshotError::Store(StoreError::CorruptRecord))
+            ),
+            "U27B_FULL_CUT missing empty encoding"
+        );
+    }
+
+    #[test]
+    fn u27b_checkpoint_retained_log_identity() {
+        for case in 0..5 {
+            let (_dir, _backend, mut store, _) = u27a_fixture();
+            let tx = store.write_transaction().expect("retained log fixture");
+            {
+                let mut logs = tx.open_table(LOGS).expect("retained log table");
+                match case {
+                    0 => {
+                        logs.remove(10).expect("missing committed log");
+                    }
+                    1 => {
+                        logs.insert(10, [0xff].as_slice()).expect("malformed committed log");
+                    }
+                    2 | 3 => {
+                        let raw = logs
+                            .get(10)
+                            .expect("original committed log")
+                            .expect("retained log")
+                            .value()
+                            .to_vec();
+                        let mut entry = decode_log(&raw).expect("valid original log");
+                        if case == 2 {
+                            entry.id.term += 1;
+                        } else {
+                            entry.id.leader_node += 1;
+                        }
+                        logs.insert(10, encode_log(&entry).expect("encoded mismatching log").as_slice())
+                            .expect("mismatching committed log");
+                    }
+                    _ => {}
+                }
+            }
+            store.commit(tx).expect("persist retained log fixture");
+            let before = u17b_snapshot(&store);
+            let writer = NativeLogWriter::new(store);
+            let result = NativeSnapshotView::acquire(writer.store.clone());
+            if case == 4 {
+                assert!(result.is_ok(), "U27B_COMMITTED_RECORD complete identity");
+            } else {
+                assert_eq!(
+                    matches!(result, Err(NativeSnapshotError::Store(StoreError::CorruptRecord))),
+                    true,
+                    "U27B_COMMITTED_RECORD case {case}"
+                );
+            }
+            assert_eq!(
+                u17b_snapshot(&writer.store.lock().expect("retained log owner")),
+                before,
+                "U27B_COMMITTED_RECORD zero writes"
+            );
+        }
+    }
+
+    #[test]
+    fn u27b_checkpoint_owner_and_read_lifetime() {
+        let (_dir, _backend, store, _) = u27a_fixture();
+        let before = u17b_snapshot(&store);
+        let writer = NativeLogWriter::new(store);
+        let view = NativeSnapshotView::acquire(writer.store.clone()).expect("owned captured view");
+        drop(writer);
+        u27b_rows(
+            view.stream(Vec::new(), u24_limits()).expect("view outlives external writer"),
+            &before,
+            u17a_id(10),
+            u17a_id(1),
+        );
+
+        struct PausedOutput {
+            entered: Option<mpsc::Sender<()>>,
+            release: mpsc::Receiver<()>,
+            bytes: Vec<u8>,
+        }
+        impl Write for PausedOutput {
+            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)?;
+                }
+                self.bytes.extend_from_slice(bytes);
+                Ok(bytes.len())
+            }
+            fn flush(&mut self) -> io::Result<()> {
+                Ok(())
+            }
+        }
+        let (_dir2, _backend2, store, _) = u27a_fixture();
+        let before = u17b_snapshot(&store);
+        let writer = NativeLogWriter::new(store);
+        let view = NativeSnapshotView::acquire(writer.store.clone()).expect("paused view");
+        let (entered_tx, entered_rx) = mpsc::channel();
+        let (release_tx, release_rx) = mpsc::channel();
+        let stream = std::thread::spawn(move || {
+            view.stream(
+                PausedOutput {
+                    entered: Some(entered_tx),
+                    release: release_rx,
+                    bytes: Vec::new(),
+                },
+                u24_limits(),
+            )
+        });
+        let entered = entered_rx.recv_timeout(Duration::from_secs(5));
+        let owner_available = writer.store.try_lock().is_ok();
+        let (done_tx, done_rx) = mpsc::channel();
+        let writer_for_thread = writer.clone();
+        let write = std::thread::spawn(move || {
+            let result = u19_runtime().block_on(writer_for_thread.write(|store| {
+                u24_apply(store, &[u17a_blank(11)]);
+                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("paused stream joined");
+        write.join().expect("writer joined after release");
+        assert!(entered.is_ok(), "U27B_OWNER output rendezvous");
+        assert!(owner_available, "U27B_OWNER owner try_lock while output paused: {owner_available}");
+        assert!(
+            matches!(done, Ok(true)),
+            "U27B_OWNER writer ACK before release: done={done:?}, owner_available={owner_available}"
+        );
+        u27b_rows(streamed.expect("paused stream completed").bytes, &before, u17a_id(10), u17a_id(1));
+        struct Broken;
+        impl Write for Broken {
+            fn write(&mut self, _: &[u8]) -> io::Result<usize> {
+                Err(io::Error::new(io::ErrorKind::BrokenPipe, "u27b output failure"))
+            }
+            fn flush(&mut self) -> io::Result<()> {
+                Ok(())
+            }
+        }
+        let result = NativeSnapshotView::acquire(writer.store.clone())
+            .expect("failed output view")
+            .stream(Broken, u24_limits());
+        assert!(
+            matches!(result, Err(NativeSnapshotError::Snapshot(snapshot::SnapshotError::Io(ref error))) if error.kind() == io::ErrorKind::BrokenPipe),
+            "U27B_OWNER typed output failure"
+        );
+        u19_runtime()
+            .block_on(writer.write(|store| {
+                u24_apply(store, &[u17a_blank(12)]);
+                Ok(())
+            }))
+            .expect("write after failed output");
+        assert!(
+            NativeSnapshotView::acquire(writer.store.clone())
+                .expect("capture after failed output")
+                .stream(Vec::new(), u24_limits())
+                .is_ok(),
+            "U27B_OWNER capture after failed output"
+        );
+    }
+
     fn u24_limits() -> snapshot::Limits {
         snapshot::Limits {
             total_bytes: 1_000_000,
@@ -2364,7 +3163,7 @@ mod tests {
         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!(matches!(done, Ok(true)), "U24_STREAM_RELEASES_WRITER done={done:?}");
         assert!(streamed.is_ok(), "stream completed");
     }
     #[test]

crates/capture-authority/src/native_snapshot.rs

@@ -6,6 +6,30 @@ use std::{
     sync::{Arc, Mutex},
 };
 
+#[cfg(test)]
+std::thread_local! {
+    static AFTER_CAPTURE: std::cell::RefCell<Option<Box<dyn FnOnce()>>> = const { std::cell::RefCell::new(None) };
+}
+
+#[cfg(test)]
+pub(crate) struct CaptureHook;
+#[cfg(test)]
+impl Drop for CaptureHook {
+    fn drop(&mut self) {
+        AFTER_CAPTURE.with(|hook| {
+            hook.borrow_mut().take();
+        });
+    }
+}
+#[cfg(test)]
+pub(crate) fn after_capture(hook: impl FnOnce() + 'static) -> CaptureHook {
+    AFTER_CAPTURE.with(|slot| {
+        assert!(slot.borrow().is_none(), "one capture hook per test thread");
+        *slot.borrow_mut() = Some(Box::new(hook));
+    });
+    CaptureHook
+}
+
 #[derive(Debug, thiserror::Error)]
 pub enum NativeSnapshotError {
     #[error(transparent)]
@@ -23,6 +47,7 @@ pub struct NativeSnapshotView {
     transaction: ReadTransaction,
     header: snapshot::Header,
     counts: [u64; 4],
+    schema: u64,
     _owner: Arc<Mutex<Store>>,
 }
 impl NativeSnapshotView {
@@ -34,19 +59,40 @@ impl NativeSnapshotView {
             }
             store.database.begin_read().map_err(StoreError::from)?
         };
+        #[cfg(test)]
+        if let Some(hook) = AFTER_CAPTURE.with(|slot| slot.borrow_mut().take()) {
+            hook();
+        }
         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 logs = transaction.open_table(crate::LOGS).map_err(StoreError::from)?;
+            let committed = crate::committed::validate_committed(&meta, &logs)?;
+            if let Some(applied) = applied {
+                let committed = committed.ok_or(StoreError::CorruptRecord)?;
+                if applied.index > committed.index || (applied.index == committed.index && applied != committed) {
+                    return Err(StoreError::CorruptRecord.into());
+                }
+            }
             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 +114,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,6 +372,15 @@ 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,6 +449,22 @@ 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,
@@ -5425,6 +5441,166 @@ 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,6 +1095,217 @@ 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 {