feat(capture): persist first-decision provenance

crates/capture-authority/src/application.rs

@@ -39,7 +39,7 @@ 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};
+use redb::{ReadTransaction, ReadableDatabase, ReadableTable, ReadableTableMetadata, TableDefinition, TableHandle};
 use std::collections::BTreeMap;
 
 pub(crate) const HEADS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("heads");
@@ -89,6 +89,36 @@ 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)?)?,
+    })
+}
+
+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 {
@@ -98,7 +128,7 @@ fn decode_apply(b: &[u8]) -> Result<CreatedApplyResult, StoreError> {
     d.finish()?;
     Ok(r)
 }
-fn read_binding(meta: &impl ReadableTable<&'static str, &'static [u8]>) -> Result<CaptureBinding, StoreError> {
+pub(crate) 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()?;
@@ -176,11 +206,12 @@ 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_operation(v.value()))
+        .map(|v| decode_stored_operation(v.value(), schema))
         .transpose()?;
     let result = if let Some(existing) = existing {
         if existing.canonical_command == canonical {
@@ -233,6 +264,9 @@ 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
     };
@@ -298,11 +332,16 @@ fn entry_command(entry: &crate::LogEntry, native: bool) -> Result<DecideCreated,
 }
 
 pub(crate) fn native_application_state(tx: &ReadTransaction) -> Result<(Option<LogId>, crate::CaptureMembership), StoreError> {
-    let meta = tx.open_table(META)?;
-    if number(required(meta.get("schema")?)?.value())? != 4 {
+    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) {
         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
@@ -315,8 +354,7 @@ pub(crate) fn native_application_state(tx: &ReadTransaction) -> Result<(Option<L
 }
 
 impl Store {
-    /// Explicitly opt an empty, bound store into native entry persistence.
-    pub fn enable_native_empty(&mut self) -> Result<(), StoreError> {
+    pub fn migrate_legacy_created_to_native(&mut self) -> Result<(), StoreError> {
         let tx = self.write_transaction()?;
         {
             let mut meta = tx.open_table(META)?;
@@ -324,8 +362,66 @@ impl Store {
             if !matches!(schema, 3 | 4) {
                 return Err(StoreError::UnsupportedSchema);
             }
-            read_binding(&meta)?;
+            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)
+    }
+
+    /// Explicitly opt an empty, bound store into native entry persistence.
+    pub fn enable_native_empty(&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);
+            }
+            read_binding(&meta)?;
+            if matches!(schema, 4 | 6) {
                 let membership = crate::CaptureMembership::decode(required(meta.get("native_membership")?)?.value())
                     .map_err(|_| StoreError::CorruptRecord)?;
                 let last = read_applied(&meta)?;
@@ -354,7 +450,7 @@ impl Store {
             }
             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())?;
+            meta.insert("schema", (if schema == 5 { 6u64 } else { 4u64 }).to_be_bytes().as_slice())?;
         }
         self.commit(tx)
     }
@@ -362,7 +458,7 @@ impl Store {
         let tx = self.write_transaction()?;
         {
             let mut meta = tx.open_table(META)?;
-            if number(required(meta.get("schema")?)?.value())? != 4 {
+            if !matches!(number(required(meta.get("schema")?)?.value())?, 4 | 6) {
                 return Err(StoreError::UnsupportedSchema);
             }
             if entries.is_empty() {
@@ -391,7 +487,7 @@ impl Store {
     }
     pub fn read_native_entry(&self, index: u64) -> Result<Option<crate::CaptureRaftEntry>, StoreError> {
         let tx = self.database.begin_read()?;
-        if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
+        if !matches!(number(required(tx.open_table(META)?.get("schema")?)?.value())?, 4 | 6) {
             return Err(StoreError::UnsupportedSchema);
         }
         tx.open_table(LOGS)?
@@ -408,14 +504,14 @@ impl Store {
     /// 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 number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
+        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 number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
+        if !matches!(number(required(tx.open_table(META)?.get("schema")?)?.value())?, 4 | 6) {
             return Err(StoreError::UnsupportedSchema);
         }
         if start > end {
@@ -429,7 +525,8 @@ impl Store {
         let response;
         {
             let mut meta = tx.open_table(META)?;
-            if number(required(meta.get("schema")?)?.value())? != 4 {
+            let schema = number(required(meta.get("schema")?)?.value())?;
+            if !matches!(schema, 4 | 6) {
                 return Err(StoreError::UnsupportedSchema);
             }
             let binding = read_binding(&meta)?;
@@ -457,7 +554,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)?;
+                    let result = apply_created(&tx, id, command.created(), &canonical, &binding, schema)?;
                     crate::CaptureResponse::created(CreatedApplyResult { log_id: id, result })?
                 }
                 openraft::EntryPayload::Membership(membership) => {
@@ -477,7 +574,7 @@ impl Store {
     }
     pub fn native_apply_response(&self, index: u64) -> Result<Option<crate::CaptureResponse>, StoreError> {
         let tx = self.database.begin_read()?;
-        if number(required(tx.open_table(META)?.get("schema")?)?.value())? != 4 {
+        if !matches!(number(required(tx.open_table(META)?.get("schema")?)?.value())?, 4 | 6) {
             return Err(StoreError::UnsupportedSchema);
         }
         let results = tx.open_table(APPLY_RESULTS)?;
@@ -495,7 +592,7 @@ impl Store {
         let tx = self.write_transaction()?;
         {
             let mut meta = tx.open_table(META)?;
-            if matches!(number(required(meta.get("schema")?)?.value())?, 3 | 4) {
+            if matches!(number(required(meta.get("schema")?)?.value())?, 3..=6) {
                 if read_binding(&meta)? == binding {
                     return Ok(());
                 }
@@ -524,7 +621,8 @@ impl Store {
         let result;
         {
             let mut meta = tx.open_table(META)?;
-            if number(required(meta.get("schema")?)?.value())? != 3 {
+            let schema = number(required(meta.get("schema")?)?.value())?;
+            if !matches!(schema, 3 | 5) {
                 return Err(StoreError::UnsupportedSchema);
             }
             let binding = read_binding(&meta)?;
@@ -547,7 +645,7 @@ impl Store {
                 return Err(StoreError::InvalidApplySequence);
             }
             let command = DecideCreated::decode(&entry.payload)?;
-            result = apply_created(&tx, id, &command, &entry.payload, &binding)?;
+            result = apply_created(&tx, id, &command, &entry.payload, &binding, schema)?;
             let mut applied = encode_id(id);
             applied.extend(result_bytes(&result));
             results.insert(id.index, applied.as_slice())?;
@@ -561,7 +659,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 | 4) {
+        if !matches!(number(required(meta.get("schema")?)?.value())?, 3..=6) {
             return Ok(None);
         }
         Ok(Some(read_binding(&meta)?))
@@ -569,7 +667,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 | 4) {
+        if !matches!(number(required(meta.get("schema")?)?.value())?, 3..=6) {
             return Ok(None);
         }
         read_applied(&meta)
@@ -598,15 +696,87 @@ 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_operation(v.value())).transpose()
+        t.get(operation.as_slice())?
+            .map(|v| decode_stored_operation(v.value(), schema))
+            .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 number(required(tx.open_table(META)?.get("schema")?)?.value())? == 4 {
+        if matches!(number(required(tx.open_table(META)?.get("schema")?)?.value())?, 4 | 6) {
             t.get(index)?
                 .map(|v| {
                     decode_response(v.value()).map(|r| {
@@ -626,26 +796,72 @@ impl Store {
 
 pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreError> {
     let meta = tx.open_table(META)?;
-    let schema = number(required(meta.get("schema")?)?.value())?;
-    let native = schema == 4;
-    if !matches!(schema, 3 | 4) {
+    if !matches!(number(required(meta.get("schema")?)?.value())?, 3..=6) {
         return Ok(());
     }
-    let binding = read_binding(&meta).map_err(|error| match error {
+    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 {
         StoreError::BindingNotPinned => StoreError::CorruptRecord,
         other => other,
     })?;
-    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)?;
+    let last = read_applied(meta)?;
+    let committed = validate_committed(meta, logs)?;
     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();
@@ -688,14 +904,14 @@ pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreErro
         }
         let c = entry_command(&entry, native)?;
         let canonical = c.encode()?;
-        let op = decode_operation(required(operations.get(c.operation_id.as_slice())?)?.value())?;
+        let op = decode_stored_operation(required(operations.get(c.operation_id.as_slice())?)?.value(), schema)?;
         let canonical_matches = op.canonical_command == canonical;
         let expected = if canonical_matches {
             op.result
         } else {
             CreatedResult::OperationConflict
         };
-        if native {
+        if full_replay {
             let replay = if let Some((previous, result)) = replay_operations.get(&c.operation_id) {
                 if *previous == canonical {
                     result.clone()
@@ -721,6 +937,7 @@ pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreErro
                         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
             };
@@ -731,7 +948,7 @@ pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreErro
         if expected != r.result {
             return Err(StoreError::CorruptRecord);
         }
-        if canonical_matches {
+        if !full_replay && canonical_matches {
             first_operations.entry(c.operation_id).or_insert(r.log_id);
         }
         observed = Some(r.log_id);
@@ -740,8 +957,8 @@ pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreErro
     if observed != last {
         return Err(StoreError::CorruptRecord);
     }
-    if native {
-        if native_application_state(tx)?.1 != expected_membership
+    if full_replay {
+        if (native && native_application_metadata(meta)?.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)?
         {
@@ -755,12 +972,15 @@ pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreErro
     }
     for row in operations.iter()? {
         let (k, v) = row?;
-        let op = decode_operation(v.value())?;
+        let op = decode_stored_operation(v.value(), schema)?;
         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,
@@ -795,7 +1015,7 @@ pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreErro
         if e.binding != binding {
             return Err(StoreError::CorruptRecord);
         }
-        let op = decode_operation(required(operations.get(k.value())?)?.value())?;
+        let op = decode_stored_operation(required(operations.get(k.value())?)?.value(), schema)?;
         let c = DecideCreated::decode(&op.canonical_command)?;
         if c.object != e.object
             || c.prepared != e.prepared
@@ -858,5 +1078,5 @@ pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreErro
         d.finish()?;
         decode_version(required(versions.get(version_key(&object, id)?.as_slice())?)?.value())?;
     }
-    Ok(())
+    Ok(first_operations)
 }

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, None) => {
-            if matches!(schema, 3 | 4) && meta.get("last_applied")?.is_none_or(|v| v.value() != [0]) {
+        (1 | 3 | 4 | 5 | 6, None) => {
+            if matches!(schema, 3..=6) && meta.get("last_applied")?.is_none_or(|v| v.value() != [0]) {
                 return Err(StoreError::CorruptRecord);
             }
             Ok(None)
         }
-        (2 | 3 | 4, Some(marker)) => {
+        (2 | 3 | 4 | 5 | 6, 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 | 4) {
+            if previous.is_none() && !matches!(number(meta.get("schema")?.ok_or(StoreError::CorruptRecord)?.value())?, 3..=6) {
                 meta.insert("schema", 2u64.to_be_bytes().as_slice())?;
             }
         }

crates/capture-authority/src/created.rs

@@ -60,6 +60,11 @@ 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

@@ -0,0 +1,95 @@
+//! 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},
+};
+
+#[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],
+    _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)?
+        };
+        let (applied, membership) = application::native_application_state(&transaction)?;
+        let header = {
+            let meta = transaction.open_table(crate::META).map_err(StoreError::from)?;
+            let identity = meta
+                .get("identity")
+                .map_err(StoreError::from)?
+                .ok_or(StoreError::CorruptRecord)?;
+            snapshot::Header {
+                cluster_id: crate::decode_identity(identity.value())?.cluster_id,
+                applied,
+                membership,
+                binding: application::read_binding(&meta)?,
+            }
+        };
+        let 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,
+            _owner: owner,
+        })
+    }
+
+    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()?)
+    }
+}