diff --git a/AGENTS.md b/AGENTS.md index 6e7bd1f9..33515427 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -43,6 +43,12 @@ Materialized graphs are optional readings. Continuum is the protocol for lawful causal-history exchange. ``` +## Trace Architecture Principle + +> Graph is truth. Delta log is transport. Roaring is index. Matrix is projection. Receipt is proof. + +Echo execution traces are not a separate event-sourcing reality. They are a bounded, append-only, canonical streaming log of WSC `NodeRow`, `EdgeRow`, and `AttRow` deltas. Trace storage is chunked. Sparse selector columns may be encoded as compressed bitsets, but they are derived indexes, not the source of truth. Prover-specific rectangular traces are downstream projections of the delta log. + ## Git Rules - **NEVER** amend commits. diff --git a/CHANGELOG.md b/CHANGELOG.md index d8b7f045..3db89c27 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,11 @@ ### Added +- `warp-core` now exposes a recovered WAL evidence segment catalog derived from + `RecoveryScanReport`, with a non-authoritative live cache in + `TrustedRuntimeWal` that marks cache-update failures as rebuild posture + without turning committed WAL transactions into failures. + - `warp-core` now exposes a narrow Edict `echo.span-ir/v1` Target IR fixture bridge that accepts strict lowercase digest-locked pre-step `continueObstructed` requirements, evaluates deterministic basis freshness @@ -847,6 +852,14 @@ Applied, Rejected, Obstructed}` with receipt evidence and typed contract ### Fixed +- `warp-core` recovered filesystem WAL ACK paths now rebuild the live evidence + catalog before returning recovered success, live catalog-update failures record + the last commit where the catalog was actually fresh, and committed evidence + segments now populate `coverings_by_range` for their exact LSN range. Rebuilt + evidence catalogs now reject malformed commit/frame evidence before admitting + `ExactCommittedWal` segments. +- `Cargo.lock` now pins `crossbeam-epoch` to `0.9.20`, clearing + `RUSTSEC-2026-0204` for the benchmark-only `rayon` dependency path. - `warp-core` evolving braid logs now reject unchecked incremental mutations: `Braid::apply` returns typed lifecycle errors, rejects duplicate member weaving and mixed revealed/sealed membership, refuses empty-frontier diff --git a/Cargo.lock b/Cargo.lock index 8839532d..60e3d6e6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -442,9 +442,9 @@ dependencies = [ [[package]] name = "crossbeam-epoch" -version = "0.9.18" +version = "0.9.20" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e" +checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f" dependencies = [ "crossbeam-utils", ] @@ -677,6 +677,10 @@ dependencies = [ "thiserror 1.0.69", ] +[[package]] +name = "echo-trace" +version = "0.1.0" + [[package]] name = "echo-ttd" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 433ffdc2..ed4ffbe9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -28,7 +28,8 @@ members = [ "crates/ttd-protocol-rs", "crates/echo-ttd", "crates/method", - "xtask" + "xtask", + "crates/echo-trace", ] resolver = "2" diff --git a/crates/echo-trace/Cargo.toml b/crates/echo-trace/Cargo.toml new file mode 100644 index 00000000..f93d3f7c --- /dev/null +++ b/crates/echo-trace/Cargo.toml @@ -0,0 +1,18 @@ +# SPDX-License-Identifier: Apache-2.0 +# © James Ross Ω FLYING•ROBOTS +[package] +name = "echo-trace" +version = "0.1.0" +edition = "2024" +license.workspace = true +repository.workspace = true +rust-version.workspace = true +description = "Canonical causal trace boundary for Echo graph delta streams" +readme = "README.md" +keywords = ["echo", "trace", "causal", "graph"] +categories = ["data-structures", "development-tools"] + +[dependencies] + +[lints] +workspace = true diff --git a/crates/echo-trace/README.md b/crates/echo-trace/README.md new file mode 100644 index 00000000..502b1f59 --- /dev/null +++ b/crates/echo-trace/README.md @@ -0,0 +1,11 @@ + + + +# echo-trace + +Canonical causal trace boundary for Echo graph delta streams. + +`echo-trace` defines the small trait and data shapes used by trace sinks that +consume ordered graph deltas, seal deterministic chunks, and return receipts. +The crate is intentionally transport-neutral: it names the boundary between a +causal producer and a trace sink without owning storage, replay, or rendering. diff --git a/crates/echo-trace/src/lib.rs b/crates/echo-trace/src/lib.rs new file mode 100644 index 00000000..2533a0e1 --- /dev/null +++ b/crates/echo-trace/src/lib.rs @@ -0,0 +1,117 @@ +// SPDX-License-Identifier: Apache-2.0 +// © James Ross Ω FLYING•ROBOTS + +//! Echo Trace +//! +//! Graph is truth. Delta log is transport. Roaring is index. +//! Matrix is projection. Receipt is proof. +//! +//! This crate defines the canonical causal delta stream boundary. + +/// Represents an execution tick identifier. +pub type TickId = u64; + +/// A digest representing a causal frontier in the WARP graph. +pub type FrontierDigest = [u8; 32]; + +/// The core graph delta, encapsulating the exact WSC rows appended in a tick. +/// Parameterized over the row types to avoid circular dependencies with `warp-core`. +pub struct CanonicalGraphDelta<'a, N, E, A> { + /// The tick at which this delta occurred. + pub tick: TickId, + /// The causal frontier digest before this delta. + pub causal_frontier_before: FrontierDigest, + /// The canonical NodeRows appended. + pub node_rows: &'a [N], + /// The canonical EdgeRows appended. + pub edge_rows: &'a [E], + /// The canonical AttRows appended. + pub att_rows: &'a [A], + /// The causal frontier digest after this delta. + pub causal_frontier_after: FrontierDigest, +} + +/// Metadata describing an execution trace run. +pub struct TraceRunMeta { + /// The starting tick of the execution trace run. + pub start_tick: TickId, + /// The root causal frontier digest. + pub root_frontier: FrontierDigest, +} + +/// A receipt for a sealed trace chunk. +pub struct TraceChunkReceipt { + /// The starting tick of this chunk. + pub start_tick: TickId, + /// The total number of ticks contained in this chunk. + pub tick_count: u64, + /// The cryptographic hash of the prior chunk. + pub prior_chunk_hash: [u8; 32], + /// The cryptographic hash of the current chunk. + pub chunk_hash: [u8; 32], +} + +/// A receipt for a completed trace run. +pub struct TraceRunReceipt { + /// The total number of chunks generated in the run. + pub chunks: usize, + /// The final cryptographic hash representing the entire trace run. + pub final_hash: [u8; 32], +} + +/// A generic error type for trace operations. +#[derive(Debug)] +pub enum TraceError { + /// An I/O error occurred while writing or reading the trace sink. + IoError, + /// The trace chunk could not be sealed. + SealError, + /// A causal boundary violation occurred. + BoundaryMismatch, +} + +/// The trace boundary trait. +/// +/// Consumes canonical graph deltas, bounded in chunked streams. +pub trait TraceSink { + /// Start a new trace run. + fn begin_run(&mut self, meta: &TraceRunMeta) -> Result<(), TraceError>; + + /// Append a canonical graph delta to the stream. + fn append_delta(&mut self, delta: &CanonicalGraphDelta<'_, N, E, A>) -> Result<(), TraceError>; + + /// Seal the current chunk and return its cryptographic receipt. + fn seal_chunk(&mut self) -> Result; + + /// Finish the entire trace run and return the final receipt. + fn finish_run(&mut self) -> Result; +} + +/// A baseline sink that incurs near-zero overhead and does nothing. +pub struct NullTraceSink; + +impl TraceSink for NullTraceSink { + fn begin_run(&mut self, _meta: &TraceRunMeta) -> Result<(), TraceError> { + Ok(()) + } + fn append_delta( + &mut self, + _delta: &CanonicalGraphDelta<'_, N, E, A>, + ) -> Result<(), TraceError> { + Ok(()) + } + fn seal_chunk(&mut self) -> Result { + Ok(TraceChunkReceipt { + start_tick: 0, + tick_count: 0, + prior_chunk_hash: [0; 32], + chunk_hash: [0; 32], + }) + } + fn finish_run(&mut self) -> Result { + Ok(TraceRunReceipt { + chunks: 0, + final_hash: [0; 32], + }) + } +} diff --git a/crates/warp-core/src/causal_wal.rs b/crates/warp-core/src/causal_wal.rs index bb713a67..8b433d88 100644 --- a/crates/warp-core/src/causal_wal.rs +++ b/crates/warp-core/src/causal_wal.rs @@ -850,6 +850,14 @@ impl WalTransactionCommit { self.compute_digest() } + pub(crate) fn validate_frame_evidence( + &self, + frames: &[WalFrame], + ) -> Result<(), WalValidationError> { + validate_transaction_frames(frames, self)?; + validate_transaction_semantics(frames, self.transaction_kind) + } + fn compute_digest(&self) -> Hash { let mut h = blake3::Hasher::new(); h.update(WAL_COMMIT_DOMAIN); @@ -4300,6 +4308,8 @@ pub enum FilesystemWalFaultTarget { AppendFrame, /// Fail the next commit flush before writing the commit marker. FlushCommit, + /// Fail after writing and syncing the next commit marker. + CommitMarkerSynced, /// Fail the next manifest publish before writing manifest material. PublishManifest, } @@ -4310,6 +4320,7 @@ pub enum FilesystemWalFaultTarget { pub struct FilesystemWalFaultPlan { append_frame: u32, flush_commit: u32, + commit_marker_synced: u32, publish_manifest: u32, } @@ -4322,16 +4333,25 @@ impl FilesystemWalFaultPlan { FilesystemWalFaultTarget::AppendFrame => Self { append_frame: 1, flush_commit: 0, + commit_marker_synced: 0, publish_manifest: 0, }, FilesystemWalFaultTarget::FlushCommit => Self { append_frame: 0, flush_commit: 1, + commit_marker_synced: 0, + publish_manifest: 0, + }, + FilesystemWalFaultTarget::CommitMarkerSynced => Self { + append_frame: 0, + flush_commit: 0, + commit_marker_synced: 1, publish_manifest: 0, }, FilesystemWalFaultTarget::PublishManifest => Self { append_frame: 0, flush_commit: 0, + commit_marker_synced: 0, publish_manifest: 1, }, } @@ -4341,6 +4361,7 @@ impl FilesystemWalFaultPlan { let remaining = match target { FilesystemWalFaultTarget::AppendFrame => &mut self.append_frame, FilesystemWalFaultTarget::FlushCommit => &mut self.flush_commit, + FilesystemWalFaultTarget::CommitMarkerSynced => &mut self.commit_marker_synced, FilesystemWalFaultTarget::PublishManifest => &mut self.publish_manifest, }; if *remaining == 0 { @@ -4590,6 +4611,15 @@ impl WalStorePort for FilesystemWalStore { FilesystemSyncBoundary::CommitFileSynced, transaction_id, )); + #[cfg(any(test, feature = "host_test"))] + if self + .fault_plan + .should_fail(FilesystemWalFaultTarget::CommitMarkerSynced) + { + return Err(WalStoreError::Io( + "injected filesystem WAL commit_marker_synced failure".to_owned(), + )); + } Ok(()) } diff --git a/crates/warp-core/src/evidence.rs b/crates/warp-core/src/evidence.rs new file mode 100644 index 00000000..23694ff2 --- /dev/null +++ b/crates/warp-core/src/evidence.rs @@ -0,0 +1,559 @@ +// SPDX-License-Identifier: Apache-2.0 +// © James Ross Ω FLYING•ROBOTS +//! Evidence and catalog layer for deriving causal segments from WAL history. + +use crate::causal_wal::{ + Lsn, RecoveryScanReport, RecoveryTailPosture, WalFrame, WalRecoveredTransaction, WalSegmentRef, + WalTransactionCommit, WalTransactionId, WalTransactionKind, WriterEpochId, +}; +use crate::wsc::WscStoreEnvelopeId; +use std::collections::BTreeMap; + +/// The unique identifier of a causal evidence segment. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct EvidenceSegmentId(pub u64); + +/// The kind of segment, denoting how it was derived and what it covers. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum EvidenceSegmentKind { + /// A base segment derived from a single committed WAL transaction. + CommittedTransaction, + /// A derived segment representing a sealed WAL storage segment. + WalStorageSegment, + /// A derived segment representing a checkpoint range. + CheckpointRange, + /// A derived segment representing a WSC bundle. + WscBundle, + /// A derived segment representing a ZK wormhole proof. + ZkWormhole, +} + +/// The compaction tier of an evidence segment. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum EvidenceTier { + /// Hot, uncompressed material (exact WAL/WSC representations). + Hot, + /// Warm, compressed material with derived indexes. + Warm, + /// Cold, proof-carrying wormhole. + Cold, +} + +/// The available posture of the material contained in an evidence segment. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum OpeningPosture { + /// Exact committed WAL representation available. + ExactCommittedWal, + /// Exact raw row/patch material available. + Exact, + /// Exact material available via sparse index lookup. + Indexed, + /// Compressed WSC material, requiring lazy open/replay. + CompressedButOpenable, + /// Proof only (ZK Wormhole), material unavailable. + ProofOnly, + /// Purged prior to a checkpoint, inherently un-openable. + PrunedWithCheckpoint, + /// Data missing unexpectedly (obstruction). + ObstructedMissingMaterial, + /// Data is corrupt and unreadable. + Corrupt, + /// Segment is known to exist but cannot be reached. + Unavailable, +} + +/// A contiguous semantic tick range covered by an evidence segment. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct TickRange { + /// The starting tick of the causal range. + pub start: u64, + /// The ending tick of the causal range. + pub end: u64, +} + +/// A key for querying derived segments by LSN range. +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct EvidenceRangeKey { + /// The first WAL Log Sequence Number in this segment's causal range. + pub first_lsn: Lsn, + /// The last WAL Log Sequence Number in this segment's causal range. + pub last_lsn: Lsn, +} + +/// A derived catalog entry mapping WAL ranges to causal state or proofs. +#[derive(Clone, Debug)] +pub struct CausalEvidenceSegment { + /// The local unique identifier for this segment. + pub id: EvidenceSegmentId, + /// The kind of this segment. + pub kind: EvidenceSegmentKind, + /// The writer epoch of the transactions covering this segment. + pub writer_epoch: WriterEpochId, + /// The first WAL Log Sequence Number in this segment's causal range. + pub first_lsn: Lsn, + /// The last WAL Log Sequence Number in this segment's causal range. + pub last_lsn: Lsn, + /// The transaction ID if this segment maps exactly to one transaction. + pub transaction_id: Option, + /// The transaction kind if this segment maps exactly to one transaction. + pub transaction_kind: Option, + /// The number of records in this segment. + pub record_count: u64, + /// The number of frames in this segment. + pub frame_count: u64, + /// The records root digest. + pub records_root: [u8; 32], + /// Digest of the previous committed transaction. + pub previous_committed_transaction_digest: [u8; 32], + /// Digest of the commit marker if this is a single transaction segment. + pub commit_digest: [u8; 32], + /// The semantic tick range covered, if applicable. + pub tick_range: Option, + /// The affected frontiers root recorded by the transaction commit marker. + pub affected_frontiers_root: [u8; 32], + /// References to the underlying WAL material backing this segment. + pub wal_segment_refs: Vec, + /// References to the deterministic WSC envelopes materialized for this segment. + pub wsc_envelope_refs: Vec, + /// The root digest of the sparse selector index (Roaring bitmap) for this segment. + pub selector_index_root: Option<[u8; 32]>, + /// The root digest of the retained evidence material. + pub retained_material_root: Option<[u8; 32]>, + /// The root of the ZK wormhole proof, if this is a Cold segment. + pub wormhole_proof_root: Option<[u8; 32]>, + /// The compaction tier of this segment. + pub tier: EvidenceTier, + /// The degree to which data within this segment can be queried. + pub opening_posture: OpeningPosture, +} + +/// Errors occurring during the extraction or indexing of causal evidence. +#[derive(Debug, thiserror::Error, PartialEq, Eq)] +pub enum EvidenceCatalogError { + /// The provided WAL transaction was invalid or malformed. + #[error("invalid WAL transaction evidence")] + InvalidTransaction, + /// A frame count mismatch occurred. + #[error("frame count mismatch: expected {expected}, observed {observed}")] + FrameCountMismatch { + /// The expected number of frames. + expected: u64, + /// The observed number of frames. + observed: usize, + }, +} + +/// A lightweight borrowed view of a committed transaction and its frames. +/// Works for both live `WalCommittedTransaction` and `WalRecoveredTransaction`. +pub struct CommittedWalView<'a> { + /// The transaction commit marker. + pub commit: &'a WalTransactionCommit, + /// The transaction frames. + pub frames: &'a [WalFrame], +} + +/// Observer trait for tailing WAL transactions into higher-level evidence. +pub trait CommittedWalObserver { + /// Observes a committed transaction, mutating the observer's internal state. + fn observe_committed_wal( + &mut self, + view: CommittedWalView<'_>, + ) -> Result<(), EvidenceCatalogError>; +} + +/// An indexed catalog mapping ranges of causal history to their underlying evidence. +#[derive(Clone, Debug)] +pub struct CausalSegmentCatalog { + /// All segments indexed by local segment ID. + pub segments_by_id: BTreeMap, + /// Index mapping commit digest to base segment ID. + pub base_by_commit_digest: BTreeMap<[u8; 32], EvidenceSegmentId>, + /// Index mapping transaction ID to base segment ID. + pub base_by_transaction_id: BTreeMap, + /// Index mapping start LSN to base segment ID. + pub base_by_lsn_start: BTreeMap, + /// Index mapping ranges to derived segment IDs covering that range. + pub coverings_by_range: BTreeMap>, + /// Next available local segment ID. + next_id: u64, +} + +impl CausalSegmentCatalog { + /// Creates a new, empty causal segment catalog. + #[must_use] + pub fn new() -> Self { + Self { + segments_by_id: BTreeMap::new(), + base_by_commit_digest: BTreeMap::new(), + base_by_transaction_id: BTreeMap::new(), + base_by_lsn_start: BTreeMap::new(), + coverings_by_range: BTreeMap::new(), + next_id: 0, + } + } + + /// Rebuilds a complete `CausalSegmentCatalog` from a recovery scan report. + /// This is a read-only projection over committed evidence that does not invoke live side effects. + pub fn from_recovery_scan(report: &RecoveryScanReport) -> Result { + let mut catalog = Self::new(); + for tx in &report.transactions { + catalog.observe_recovered_transaction(tx)?; + } + catalog.finish(report.tail_posture)?; + Ok(catalog) + } + + /// Observes a single recovered transaction. + pub fn observe_recovered_transaction( + &mut self, + tx: &WalRecoveredTransaction, + ) -> Result<(), EvidenceCatalogError> { + self.observe_committed_wal(CommittedWalView { + commit: &tx.commit, + frames: &tx.frames, + }) + } + + /// Inserts a base segment into the layered indexes. + pub fn insert_base_segment(&mut self, segment: CausalEvidenceSegment) { + let id = segment.id; + if segment.kind == EvidenceSegmentKind::CommittedTransaction { + self.base_by_commit_digest.insert(segment.commit_digest, id); + if let Some(tx_id) = segment.transaction_id { + self.base_by_transaction_id.insert(tx_id, id); + } + self.base_by_lsn_start.insert(segment.first_lsn, id); + self.coverings_by_range + .entry(EvidenceRangeKey { + first_lsn: segment.first_lsn, + last_lsn: segment.last_lsn, + }) + .or_default() + .push(id); + } + self.segments_by_id.insert(id, segment); + } + + /// Concludes the catalog build, optionally recording truncation intent based on tail posture. + pub fn finish( + &mut self, + _tail_posture: RecoveryTailPosture, + ) -> Result<(), EvidenceCatalogError> { + Ok(()) + } + + fn next_id(&mut self) -> EvidenceSegmentId { + let id = self.next_id; + self.next_id += 1; + EvidenceSegmentId(id) + } +} + +impl Default for CausalSegmentCatalog { + fn default() -> Self { + Self::new() + } +} + +impl CommittedWalObserver for CausalSegmentCatalog { + fn observe_committed_wal( + &mut self, + view: CommittedWalView<'_>, + ) -> Result<(), EvidenceCatalogError> { + let commit = view.commit; + let frame_count = view.frames.len() as u64; + + if view.commit.record_count != view.frames.len() as u64 { + return Err(EvidenceCatalogError::FrameCountMismatch { + expected: view.commit.record_count, + observed: view.frames.len(), + }); + } + commit + .validate_frame_evidence(view.frames) + .map_err(|_| EvidenceCatalogError::InvalidTransaction)?; + + let id = self.next_id(); + let segment = CausalEvidenceSegment { + id, + kind: EvidenceSegmentKind::CommittedTransaction, + writer_epoch: commit.writer_epoch, + first_lsn: commit.first_lsn, + last_lsn: commit.last_lsn, + transaction_id: Some(commit.transaction_id), + transaction_kind: Some(commit.transaction_kind), + record_count: commit.record_count, + frame_count, + records_root: commit.records_root, + affected_frontiers_root: commit.affected_frontiers_root, + previous_committed_transaction_digest: commit.previous_committed_transaction_digest, + commit_digest: commit.commit_digest, + tick_range: None, + wal_segment_refs: Vec::new(), + wsc_envelope_refs: Vec::new(), + selector_index_root: None, + retained_material_root: None, + wormhole_proof_root: None, + tier: EvidenceTier::Hot, + opening_posture: OpeningPosture::ExactCommittedWal, + }; + self.insert_base_segment(segment); + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::causal_wal::{ + AffectedFrontier, AffectedFrontierKind, PayloadCodecId, PayloadSchemaId, + WalAppendAuthority, WalDurabilityMode, WalRecordKind, WalSegmentId, WalTransactionBuilder, + }; + use crate::{causal_wal::Lsn, Hash}; + + fn test_hash(label: &[u8]) -> Hash { + *blake3::hash(label).as_bytes() + } + + fn test_wal_shape( + tx_kind: WalTransactionKind, + ) -> (WalAppendAuthority, WalRecordKind, AffectedFrontierKind) { + match tx_kind { + WalTransactionKind::SubmissionIntake => ( + WalAppendAuthority::SubmissionIntake, + WalRecordKind::SubmissionAcceptedRecorded, + AffectedFrontierKind::SubmissionQueue, + ), + WalTransactionKind::SchedulerTick => ( + WalAppendAuthority::TrustedScheduler, + WalRecordKind::TickReceiptRecorded, + AffectedFrontierKind::ReceiptIndex, + ), + WalTransactionKind::RuntimePosture => ( + WalAppendAuthority::RuntimeControl, + WalRecordKind::TrustedRuntimeControlRecorded, + AffectedFrontierKind::RuntimeControl, + ), + WalTransactionKind::Checkpoint => ( + WalAppendAuthority::Recovery, + WalRecordKind::CheckpointPublicationRecorded, + AffectedFrontierKind::CheckpointIndex, + ), + WalTransactionKind::MaterializationOutbox => ( + WalAppendAuthority::TrustedScheduler, + WalRecordKind::MaterializationIntentRecorded, + AffectedFrontierKind::ReceiptIndex, + ), + WalTransactionKind::TopologyIntent => ( + WalAppendAuthority::TrustedScheduler, + WalRecordKind::TopologyStrandForkRecorded, + AffectedFrontierKind::TopologyIndex, + ), + } + } + + fn make_test_commit( + tx_kind: WalTransactionKind, + tx_id_hash: &[u8], + start_lsn: Lsn, + ) -> WalRecoveredTransaction { + let (authority, record_kind, frontier_kind) = test_wal_shape(tx_kind); + let transaction_id = WalTransactionId::from_hash(test_hash(tx_id_hash)); + let mut builder = WalTransactionBuilder::new( + WriterEpochId::from_hash(test_hash(b"test-writer-epoch")), + WalSegmentId::from_raw(1), + transaction_id, + tx_kind, + authority, + start_lsn, + test_hash(b"previous-frame"), + test_hash(b"previous-commit"), + WalDurabilityMode::StrictFilesystem, + PayloadCodecId::from_hash(test_hash(b"payload-codec")), + PayloadSchemaId::from_hash(test_hash(b"payload-schema")), + 1, + 1, + test_hash(b"digest-domain"), + ); + builder + .push_record(record_kind, tx_id_hash.to_vec()) + .expect("test WAL record should build"); + let transaction = builder + .commit(vec![AffectedFrontier { + kind: frontier_kind, + before_digest: test_hash(b"frontier-before"), + after_digest: test_hash(b"frontier-after"), + }]) + .expect("test WAL transaction should validate"); + WalRecoveredTransaction { + commit: transaction.commit, + frames: transaction.frames, + } + } + + #[test] + fn test_committed_transactions_become_catalog_segments() { + let tx = make_test_commit( + WalTransactionKind::SubmissionIntake, + b"tx1", + Lsn::from_raw(1), + ); + let report = RecoveryScanReport { + transactions: vec![tx], + tail_posture: RecoveryTailPosture::Clean, + }; + + let catalog = CausalSegmentCatalog::from_recovery_scan(&report).unwrap(); + assert_eq!(catalog.segments_by_id.len(), 1); + let segment = catalog.segments_by_id.values().next().unwrap(); + assert_eq!(segment.kind, EvidenceSegmentKind::CommittedTransaction); + } + + #[test] + fn test_multiple_transactions_produce_multiple_segments() { + let tx1 = make_test_commit( + WalTransactionKind::SubmissionIntake, + b"tx1", + Lsn::from_raw(1), + ); + let tx2 = make_test_commit(WalTransactionKind::SchedulerTick, b"tx2", Lsn::from_raw(2)); + + let report = RecoveryScanReport { + transactions: vec![tx1.clone(), tx2.clone()], + tail_posture: RecoveryTailPosture::Clean, + }; + + let catalog = CausalSegmentCatalog::from_recovery_scan(&report).unwrap(); + assert_eq!(catalog.segments_by_id.len(), 2); + + let seg1 = catalog + .base_by_commit_digest + .get(&tx1.commit.commit_digest) + .and_then(|id| catalog.segments_by_id.get(id)) + .unwrap(); + assert_eq!( + seg1.transaction_kind, + Some(WalTransactionKind::SubmissionIntake) + ); + + let seg2 = catalog + .base_by_commit_digest + .get(&tx2.commit.commit_digest) + .and_then(|id| catalog.segments_by_id.get(id)) + .unwrap(); + assert_eq!( + seg2.transaction_kind, + Some(WalTransactionKind::SchedulerTick) + ); + } + + #[test] + fn test_base_segment_preserves_transaction_properties() { + let tx = make_test_commit( + WalTransactionKind::SubmissionIntake, + b"tx1", + Lsn::from_raw(1), + ); + let report = RecoveryScanReport { + transactions: vec![tx.clone()], + tail_posture: RecoveryTailPosture::Clean, + }; + let catalog = CausalSegmentCatalog::from_recovery_scan(&report).unwrap(); + let segment = catalog.segments_by_id.values().next().unwrap(); + + // base segment preserves transaction_kind + assert_eq!(segment.transaction_kind, Some(tx.commit.transaction_kind)); + // base segment lsn range equals commit.first_lsn..commit.last_lsn + assert_eq!(segment.first_lsn, tx.commit.first_lsn); + assert_eq!(segment.last_lsn, tx.commit.last_lsn); + // base segment records_root equals commit.records_root + assert_eq!(segment.records_root, tx.commit.records_root); + // base segment affected_frontiers_root equals commit.affected_frontiers_root + assert_eq!( + segment.affected_frontiers_root, + tx.commit.affected_frontiers_root + ); + } + + #[test] + fn test_base_segments_populate_covering_range_index() { + let tx = make_test_commit( + WalTransactionKind::SubmissionIntake, + b"tx1", + Lsn::from_raw(7), + ); + let report = RecoveryScanReport { + transactions: vec![tx.clone()], + tail_posture: RecoveryTailPosture::Clean, + }; + + let catalog = CausalSegmentCatalog::from_recovery_scan(&report).unwrap(); + let segment = catalog.segments_by_id.values().next().unwrap(); + let range_key = EvidenceRangeKey { + first_lsn: tx.commit.first_lsn, + last_lsn: tx.commit.last_lsn, + }; + + assert_eq!( + catalog + .coverings_by_range + .get(&range_key) + .map(Vec::as_slice), + Some(&[segment.id][..]) + ); + } + + #[test] + fn test_malformed_recovered_transaction_is_rejected() { + let mut tx = make_test_commit( + WalTransactionKind::SubmissionIntake, + b"tx1", + Lsn::from_raw(7), + ); + tx.commit.commit_digest = test_hash(b"forged-commit-digest"); + let report = RecoveryScanReport { + transactions: vec![tx], + tail_posture: RecoveryTailPosture::Clean, + }; + + assert!(matches!( + CausalSegmentCatalog::from_recovery_scan(&report), + Err(EvidenceCatalogError::InvalidTransaction) + )); + } + + #[test] + fn test_uncommitted_tail_frames_do_not_become_catalog_segments() { + let report = RecoveryScanReport { + transactions: vec![], + tail_posture: RecoveryTailPosture::WouldTruncateAfter(Lsn::from_raw(5)), + }; + + let catalog = CausalSegmentCatalog::from_recovery_scan(&report).unwrap(); + assert!(catalog.segments_by_id.is_empty()); + } + + #[test] + fn test_rebuilding_catalog_yields_identical_segments() { + let tx1 = make_test_commit( + WalTransactionKind::SubmissionIntake, + b"tx1", + Lsn::from_raw(1), + ); + let tx2 = make_test_commit(WalTransactionKind::SchedulerTick, b"tx2", Lsn::from_raw(2)); + let report = RecoveryScanReport { + transactions: vec![tx1, tx2], + tail_posture: RecoveryTailPosture::Clean, + }; + + let catalog1 = CausalSegmentCatalog::from_recovery_scan(&report).unwrap(); + let catalog2 = CausalSegmentCatalog::from_recovery_scan(&report).unwrap(); + + assert_eq!(catalog1.segments_by_id.len(), catalog2.segments_by_id.len()); + for (id, seg1) in &catalog1.segments_by_id { + let seg2 = catalog2.segments_by_id.get(id).unwrap(); + assert_eq!(seg1.commit_digest, seg2.commit_digest); + assert_eq!(seg1.first_lsn, seg2.first_lsn); + } + } +} diff --git a/crates/warp-core/src/lib.rs b/crates/warp-core/src/lib.rs index 383acbbc..bbdf40b1 100644 --- a/crates/warp-core/src/lib.rs +++ b/crates/warp-core/src/lib.rs @@ -56,6 +56,7 @@ pub mod domain; mod dynamic_binding; mod edict_target_ir; mod engine_impl; +pub mod evidence; mod footprint; /// Footprint enforcement guard for parallel execution. /// @@ -399,8 +400,9 @@ pub use tick_patch::{ }; #[cfg(all(feature = "native_rule_bootstrap", feature = "trusted_runtime"))] pub use trusted_runtime_host::{ - TrustedRuntimeApp, TrustedRuntimeHost, TrustedRuntimeHostError, TrustedRuntimeHostRunReport, - TrustedRuntimeWal, TrustedRuntimeWalConfig, TrustedRuntimeWalError, TrustedRuntimeWalStoreKind, + EvidenceCatalogPosture, TrustedRuntimeApp, TrustedRuntimeHost, TrustedRuntimeHostError, + TrustedRuntimeHostRunReport, TrustedRuntimeWal, TrustedRuntimeWalConfig, + TrustedRuntimeWalError, TrustedRuntimeWalStoreKind, }; pub use tx::TxId; pub use warp_state::{WarpInstance, WarpState}; diff --git a/crates/warp-core/src/trusted_runtime_host.rs b/crates/warp-core/src/trusted_runtime_host.rs index d83446f9..3d9ee78f 100644 --- a/crates/warp-core/src/trusted_runtime_host.rs +++ b/crates/warp-core/src/trusted_runtime_host.rs @@ -83,6 +83,9 @@ pub enum TrustedRuntimeWalError { /// WAL recovery failed while rebuilding runtime evidence. #[error("trusted runtime WAL recovery error: {0}")] Recovery(#[from] WalRecoveryError), + /// Evidence catalog operations failed. + #[error("trusted runtime evidence catalog error: {0}")] + EvidenceCatalog(#[from] crate::evidence::EvidenceCatalogError), /// Runtime outcome evidence could not be matched to the receipt correlation. #[error( "trusted runtime WAL tick outcome unavailable for submission {submission_id:?} receipt {receipt_digest:?}" @@ -509,10 +512,28 @@ impl TrustedRuntimeHost { } } +/// Represents the live cache posture of the derived evidence catalog. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum EvidenceCatalogPosture { + /// The live catalog is synchronized with committed history. + Fresh, + /// The live catalog failed to update and requires recovery rebuild. + NeedsRebuild { + /// A digest indicating the reason for the rebuild. + reason: Hash, + /// The last transaction digest where the catalog was known to be fresh. + last_good_commit: Hash, + }, +} + /// Minimal trusted-runtime WAL adapter for ACK-boundary integration tests. #[derive(Clone, Debug)] pub struct TrustedRuntimeWal { store: TrustedRuntimeWalStore, + evidence_catalog: Option, + evidence_catalog_posture: EvidenceCatalogPosture, + #[cfg(any(test, feature = "host_test"))] + fail_next_evidence_catalog_update: bool, writer_epoch: WriterEpochId, segment_id: WalSegmentId, next_lsn: Lsn, @@ -541,8 +562,10 @@ impl TrustedRuntimeWal { pub fn from_config(config: TrustedRuntimeWalConfig) -> Result { let TrustedRuntimeWalConfig { store, next_lsn } = config; let mut store = TrustedRuntimeWalStore::open(store)?; - let recovered_cursor = - TrustedRuntimeWalCursor::from_recovery(&store.recover_for_writer()?)?; + let recovery_report = store.recover_for_writer()?; + let recovered_cursor = TrustedRuntimeWalCursor::from_recovery(&recovery_report)?; + let evidence_catalog = + crate::evidence::CausalSegmentCatalog::from_recovery_scan(&recovery_report)?; let next_lsn = if recovered_cursor.has_committed_history { recovered_cursor.next_lsn } else { @@ -579,6 +602,10 @@ impl TrustedRuntimeWal { submission_frontier_digest: recovered_cursor.submission_frontier_digest, receipt_frontier_digest: recovered_cursor.receipt_frontier_digest, runtime_state_frontier_digest: recovered_cursor.runtime_state_frontier_digest, + evidence_catalog: Some(evidence_catalog), + evidence_catalog_posture: EvidenceCatalogPosture::Fresh, + #[cfg(any(test, feature = "host_test"))] + fail_next_evidence_catalog_update: false, }) } @@ -620,6 +647,27 @@ impl TrustedRuntimeWal { }) } + /// Recovers the causal segment catalog from committed WAL transactions. + pub fn recover_evidence_catalog_read_only( + &self, + ) -> Result { + let report = self.store.recover_read_only()?; + crate::evidence::CausalSegmentCatalog::from_recovery_scan(&report) + .map_err(TrustedRuntimeWalError::EvidenceCatalog) + } + + /// Returns the live evidence catalog if it exists. + #[must_use] + pub fn evidence_catalog(&self) -> Option<&crate::evidence::CausalSegmentCatalog> { + self.evidence_catalog.as_ref() + } + + /// Returns the current posture of the live evidence catalog cache. + #[must_use] + pub fn evidence_catalog_posture(&self) -> &EvidenceCatalogPosture { + &self.evidence_catalog_posture + } + /// Returns the number of committed submission-intake transactions. #[must_use] pub fn submission_acceptance_count(&self) -> usize { @@ -674,6 +722,18 @@ impl TrustedRuntimeWal { fn refresh_cursor_from_store_for_writer(&mut self) -> Result<(), TrustedRuntimeWalError> { let report = self.store.recover_for_writer()?; let cursor = TrustedRuntimeWalCursor::from_recovery(&report)?; + match crate::evidence::CausalSegmentCatalog::from_recovery_scan(&report) { + Ok(catalog) => { + self.evidence_catalog = Some(catalog); + self.evidence_catalog_posture = EvidenceCatalogPosture::Fresh; + } + Err(_) => { + self.evidence_catalog_posture = EvidenceCatalogPosture::NeedsRebuild { + reason: *blake3::hash(b"catalog_recovery_rebuild_error").as_bytes(), + last_good_commit: self.previous_committed_transaction_digest, + }; + } + } self.next_lsn = cursor.next_lsn; self.previous_frame_digest = cursor.previous_frame_digest; self.previous_committed_transaction_digest = cursor.previous_committed_transaction_digest; @@ -818,14 +878,44 @@ impl TrustedRuntimeWal { .last_lsn .checked_next() .ok_or(WalBuildError::LsnOverflow)?; + let last_good_commit = self.previous_committed_transaction_digest; let commit = transaction.commit.clone(); + let frames = transaction.frames.clone(); self.store.append_transaction(transaction)?; self.next_lsn = next_lsn; self.previous_frame_digest = last_frame_digest; self.previous_committed_transaction_digest = commit.commit_digest; + self.try_update_evidence_catalog_after_commit(&commit, &frames, last_good_commit); Ok(commit) } + fn try_update_evidence_catalog_after_commit( + &mut self, + commit: &WalTransactionCommit, + frames: &[crate::causal_wal::WalFrame], + last_good_commit: Hash, + ) { + use crate::evidence::CommittedWalObserver; + #[cfg(any(test, feature = "host_test"))] + if self.fail_next_evidence_catalog_update { + self.fail_next_evidence_catalog_update = false; + self.evidence_catalog_posture = EvidenceCatalogPosture::NeedsRebuild { + reason: *blake3::hash(b"catalog_update_error").as_bytes(), + last_good_commit, + }; + return; + } + if let Some(catalog) = self.evidence_catalog.as_mut() { + let view = crate::evidence::CommittedWalView { commit, frames }; + if catalog.observe_committed_wal(view).is_err() { + self.evidence_catalog_posture = EvidenceCatalogPosture::NeedsRebuild { + reason: *blake3::hash(b"catalog_update_error").as_bytes(), + last_good_commit, + }; + } + } + } + fn builder( &self, kind: WalTransactionKind, @@ -1225,6 +1315,11 @@ impl TrustedRuntimeWal { self.previous_frame_digest = last_frame_digest; Ok(()) } + + /// Forces the next live evidence catalog update to fail after WAL commit. + pub fn fail_next_evidence_catalog_update_for_test(&mut self) { + self.fail_next_evidence_catalog_update = true; + } } /// App-facing handle for a trusted local runtime host. diff --git a/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs b/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs index f5d9e30a..1b30580e 100644 --- a/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs +++ b/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs @@ -1520,3 +1520,151 @@ fn runtime_wal_ack_recover_read_only_exposes_recovery_certificate() { assert!(recovery.certificate.first_lsn.is_some()); assert!(recovery.certificate.last_lsn.is_some()); } + +use warp_core::EvidenceCatalogPosture; + +#[test] +fn runtime_wal_live_evidence_catalog_matches_read_only_recovery_after_submission() { + let (runtime, worldline_id) = runtime(); + let mut host = + TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize"); + host.enable_in_memory_runtime_wal() + .expect("runtime WAL should initialize"); + host.register_contract_package(package()) + .expect("host should install package"); + + let _submission = { + let mut app = host.app(); + app.submit_intent_with_runtime_wal_ack(eint_envelope(worldline_id)) + .expect("submission acceptance should commit before ACK") + }; + + let wal = host.runtime_wal().expect("runtime WAL should exist"); + + // Live catalog must exist and be fresh + let live_catalog = wal.evidence_catalog().expect("live catalog should exist"); + assert_eq!( + *wal.evidence_catalog_posture(), + EvidenceCatalogPosture::Fresh, + "live catalog should be fresh" + ); + + // Recovered catalog from pristine read-only scan + let recovered_catalog = wal + .recover_evidence_catalog_read_only() + .expect("rebuild should succeed"); + + assert_eq!( + live_catalog.segments_by_id.len(), + 1, + "submission should produce 1 base segment" + ); + assert_eq!( + live_catalog.segments_by_id.len(), + recovered_catalog.segments_by_id.len(), + "live catalog must match recovered catalog length" + ); + + for (id, live_seg) in &live_catalog.segments_by_id { + let rec_seg = recovered_catalog + .segments_by_id + .get(id) + .expect("recovered catalog missing segment"); + assert_eq!( + live_seg.commit_digest, rec_seg.commit_digest, + "segment commit digest must match" + ); + } +} + +#[test] +fn runtime_wal_live_evidence_catalog_rebuilds_after_recovered_filesystem_ack() { + let wal_root = temp_runtime_wal_dir("catalog-recovered-filesystem-ack"); + let (runtime, worldline_id) = runtime(); + let mut host = + TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize"); + host.enable_runtime_wal( + TrustedRuntimeWalConfig::filesystem_with_fault_plan_for_test( + &wal_root, + FilesystemWalFaultPlan::fail_next(FilesystemWalFaultTarget::CommitMarkerSynced), + ), + ) + .expect("host should configure faulting filesystem runtime WAL adapter"); + + let _submission = { + let mut app = host.app(); + app.submit_intent_with_runtime_wal_ack(eint_envelope(worldline_id)) + .expect("recoverable commit marker failure should still ACK") + }; + + let wal = host.runtime_wal().expect("runtime WAL should exist"); + assert_eq!( + *wal.evidence_catalog_posture(), + EvidenceCatalogPosture::Fresh, + "recovered ACK should leave the live catalog fresh" + ); + + let live_catalog = wal.evidence_catalog().expect("live catalog should exist"); + let recovered_catalog = wal + .recover_evidence_catalog_read_only() + .expect("read-only rebuild should succeed"); + assert_eq!( + recovered_catalog.segments_by_id.len(), + 1, + "recovered WAL should contain the committed submission segment" + ); + assert_eq!( + live_catalog.segments_by_id.len(), + recovered_catalog.segments_by_id.len(), + "live catalog must include the recovered committed submission" + ); +} + +#[test] +fn runtime_wal_live_evidence_catalog_failure_marks_needs_rebuild_without_failing_commit() { + let (runtime, worldline_a, worldline_b) = runtime_pair(); + let mut host = + TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize"); + host.enable_in_memory_runtime_wal() + .expect("runtime WAL should initialize"); + + let _first = { + let mut app = host.app(); + app.submit_intent_with_runtime_wal_ack(eint_envelope(worldline_a)) + .expect("first submission acceptance should commit before ACK") + }; + let last_good_commit = host + .runtime_wal() + .expect("runtime WAL should exist") + .commits() + .last() + .expect("first commit should exist") + .commit_digest; + + let mut faulting_wal = host + .runtime_wal() + .expect("runtime WAL should exist") + .clone(); + faulting_wal.fail_next_evidence_catalog_update_for_test(); + host.replace_runtime_wal_for_test(faulting_wal); + + let _second = { + let mut app = host.app(); + app.submit_intent_with_runtime_wal_ack(eint_envelope(worldline_b)) + .expect("catalog update failure should not reject committed WAL acceptance") + }; + + let wal = host.runtime_wal().expect("runtime WAL should exist"); + assert_eq!( + wal.commits().len(), + 2, + "WAL commit should succeed even when the derived catalog fails" + ); + assert!(matches!( + wal.evidence_catalog_posture(), + EvidenceCatalogPosture::NeedsRebuild { + last_good_commit: observed, + .. + } if *observed == last_good_commit + )); +}