test(capture): specify atomic populated native migration

Cargo.lock

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

crates/capture-authority/Cargo.toml

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

crates/capture-authority/src/application.rs

@@ -1,45 +1,8 @@
-fn read_native_range_in<R: std::ops::RangeBounds<u64>>(
-    tx: &redb::ReadTransaction,
-    range: R,
-) -> Result<Vec<crate::CaptureRaftEntry>, StoreError> {
-    use std::ops::Bound::{Excluded, Included, Unbounded};
-    let logs = tx.open_table(LOGS)?;
-    let start = match range.start_bound() {
-        Included(value) => *value,
-        Excluded(value) => match value.checked_add(1) {
-            Some(value) => value,
-            None => return Ok(Vec::new()),
-        },
-        Unbounded => 0,
-    };
-    let end = match range.end_bound() {
-        Included(value) => *value,
-        Excluded(value) => match value.checked_sub(1) {
-            Some(value) => value,
-            None => return Ok(Vec::new()),
-        },
-        Unbounded => u64::MAX,
-    };
-    if start > end {
-        return Ok(Vec::new());
-    }
-    let mut entries = Vec::new();
-    for row in logs.range(start..=end)? {
-        let (key, value) = row?;
-        let entry = decode_log(value.value())?;
-        if entry.id.index != key.value() {
-            return Err(StoreError::CorruptRecord);
-        }
-        entries.push(decode_native_entry(entry)?);
-    }
-    Ok(entries)
-}
-
 use crate::{
     LOGS, LogId, META, Store, StoreError, committed::validate_committed, created::*, decode_log, encode_id, encode_log,
     next_index, number,
 };
-use redb::{ReadTransaction, ReadableDatabase, ReadableTable, ReadableTableMetadata, TableDefinition, TableHandle};
+use redb::{ReadTransaction, ReadableDatabase, ReadableTable, ReadableTableMetadata, TableDefinition};
 use std::collections::BTreeMap;
 
 pub(crate) const HEADS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("heads");
@@ -89,40 +52,6 @@ fn decode_operation(b: &[u8]) -> Result<CreatedOperation, StoreError> {
         result,
     })
 }
-fn decode_provenance(b: &[u8]) -> Result<CreatedOperationProvenance, StoreError> {
-    if b.len() > MAX_OBJECT_KEY_LENGTH + 512 || b.get(..5) != Some(b"COPR\x01") {
-        return Err(StoreError::CorruptRecord);
-    }
-    let encoded_id = b.get(5..29).ok_or(StoreError::CorruptRecord)?;
-    if encoded_id == [0; 24] {
-        return Err(StoreError::CorruptRecord);
-    }
-    Ok(CreatedOperationProvenance {
-        first_decision_log_id: crate::decode_id(encoded_id)?,
-        operation: decode_operation(b.get(29..).ok_or(StoreError::CorruptRecord)?)?,
-    })
-}
-
-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)
-    } else {
-        decode_operation(b)
-    }
-}
-
-fn encode_provenance(first: LogId, operation: &[u8]) -> Result<Vec<u8>, StoreError> {
-    let mut bytes = b"COPR\x01".to_vec();
-    bytes.extend(encode_id(first));
-    bytes.extend(operation);
-    decode_provenance(&bytes)?;
-    Ok(bytes)
-}
-
 fn decode_apply(b: &[u8]) -> Result<CreatedApplyResult, StoreError> {
     let mut d = Decoder::new(b);
     let r = CreatedApplyResult {
@@ -132,7 +61,7 @@ fn decode_apply(b: &[u8]) -> Result<CreatedApplyResult, StoreError> {
     d.finish()?;
     Ok(r)
 }
-pub(crate) fn read_binding(meta: &impl ReadableTable<&'static str, &'static [u8]>) -> Result<CaptureBinding, StoreError> {
+fn read_binding(meta: &impl ReadableTable<&'static str, &'static [u8]>) -> Result<CaptureBinding, StoreError> {
     let b = meta.get("capture_binding")?.ok_or(StoreError::BindingNotPinned)?;
     let mut d = Decoder::new(b.value());
     let binding = d.binding()?;
@@ -210,12 +139,11 @@ fn apply_created(
     command: &DecideCreated,
     canonical: &[u8],
     binding: &CaptureBinding,
-    schema: u64,
 ) -> Result<CreatedResult, StoreError> {
     let mut operations = tx.open_table(OPERATIONS)?;
     let existing = operations
         .get(command.operation_id.as_slice())?
-        .map(|v| decode_stored_operation(v.value(), schema))
+        .map(|v| decode_operation(v.value()))
         .transpose()?;
     let result = if let Some(existing) = existing {
         if existing.canonical_command == canonical {
@@ -268,9 +196,6 @@ fn apply_created(
             .to_vec();
         op.extend(canonical);
         op.extend(result_bytes(&outcome));
-        if matches!(schema, 5 | 6) {
-            op = encode_provenance(id, &op)?;
-        }
         operations.insert(command.operation_id.as_slice(), op.as_slice())?;
         outcome
     };
@@ -336,16 +261,11 @@ fn entry_command(entry: &crate::LogEntry, native: bool) -> Result<DecideCreated,
 }
 
 pub(crate) fn native_application_state(tx: &ReadTransaction) -> Result<(Option<LogId>, crate::CaptureMembership), StoreError> {
-    native_application_metadata(&tx.open_table(META)?)
-}
-
-fn native_application_metadata(
-    meta: &impl ReadableTable<&'static str, &'static [u8]>,
-) -> Result<(Option<LogId>, crate::CaptureMembership), StoreError> {
-    if !matches!(number(required(meta.get("schema")?)?.value())?, 4 | 6) {
+    let meta = tx.open_table(META)?;
+    if number(required(meta.get("schema")?)?.value())? != 4 {
         return Err(StoreError::UnsupportedSchema);
     }
-    let last = read_applied(meta)?;
+    let last = read_applied(&meta)?;
     let membership = crate::CaptureMembership::decode(required(meta.get("native_membership")?)?.value())
         .map_err(|_| StoreError::CorruptRecord)?;
     if membership
@@ -359,60 +279,7 @@ fn native_application_metadata(
 
 impl Store {
     pub fn migrate_legacy_created_to_native(&mut self) -> Result<(), StoreError> {
-        let tx = self.write_transaction()?;
-        {
-            let mut meta = tx.open_table(META)?;
-            let schema = number(required(meta.get("schema")?)?.value())?;
-            if !matches!(schema, 3 | 4) {
-                return Err(StoreError::UnsupportedSchema);
-            }
-            let mut logs = tx.open_table(LOGS)?;
-            let heads = tx.open_table(HEADS)?;
-            let versions = tx.open_table(VERSIONS)?;
-            let events = tx.open_table(EVENTS)?;
-            let operations = tx.open_table(OPERATIONS)?;
-            let mut results = tx.open_table(APPLY_RESULTS)?;
-            validate_application_tables(&meta, &logs, &heads, &versions, &events, &operations, &results, false)?;
-            if schema == 4 {
-                return Ok(());
-            }
-            if meta.get("native_membership")?.is_some() {
-                return Err(StoreError::CorruptRecord);
-            }
-            // Validate the entire retained suffix before replacing any record.
-            let mut native_logs = Vec::new();
-            for row in logs.iter()? {
-                let (index, bytes) = row?;
-                let entry = decode_log(bytes.value())?;
-                let command = DecideCreated::decode(&entry.payload)?;
-                if entry.id.index != index.value() || command.encode()? != entry.payload {
-                    return Err(StoreError::CorruptRecord);
-                }
-                let native = crate::CaptureRaftEntry {
-                    log_id: entry.id.into(),
-                    payload: openraft::EntryPayload::Normal(crate::CaptureCommand::try_created(command)?),
-                };
-                native_logs.push((index.value(), encode_native_entry(&native)?));
-            }
-            let mut native_results = Vec::new();
-            for row in results.iter()? {
-                let (index, bytes) = row?;
-                let response = crate::CaptureResponse::created(decode_apply(bytes.value())?)?;
-                native_results.push((index.value(), encode_response(&response)?));
-            }
-            for (index, bytes) in native_logs {
-                logs.insert(index, bytes.as_slice())?;
-            }
-            for (index, bytes) in native_results {
-                results.insert(index, bytes.as_slice())?;
-            }
-            let membership = crate::CaptureMembership::new(None, vec![], BTreeMap::new())?;
-            meta.insert("native_membership", membership.encode()?.as_slice())?;
-            meta.insert("schema", 4u64.to_be_bytes().as_slice())?;
-            // Replay native decisions inside the same durable writer boundary.
-            validate_application_tables(&meta, &logs, &heads, &versions, &events, &operations, &results, false)?;
-        }
-        self.commit(tx)
+        Err(StoreError::UnsupportedSchema)
     }
 
     /// Explicitly opt an empty, bound store into native entry persistence.
@@ -421,11 +288,11 @@ impl Store {
         {
             let mut meta = tx.open_table(META)?;
             let schema = number(required(meta.get("schema")?)?.value())?;
-            if !matches!(schema, 3..=6) {
+            if !matches!(schema, 3 | 4) {
                 return Err(StoreError::UnsupportedSchema);
             }
             read_binding(&meta)?;
-            if matches!(schema, 4 | 6) {
+            if schema == 4 {
                 let membership = crate::CaptureMembership::decode(required(meta.get("native_membership")?)?.value())
                     .map_err(|_| StoreError::CorruptRecord)?;
                 let last = read_applied(&meta)?;
@@ -454,7 +321,7 @@ impl Store {
             }
             let membership = crate::CaptureMembership::new(None, vec![], BTreeMap::new())?;
             meta.insert("native_membership", membership.encode()?.as_slice())?;
-            meta.insert("schema", (if schema == 5 { 6u64 } else { 4u64 }).to_be_bytes().as_slice())?;
+            meta.insert("schema", 4u64.to_be_bytes().as_slice())?;
         }
         self.commit(tx)
     }
@@ -462,7 +329,7 @@ impl Store {
         let tx = self.write_transaction()?;
         {
             let mut meta = tx.open_table(META)?;
-            if !matches!(number(required(meta.get("schema")?)?.value())?, 4 | 6) {
+            if number(required(meta.get("schema")?)?.value())? != 4 {
                 return Err(StoreError::UnsupportedSchema);
             }
             if entries.is_empty() {
@@ -491,7 +358,7 @@ impl Store {
     }
     pub fn read_native_entry(&self, index: u64) -> Result<Option<crate::CaptureRaftEntry>, StoreError> {
         let tx = self.database.begin_read()?;
-        if !matches!(number(required(tx.open_table(META)?.get("schema")?)?.value())?, 4 | 6) {
+        if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
             return Err(StoreError::UnsupportedSchema);
         }
         tx.open_table(LOGS)?
@@ -505,32 +372,32 @@ impl Store {
             })
             .transpose()
     }
-    /// Read typed native entries within the requested bounds.
-    pub fn read_native_range<R: std::ops::RangeBounds<u64>>(&self, range: R) -> Result<Vec<crate::CaptureRaftEntry>, StoreError> {
-        let tx = self.database.begin_read()?;
-        if !matches!(number(required(tx.open_table(META)?.get("schema")?)?.value())?, 4 | 6) {
-            return Err(StoreError::UnsupportedSchema);
-        }
-        read_native_range_in(&tx, range)
-    }
     pub fn read_native_entries(&self, start: u64, end: u64) -> Result<Vec<crate::CaptureRaftEntry>, StoreError> {
         let tx = self.database.begin_read()?;
-        if !matches!(number(required(tx.open_table(META)?.get("schema")?)?.value())?, 4 | 6) {
+        if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
             return Err(StoreError::UnsupportedSchema);
         }
         if start > end {
             return Err(StoreError::InvalidLogSequence);
         }
-        read_native_range_in(&tx, start..end)
+        let logs = tx.open_table(LOGS)?;
+        let mut entries = Vec::new();
+        for row in logs.range(start..end)? {
+            let (key, value) = row?;
+            let entry = decode_log(value.value())?;
+            if entry.id.index != key.value() {
+                return Err(StoreError::CorruptRecord);
+            }
+            entries.push(decode_native_entry(entry)?);
+        }
+        Ok(entries)
     }
-
     pub fn apply_next_native(&mut self, id: LogId) -> Result<crate::CaptureResponse, StoreError> {
         let tx = self.write_transaction()?;
         let response;
         {
             let mut meta = tx.open_table(META)?;
-            let schema = number(required(meta.get("schema")?)?.value())?;
-            if !matches!(schema, 4 | 6) {
+            if number(required(meta.get("schema")?)?.value())? != 4 {
                 return Err(StoreError::UnsupportedSchema);
             }
             let binding = read_binding(&meta)?;
@@ -558,7 +425,7 @@ impl Store {
                 openraft::EntryPayload::Blank => crate::CaptureResponse::blank(id)?,
                 openraft::EntryPayload::Normal(command) => {
                     let canonical = command.created().encode()?;
-                    let result = apply_created(&tx, id, command.created(), &canonical, &binding, schema)?;
+                    let result = apply_created(&tx, id, command.created(), &canonical, &binding)?;
                     crate::CaptureResponse::created(CreatedApplyResult { log_id: id, result })?
                 }
                 openraft::EntryPayload::Membership(membership) => {
@@ -578,7 +445,7 @@ impl Store {
     }
     pub fn native_apply_response(&self, index: u64) -> Result<Option<crate::CaptureResponse>, StoreError> {
         let tx = self.database.begin_read()?;
-        if !matches!(number(required(tx.open_table(META)?.get("schema")?)?.value())?, 4 | 6) {
+        if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
             return Err(StoreError::UnsupportedSchema);
         }
         let results = tx.open_table(APPLY_RESULTS)?;
@@ -596,7 +463,7 @@ impl Store {
         let tx = self.write_transaction()?;
         {
             let mut meta = tx.open_table(META)?;
-            if matches!(number(required(meta.get("schema")?)?.value())?, 3..=6) {
+            if matches!(number(required(meta.get("schema")?)?.value())?, 3 | 4) {
                 if read_binding(&meta)? == binding {
                     return Ok(());
                 }
@@ -625,8 +492,7 @@ impl Store {
         let result;
         {
             let mut meta = tx.open_table(META)?;
-            let schema = number(required(meta.get("schema")?)?.value())?;
-            if !matches!(schema, 3 | 5) {
+            if number(required(meta.get("schema")?)?.value())? != 3 {
                 return Err(StoreError::UnsupportedSchema);
             }
             let binding = read_binding(&meta)?;
@@ -649,7 +515,7 @@ impl Store {
                 return Err(StoreError::InvalidApplySequence);
             }
             let command = DecideCreated::decode(&entry.payload)?;
-            result = apply_created(&tx, id, &command, &entry.payload, &binding, schema)?;
+            result = apply_created(&tx, id, &command, &entry.payload, &binding)?;
             let mut applied = encode_id(id);
             applied.extend(result_bytes(&result));
             results.insert(id.index, applied.as_slice())?;
@@ -663,7 +529,7 @@ impl Store {
     pub fn capture_binding(&self) -> Result<Option<CaptureBinding>, StoreError> {
         let tx = self.database.begin_read()?;
         let meta = tx.open_table(META)?;
-        if !matches!(number(required(meta.get("schema")?)?.value())?, 3..=6) {
+        if !matches!(number(required(meta.get("schema")?)?.value())?, 3 | 4) {
             return Ok(None);
         }
         Ok(Some(read_binding(&meta)?))
@@ -671,7 +537,7 @@ impl Store {
     pub fn last_applied(&self) -> Result<Option<LogId>, StoreError> {
         let tx = self.database.begin_read()?;
         let meta = tx.open_table(META)?;
-        if !matches!(number(required(meta.get("schema")?)?.value())?, 3..=6) {
+        if !matches!(number(required(meta.get("schema")?)?.value())?, 3 | 4) {
             return Ok(None);
         }
         read_applied(&meta)
@@ -700,87 +566,15 @@ impl Store {
         let t = tx.open_table(EVENTS)?;
         t.get(operation.as_slice())?.map(|v| decode_event(v.value())).transpose()
     }
-    /// Read the persisted first decision, without fabricating legacy provenance.
-    pub fn created_operation_provenance(&self, operation: [u8; 16]) -> Result<Option<CreatedOperationProvenance>, StoreError> {
-        let tx = self.database.begin_read()?;
-        let schema = number(required(tx.open_table(META)?.get("schema")?)?.value())?;
-        if !matches!(schema, 5 | 6) {
-            return Err(StoreError::UnsupportedSchema);
-        }
-        tx.open_table(OPERATIONS)?
-            .get(operation.as_slice())?
-            .map(|value| decode_provenance(value.value()))
-            .transpose()
-    }
-
-    /// Explicitly promote a fully verified legacy or native Created history.
-    pub fn promote_created_provenance(&mut self) -> Result<(), StoreError> {
-        let tx = self.write_transaction()?;
-        {
-            let mut meta = tx.open_table(META)?;
-            let schema = number(required(meta.get("schema")?)?.value())?;
-            if !matches!(schema, 3..=6) {
-                return Err(StoreError::UnsupportedSchema);
-            }
-            let tables = tx
-                .list_tables()?
-                .map(|table| table.name().to_owned())
-                .collect::<std::collections::BTreeSet<_>>();
-            if [
-                "meta",
-                "vote",
-                "logs",
-                "heads",
-                "versions",
-                "events",
-                "operations",
-                "apply_results",
-            ]
-            .iter()
-            .any(|name| !tables.contains(*name))
-            {
-                return Err(StoreError::CorruptRecord);
-            }
-            crate::validate_identity(crate::decode_identity(required(meta.get("identity")?)?.value())?)?;
-            for row in tx.open_table(crate::VOTE)?.iter()? {
-                let (key, value) = row?;
-                if key.value() != 0 {
-                    return Err(StoreError::CorruptRecord);
-                }
-                crate::decode_vote(value.value())?;
-            }
-            let logs = tx.open_table(LOGS)?;
-            let heads = tx.open_table(HEADS)?;
-            let versions = tx.open_table(VERSIONS)?;
-            let events = tx.open_table(EVENTS)?;
-            let mut operations = tx.open_table(OPERATIONS)?;
-            let results = tx.open_table(APPLY_RESULTS)?;
-            let first = validate_application_tables(&meta, &logs, &heads, &versions, &events, &operations, &results, true)?;
-            if matches!(schema, 5 | 6) {
-                return Ok(());
-            }
-            // Keep one encoded row at a time; the replay map contains only first LogIds.
-            for (operation, id) in first {
-                let original = required(operations.get(operation.as_slice())?)?.value().to_vec();
-                operations.insert(operation.as_slice(), encode_provenance(id, &original)?.as_slice())?;
-            }
-            meta.insert("schema", (if schema == 3 { 5u64 } else { 6u64 }).to_be_bytes().as_slice())?;
-        }
-        self.commit(tx)
-    }
-
     pub fn created_operation(&self, operation: [u8; 16]) -> Result<Option<CreatedOperation>, StoreError> {
         let tx = self.database.begin_read()?;
-        let schema = number(required(tx.open_table(META)?.get("schema")?)?.value())?;
         let t = tx.open_table(OPERATIONS)?;
-        t.get(operation.as_slice())?
-            .map(|v| decode_stored_operation(v.value(), schema))
-            .transpose()
+        t.get(operation.as_slice())?.map(|v| decode_operation(v.value())).transpose()
     }
     pub fn created_apply_result(&self, index: u64) -> Result<Option<CreatedApplyResult>, StoreError> {
         let tx = self.database.begin_read()?;
         let t = tx.open_table(APPLY_RESULTS)?;
-        if matches!(number(required(tx.open_table(META)?.get("schema")?)?.value())?, 4 | 6) {
+        if number(required(tx.open_table(META)?.get("schema")?)?.value())? == 4 {
             t.get(index)?
                 .map(|v| {
                     decode_response(v.value()).map(|r| {
@@ -800,72 +594,26 @@ impl Store {
 
 pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreError> {
     let meta = tx.open_table(META)?;
-    if !matches!(number(required(meta.get("schema")?)?.value())?, 3..=6) {
+    let schema = number(required(meta.get("schema")?)?.value())?;
+    let native = schema == 4;
+    if !matches!(schema, 3 | 4) {
         return Ok(());
     }
-    validate_application_tables(
-        &tx.open_table(META)?,
-        &tx.open_table(LOGS)?,
-        &tx.open_table(HEADS)?,
-        &tx.open_table(VERSIONS)?,
-        &tx.open_table(EVENTS)?,
-        &tx.open_table(OPERATIONS)?,
-        &tx.open_table(APPLY_RESULTS)?,
-        false,
-    )
-    .map(|_| ())
-}
-
-fn validate_application_tables(
-    meta: &impl ReadableTable<&'static str, &'static [u8]>,
-    logs: &impl ReadableTable<u64, &'static [u8]>,
-    heads: &impl ReadableTable<&'static [u8], &'static [u8]>,
-    versions: &impl ReadableTable<&'static [u8], &'static [u8]>,
-    events: &impl ReadableTable<&'static [u8], &'static [u8]>,
-    operations: &impl ReadableTable<&'static [u8], &'static [u8]>,
-    results: &impl ReadableTable<u64, &'static [u8]>,
-    promoting_provenance: bool,
-) -> Result<BTreeMap<[u8; 16], LogId>, StoreError> {
-    let schema = number(required(meta.get("schema")?)?.value())?;
-    let native = matches!(schema, 4 | 6);
-    let full_replay = native || schema == 5 || promoting_provenance;
-    if !matches!(schema, 3..=6) {
-        return Ok(BTreeMap::new());
-    }
-    let binding = read_binding(meta).map_err(|error| match error {
+    let binding = read_binding(&meta).map_err(|error| match error {
         StoreError::BindingNotPinned => StoreError::CorruptRecord,
         other => other,
     })?;
-    let last = read_applied(meta)?;
-    let committed = validate_committed(meta, logs)?;
+    let last = read_applied(&meta)?;
+    let logs = tx.open_table(LOGS)?;
+    let committed = validate_committed(&meta, &logs)?;
+    let heads = tx.open_table(HEADS)?;
+    let versions = tx.open_table(VERSIONS)?;
+    let events = tx.open_table(EVENTS)?;
+    let operations = tx.open_table(OPERATIONS)?;
+    let results = tx.open_table(APPLY_RESULTS)?;
     if last.is_some_and(|last| committed.is_none_or(|c| last.index > c.index)) {
         return Err(StoreError::CorruptRecord);
     }
-    if !native && meta.get("native_membership")?.is_some() {
-        return Err(StoreError::CorruptRecord);
-    }
-    let mut log_next = Some(0);
-    let mut tail = None;
-    for row in logs.iter()? {
-        let (index, bytes) = row?;
-        let entry = decode_log(bytes.value())?;
-        if log_next != Some(index.value()) || entry.id.index != index.value() {
-            return Err(StoreError::CorruptRecord);
-        }
-        if native {
-            decode_native_entry(entry.clone())?;
-        } else if promoting_provenance || schema == 5 {
-            let command = DecideCreated::decode(&entry.payload)?;
-            if command.encode()? != entry.payload {
-                return Err(StoreError::CorruptRecord);
-            }
-        }
-        log_next = index.value().checked_add(1);
-        tail = Some(entry.id);
-    }
-    if meta.get("tail")?.map(|value| crate::decode_id(value.value())).transpose()? != tail {
-        return Err(StoreError::CorruptRecord);
-    }
     let mut next = Some(0);
     let mut observed = None;
     let mut first_operations = BTreeMap::new();
@@ -908,14 +656,14 @@ fn validate_application_tables(
         }
         let c = entry_command(&entry, native)?;
         let canonical = c.encode()?;
-        let op = decode_stored_operation(required(operations.get(c.operation_id.as_slice())?)?.value(), schema)?;
+        let op = decode_operation(required(operations.get(c.operation_id.as_slice())?)?.value())?;
         let canonical_matches = op.canonical_command == canonical;
         let expected = if canonical_matches {
             op.result
         } else {
             CreatedResult::OperationConflict
         };
-        if full_replay {
+        if native {
             let replay = if let Some((previous, result)) = replay_operations.get(&c.operation_id) {
                 if *previous == canonical {
                     result.clone()
@@ -941,7 +689,6 @@ fn validate_application_tables(
                         decision_log_id: r.log_id,
                     }
                 };
-                first_operations.insert(c.operation_id, r.log_id);
                 replay_operations.insert(c.operation_id, (canonical.clone(), result.clone()));
                 result
             };
@@ -952,7 +699,7 @@ fn validate_application_tables(
         if expected != r.result {
             return Err(StoreError::CorruptRecord);
         }
-        if !full_replay && canonical_matches {
+        if canonical_matches {
             first_operations.entry(c.operation_id).or_insert(r.log_id);
         }
         observed = Some(r.log_id);
@@ -961,8 +708,8 @@ fn validate_application_tables(
     if observed != last {
         return Err(StoreError::CorruptRecord);
     }
-    if full_replay {
-        if (native && native_application_metadata(meta)?.1 != expected_membership)
+    if native {
+        if native_application_state(tx)?.1 != expected_membership
             || heads.len()? != u64::try_from(replay_heads.len()).map_err(|_| StoreError::CorruptRecord)?
             || versions.len()? != u64::try_from(replay_versions.len()).map_err(|_| StoreError::CorruptRecord)?
         {
@@ -976,15 +723,12 @@ fn validate_application_tables(
     }
     for row in operations.iter()? {
         let (k, v) = row?;
-        let op = decode_stored_operation(v.value(), schema)?;
+        let op = decode_operation(v.value())?;
         let c = DecideCreated::decode(&op.canonical_command)?;
         if k.value() != c.operation_id {
             return Err(StoreError::CorruptRecord);
         }
         let first = required(first_operations.get(&c.operation_id))?;
-        if matches!(schema, 5 | 6) && decode_provenance(v.value())?.first_decision_log_id != *first {
-            return Err(StoreError::CorruptRecord);
-        }
         if let CreatedResult::Created {
             version_id,
             event_id,
@@ -1019,7 +763,7 @@ fn validate_application_tables(
         if e.binding != binding {
             return Err(StoreError::CorruptRecord);
         }
-        let op = decode_stored_operation(required(operations.get(k.value())?)?.value(), schema)?;
+        let op = decode_operation(required(operations.get(k.value())?)?.value())?;
         let c = DecideCreated::decode(&op.canonical_command)?;
         if c.object != e.object
             || c.prepared != e.prepared
@@ -1082,5 +826,5 @@ fn validate_application_tables(
         d.finish()?;
         decode_version(required(versions.get(version_key(&object, id)?.as_slice())?)?.value())?;
     }
-    Ok(first_operations)
+    Ok(())
 }

crates/capture-authority/src/committed.rs

@@ -8,13 +8,13 @@ pub(crate) fn validate_committed(
     let schema = number(meta.get("schema")?.ok_or(StoreError::CorruptRecord)?.value())?;
     let marker = meta.get("committed")?;
     match (schema, marker) {
-        (1 | 3 | 4 | 5 | 6, None) => {
-            if matches!(schema, 3..=6) && meta.get("last_applied")?.is_none_or(|v| v.value() != [0]) {
+        (1 | 3 | 4, None) => {
+            if matches!(schema, 3 | 4) && meta.get("last_applied")?.is_none_or(|v| v.value() != [0]) {
                 return Err(StoreError::CorruptRecord);
             }
             Ok(None)
         }
-        (2 | 3 | 4 | 5 | 6, Some(marker)) => {
+        (2 | 3 | 4, Some(marker)) => {
             let id = decode_id(marker.value())?;
             let record = logs.get(id.index)?.ok_or(StoreError::CorruptRecord)?;
             if decode_log(record.value())?.id != id {
@@ -54,7 +54,7 @@ impl Store {
                 }
             }
             meta.insert("committed", encode_id(id).as_slice())?;
-            if previous.is_none() && !matches!(number(meta.get("schema")?.ok_or(StoreError::CorruptRecord)?.value())?, 3..=6) {
+            if previous.is_none() && !matches!(number(meta.get("schema")?.ok_or(StoreError::CorruptRecord)?.value())?, 3 | 4) {
                 meta.insert("schema", 2u64.to_be_bytes().as_slice())?;
             }
         }

crates/capture-authority/src/created.rs

@@ -60,11 +60,6 @@ pub struct CreatedOperation {
     pub result: CreatedResult,
 }
 #[derive(Debug, Clone, PartialEq, Eq)]
-pub struct CreatedOperationProvenance {
-    pub operation: CreatedOperation,
-    pub first_decision_log_id: LogId,
-}
-#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct CreatedApplyResult {
     pub log_id: LogId,
     pub result: CreatedResult,

crates/capture-authority/src/native_snapshot.rs

@@ -1,154 +0,0 @@
-//! Owned consistent native application snapshot view.
-use crate::{Store, StoreError, application, snapshot};
-use redb::{ReadTransaction, ReadableDatabase, ReadableTable, ReadableTableMetadata};
-use std::{
-    io::Write,
-    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)]
-    Store(#[from] StoreError),
-    #[error(transparent)]
-    Snapshot(#[from] snapshot::SnapshotError),
-    #[error("native snapshot Store lock poisoned")]
-    Poisoned,
-    #[error("native snapshot worker failed")]
-    Worker(#[source] tokio::task::JoinError),
-}
-
-pub struct NativeSnapshotView {
-    // Drop the transaction before the owner so the Database stays open throughout the view.
-    transaction: ReadTransaction,
-    header: snapshot::Header,
-    counts: [u64; 4],
-    schema: u64,
-    _owner: Arc<Mutex<Store>>,
-}
-impl NativeSnapshotView {
-    pub(crate) fn acquire(owner: Arc<Mutex<Store>>) -> Result<Self, NativeSnapshotError> {
-        let transaction = {
-            let store = owner.lock().map_err(|_| NativeSnapshotError::Poisoned)?;
-            if !store.writable {
-                return Err(StoreError::WriteDisabled.into());
-            }
-            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, 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)?;
-            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 [
-            application::HEADS,
-            application::VERSIONS,
-            application::EVENTS,
-            application::OPERATIONS,
-        ]
-        .into_iter()
-        .enumerate()
-        {
-            counts[index] = transaction
-                .open_table(definition)
-                .map_err(StoreError::from)?
-                .len()
-                .map_err(StoreError::from)?;
-        }
-        Ok(Self {
-            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 [
-            (application::HEADS, snapshot::Section::Heads),
-            (application::VERSIONS, snapshot::Section::Versions),
-            (application::EVENTS, snapshot::Section::Events),
-            (application::OPERATIONS, snapshot::Section::Operations),
-        ] {
-            let table = self.transaction.open_table(definition).map_err(StoreError::from)?;
-            for row in table.iter().map_err(StoreError::from)? {
-                let (key, value) = row.map_err(StoreError::from)?;
-                encoder.row(snapshot::Row {
-                    section,
-                    key: key.value(),
-                    value: value.value(),
-                })?;
-            }
-        }
-        Ok(encoder.finish()?)
-    }
-}

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 {

docs/architecture/README.md

@@ -32,7 +32,6 @@ Required headings and strings in these files are asserted by `scripts/check_arch
 
 | Document | Use this when |
 |---|---|
-| [capture-checkpoint-contract.md](capture-checkpoint-contract.md) | changing Capture checkpoint capture, provenance, installation, recovery, or log cleanup |
 | [local-descriptor-cache.md](local-descriptor-cache.md) | changing descriptor cache keys/open modes or invalidation after local mutations |
 | [erasure-coding.md](erasure-coding.md) | changing anything under `crates/ecstore/src/erasure/`, `crates/filemeta/`, `crates/ecstore/src/set_disk/`, storage-class or layout code, or any decode, quorum, or heal boundary (normative spec) |
 | [placement-repair-invariants.md](placement-repair-invariants.md) | changing anything that resolves an object to a pool, set, or disk, or that admits scanner or heal work |

docs/architecture/capture-checkpoint-contract.md

@@ -1,211 +0,0 @@
-**Use this when:** changing Capture checkpoint capture, provenance, installation, recovery, or log cleanup.
-
-# Capture checkpoint contract
-
-This document defines the requirements for a recoverable Capture checkpoint.
-It is a design contract, not evidence that production capture, installation,
-purge, quorum integration, or S3 behavior implements these requirements.
-
-## Existing evidence and the missing boundary
-
-In `crates/capture-authority/src/application.rs`, `apply_created` inserts one
-canonical command and result on the first decision for an operation ID. An
-identical command returns that stored result; a different command returns
-`OperationConflict` without replacing the canonical record. The first result
-remains authoritative even when later state would produce a different result.
-
-`CreatedOperation` in `crates/capture-authority/src/created.rs` contains command
-bytes and a result. Only `CreatedResult::Created` carries `decision_log_id`.
-Negative results carry no first-decision ID. Heads, versions, events, and
-operations therefore cannot reconstruct the source of every first decision.
-For example, a stored `HeadMismatch` records an observed head, but does not
-prove that the head existed at the original decision cut. Later retries do
-not supply the missing historical evidence.
-
-`validate_application` in `crates/capture-authority/src/application.rs` derives
-first operation IDs from retained log/application records and, for native
-schema 4, replays decisions from the beginning using in-memory maps and sets.
-That intact-prefix validator is evidence for existing behavior; it is not a
-checkpoint-baseline validator or a bounded recovery algorithm.
-
-The `Row` comment and `StagingSink` in
-`crates/capture-authority/src/snapshot.rs` reserve operation provenance for
-caller validation and require isolated staging. Rows remain opaque to the
-framing decoder. A framing digest detects byte corruption; it does not establish
-semantic consistency, historical truth, or authority to install a checkpoint.
-
-## Canonical operation provenance
-
-Checkpoint-capable storage must use an explicitly versioned operation envelope
-containing the full first-decision `LogId` and the existing canonical operation
-bytes, including their stored result. Preserve those bytes and result; do not
-recompute a negative result against checkpoint-time heads or binding state.
-The envelope version is a storage capability boundary, distinct from the
-snapshot framing version. The existing CSNP v1 framing and header must not be
-silently extended. Header capability metadata requires a declared framing
-revision or an authenticated, versioned manifest whose consistency with the
-captured header is validated. Distinct storage and framing versions do not
-permit silently changing the v1 format.
-
-The first insertion must atomically record the envelope, application response,
-applied position, and any successful head/version/event changes in the same
-write transaction. Its first-decision ID must identify that insertion's log
-entry. An identical-command retry returns the original result and leaves both
-bytes and first-decision ID unchanged. A conflicting-command retry records its
-own application response as `OperationConflict`; it cannot overwrite the
-canonical envelope or claim to be the first decision.
-
-```mermaid
-flowchart TD
-    A[Committed command and operation ID] --> B{Canonical operation exists?}
-    B -->|No| C[Evaluate once at this log ID]
-    C --> D[Atomically persist first ID, original bytes and result]
-    B -->|Yes| E{Same canonical command bytes?}
-    E -->|Yes| F[Return original result; preserve first ID]
-    E -->|No| G[Return OperationConflict; preserve canonical record]
-```
-
-## Compatibility and promotion
-
-The existing schema 3 to schema 4 migration remains unchanged. Adding provenance
-requires a separately declared storage capability with explicit readers, writers,
-upgrade, and downgrade rules. A schema 4 label alone must not imply provenance
-support. Unknown envelope versions and stores missing required provenance must
-be rejected by checkpoint installation and purge authorization.
-
-An old schema 4 store may backfill first IDs only from its complete, intact,
-validated prefix: replay and validate the original command/results, identify the
-first canonical insertion for each operation, then atomically persist the
-capability and envelopes. A retry's ID or a successful event's ID cannot stand
-in for missing negative provenance. If the prefix has already been removed or
-cannot be validated, refuse promotion and cleanup; do not invent IDs. Readers
-without the new capability must refuse the upgraded format rather than silently
-interpret it as the old operation encoding. Retaining all history forever is
-not the checkpoint design: validated promotion supplies a recoverable baseline
-that permits bounded suffix retention.
-
-## Capture cut and resource ownership
-
-Capture must read the header, heads, versions, events, and operations through
-one owned read transaction. The view must retain the owning store/database
-handle for the entire transaction, cursor, and output lifetime; an owned
-transaction type alone does not prove database lifetime safety. Keep the store
-`Arc` alive so dropping another handle cannot invalidate the view. These are
-ownership requirements, not a claim about unverified library behavior.
-
-The header must describe that same committed, applied cut with its full `LogId`,
-full effective membership including its decision position and node metadata,
-cluster identity, binding identity/revision, and supported storage/envelope
-capabilities. Membership and binding must come from the captured transaction,
-not from a later live read. Never capture beyond the durable committed position.
-A four-table view without this cut and metadata is not independently installable.
-
-Emit rows incrementally in deterministic section/key order. Bound row size,
-section and total bytes, counts, temporary disk consumption, and concurrent
-capture admission. Use checked arithmetic. Do not collect the whole snapshot
-in RAM. Cancellation, disk exhaustion, or limit failure must leave the live
-store unchanged and release task-owned staging resources.
-
-## Isolated validation and trust
-
-Decode only into isolated disk staging. Check framing, limits, section order,
-unique keys, complete header metadata, capability versions, cluster/binding
-compatibility, canonical command/result encoding, envelope IDs at or before the
-cut, and all head/version/event/operation relationships. A successful operation
-must agree with its version, event, and first decision. Negative operations must
-have no success event or side effect attributable to that decision.
-
-These checks still cannot prove the historical predicate behind a negative
-result. Neither a digest nor a first ID proves that a `HeadMismatch`,
-`BindingMismatch`, or `VersionConflict` was justified when first decided.
-Remote installation therefore requires an explicitly authenticated and
-authorized Raft authority for the claimed cluster, cut, and membership, or a
-complete historical proof validated from a trusted baseline. Reject input that
-has neither. Authentication and authorization must precede publication and
-purge; a caller's assertion of trust is insufficient. This document does not
-claim that this remote Raft authentication boundary is implemented.
-
-Replay validation must use disk-backed staging/indexes and bounded buffers,
-including duplicate-operation and version/head tracking. Putting the input on
-disk while retaining replay maps proportional to history does not meet this
-requirement. The existing `validate_application` RAM replay cannot simply be
-reused to claim bounded checkpoint installation.
-
-## Atomic publication and recovery
-
-After isolated validation, publication must have one durable atomic point that
-selects the complete checkpoint generation, its application state, cut,
-membership, binding, capability, and recovery metadata. A pointer switch is
-acceptable only when all referenced bytes are already durable and recovery
-cannot observe a partial generation. Serialize publication with application so
-no committed suffix write is lost between validation and the switch.
-
-Preserve local node identity and persisted vote/hard state. Remote checkpoint
-metadata cannot overwrite either. Validate cluster identity, binding, full cut
-identity, and membership before replacing state. Reject rollback behind the
-local durable application/checkpoint floor; an equal index with a different
-term/leader identity is a conflict, not an equivalent cut. Retain valid local
-suffix and commit state consistently with the installed baseline, rather than
-resetting them from remote metadata.
-
-On reopen, select only a fully published generation and verify its recoverable
-baseline before replay. The suffix starts exactly at cut.index + 1, with checked
-index arithmetic; the complete cut ID must agree wherever the boundary is
-retained. Validate contiguous suffix indices, entry identities, committed and
-applied bounds, typed application responses, canonical retry behavior, and
-membership transitions relative to checkpoint state. Reject gaps, conflicting
-boundary IDs, stale membership, or a suffix belonging to another cluster or
-binding. Recovery must begin from checkpoint heads and canonical operations,
-not from empty state or the assumption that the retained log starts at zero.
-
-Purge is authorized only after the published checkpoint can be reopened and
-validated as a recovery baseline. Delete only entries/results covered by that
-exact cut, retaining everything needed for the continuous suffix and local
-Raft durability. Re-check the selected generation and cut immediately before
-cleanup. Persist a recoverable purge boundary so interrupted cleanup is
-idempotent. Never delete the only evidence before provenance promotion or
-checkpoint durability succeeds.
-
-```mermaid
-flowchart TD
-    A[One owned transaction and store lifetime] --> B[Stream captured cut to isolated disk staging]
-    B --> C[Validate framing, semantics, provenance and authorized source]
-    C --> D[Make complete generation durable]
-    D --> E[Single atomic publication point]
-    E --> F[Reopen baseline and validate continuous suffix]
-    F --> G[Authorize purge only through that exact cut]
-    G --> H[Idempotent cleanup; preserve recoverable baseline and suffix]
-    B -->|Failure| I[Discard staging; live generation remains usable]
-    C -->|Failure| I
-    D -->|Before publication crash| I
-```
-
-A crash before publication leaves the previous generation recoverable. A crash
-after publication must recover the new complete baseline, including when purge
-has not begun or is only partly complete. Orphan staging is never selected as
-live state. Old generations may be reclaimed only after no active view needs
-them and the selected generation remains recoverable.
-
-## Behavioral verification requirements
-
-The following are required behavior checks, not reports of tests run or gates
-passed. Existing symbols above anchor the semantics; future checkpoint paths
-must demonstrate the additional requirements directly.
-
-| Behavior | Required observation |
-|---|---|
-| First negative result | Original command/result and first full ID survive capture, install, and reopen; later state does not re-evaluate it. |
-| Identical and conflicting retries | `apply_created` semantics hold across the checkpoint boundary: original result/ID retained; conflict cannot overwrite the canonical record. |
-| Old schema 4 promotion | An intact prefix validates and supplies first IDs; missing/corrupt prefix refuses promotion and purge; schema 3 to 4 behavior stays unchanged. |
-| Snapshot cut under concurrent writes | Header, membership, binding, and all rows describe one transaction cut; later writes remain in the suffix. |
-| View lifetime | Dropping other store handles during row emission cannot invalidate the retained transaction or database. |
-| Framing versus authority | Valid digest with forged negative history is insufficient; unauthenticated or unauthorized remote input is rejected. |
-| Cross-record corruption | Altered command, first ID, event, version, head, membership, or binding fails isolated validation before publication. |
-| Recovery boundary | Exact cut plus continuous suffix recovers; gaps, conflicting full IDs, cross-cluster input, and rollback are rejected. |
-| Local Raft state | Installation preserves local identity, vote/hard state, and compatible durable suffix/commit information. |
-| Resource exhaustion | Oversize rows, arithmetic overflow, disk exhaustion, cancellation, and large histories fail safely within declared memory/disk bounds. |
-| Crash publication | Crashes before and after the atomic point select a complete old or new generation, never a mixture. |
-| Interrupted purge | Reopen succeeds before cleanup is authorized; each interruption resumes without deleting the recoverable baseline or required suffix. |
-
-Documentation path and whitespace guards only validate document hygiene. They
-cannot establish these production behaviors or the remote trust boundary.