From dfa5d50565371cf3e612cae331fbf6831ea3f9bd Mon Sep 17 00:00:00 2001 From: overtrue Date: Sun, 11 Oct 2026 18:58:12 +0800 Subject: [PATCH] feat(capture): persist first-decision provenance --- crates/capture-authority/src/application.rs | 211 +++++- crates/capture-authority/src/committed.rs | 8 +- crates/capture-authority/src/created.rs | 5 + crates/capture-authority/src/lib.rs | 689 +++++++++++++++++++- 4 files changed, 866 insertions(+), 47 deletions(-) diff --git a/crates/capture-authority/src/application.rs b/crates/capture-authority/src/application.rs index e75e37c08..e3c77e62c 100644 --- a/crates/capture-authority/src/application.rs +++ b/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 { result, }) } +fn decode_provenance(b: &[u8]) -> Result { + 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 { + if matches!(schema, 5 | 6) { + Ok(decode_provenance(b)?.operation) + } else { + decode_operation(b) + } +} + +fn encode_provenance(first: LogId, operation: &[u8]) -> Result, 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 { let mut d = Decoder::new(b); let r = CreatedApplyResult { @@ -176,11 +206,12 @@ fn apply_created( command: &DecideCreated, canonical: &[u8], binding: &CaptureBinding, + schema: u64, ) -> Result { 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 }; @@ -304,7 +338,7 @@ pub(crate) fn native_application_state(tx: &ReadTransaction) -> Result<(Option, ) -> Result<(Option, crate::CaptureMembership), StoreError> { - if number(required(meta.get("schema")?)?.value())? != 4 { + if !matches!(number(required(meta.get("schema")?)?.value())?, 4 | 6) { return Err(StoreError::UnsupportedSchema); } let last = read_applied(meta)?; @@ -334,7 +368,7 @@ impl Store { 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)?; + validate_application_tables(&meta, &logs, &heads, &versions, &events, &operations, &results, false)?; if schema == 4 { return Ok(()); } @@ -372,7 +406,7 @@ impl Store { 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)?; + validate_application_tables(&meta, &logs, &heads, &versions, &events, &operations, &results, false)?; } self.commit(tx) } @@ -383,11 +417,11 @@ impl Store { { let mut meta = tx.open_table(META)?; let schema = number(required(meta.get("schema")?)?.value())?; - if !matches!(schema, 3 | 4) { + if !matches!(schema, 3..=6) { return Err(StoreError::UnsupportedSchema); } read_binding(&meta)?; - if schema == 4 { + 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)?; @@ -416,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) } @@ -424,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() { @@ -453,7 +487,7 @@ impl Store { } pub fn read_native_entry(&self, index: u64) -> Result, 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)? @@ -470,14 +504,14 @@ impl Store { /// Read typed native entries within the requested bounds. pub fn read_native_range>(&self, range: R) -> Result, 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, 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 { @@ -491,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)?; @@ -519,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) => { @@ -539,7 +574,7 @@ impl Store { } pub fn native_apply_response(&self, index: u64) -> Result, 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)?; @@ -557,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(()); } @@ -586,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)?; @@ -609,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())?; @@ -623,7 +659,7 @@ impl Store { pub fn capture_binding(&self) -> Result, 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)?)) @@ -631,7 +667,7 @@ impl Store { pub fn last_applied(&self) -> Result, 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) @@ -660,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, 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::>(); + 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, 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, 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| { @@ -688,7 +796,7 @@ 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 | 4) { + if !matches!(number(required(meta.get("schema")?)?.value())?, 3..=6) { return Ok(()); } validate_application_tables( @@ -699,7 +807,9 @@ pub(crate) fn validate_application(tx: &ReadTransaction) -> Result<(), StoreErro &tx.open_table(EVENTS)?, &tx.open_table(OPERATIONS)?, &tx.open_table(APPLY_RESULTS)?, + false, ) + .map(|_| ()) } fn validate_application_tables( @@ -710,11 +820,13 @@ fn validate_application_tables( events: &impl ReadableTable<&'static [u8], &'static [u8]>, operations: &impl ReadableTable<&'static [u8], &'static [u8]>, results: &impl ReadableTable, -) -> Result<(), StoreError> { + promoting_provenance: bool, +) -> Result, StoreError> { let schema = number(required(meta.get("schema")?)?.value())?; - let native = schema == 4; - if !matches!(schema, 3 | 4) { - return Ok(()); + 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, @@ -725,6 +837,31 @@ fn validate_application_tables( 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(); @@ -767,14 +904,14 @@ fn validate_application_tables( } 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() @@ -800,6 +937,7 @@ 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 }; @@ -810,7 +948,7 @@ fn validate_application_tables( 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); @@ -819,8 +957,8 @@ fn validate_application_tables( if observed != last { return Err(StoreError::CorruptRecord); } - if native { - if native_application_metadata(meta)?.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)? { @@ -834,12 +972,15 @@ fn validate_application_tables( } 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, @@ -874,7 +1015,7 @@ fn validate_application_tables( 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 @@ -937,5 +1078,5 @@ fn validate_application_tables( d.finish()?; decode_version(required(versions.get(version_key(&object, id)?.as_slice())?)?.value())?; } - Ok(()) + Ok(first_operations) } diff --git a/crates/capture-authority/src/committed.rs b/crates/capture-authority/src/committed.rs index a55ebc613..418d5ad7f 100644 --- a/crates/capture-authority/src/committed.rs +++ b/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())?; } } diff --git a/crates/capture-authority/src/created.rs b/crates/capture-authority/src/created.rs index d81f3c86d..7f875b151 100644 --- a/crates/capture-authority/src/created.rs +++ b/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, diff --git a/crates/capture-authority/src/lib.rs b/crates/capture-authority/src/lib.rs index d25369bf6..50e6281a9 100644 --- a/crates/capture-authority/src/lib.rs +++ b/crates/capture-authority/src/lib.rs @@ -15,8 +15,8 @@ pub use native_snapshot::{NativeSnapshotError, NativeSnapshotView}; mod raft_types; pub mod snapshot; pub use created::{ - CaptureBinding, CreatedApplyResult, CreatedEvent, CreatedOperation, CreatedResult, CreatedVersion, DecideCreated, - MAX_OBJECT_KEY_LENGTH, ObjectIdentity, PreparedIdentity, + CaptureBinding, CreatedApplyResult, CreatedEvent, CreatedOperation, CreatedOperationProvenance, CreatedResult, + CreatedVersion, DecideCreated, MAX_OBJECT_KEY_LENGTH, ObjectIdentity, PreparedIdentity, }; pub use node::{CaptureNode, CaptureNodeError}; pub use raft_types::{ @@ -381,7 +381,7 @@ impl Store { let tx = store.database.begin_read()?; let meta = tx.open_table(META)?; let schema = meta.get("schema")?.ok_or(StoreError::CorruptRecord)?; - if !matches!(number(schema.value())?, 1..=4) { + if !matches!(number(schema.value())?, 1..=6) { return Err(StoreError::UnsupportedSchema); } let identity = meta.get("identity")?.ok_or(StoreError::CorruptRecord)?; @@ -403,13 +403,13 @@ impl Store { let (key, value) = row?; let entry = decode_log(value.value())?; if next != Some(key.value()) || entry.id.index != key.value() { - return Err(if number(schema.value())? == 4 { + return Err(if matches!(number(schema.value())?, 4 | 6) { StoreError::CorruptRecord } else { StoreError::InvalidLogSequence }); } - if number(schema.value())? == 4 { + if matches!(number(schema.value())?, 4 | 6) { application::decode_native_entry(entry.clone())?; } next = key.value().checked_add(1); @@ -419,7 +419,7 @@ impl Store { application::validate_application(&tx)?; let tail = meta.get("tail")?.map(|v| decode_id(v.value())).transpose()?; if tail != last { - return Err(if number(schema.value())? == 4 { + return Err(if matches!(number(schema.value())?, 4 | 6) { StoreError::CorruptRecord } else { StoreError::InvalidLogSequence @@ -465,7 +465,10 @@ impl Store { return Err(StoreError::WriteDisabled); } let tx = self.database.begin_read()?; - if number(tx.open_table(META)?.get("schema")?.ok_or(StoreError::CorruptRecord)?.value())? == 4 { + if matches!( + number(tx.open_table(META)?.get("schema")?.ok_or(StoreError::CorruptRecord)?.value())?, + 4 | 6 + ) { return Err(StoreError::UnsupportedSchema); } drop(tx); @@ -1236,6 +1239,624 @@ mod tests { u19_assert_io(error); } + fn provenance_promote(store: &mut Store) { + let result = store.promote_created_provenance(); + assert_eq!(result.as_ref().map(|_| ()).map_err(StoreError::code), Ok(()), "U26_PROVENANCE_ENABLED"); + } + + fn provenance_assert(store: &Store, command: &DecideCreated, first: LogId, result: &CreatedResult) { + let actual = store + .created_operation_provenance(command.operation_id) + .expect("provenance read") + .expect("permanent operation"); + assert_eq!(actual.first_decision_log_id, first, "U26_FIRST_ID_STICKY"); + assert_eq!(actual.operation.canonical_command, command.encode().expect("canonical bytes")); + assert_eq!(&actual.operation.result, result); + assert_eq!( + store.created_operation(command.operation_id).expect("compatible projection"), + Some(actual.operation) + ); + } + + fn provenance_commands() -> Vec { + let success = u17a_command(); + let mut conflict = success.clone(); + conflict.binding_revision = 99; + let mut binding = conflict.clone(); + binding.operation_id = [11; 16]; + binding.expected_head = Some([12; 16]); + let mut head = success.clone(); + head.operation_id = [13; 16]; + let mut version = success.clone(); + version.operation_id = [14; 16]; + version.expected_head = Some(success.prepared.version_id); + vec![ + success.clone(), + success, + conflict, + binding.clone(), + head.clone(), + version.clone(), + binding, + head, + version, + ] + } + + fn provenance_history(native: bool) -> (tempfile::TempDir, CrashBackend, Store, Vec) { + let (dir, backend, mut store) = u17a_bound(); + store.save_vote(vote()).expect("nonempty provenance fixture vote"); + let commands = provenance_commands(); + if native { + store.enable_native_empty().expect("native fixture"); + store + .append_native( + &commands + .iter() + .enumerate() + .map(|(i, c)| u17a_created(u64::try_from(i).expect("index"), c.clone())) + .collect::>(), + ) + .expect("native prefix"); + } else { + store + .append( + &commands + .iter() + .enumerate() + .map(|(i, c)| LogEntry { + id: u17a_id(u64::try_from(i).expect("index")), + payload: c.encode().expect("legacy canonical"), + }) + .collect::>(), + ) + .expect("legacy prefix"); + } + store.save_committed(u17a_id(8)).expect("complete commit"); + for i in 0..9 { + if native { + store.apply_next_native(u17a_id(i)).expect("native apply"); + } else { + store.apply_next_created(u17a_id(i)).expect("legacy apply"); + } + } + (dir, backend, store, commands) + } + + #[test] + fn provenance_first_decisions_and_retries_reopen() { + let (_dir, backend, mut store) = u17a_store(); + provenance_promote(&mut store); + assert_eq!(committed_schema(&store), 6); + let commands = provenance_commands(); + let entries = commands + .iter() + .enumerate() + .map(|(i, c)| u17a_created(u64::try_from(i).expect("index"), c.clone())) + .collect::>(); + store.append_native(&entries).expect("Created entries"); + store.save_committed(u17a_id(8)).expect("committed Created"); + for i in 0..9 { + store.apply_next_native(u17a_id(i)).expect("decision and retry"); + } + let expected = [ + ( + 0usize, + 0u64, + CreatedResult::Created { + version_id: [3; 16], + event_id: [1; 16], + decision_log_id: u17a_id(0), + }, + ), + (3, 3, CreatedResult::BindingMismatch), + ( + 4, + 4, + CreatedResult::HeadMismatch { + actual_head: Some([3; 16]), + }, + ), + (5, 5, CreatedResult::VersionConflict), + ]; + for (index, first, result) in &expected { + provenance_assert(&store, &commands[*index], u17a_id(*first), result); + } + assert_eq!( + store + .native_apply_response(2) + .expect("conflict receipt") + .expect("receipt") + .created_result(), + Some(&CreatedResult::OperationConflict) + ); + drop(store); + let store = Store::open(&backend.path, identity()).expect("reopen full provenance replay"); + for (index, first, result) in &expected { + provenance_assert(&store, &commands[*index], u17a_id(*first), result); + } + assert_eq!(store.created_event(commands[3].operation_id).expect("negative event"), None); + assert_eq!(encode_id(u17a_id(3)).len(), 24); + } + + #[test] + fn provenance_native_controls_preserve_state() { + let (_dir, backend, mut store) = u17a_store(); + provenance_promote(&mut store); + let (member, membership) = u17a_membership(1); + let entries = vec![u17a_blank(0), member, u17a_created(2, u17a_command()), u17a_blank(3)]; + let writer = NativeLogWriter::new(store); + let runtime = u19_runtime(); + runtime.block_on(writer.append_native(entries.clone())).expect("async append"); + let responses = runtime + .block_on(writer.apply_committed(entries)) + .expect("async committed apply"); + assert_eq!(responses[0], CaptureResponse::blank(u17a_id(0)).expect("blank")); + assert_eq!(responses[1], CaptureResponse::membership(u17a_id(1)).expect("membership")); + assert_eq!( + runtime.block_on(writer.applied_state()).expect("applied pair"), + (Some(u17a_id(3)), membership.clone()) + ); + { + let store = writer.store.lock().expect("store"); + assert_eq!( + store + .database + .begin_read() + .expect("read") + .open_table(application::OPERATIONS) + .expect("operations") + .len() + .expect("count"), + 1 + ); + assert_eq!(store.created_apply_result(0).expect("blank projection"), None); + assert_eq!(store.created_apply_result(1).expect("membership projection"), None); + } + let view = runtime.block_on(writer.snapshot_view()).expect("schema6 native view"); + drop(view); + backend.cut(); + drop(writer); + let store = Store::open(&backend.path, identity()).expect("async ACK cut reopen"); + assert_eq!(store.native_application_state().expect("recovered pair"), (Some(u17a_id(3)), membership)); + for (i, response) in responses.iter().enumerate() { + assert_eq!( + store + .native_apply_response(u64::try_from(i).expect("index")) + .expect("receipt") + .as_ref(), + Some(response), + "U26_DURABLE_ACK" + ); + } + assert_eq!( + store + .created_operation_provenance([1; 16]) + .expect("durable operation") + .expect("operation") + .first_decision_log_id, + u17a_id(2), + "U26_DURABLE_ACK" + ); + } + + #[test] + fn provenance_promotes_schema3_and_schema4_exact_history() { + for native in [false, true] { + let (_dir, backend, mut store, commands) = provenance_history(native); + let old = u17b_snapshot(&store); + assert_eq!( + store + .created_operation_provenance([1; 16]) + .expect_err("old schema has no provenance") + .code(), + "unsupported_schema" + ); + provenance_promote(&mut store); + assert_eq!(committed_schema(&store), if native { 6 } else { 5 }); + let new = u17b_snapshot(&store); + let metadata_without_schema = |rows: &Vec<(Vec, Vec)>| { + rows.iter() + .filter(|(key, _)| key.as_slice() != b"schema") + .cloned() + .collect::>() + }; + assert_eq!( + metadata_without_schema(&old[0]), + metadata_without_schema(&new[0]), + "only schema metadata changes" + ); + for i in 1..6 { + assert_eq!(old[i], new[i], "promotion preserves logs, responses and business state"); + } + assert_eq!(old[7], new[7], "promotion preserves vote"); + for (old_row, new_row) in old[6].iter().zip(&new[6]) { + assert_eq!(old_row.0, new_row.0); + assert_eq!(&new_row.1[..5], b"COPR\x01"); + assert_eq!(&new_row.1[29..], old_row.1.as_slice(), "canonical old value must remain byte exact"); + } + for (i, first) in [(0usize, 0u64), (3, 3), (4, 4), (5, 5)] { + let row = store + .created_operation_provenance(commands[i].operation_id) + .expect("promoted read") + .expect("operation"); + assert_eq!(row.first_decision_log_id, u17a_id(first), "U26_PROMOTION_FIRST_ID"); + } + let once = u17b_snapshot(&store); + provenance_promote(&mut store); + assert_eq!(u17b_snapshot(&store), once); + drop(store); + let mut store = Store::open(&backend.path, identity()).expect("reopen promoted history"); + let mut suffix = commands[0].clone(); + suffix.operation_id = [20; 16]; + suffix.prepared.version_id = [21; 16]; + suffix.expected_head = Some([3; 16]); + if native { + store + .append_native(&[u17a_created(9, suffix.clone())]) + .expect("native suffix"); + } else { + store + .append(&[LogEntry { + id: u17a_id(9), + payload: suffix.encode().expect("suffix"), + }]) + .expect("legacy suffix"); + } + store.save_committed(u17a_id(9)).expect("suffix committed"); + if native { + store.apply_next_native(u17a_id(9)).expect("native suffix apply"); + } else { + store.apply_next_created(u17a_id(9)).expect("legacy suffix apply"); + } + assert_eq!( + store + .created_operation_provenance(suffix.operation_id) + .expect("suffix provenance") + .expect("suffix operation") + .first_decision_log_id, + u17a_id(9) + ); + } + } + + #[test] + fn provenance_rejects_incomplete_or_forged_prefix_atomically() { + use application::{APPLY_RESULTS, EVENTS, HEADS, OPERATIONS, VERSIONS}; + for native in [false, true] { + for corruption in 0..9 { + let (_dir, _backend, mut store, commands) = provenance_history(native); + let tx = store.database.begin_write().expect("corruption transaction"); + match corruption { + 0 => { + tx.open_table(LOGS).expect("logs").remove(0).expect("missing prefix"); + } + 1 => { + tx.open_table(APPLY_RESULTS) + .expect("responses") + .remove(3) + .expect("missing negative response"); + } + 2 => { + let mut logs = tx.open_table(LOGS).expect("logs"); + let raw = logs.get(3).expect("get").expect("row").value().to_vec(); + let mut entry = decode_log(&raw).expect("log"); + entry.id.term += 1; + logs.insert(3, encode_log(&entry).expect("encode").as_slice()) + .expect("wrong full ID"); + } + 3 => { + let mut operations = tx.open_table(OPERATIONS).expect("operations"); + let raw = operations + .get(commands[3].operation_id.as_slice()) + .expect("get") + .expect("row") + .value() + .to_vec(); + operations.insert([99; 16].as_slice(), raw.as_slice()).expect("orphan"); + } + 4 => { + tx.open_table(EVENTS) + .expect("events") + .remove([1; 16].as_slice()) + .expect("missing event"); + } + 5 => { + let mut meta = tx.open_table(META).expect("meta"); + meta.insert("last_applied", encode_id(u17a_id(7)).as_slice()) + .expect("wrong applied ID"); + } + 6 => { + tx.open_table(HEADS) + .expect("heads") + .remove(crate::created::object_bytes(&commands[0].object).expect("object").as_slice()) + .expect("missing head"); + } + 7 => { + let mut versions = tx.open_table(VERSIONS).expect("versions"); + let key = versions + .iter() + .expect("iter") + .next() + .expect("version") + .expect("row") + .0 + .value() + .to_vec(); + versions.remove(key.as_slice()).expect("missing version"); + } + _ => { + let mut logs = tx.open_table(LOGS).expect("logs"); + let raw = logs.get(2).expect("get").expect("row").value().to_vec(); + let mut entry = decode_log(&raw).expect("log"); + entry.payload = if native { + decode_log(logs.get(3).expect("source log").expect("source row").value()) + .expect("source entry") + .payload + } else { + commands[3].encode().expect("canonical") + }; + logs.insert(2, encode_log(&entry).expect("encode").as_slice()) + .expect("forged early conflict"); + } + } + tx.commit().expect("inject corruption"); + let before = u17b_snapshot(&store); + assert_eq!( + store + .promote_created_provenance() + .expect_err("corrupt promotion refused") + .code(), + "corrupt_record" + ); + assert_eq!(u17b_snapshot(&store), before, "rejection must have zero writes"); + } + } + } + + #[test] + fn provenance_empty_first_commit_never_downgrades() { + for native in [false, true] { + let (_dir, backend, mut store) = u17a_bound(); + provenance_promote(&mut store); + assert_eq!(committed_schema(&store), 5); + if native { + store.enable_native_empty().expect("empty5 explicit native6"); + store.append_native(&[u17a_blank(0)]).expect("blank"); + } else { + store + .append(&[LogEntry { + id: u17a_id(0), + payload: u17a_command().encode().expect("command"), + }]) + .expect("legacy"); + } + store.save_committed(u17a_id(0)).expect("first committed"); + assert_eq!(committed_schema(&store), if native { 6 } else { 5 }, "U26_SCHEMA_NEVER_DOWNGRADES"); + if native { + store.apply_next_native(u17a_id(0)).expect("blank apply"); + } else { + store.apply_next_created(u17a_id(0)).expect("legacy apply"); + assert_eq!( + store.enable_native_empty().expect_err("populated5 forbidden").code(), + "unsupported_schema" + ); + assert_eq!( + store + .migrate_legacy_created_to_native() + .expect_err("U17B cannot reinterpret5") + .code(), + "unsupported_schema" + ); + } + drop(store); + let store = Store::open(&backend.path, identity()).expect("reopen schema"); + assert_eq!(committed_schema(&store), if native { 6 } else { 5 }, "U26_SCHEMA_NEVER_DOWNGRADES"); + if !native { + let writer = NativeLogWriter::new(store); + assert!( + u19_runtime().block_on(writer.snapshot_view()).is_err(), + "schema5 cannot expose a native view" + ); + } + } + } + + #[test] + fn provenance_old_binary_and_unknown_versions_reject() { + for native in [false, true] { + for corruption in 0..10 { + let (_dir, backend, mut store, _) = provenance_history(native); + provenance_promote(&mut store); + let tx = store.database.begin_write().expect("corruption"); + if corruption == 0 { + tx.open_table(META) + .expect("meta") + .insert("schema", 7u64.to_be_bytes().as_slice()) + .expect("future schema"); + } else { + let mut operations = tx.open_table(application::OPERATIONS).expect("operations"); + let key = [11; 16]; + let mut raw = operations + .get(key.as_slice()) + .expect("get") + .expect("negative") + .value() + .to_vec(); + match corruption { + 1 => raw[4] = 2, + 2 => { + raw.pop(); + } + 3 => raw.push(0), + 4 => raw[5..29].fill(0), + 5 => raw[5..29].copy_from_slice(&encode_id(u17a_id(6))), + 6 => { + let mut id = u17a_id(3); + id.leader_node += 1; + raw[5..29].copy_from_slice(&encode_id(id)); + } + 7 => { + let mut id = u17a_id(3); + id.term += 1; + raw[5..29].copy_from_slice(&encode_id(id)); + } + 8 => raw[29..37].copy_from_slice(&u64::MAX.to_be_bytes()), + _ => raw = raw[29..].to_vec(), + } + operations.insert(key.as_slice(), raw.as_slice()).expect("bad envelope"); + } + tx.commit().expect("persist corruption"); + drop(store); + assert_eq!( + Store::open(&backend.path, identity()) + .err() + .expect("strict open rejects") + .code(), + if corruption == 0 { + "unsupported_schema" + } else { + "corrupt_record" + } + ); + } + } + // The independent accepted old-binary probe is run by the root task. + } + + #[test] + fn provenance_promotion_and_apply_powercut_atomic() { + for native in [false, true] { + for promotion in [false, true] { + let fixture = || { + let (dir, backend, mut store, _) = provenance_history(native); + if !promotion { + provenance_promote(&mut store); + let mut c = u17a_command(); + c.operation_id = [30; 16]; + c.binding_revision = 99; + if native { + store.append_native(&[u17a_created(9, c)]).expect("pending native"); + } else { + store + .append(&[LogEntry { + id: u17a_id(9), + payload: c.encode().expect("pending canonical"), + }]) + .expect("pending legacy"); + } + store.save_committed(u17a_id(9)).expect("pending commit"); + } + (dir, backend, store) + }; + let action = |store: &mut Store| -> Result<(), StoreError> { + if promotion { + store.promote_created_provenance() + } else if native { + store.apply_next_native(u17a_id(9)).map(|_| ()) + } else { + store.apply_next_created(u17a_id(9)).map(|_| ()) + } + }; + let (_dir, backend, mut store) = fixture(); + let old = u17b_snapshot(&store); + backend.media.lock().expect("reset syncs").syncs = 0; + let result = action(&mut store); + assert_eq!(result.as_ref().map(|_| ()).map_err(StoreError::code), Ok(()), "U26_PROVENANCE_ENABLED"); + let reached = backend.media.lock().expect("syncs").syncs; + assert_eq!(reached > 0, true, "U26_DURABLE_ACK"); + let new = u17b_snapshot(&store); + backend.cut(); + drop(store); + let database = Database::open(&backend.path).expect("ACK database recovery"); + assert_eq!(u17b_snapshot_database(&database), new, "U26_DURABLE_ACK"); + drop(database); + let (_dir, backend, mut store) = fixture(); + backend.arm(1, false, false); + assert_eq!( + action(&mut store).expect_err("uncertain provenance commit").code(), + "commit_indeterminate" + ); + let fenced = u17b_snapshot(&store); + let mutations: [fn(&mut Store) -> Result<(), StoreError>; 11] = [ + Store::promote_created_provenance, + |store| store.append(&[]), + |store| store.append_native(&[]), + |store| store.save_vote(vote()), + |store| store.save_committed(u17a_id(8)), + |store| store.apply_next_created(u17a_id(0)).map(|_| ()), + |store| store.apply_next_native(u17a_id(0)).map(|_| ()), + |store| store.pin_capture_binding(u17a_binding()), + Store::enable_native_empty, + Store::migrate_legacy_created_to_native, + |store| store.truncate_suffix(9), + ]; + for mutation in mutations { + assert_eq!( + mutation(&mut store) + .expect_err("uncertain handle must fence every mutation") + .code(), + "write_disabled", + "U26_WRITE_FENCE" + ); + assert_eq!(u17b_snapshot(&store), fenced, "fenced mutation must not change any table"); + } + let writer = NativeLogWriter::new(store); + let result = u19_runtime().block_on(writer.apply_committed(vec![])); + assert!( + matches!(result, Err(NativeWriteError::Store(StoreError::WriteDisabled))), + "U26_WRITE_FENCE async apply" + ); + assert_eq!(u17b_snapshot(&writer.store.lock().expect("fenced writer")), fenced); + drop(writer); + for sync in 1..=reached { + for after in [false, true] { + let (_dir, backend, mut store) = fixture(); + assert_eq!(u17b_snapshot(&store), old); + backend.arm(sync, after, true); + let result = action(&mut store); + let actual_syncs = backend.media.lock().expect("reached cut").syncs; + assert!(actual_syncs >= sync, "cut must actually reach target sync"); + backend.cut(); + drop(store); + match Database::open(&backend.path) { + Ok(database) => { + let actual = u17b_snapshot_database(&database); + assert!(actual == old || actual == new, "physical all-table atomicity"); + if result.is_ok() { + assert_eq!(actual, new, "U26_DURABLE_ACK"); + } + let recovered = Store::from_database(database, identity()) + .expect("logical corruption cannot hide mixed state"); + assert_eq!(u17b_snapshot(&recovered), actual); + } + Err(redb::DatabaseError::Storage(redb::StorageError::Corrupted(_))) => { + assert!(result.is_err(), "U26_DURABLE_ACK"); + } + Err(error) => panic!("unexpected physical recovery error: {error:?}"), + } + } + } + } + let (_dir, _backend, mut expected_store, _) = provenance_history(native); + provenance_promote(&mut expected_store); + let expected = u17b_snapshot(&expected_store); + let dir = tempfile::tempdir().expect("child provenance directory"); + let path = dir.path().join("db"); + let mut store = Store::initialize(&path, identity()).expect("child database"); + store.pin_capture_binding(u17a_binding()).expect("child binding"); + drop(store); + // Process termination after ACK is separate from the physical media cuts above. + let mut process = child(&path, if native { "provenance-native" } else { "provenance-legacy" }); + ack(&mut process); + process.kill().expect("SIGKILL acknowledged child"); + assert!(!process.wait().expect("reap killed child").success()); + let reopened = Store::open(&path, identity()).expect("independent child ACK reopen"); + assert_eq!(committed_schema(&reopened), if native { 6 } else { 5 }); + assert_eq!(u17b_snapshot(&reopened), expected, "U26_DURABLE_ACK complete child state"); + } + } + fn u17b_snapshot(store: &Store) -> Vec, Vec)>> { u17b_snapshot_database(&store.database) } @@ -1278,6 +1899,17 @@ mod tests { .collect(), ); } + snapshot.push( + tx.open_table(VOTE) + .expect("snapshot vote table") + .iter() + .expect("vote iterator") + .map(|row| { + let (key, value) = row.expect("vote row"); + (key.value().to_be_bytes().to_vec(), value.value().to_vec()) + }) + .collect(), + ); snapshot } fn u17b_history() -> (tempfile::TempDir, CrashBackend, Store, Vec) { @@ -3440,7 +4072,48 @@ mod tests { assert!(Store::open(path, identity()).is_err(), "U08_INVARIANT exclusive lock absent"); } else { let mut store = Store::open(path, identity()).expect("child open"); - if std::env::var("U08_CHILD_MODE").expect("mode") == "committed" { + let mode = std::env::var("U08_CHILD_MODE").expect("mode"); + if matches!(mode.as_str(), "provenance-native" | "provenance-legacy") { + store.save_vote(vote()).expect("child provenance vote ACK"); + let native = mode == "provenance-native"; + if native { + store.enable_native_empty().expect("child native enable"); + } + provenance_promote(&mut store); + let commands = provenance_commands(); + if native { + store + .append_native( + &commands + .iter() + .enumerate() + .map(|(i, command)| u17a_created(u64::try_from(i).expect("child index"), command.clone())) + .collect::>(), + ) + .expect("child native append"); + } else { + store + .append( + &commands + .iter() + .enumerate() + .map(|(i, command)| LogEntry { + id: u17a_id(u64::try_from(i).expect("child index")), + payload: command.encode().expect("child canonical"), + }) + .collect::>(), + ) + .expect("child legacy append"); + } + store.save_committed(u17a_id(8)).expect("child committed ACK"); + for index in 0..9 { + if native { + store.apply_next_native(u17a_id(index)).expect("child native apply ACK"); + } else { + store.apply_next_created(u17a_id(index)).expect("child legacy apply ACK"); + } + } + } else if mode == "committed" { store.save_committed(truncate_batch()[2].id).expect("child committed ACK"); } else if std::env::var("U08_CHILD_MODE").expect("mode") == "truncate" { store.truncate_suffix(2).expect("truncate ACK");