From eea3466c7ed732f76ec6b0f9571568b23c954226 Mon Sep 17 00:00:00 2001 From: James Ross Date: Thu, 30 Jul 2026 05:45:47 -0700 Subject: [PATCH 01/11] test: require durable filesystem writer epochs --- .../tests/causal_wal_hardening_tests.rs | 184 ++++++++++++++++++ 1 file changed, 184 insertions(+) diff --git a/crates/warp-core/tests/causal_wal_hardening_tests.rs b/crates/warp-core/tests/causal_wal_hardening_tests.rs index 3e17d7c0..27adb825 100644 --- a/crates/warp-core/tests/causal_wal_hardening_tests.rs +++ b/crates/warp-core/tests/causal_wal_hardening_tests.rs @@ -50,6 +50,7 @@ use std::fs::{self, OpenOptions}; use std::io::ErrorKind; use std::io::{Read, Write}; use std::path::{Path, PathBuf}; +use std::process::Command; use std::sync::atomic::{AtomicU64, Ordering}; const WAL_SEGMENT_RECORD_MAGIC: &[u8; 8] = b"ECWALR1!"; @@ -881,6 +882,189 @@ fn writer_epoch_chain_gap_rejected() { assert!(matches!(error, WalStoreError::WriterEpochChainGap)); } +#[test] +fn filesystem_closed_writer_epoch_chain_survives_reopen() { + let root = temp_wal_root("writer-epoch-closed-reopen"); + let transaction = submission_transaction("epoch-closed-reopen", Lsn::from_raw(0)); + let previous_commit_digest = transaction.commit.commit_digest; + { + let mut store = must_ok(FilesystemWalStore::open(&root, WalSegmentId::from_raw(1))); + must_ok(store.acquire_writer_epoch(writer_epoch_request())); + must_ok(store.append_transaction(transaction)); + must_ok(store.close_epoch(epoch_id())); + } + + let mut reopened = must_ok(FilesystemWalStore::open(&root, WalSegmentId::from_raw(1))); + let epoch = must_ok(reopened.acquire_writer_epoch(writer_epoch_request_for( + "2", + Lsn::from_raw(2), + Some(epoch_id()), + Some(previous_commit_digest), + ))); + + assert_eq!(epoch.epoch_id, epoch_id_for("2")); + assert_eq!(epoch.previous_epoch_id, Some(epoch_id())); + assert_eq!( + epoch.previous_epoch_final_commit_digest, + Some(previous_commit_digest) + ); + drop(reopened); + must_ok(fs::remove_dir_all(root)); +} + +#[test] +fn filesystem_active_writer_epoch_and_final_commit_survive_reopen() { + let root = temp_wal_root("writer-epoch-active-reopen"); + { + let mut store = must_ok(FilesystemWalStore::open(&root, WalSegmentId::from_raw(1))); + must_ok(store.acquire_writer_epoch(writer_epoch_request())); + must_ok(store.append_transaction(submission_transaction( + "epoch-active-reopen", + Lsn::from_raw(0), + ))); + } + + let mut reopened = must_ok(FilesystemWalStore::open(&root, WalSegmentId::from_raw(1))); + let error = must_err( + reopened.acquire_writer_epoch(writer_epoch_request()), + "recovered active writer epoch must refuse overlap", + ); + + assert!(matches!(error, WalStoreError::WriterEpochAlreadyActive)); + drop(reopened); + must_ok(fs::remove_dir_all(root)); +} + +#[test] +fn duplicate_writer_epoch_id_is_a_chain_gap() { + let mut store = InMemoryWalStore::new(); + must_ok(store.acquire_writer_epoch(writer_epoch_request())); + let transaction = submission_transaction("epoch-duplicate", Lsn::from_raw(0)); + let previous_commit_digest = transaction.commit.commit_digest; + must_ok(store.append_transaction(transaction)); + must_ok(store.close_epoch(epoch_id())); + let mut duplicate = writer_epoch_request_for( + "2", + Lsn::from_raw(2), + Some(epoch_id()), + Some(previous_commit_digest), + ); + duplicate.epoch_id = epoch_id(); + + let error = must_err( + store.acquire_writer_epoch(duplicate), + "duplicate writer epoch id must reject", + ); + + assert!(matches!(error, WalStoreError::WriterEpochChainGap)); +} + +#[test] +fn stale_writer_epoch_link_is_a_chain_gap() { + let mut store = InMemoryWalStore::new(); + must_ok(store.acquire_writer_epoch(writer_epoch_request())); + let first = submission_transaction("epoch-stale-first", Lsn::from_raw(0)); + let first_commit_digest = first.commit.commit_digest; + must_ok(store.append_transaction(first)); + must_ok(store.close_epoch(epoch_id())); + must_ok(store.acquire_writer_epoch(writer_epoch_request_for( + "2", + Lsn::from_raw(2), + Some(epoch_id()), + Some(first_commit_digest), + ))); + must_ok(store.close_epoch(epoch_id_for("2"))); + + let error = must_err( + store.acquire_writer_epoch(writer_epoch_request_for( + "3", + Lsn::from_raw(3), + Some(epoch_id()), + Some(first_commit_digest), + )), + "a successor must cite the latest closed writer epoch", + ); + + assert!(matches!(error, WalStoreError::WriterEpochChainGap)); +} + +#[test] +fn fixed_seed_filesystem_writer_epoch_chain_survives_bounded_reopens() { + const EPOCH_COUNT: u64 = 16; + + let root = temp_wal_root("writer-epoch-fixed-seed"); + let mut previous_epoch_id = None; + for ordinal in 0..EPOCH_COUNT { + let mut store = must_ok(FilesystemWalStore::open(&root, WalSegmentId::from_raw(1))); + let label = format!("fixed-seed-{ordinal:02}"); + let request = + writer_epoch_request_for(&label, Lsn::from_raw(ordinal), previous_epoch_id, None); + let epoch = must_ok(store.acquire_writer_epoch(request)); + must_ok(store.close_epoch(epoch.epoch_id)); + previous_epoch_id = Some(epoch.epoch_id); + } + + assert_eq!(previous_epoch_id, Some(epoch_id_for("fixed-seed-15"))); + must_ok(fs::remove_dir_all(root)); +} + +#[test] +#[ignore = "child entrypoint exercised by the independent-process writer-epoch test"] +fn emit_filesystem_writer_epoch_process_step() { + let root = PathBuf::from( + std::env::var_os("ECHO_WRITER_EPOCH_TEST_ROOT") + .unwrap_or_else(|| panic!("child writer-epoch root is required")), + ); + let phase = std::env::var("ECHO_WRITER_EPOCH_TEST_PHASE") + .unwrap_or_else(|error| panic!("child writer-epoch phase is required: {error}")); + let mut store = must_ok(FilesystemWalStore::open(&root, WalSegmentId::from_raw(1))); + match phase.as_str() { + "first" => { + must_ok(store.acquire_writer_epoch(writer_epoch_request())); + must_ok( + store.append_transaction(submission_transaction("epoch-process", Lsn::from_raw(0))), + ); + must_ok(store.close_epoch(epoch_id())); + } + "second" => { + let previous_commit_digest = submission_transaction("epoch-process", Lsn::from_raw(0)) + .commit + .commit_digest; + must_ok(store.acquire_writer_epoch(writer_epoch_request_for( + "2", + Lsn::from_raw(2), + Some(epoch_id()), + Some(previous_commit_digest), + ))); + must_ok(store.close_epoch(epoch_id_for("2"))); + } + other => panic!("unknown child writer-epoch phase `{other}`"), + } +} + +#[test] +fn filesystem_writer_epoch_chain_crosses_independent_processes() { + let root = temp_wal_root("writer-epoch-process"); + let executable = must_ok(std::env::current_exe()); + for phase in ["first", "second"] { + let status = must_ok( + Command::new(&executable) + .args([ + "--ignored", + "--exact", + "emit_filesystem_writer_epoch_process_step", + "--nocapture", + ]) + .env("ECHO_WRITER_EPOCH_TEST_ROOT", &root) + .env("ECHO_WRITER_EPOCH_TEST_PHASE", phase) + .status(), + ); + assert!(status.success(), "child phase `{phase}` must succeed"); + } + + must_ok(fs::remove_dir_all(root)); +} + #[test] fn interleaved_transactions_rejected() { let first = submission_transaction("interleaved:first", Lsn::from_raw(0)); From 3661d5ad85fe61e1e1392faa26fec09a248302fa Mon Sep 17 00:00:00 2001 From: James Ross Date: Thu, 30 Jul 2026 06:08:31 -0700 Subject: [PATCH 02/11] feat: persist filesystem WAL writer epochs --- crates/warp-core/src/causal_wal.rs | 503 +++++++++++++++++- crates/warp-core/src/trusted_runtime_host.rs | 79 ++- .../tests/causal_wal_hardening_tests.rs | 32 ++ .../tests/external_action_protocol_tests.rs | 126 +++++ 4 files changed, 719 insertions(+), 21 deletions(-) diff --git a/crates/warp-core/src/causal_wal.rs b/crates/warp-core/src/causal_wal.rs index f5a24081..70753cf1 100644 --- a/crates/warp-core/src/causal_wal.rs +++ b/crates/warp-core/src/causal_wal.rs @@ -72,6 +72,9 @@ use crate::{CausalTickReceiptRef, CAUSAL_TICK_RECEIPT_REF_LEN}; const WAL_FRAME_DOMAIN: &[u8] = b"echo:causal_wal:frame:v1\0"; const WAL_PAYLOAD_DOMAIN: &[u8] = b"echo:causal_wal:payload:v1\0"; +const WAL_WRITER_EPOCH_LEDGER_DOMAIN: &[u8] = b"echo:causal_wal:writer_epoch_ledger:v1\0"; +const WAL_WRITER_EPOCH_LEDGER_MAGIC: &[u8; 8] = b"EWEP0001"; +const WAL_WRITER_EPOCH_LEDGER_VERSION: u16 = 1; const WAL_TICK_RECEIPT_MAGIC_V2: &[u8; 8] = b"ETICK002"; const WAL_TICK_RECEIPT_BATCH_MAGIC_V3: &[u8; 8] = b"ETICK003"; const WAL_RECEIPT_CORRELATION_MAGIC_V2: &[u8; 8] = b"ERCOR002"; @@ -843,6 +846,13 @@ struct WriterEpochClosure { final_commit_digest: Option, } +#[derive(Clone, Debug, Default, PartialEq, Eq)] +struct WriterEpochLedger { + active_epoch: Option, + closed_epochs: Vec, + epoch_closures: BTreeMap, +} + /// Canonical WAL record payload. #[derive(Clone, Debug, PartialEq, Eq)] pub struct WalRecordPayload { @@ -1550,12 +1560,20 @@ fn validate_writer_epoch_request( if active_epoch.is_some() { return Err(WalStoreError::WriterEpochAlreadyActive); } + if closed_epochs + .iter() + .any(|epoch| epoch.epoch_id == request.epoch_id) + { + return Err(WalStoreError::WriterEpochChainGap); + } match request.previous_epoch_id { Some(previous_epoch_id) => { let previous_epoch = closed_epochs - .iter() - .find(|epoch| epoch.epoch_id == previous_epoch_id) + .last() .ok_or(WalStoreError::UnknownPreviousWriterEpoch)?; + if previous_epoch.epoch_id != previous_epoch_id { + return Err(WalStoreError::WriterEpochChainGap); + } let previous_closure = epoch_closures .get(&previous_epoch_id) .copied() @@ -1567,6 +1585,8 @@ fn validate_writer_epoch_request( if request.started_at_lsn <= final_lsn { return Err(WalStoreError::WriterEpochLsnRegression); } + } else if request.started_at_lsn <= previous_epoch.started_at_lsn { + return Err(WalStoreError::WriterEpochLsnRegression); } if request.storage_fencing_token == previous_epoch.storage_fencing_token || request.lease_or_lock_evidence == previous_epoch.lease_or_lock_evidence @@ -5551,13 +5571,14 @@ impl FilesystemWalFaultPlan { } /// Local filesystem WAL store backed by segment files. -#[derive(Clone, Debug)] +#[derive(Debug)] pub struct FilesystemWalStore { root: PathBuf, segment_id: WalSegmentId, active_epoch: Option, closed_epochs: Vec, epoch_closures: BTreeMap, + writer_lock: Option, manifests: Vec, sync_evidence: Vec, #[cfg(any(test, feature = "host_test"))] @@ -5596,12 +5617,18 @@ impl FilesystemWalStore { segment_id, )); } + let mut writer_epoch_ledger = read_writer_epoch_ledger(&root)?; + reconcile_writer_epoch_closures( + &mut writer_epoch_ledger, + &read_filesystem_segments(&root)?.1, + )?; Ok(Self { root, segment_id, - active_epoch: None, - closed_epochs: Vec::new(), - epoch_closures: BTreeMap::new(), + active_epoch: writer_epoch_ledger.active_epoch, + closed_epochs: writer_epoch_ledger.closed_epochs, + epoch_closures: writer_epoch_ledger.epoch_closures, + writer_lock: None, manifests: Vec::new(), sync_evidence, #[cfg(any(test, feature = "host_test"))] @@ -5639,6 +5666,145 @@ impl FilesystemWalStore { self.segment_id } + fn writer_epoch_ledger(&self) -> WriterEpochLedger { + WriterEpochLedger { + active_epoch: self.active_epoch.clone(), + closed_epochs: self.closed_epochs.clone(), + epoch_closures: self.epoch_closures.clone(), + } + } + + fn install_writer_epoch_ledger(&mut self, ledger: WriterEpochLedger) { + self.active_epoch = ledger.active_epoch; + self.closed_epochs = ledger.closed_epochs; + self.epoch_closures = ledger.epoch_closures; + } + + fn reload_writer_epoch_ledger(&mut self) -> Result<(), WalStoreError> { + let mut ledger = read_writer_epoch_ledger(&self.root)?; + reconcile_writer_epoch_closures(&mut ledger, &read_filesystem_segments(&self.root)?.1)?; + self.install_writer_epoch_ledger(ledger); + Ok(()) + } + + fn persist_writer_epoch_ledger(&self) -> Result<(), WalStoreError> { + write_writer_epoch_ledger_atomic(&self.root, &self.writer_epoch_ledger()) + } + + fn ensure_writer_lock(&self) -> Result<(), WalStoreError> { + if self.writer_lock.is_some() { + Ok(()) + } else { + Err(WalStoreError::WriterEpochLeaseUnavailable) + } + } + + /// Acquires a fresh, durably linked writer epoch for a recovered host. + /// + /// The filesystem lease is acquired before the persisted epoch ledger is + /// reread. An active epoch left by a terminated process is closed under + /// that lease before the successor is derived and admitted. A concurrently + /// live writer retains the lease and prevents takeover. + pub fn acquire_fresh_writer_epoch( + &mut self, + minimum_started_at_lsn: Lsn, + ) -> Result { + if self.writer_lock.is_some() { + return Err(WalStoreError::WriterEpochAlreadyActive); + } + let writer_lock = acquire_writer_epoch_lock(&self.root)?; + self.reload_writer_epoch_ledger()?; + + if self.active_epoch.is_some() { + let previous_ledger = self.writer_epoch_ledger(); + let recovered_active_epoch = self + .active_epoch + .take() + .ok_or(WalStoreError::NoActiveWriterEpoch)?; + self.epoch_closures + .entry(recovered_active_epoch.epoch_id) + .or_default(); + self.closed_epochs.push(recovered_active_epoch); + if let Err(error) = self.persist_writer_epoch_ledger() { + self.install_writer_epoch_ledger(previous_ledger); + return Err(error); + } + } + + let previous_epoch = self.closed_epochs.last(); + let previous_epoch_id = previous_epoch.map(|epoch| epoch.epoch_id); + let previous_closure = previous_epoch + .and_then(|epoch| self.epoch_closures.get(&epoch.epoch_id)) + .copied() + .unwrap_or_default(); + let required_started_at_lsn = previous_closure + .final_lsn + .or_else(|| previous_epoch.map(|epoch| epoch.started_at_lsn)) + .and_then(Lsn::checked_next) + .unwrap_or(minimum_started_at_lsn); + let started_at_lsn = minimum_started_at_lsn.max(required_started_at_lsn); + let ordinal = u64::try_from(self.closed_epochs.len()) + .map_err(|_| WalStoreError::WriterEpochChainGap)? + .checked_add(1) + .ok_or(WalStoreError::WriterEpochChainGap)?; + let previous_epoch_final_commit_digest = previous_closure.final_commit_digest; + let epoch_id = WriterEpochId::from_hash(derive_filesystem_writer_epoch_evidence( + "epoch", + ordinal, + started_at_lsn, + previous_epoch_id, + previous_epoch_final_commit_digest, + )); + let request = WriterEpochRequest { + epoch_id, + storage_fencing_token: derive_filesystem_writer_epoch_evidence( + "fencing", + ordinal, + started_at_lsn, + previous_epoch_id, + previous_epoch_final_commit_digest, + ), + process_identity: derive_filesystem_writer_epoch_evidence( + "process", + ordinal, + started_at_lsn, + previous_epoch_id, + previous_epoch_final_commit_digest, + ), + host_identity: derive_filesystem_writer_epoch_evidence( + "host", + ordinal, + started_at_lsn, + previous_epoch_id, + previous_epoch_final_commit_digest, + ), + started_at_lsn, + previous_epoch_id, + previous_epoch_final_commit_digest, + lease_or_lock_evidence: derive_filesystem_writer_epoch_evidence( + "lease", + ordinal, + started_at_lsn, + previous_epoch_id, + previous_epoch_final_commit_digest, + ), + }; + let epoch = validate_writer_epoch_request( + None, + &self.closed_epochs, + &self.epoch_closures, + request, + )?; + let closed_ledger = self.writer_epoch_ledger(); + self.active_epoch = Some(epoch.clone()); + if let Err(error) = self.persist_writer_epoch_ledger() { + self.install_writer_epoch_ledger(closed_ledger); + return Err(error); + } + self.writer_lock = Some(writer_lock); + Ok(epoch) + } + /// Returns the WAL root directory. #[must_use] pub fn root(&self) -> &Path { @@ -5679,6 +5845,7 @@ impl FilesystemWalStore { admission_kernel_capability: Option, external_action_coordinator_capability: Option, ) -> Result<(), WalStoreError> { + self.ensure_writer_lock()?; let active_epoch = self .active_epoch .as_ref() @@ -5709,6 +5876,7 @@ impl FilesystemWalStore { final_commit_digest: Some(commit.commit_digest), }, ); + self.persist_writer_epoch_ledger()?; self.sync_evidence.push(FilesystemSyncEvidence::transaction( FilesystemSyncBoundary::CommitFileSynced, transaction_id, @@ -5739,6 +5907,7 @@ impl FilesystemWalStore { &mut self, epoch_id: WriterEpochId, ) -> Result { + self.ensure_writer_lock()?; let active_epoch = self .active_epoch .as_ref() @@ -5775,6 +5944,14 @@ impl WalStorePort for FilesystemWalStore { &mut self, request: WriterEpochRequest, ) -> Result { + let writer_lock = if self.writer_lock.is_some() { + None + } else { + Some(acquire_writer_epoch_lock(&self.root)?) + }; + if writer_lock.is_some() { + self.reload_writer_epoch_ledger()?; + } let epoch = validate_writer_epoch_request( self.active_epoch.as_ref(), &self.closed_epochs, @@ -5782,6 +5959,13 @@ impl WalStorePort for FilesystemWalStore { request, )?; self.active_epoch = Some(epoch.clone()); + if let Err(error) = self.persist_writer_epoch_ledger() { + self.active_epoch = None; + return Err(error); + } + if let Some(writer_lock) = writer_lock { + self.writer_lock = Some(writer_lock); + } Ok(epoch) } @@ -5790,6 +5974,7 @@ impl WalStorePort for FilesystemWalStore { epoch_id: WriterEpochId, frame: WalFrame, ) -> Result<(), WalStoreError> { + self.ensure_writer_lock()?; let active_epoch = self .active_epoch .as_ref() @@ -5883,6 +6068,7 @@ impl WalStorePort for FilesystemWalStore { epoch_id: WriterEpochId, manifest: WalManifest, ) -> Result<(), WalStoreError> { + self.ensure_writer_lock()?; let active_epoch = self .active_epoch .as_ref() @@ -5911,6 +6097,8 @@ impl WalStorePort for FilesystemWalStore { } fn close_epoch(&mut self, epoch_id: WriterEpochId) -> Result<(), WalStoreError> { + self.ensure_writer_lock()?; + let previous_ledger = self.writer_epoch_ledger(); let epoch = self .active_epoch .take() @@ -5921,6 +6109,11 @@ impl WalStorePort for FilesystemWalStore { } self.epoch_closures.entry(epoch_id).or_default(); self.closed_epochs.push(epoch); + if let Err(error) = self.persist_writer_epoch_ledger() { + self.install_writer_epoch_ledger(previous_ledger); + return Err(error); + } + self.writer_lock = None; Ok(()) } } @@ -7941,6 +8134,295 @@ fn write_manifest_atomic(root: &Path, manifest: &WalManifest) -> Result<(), WalS sync_directory_store(root) } +fn writer_epoch_request_from_epoch(epoch: &WriterEpoch) -> WriterEpochRequest { + WriterEpochRequest { + epoch_id: epoch.epoch_id, + storage_fencing_token: epoch.storage_fencing_token, + process_identity: epoch.process_identity, + host_identity: epoch.host_identity, + started_at_lsn: epoch.started_at_lsn, + previous_epoch_id: epoch.previous_epoch_id, + previous_epoch_final_commit_digest: epoch.previous_epoch_final_commit_digest, + lease_or_lock_evidence: epoch.lease_or_lock_evidence, + } +} + +fn push_writer_epoch(bytes: &mut Vec, epoch: &WriterEpoch) { + push_hash(bytes, &epoch.epoch_id.as_hash()); + push_hash(bytes, &epoch.storage_fencing_token); + push_hash(bytes, &epoch.process_identity); + push_hash(bytes, &epoch.host_identity); + bytes.extend_from_slice(&epoch.started_at_lsn.as_u64().to_le_bytes()); + push_optional_hash(bytes, epoch.previous_epoch_id.map(WriterEpochId::as_hash)); + push_optional_hash(bytes, epoch.previous_epoch_final_commit_digest); + push_hash(bytes, &epoch.lease_or_lock_evidence); +} + +fn read_writer_epoch(cursor: &mut WalPayloadCursor<'_>) -> Result { + Ok(WriterEpoch { + epoch_id: WriterEpochId::from_hash(cursor.read_hash()?), + storage_fencing_token: cursor.read_hash()?, + process_identity: cursor.read_hash()?, + host_identity: cursor.read_hash()?, + started_at_lsn: Lsn::from_raw(cursor.read_u64()?), + previous_epoch_id: cursor.read_optional_hash()?.map(WriterEpochId::from_hash), + previous_epoch_final_commit_digest: cursor.read_optional_hash()?, + lease_or_lock_evidence: cursor.read_hash()?, + }) +} + +fn push_writer_epoch_closure(bytes: &mut Vec, closure: WriterEpochClosure) { + push_optional_lsn(bytes, closure.final_lsn); + push_optional_hash(bytes, closure.final_commit_digest); +} + +fn read_writer_epoch_closure( + cursor: &mut WalPayloadCursor<'_>, +) -> Result { + Ok(WriterEpochClosure { + final_lsn: cursor.read_optional_lsn()?, + final_commit_digest: cursor.read_optional_hash()?, + }) +} + +fn validate_writer_epoch_closure(closure: WriterEpochClosure) -> Result<(), WalStoreError> { + if closure.final_lsn.is_some() != closure.final_commit_digest.is_some() { + return Err(WalStoreError::WriterEpochFinalCommitDigestMismatch); + } + Ok(()) +} + +fn encode_writer_epoch_ledger(ledger: &WriterEpochLedger) -> Vec { + let mut payload = Vec::new(); + payload.extend_from_slice(&WAL_WRITER_EPOCH_LEDGER_VERSION.to_le_bytes()); + payload.extend_from_slice(&len_u64(ledger.closed_epochs.len()).to_le_bytes()); + for epoch in &ledger.closed_epochs { + push_writer_epoch(&mut payload, epoch); + push_writer_epoch_closure( + &mut payload, + ledger + .epoch_closures + .get(&epoch.epoch_id) + .copied() + .unwrap_or_default(), + ); + } + match &ledger.active_epoch { + Some(epoch) => { + payload.push(1); + push_writer_epoch(&mut payload, epoch); + push_writer_epoch_closure( + &mut payload, + ledger + .epoch_closures + .get(&epoch.epoch_id) + .copied() + .unwrap_or_default(), + ); + } + None => payload.push(0), + } + payload +} + +fn decode_writer_epoch_ledger(payload: &[u8]) -> Result { + let mut cursor = WalPayloadCursor::new(payload); + if cursor.read_u16()? != WAL_WRITER_EPOCH_LEDGER_VERSION { + return Err(WalDecodeError::InvalidRecordMagic { + record_kind: "writer epoch ledger", + } + .into()); + } + let closed_len = + usize::try_from(cursor.read_u64()?).map_err(|_| WalDecodeError::UnexpectedEof)?; + let mut ledger = WriterEpochLedger::default(); + for _ in 0..closed_len { + let epoch = read_writer_epoch(&mut cursor)?; + let closure = read_writer_epoch_closure(&mut cursor)?; + validate_writer_epoch_closure(closure)?; + let validated = validate_writer_epoch_request( + None, + &ledger.closed_epochs, + &ledger.epoch_closures, + writer_epoch_request_from_epoch(&epoch), + )?; + if validated != epoch { + return Err(WalStoreError::WriterEpochChainGap); + } + ledger.epoch_closures.insert(epoch.epoch_id, closure); + ledger.closed_epochs.push(epoch); + } + match cursor.read_u8()? { + 0 => {} + 1 => { + let epoch = read_writer_epoch(&mut cursor)?; + let closure = read_writer_epoch_closure(&mut cursor)?; + validate_writer_epoch_closure(closure)?; + let validated = validate_writer_epoch_request( + None, + &ledger.closed_epochs, + &ledger.epoch_closures, + writer_epoch_request_from_epoch(&epoch), + )?; + if validated != epoch { + return Err(WalStoreError::WriterEpochChainGap); + } + ledger.epoch_closures.insert(epoch.epoch_id, closure); + ledger.active_epoch = Some(epoch); + } + code => { + return Err(WalDecodeError::UnknownEnumCode { + enum_name: "Option", + code, + } + .into()); + } + } + cursor.finish()?; + Ok(ledger) +} + +fn writer_epoch_ledger_digest(payload: &[u8]) -> Hash { + let mut hasher = blake3::Hasher::new(); + hasher.update(WAL_WRITER_EPOCH_LEDGER_DOMAIN); + update_len_prefixed(&mut hasher, payload); + hasher.finalize().into() +} + +fn derive_filesystem_writer_epoch_evidence( + label: &str, + ordinal: u64, + started_at_lsn: Lsn, + previous_epoch_id: Option, + previous_epoch_final_commit_digest: Option, +) -> Hash { + let mut hasher = blake3::Hasher::new(); + hasher.update(WAL_WRITER_EPOCH_LEDGER_DOMAIN); + update_len_prefixed(&mut hasher, label.as_bytes()); + hasher.update(&ordinal.to_le_bytes()); + hasher.update(&started_at_lsn.as_u64().to_le_bytes()); + match previous_epoch_id { + Some(epoch_id) => { + hasher.update(&[1]); + hasher.update(&epoch_id.as_hash()); + } + None => { + hasher.update(&[0]); + } + } + match previous_epoch_final_commit_digest { + Some(digest) => { + hasher.update(&[1]); + hasher.update(&digest); + } + None => { + hasher.update(&[0]); + } + } + hasher.finalize().into() +} + +fn writer_epoch_ledger_file_bytes(ledger: &WriterEpochLedger) -> Vec { + let payload = encode_writer_epoch_ledger(ledger); + let mut bytes = Vec::new(); + bytes.extend_from_slice(WAL_WRITER_EPOCH_LEDGER_MAGIC); + bytes.extend_from_slice(&len_u64(payload.len()).to_le_bytes()); + bytes.extend_from_slice(&payload); + bytes.extend_from_slice(&writer_epoch_ledger_digest(&payload)); + bytes +} + +fn read_writer_epoch_ledger(root: &Path) -> Result { + let path = root.join("writer-epochs.ecwal"); + if !path.exists() { + return Ok(WriterEpochLedger::default()); + } + let bytes = fs::read(path)?; + let mut cursor = WalPayloadCursor::new(&bytes); + if cursor.read_exact(WAL_WRITER_EPOCH_LEDGER_MAGIC.len())? != WAL_WRITER_EPOCH_LEDGER_MAGIC { + return Err(WalDecodeError::InvalidRecordMagic { + record_kind: "writer epoch ledger", + } + .into()); + } + let payload_len = + usize::try_from(cursor.read_u64()?).map_err(|_| WalDecodeError::UnexpectedEof)?; + let payload = cursor.read_exact(payload_len)?; + let stored_digest = cursor.read_hash()?; + cursor.finish()?; + if stored_digest != writer_epoch_ledger_digest(payload) { + return Err(WalStoreError::WriterEpochLedgerDigestMismatch); + } + decode_writer_epoch_ledger(payload) +} + +fn write_writer_epoch_ledger_atomic( + root: &Path, + ledger: &WriterEpochLedger, +) -> Result<(), WalStoreError> { + let path = root.join("writer-epochs.ecwal"); + let temp = root.join(".writer-epochs.ecwal.tmp"); + let bytes = writer_epoch_ledger_file_bytes(ledger); + { + let mut file = File::create(&temp)?; + file.write_all(&bytes)?; + file.sync_all()?; + } + fs::rename(temp, path)?; + sync_directory_store(root) +} + +fn reconcile_writer_epoch_closures( + ledger: &mut WriterEpochLedger, + commits: &[WalTransactionCommit], +) -> Result<(), WalStoreError> { + if ledger.active_epoch.is_none() && ledger.closed_epochs.is_empty() && !commits.is_empty() { + return Err(WalStoreError::MissingWriterEpochLedger); + } + for commit in commits { + let known_epoch = ledger + .active_epoch + .as_ref() + .is_some_and(|epoch| epoch.epoch_id == commit.writer_epoch) + || ledger + .closed_epochs + .iter() + .any(|epoch| epoch.epoch_id == commit.writer_epoch); + if !known_epoch { + return Err(WalStoreError::UnknownPreviousWriterEpoch); + } + let closure = ledger + .epoch_closures + .entry(commit.writer_epoch) + .or_default(); + if closure + .final_lsn + .is_none_or(|final_lsn| commit.last_lsn > final_lsn) + { + closure.final_lsn = Some(commit.last_lsn); + closure.final_commit_digest = Some(commit.commit_digest); + } + } + Ok(()) +} + +fn acquire_writer_epoch_lock(root: &Path) -> Result { + let lock_path = root.join("writer-epoch.lock"); + let lock = OpenOptions::new() + .create(true) + .read(true) + .write(true) + .truncate(false) + .open(lock_path)?; + if let Err(error) = lock.try_lock() { + return match error { + std::fs::TryLockError::WouldBlock => Err(WalStoreError::WriterEpochLeaseUnavailable), + std::fs::TryLockError::Error(error) => Err(error.into()), + }; + } + Ok(lock) +} + fn sync_directory_store(path: &Path) -> Result<(), WalStoreError> { File::open(path)?.sync_all()?; Ok(()) @@ -9359,6 +9841,15 @@ pub enum WalStoreError { /// New epoch starts at or before the previous closed epoch final LSN. #[error("WAL writer epoch LSN regression")] WriterEpochLsnRegression, + /// Another filesystem WAL writer holds the storage lease. + #[error("filesystem WAL writer epoch lease is unavailable")] + WriterEpochLeaseUnavailable, + /// Filesystem WAL commits exist without their writer-epoch ledger. + #[error("filesystem WAL writer epoch ledger is missing")] + MissingWriterEpochLedger, + /// Filesystem WAL writer-epoch ledger digest does not match its payload. + #[error("filesystem WAL writer epoch ledger digest mismatch")] + WriterEpochLedgerDigestMismatch, /// Validation failed. #[error(transparent)] Validation(#[from] WalValidationError), diff --git a/crates/warp-core/src/trusted_runtime_host.rs b/crates/warp-core/src/trusted_runtime_host.rs index 0d06bfa4..e2b74307 100644 --- a/crates/warp-core/src/trusted_runtime_host.rs +++ b/crates/warp-core/src/trusted_runtime_host.rs @@ -346,6 +346,9 @@ pub enum TrustedRuntimeWalError { /// Number of transactions in the attempted durable batch. transaction_count: usize, }, + /// The in-memory WAL rollback snapshot was unexpectedly unavailable. + #[error("trusted runtime in-memory WAL rollback snapshot is unavailable")] + InMemoryRollbackSnapshotUnavailable, /// Per-Action receipt records do not describe one exact scheduler Tick. #[error("trusted runtime WAL scheduler Tick batch is internally inconsistent")] SchedulerTickBatchMismatch, @@ -2097,7 +2100,7 @@ impl TrustedRuntimeHost { } .into()); } - let runtime_wal_before = runtime_wal.clone(); + let runtime_wal_before = runtime_wal.in_memory_rollback_snapshot(); for group in tick_wal_groups.values() { let Some((first_correlation, _, state_delta, state_delta_digest)) = group.first() else { @@ -2130,7 +2133,8 @@ impl TrustedRuntimeHost { continue; } if !runtime_wal.uses_filesystem_store() { - *runtime_wal = runtime_wal_before; + *runtime_wal = runtime_wal_before + .ok_or(TrustedRuntimeWalError::InMemoryRollbackSnapshotUnavailable)?; } self.runtime = runtime_before; self.provenance = provenance_before; @@ -2417,7 +2421,7 @@ impl CausalAnchorClaimProjection { } /// Minimal trusted-runtime WAL adapter for ACK-boundary integration tests. -#[derive(Clone, Debug)] +#[derive(Debug)] pub struct TrustedRuntimeWal { store: TrustedRuntimeWalStore, evidence_catalog: Option, @@ -2477,17 +2481,9 @@ impl TrustedRuntimeWal { } else { next_lsn }; - let writer_epoch = WriterEpochId::from_hash(trusted_runtime_wal_digest("writer-epoch")); - store.acquire_writer_epoch(WriterEpochRequest { - epoch_id: writer_epoch, - storage_fencing_token: trusted_runtime_wal_digest("fencing-token"), - process_identity: trusted_runtime_wal_digest("process"), - host_identity: trusted_runtime_wal_digest("host"), - started_at_lsn: next_lsn, - previous_epoch_id: None, - previous_epoch_final_commit_digest: None, - lease_or_lock_evidence: trusted_runtime_wal_digest("lease"), - })?; + let writer_epoch = store.acquire_runtime_writer_epoch(next_lsn)?; + let next_lsn = writer_epoch.started_at_lsn; + let writer_epoch = writer_epoch.epoch_id; let durability_mode = store.durability_mode(); Ok(Self { store, @@ -2550,6 +2546,40 @@ impl TrustedRuntimeWal { self.store.cloned_in_memory_store() } + fn in_memory_rollback_snapshot(&self) -> Option { + Some(Self { + store: TrustedRuntimeWalStore::InMemory(self.store.cloned_in_memory_store()?), + evidence_catalog: self.evidence_catalog.clone(), + evidence_catalog_posture: self.evidence_catalog_posture.clone(), + #[cfg(any(test, feature = "host_test"))] + fail_next_evidence_catalog_update: self.fail_next_evidence_catalog_update, + #[cfg(any(test, feature = "host_test"))] + recover_read_only_call_count: std::cell::Cell::new( + self.recover_read_only_call_count.get(), + ), + writer_epoch: self.writer_epoch, + segment_id: self.segment_id, + next_lsn: self.next_lsn, + previous_frame_digest: self.previous_frame_digest, + previous_committed_transaction_digest: self.previous_committed_transaction_digest, + durability_mode: self.durability_mode, + payload_codec_id: self.payload_codec_id, + payload_schema_id: self.payload_schema_id, + digest_domain: self.digest_domain, + submission_frontier_digest: self.submission_frontier_digest, + receipt_frontier_digest: self.receipt_frontier_digest, + runtime_state_frontier_digest: self.runtime_state_frontier_digest, + executable_operation_catalog_frontier_digest: self + .executable_operation_catalog_frontier_digest, + executable_operation_receipt_frontier_digest: self + .executable_operation_receipt_frontier_digest, + causal_anchor_frontier_digest: self.causal_anchor_frontier_digest, + causal_history_frontier_digest: self.causal_history_frontier_digest, + causal_anchor_claim_projection: self.causal_anchor_claim_projection.clone(), + durable_submission_acceptances: self.durable_submission_acceptances.clone(), + }) + } + /// Re-runs state-delta recovery after repeating the last scheduler /// transaction, for adversarial transaction-atomicity tests. #[cfg(any(test, feature = "host_test"))] @@ -3305,7 +3335,7 @@ impl TrustedRuntimeWal { } } -#[derive(Clone, Debug)] +#[derive(Debug)] enum TrustedRuntimeWalStore { InMemory(InMemoryWalStore), Filesystem(FilesystemWalStore), @@ -3353,6 +3383,25 @@ impl TrustedRuntimeWalStore { } } + fn acquire_runtime_writer_epoch( + &mut self, + next_lsn: Lsn, + ) -> Result { + match self { + Self::InMemory(store) => store.acquire_writer_epoch(WriterEpochRequest { + epoch_id: WriterEpochId::from_hash(trusted_runtime_wal_digest("writer-epoch")), + storage_fencing_token: trusted_runtime_wal_digest("fencing-token"), + process_identity: trusted_runtime_wal_digest("process"), + host_identity: trusted_runtime_wal_digest("host"), + started_at_lsn: next_lsn, + previous_epoch_id: None, + previous_epoch_final_commit_digest: None, + lease_or_lock_evidence: trusted_runtime_wal_digest("lease"), + }), + Self::Filesystem(store) => store.acquire_fresh_writer_epoch(next_lsn), + } + } + fn recover_for_writer(&self) -> Result { match self { Self::InMemory(store) => recover_runtime_wal_store_read_only(store), diff --git a/crates/warp-core/tests/causal_wal_hardening_tests.rs b/crates/warp-core/tests/causal_wal_hardening_tests.rs index 27adb825..f193752c 100644 --- a/crates/warp-core/tests/causal_wal_hardening_tests.rs +++ b/crates/warp-core/tests/causal_wal_hardening_tests.rs @@ -935,6 +935,38 @@ fn filesystem_active_writer_epoch_and_final_commit_survive_reopen() { must_ok(fs::remove_dir_all(root)); } +#[test] +fn filesystem_writer_lease_refuses_overlap_before_takeover() { + let root = temp_wal_root("writer-epoch-overlap"); + let mut active = must_ok(FilesystemWalStore::open(&root, WalSegmentId::from_raw(1))); + must_ok(active.acquire_writer_epoch(writer_epoch_request())); + let mut contender = must_ok(FilesystemWalStore::open(&root, WalSegmentId::from_raw(1))); + + let error = must_err( + contender.acquire_fresh_writer_epoch(Lsn::from_raw(0)), + "a live filesystem writer lease must refuse overlap", + ); + assert!(matches!(error, WalStoreError::WriterEpochLeaseUnavailable)); + let append_error = must_err( + contender.append_transaction(submission_transaction( + "writer-epoch-overlap", + Lsn::from_raw(0), + )), + "a store without the filesystem writer lease must not append", + ); + assert!(matches!( + append_error, + WalStoreError::WriterEpochLeaseUnavailable + )); + + drop(active); + let successor = must_ok(contender.acquire_fresh_writer_epoch(Lsn::from_raw(0))); + assert_eq!(successor.previous_epoch_id, Some(epoch_id())); + assert!(successor.started_at_lsn > Lsn::from_raw(0)); + drop(contender); + must_ok(fs::remove_dir_all(root)); +} + #[test] fn duplicate_writer_epoch_id_is_a_chain_gap() { let mut store = InMemoryWalStore::new(); diff --git a/crates/warp-core/tests/external_action_protocol_tests.rs b/crates/warp-core/tests/external_action_protocol_tests.rs index 8dda756c..ec20e101 100644 --- a/crates/warp-core/tests/external_action_protocol_tests.rs +++ b/crates/warp-core/tests/external_action_protocol_tests.rs @@ -6,6 +6,7 @@ use std::cell::Cell; use std::path::PathBuf; +use std::process::Command; use std::sync::atomic::{AtomicU64, Ordering}; use warp_core::causal_wal::{ @@ -103,6 +104,23 @@ fn context_with_durability( } } +fn filesystem_context( + label: &str, + writer_epoch: WriterEpochId, +) -> ExternalActionTransactionContextV1 { + ExternalActionTransactionContextV1 { + writer_epoch, + segment_id: WalSegmentId::from_raw(1), + transaction_id: WalTransactionId::from_hash(digest(label)), + durability_mode: WalDurabilityMode::StrictFilesystem, + payload_codec_id: PayloadCodecId::from_hash(digest("external-action:codec")), + payload_schema_id: PayloadSchemaId::from_hash(digest("external-action:schema")), + payload_schema_version: 1, + canonical_encoding_version: 1, + digest_domain: digest("external-action:domain"), + } +} + static TEMP_WAL_COUNTER: AtomicU64 = AtomicU64::new(0); struct TempWalDir(PathBuf); @@ -891,6 +909,114 @@ fn filesystem_reopen_recovers_settlement_without_adapter_reexecution() { assert_eq!(reopened.read_commits().len(), commits_before); } +#[test] +#[ignore = "child entrypoint exercised by the independent-process external-action test"] +fn emit_filesystem_external_action_process_step() { + let root = PathBuf::from( + std::env::var_os("ECHO_EXTERNAL_ACTION_PROCESS_ROOT") + .unwrap_or_else(|| panic!("missing child WAL root")), + ); + let phase = std::env::var("ECHO_EXTERNAL_ACTION_PROCESS_PHASE") + .unwrap_or_else(|_| panic!("missing child phase")); + let request = request_with("filesystem-process", 31, 128); + let mut store = must_ok(FilesystemWalStore::open(&root, WalSegmentId::from_raw(1))); + + match phase.as_str() { + "request" => { + let epoch = must_ok(store.acquire_fresh_writer_epoch(Lsn::from_raw(0))); + let mut coordinator = coordinator(&store); + let recorded = must_ok(record_external_action_request( + &mut store, + &mut coordinator, + filesystem_context("request:filesystem-process", epoch.epoch_id), + request, + )); + assert_eq!(recorded.request(), request); + } + "claim" => { + let epoch = must_ok(store.acquire_fresh_writer_epoch(Lsn::from_raw(0))); + let mut coordinator = coordinator(&store); + let recorded = must_ok(coordinator.recorded_request(request.request_id())); + let grant = must_ok(claim_external_action( + &mut store, + &mut coordinator, + filesystem_context("claim:filesystem-process", epoch.epoch_id), + recorded, + authorization(&request), + request.basis_digest, + 0, + digest("claim:filesystem-process:lease"), + )); + assert_eq!(grant.request(), request); + } + "settlement" => { + let epoch = must_ok(store.acquire_fresh_writer_epoch(Lsn::from_raw(0))); + let mut coordinator = coordinator(&store); + let grant = must_ok(coordinator.claim_grant(request.request_id())); + let settlement_candidate = candidate( + &grant, + ExternalActionSettlementKindV1::Succeeded, + b"independent-process-result".to_vec(), + ); + let admitted = must_ok(admit_external_action_settlement( + &mut store, + &mut coordinator, + filesystem_context("settlement:filesystem-process", epoch.epoch_id), + grant, + settlement_candidate, + )); + assert_eq!( + admitted.settlement().canonical_result_bytes.as_slice(), + b"independent-process-result" + ); + } + "replay" => { + let commits_before = store.read_commits().len(); + let coordinator = coordinator(&store); + let admitted = must_ok(coordinator.admitted_settlement(request.request_id())); + assert_eq!( + admitted.settlement().canonical_result_bytes.as_slice(), + b"independent-process-result" + ); + assert_eq!(store.read_commits().len(), commits_before); + } + other => panic!("unknown child phase {other}"), + } +} + +#[test] +fn filesystem_external_action_lifecycle_crosses_independent_processes() { + let wal_dir = TempWalDir::new("independent-processes"); + let executable = must_ok(std::env::current_exe()); + for phase in ["request", "claim", "settlement", "replay"] { + let status = must_ok( + Command::new(&executable) + .arg("--ignored") + .arg("--exact") + .arg("emit_filesystem_external_action_process_step") + .arg("--nocapture") + .env("ECHO_EXTERNAL_ACTION_PROCESS_ROOT", &wal_dir.0) + .env("ECHO_EXTERNAL_ACTION_PROCESS_PHASE", phase) + .status(), + ); + assert!(status.success(), "child phase {phase} failed: {status}"); + } + + let report = must_ok(recover_filesystem_store( + &wal_dir.0, + RecoveryAccessMode::ReadOnly, + )); + let request = request_with("filesystem-process", 31, 128); + let recovered = must_ok(observe_external_actions(&report)); + let entry = recovered + .get(request.request_id()) + .unwrap_or_else(|| panic!("missing recovered independent-process request")); + assert_eq!( + entry.posture, + RecoveredExternalActionPostureV1::Settled(ExternalActionSettlementKindV1::Succeeded) + ); +} + #[test] fn filesystem_scan_failure_cannot_be_admitted_as_genesis() { let wal_dir = TempWalDir::new("corrupt-snapshot"); From 7030a6afd21d4a9a5d393542ff1589b51fb5a9c4 Mon Sep 17 00:00:00 2001 From: James Ross Date: Thu, 30 Jul 2026 06:09:04 -0700 Subject: [PATCH 03/11] docs: specify durable writer epoch recovery --- docs/topics/WAL.md | 36 +++++++++++++++++++++++++++++++++++- 1 file changed, 35 insertions(+), 1 deletion(-) diff --git a/docs/topics/WAL.md b/docs/topics/WAL.md index ee8dce68..f24baffa 100644 --- a/docs/topics/WAL.md +++ b/docs/topics/WAL.md @@ -17,7 +17,7 @@ Echo may only claim what its WAL can recover. ## What We Found -The current runtime WAL evidence says eleven concrete things. +The current runtime WAL evidence says twelve concrete things. First, accepted-submission evidence is not just an in-memory editor event. The WAL-backed ACK path, `submit_intent_with_runtime_wal_ack(...)`, returns only @@ -194,6 +194,25 @@ and a basis commitment; operation-specific schema validation runs before the generic settlement transaction. Replay consumes those retained bytes without opening the workspace again. +Twelfth, strict filesystem WAL writer authority survives process loss as a +durable, checksummed epoch ledger rather than process memory. Epoch acquisition +is persisted before the store returns authority. Every committed transaction +then advances that epoch's final LSN and commit-digest evidence, and explicit +closure persists the complete predecessor link. Reopen restores closed and +active epochs and reconciles a commit that reached its segment sync boundary +before the ledger snapshot. + +The filesystem store holds an operating-system writer lease for the complete +active epoch. A second live process cannot append, close, or replace that +epoch. After process loss releases the lease, the trusted runtime closes the +recovered active epoch under the newly acquired lease and derives a fresh +successor with a monotonic start LSN, new epoch identity, fencing token, and +lease evidence bound to the exact latest predecessor and its final commit. +Duplicate identities, stale or missing predecessor links, reused fencing +evidence, LSN regression, corrupted ledgers, and commits without their epoch +ledger fail closed before append. This fencing is generic WAL authority; it +contains no application vocabulary or callback. + ## Boundaries The WAL belongs to the trusted runtime host. Application-facing code can submit @@ -391,6 +410,21 @@ metadata are excluded from the application-state root because the obstruction legitimately extends causal history. It uses the strict filesystem adapter and no native application callback. +Filesystem writer-epoch and external-action process-loss witnesses live in +`crates/warp-core/tests/causal_wal_hardening_tests.rs` and +`crates/warp-core/tests/external_action_protocol_tests.rs`. Read these first: + +- `filesystem_writer_lease_refuses_overlap_before_takeover` +- `filesystem_writer_epoch_chain_crosses_independent_processes` +- `fixed_seed_filesystem_writer_epoch_chain_survives_bounded_reopens` +- `filesystem_external_action_lifecycle_crosses_independent_processes` + +The last witness uses four independent processes. Three processes durably +record the request, claim, and settlement under successively fenced writer +epochs. The fourth reconstructs and consumes the settled result without +acquiring writer authority, appending a transaction, or reissuing an adapter +effect. + For a successful projected Action, the decided-Tick transaction also retains the compiler-owned projection identity, output type coordinate, exact canonical application-result bytes, and their domain-separated result identity. Recovery From 3ba3a21fb59743b4e35df1b11269cebe56d6f786 Mon Sep 17 00:00:00 2001 From: James Ross Date: Thu, 30 Jul 2026 06:09:11 -0700 Subject: [PATCH 04/11] docs: record filesystem writer epoch fencing --- CHANGELOG.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9a0a187c..a9d3b4fc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,14 @@ ### Added +- Strict filesystem WAL stores now persist a checksummed writer-epoch ledger + containing active and closed epochs plus exact final LSN and commit-digest + evidence. An OS-backed writer lease refuses overlapping processes before + append; a recovered trusted host closes only an abandoned epoch under that + lease and derives a fresh, monotonically linked successor. Duplicate, stale, + skipped, regressed, or unfenced epoch chains fail closed. Independent-process + witnesses carry an external-action request, claim, settlement, and + effect-free replay across successive host processes. - Echo now consumes the exact independently verified Edict `workspace.patch.applyValidated@1` request through a capability-rooted single-file patch adapter. The request binds the prior bounded-observation From f1a74012fd727976df9926e33d3b2f77735991e4 Mon Sep 17 00:00:00 2001 From: James Ross Date: Thu, 30 Jul 2026 06:11:50 -0700 Subject: [PATCH 05/11] fix: keep filesystem WAL authority single-owner --- crates/warp-core/src/trusted_runtime_host.rs | 7 +++++++ crates/warp-core/tests/trusted_runtime_host_loop_tests.rs | 3 ++- 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/crates/warp-core/src/trusted_runtime_host.rs b/crates/warp-core/src/trusted_runtime_host.rs index e2b74307..073f39cf 100644 --- a/crates/warp-core/src/trusted_runtime_host.rs +++ b/crates/warp-core/src/trusted_runtime_host.rs @@ -2580,6 +2580,13 @@ impl TrustedRuntimeWal { }) } + /// Returns an in-memory-only copy for host rollback tests. + #[cfg(any(test, feature = "host_test"))] + #[must_use] + pub fn cloned_in_memory_for_test(&self) -> Option { + self.in_memory_rollback_snapshot() + } + /// Re-runs state-delta recovery after repeating the last scheduler /// transaction, for adversarial transaction-atomicity tests. #[cfg(any(test, feature = "host_test"))] 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 05890c34..c2bdc959 100644 --- a/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs +++ b/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs @@ -2046,7 +2046,8 @@ fn runtime_wal_live_evidence_catalog_failure_marks_needs_rebuild_without_failing let mut faulting_wal = host .runtime_wal() .expect("runtime WAL should exist") - .clone(); + .cloned_in_memory_for_test() + .expect("test WAL should be in-memory"); faulting_wal.fail_next_evidence_catalog_update_for_test(); host.replace_runtime_wal_for_test(faulting_wal); From 88a5c7336c9818bc4437221644f4be1a9c553f15 Mon Sep 17 00:00:00 2001 From: James Ross Date: Thu, 30 Jul 2026 06:14:37 -0700 Subject: [PATCH 06/11] test: reject missing writer epoch evidence --- .../tests/causal_wal_hardening_tests.rs | 49 +++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/crates/warp-core/tests/causal_wal_hardening_tests.rs b/crates/warp-core/tests/causal_wal_hardening_tests.rs index f193752c..aa58f3ed 100644 --- a/crates/warp-core/tests/causal_wal_hardening_tests.rs +++ b/crates/warp-core/tests/causal_wal_hardening_tests.rs @@ -967,6 +967,55 @@ fn filesystem_writer_lease_refuses_overlap_before_takeover() { must_ok(fs::remove_dir_all(root)); } +#[test] +fn filesystem_writer_epoch_ledger_digest_mismatch_fails_closed() { + let root = temp_wal_root("writer-epoch-ledger-digest"); + { + let mut store = must_ok(FilesystemWalStore::open(&root, WalSegmentId::from_raw(1))); + must_ok(store.acquire_writer_epoch(writer_epoch_request())); + must_ok(store.close_epoch(epoch_id())); + } + let ledger_path = root.join("writer-epochs.ecwal"); + let mut bytes = must_ok(fs::read(&ledger_path)); + let last = bytes + .last_mut() + .unwrap_or_else(|| panic!("writer-epoch ledger must not be empty")); + *last ^= 0x01; + must_ok(fs::write(&ledger_path, bytes)); + + let error = must_err( + FilesystemWalStore::open(&root, WalSegmentId::from_raw(1)), + "corrupt writer-epoch ledger must fail closed", + ); + assert!(matches!( + error, + WalStoreError::WriterEpochLedgerDigestMismatch + )); + must_ok(fs::remove_dir_all(root)); +} + +#[test] +fn filesystem_commits_without_writer_epoch_ledger_fail_closed() { + let root = temp_wal_root("writer-epoch-ledger-missing"); + { + let mut store = must_ok(FilesystemWalStore::open(&root, WalSegmentId::from_raw(1))); + must_ok(store.acquire_writer_epoch(writer_epoch_request())); + must_ok(store.append_transaction(submission_transaction( + "writer-epoch-ledger-missing", + Lsn::from_raw(0), + ))); + must_ok(store.close_epoch(epoch_id())); + } + must_ok(fs::remove_file(root.join("writer-epochs.ecwal"))); + + let error = must_err( + FilesystemWalStore::open(&root, WalSegmentId::from_raw(1)), + "committed WAL without its writer-epoch ledger must fail closed", + ); + assert!(matches!(error, WalStoreError::MissingWriterEpochLedger)); + must_ok(fs::remove_dir_all(root)); +} + #[test] fn duplicate_writer_epoch_id_is_a_chain_gap() { let mut store = InMemoryWalStore::new(); From 7421ce9f8b1640369a48030230fade8eaa882cf0 Mon Sep 17 00:00:00 2001 From: James Ross Date: Thu, 30 Jul 2026 06:23:57 -0700 Subject: [PATCH 07/11] test: admit writer epoch process fixtures --- .ban-nondeterminism-allowlist | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.ban-nondeterminism-allowlist b/.ban-nondeterminism-allowlist index 2be9b786..a3be21c7 100644 --- a/.ban-nondeterminism-allowlist +++ b/.ban-nondeterminism-allowlist @@ -11,7 +11,9 @@ std-fs crates/warp-core/src/causal_wal.rs native WAL filesystem adapter pending std-fs crates/warp-core/src/wsc/store.rs native WSC filesystem store adapter pending extraction to a boundary crate. std-fs crates/warp-core/src/wsc/view.rs native WSC file-open helper pending extraction to a boundary adapter. std-fs crates/warp-core/src/causal_wal_tests.rs WAL filesystem fixture I/O only. +std-env crates/warp-core/tests/causal_wal_hardening_tests.rs WAL independent-process fixture environment handoff only. std-fs crates/warp-core/tests/causal_wal_hardening_tests.rs WAL filesystem fixture I/O only. +std-process crates/warp-core/tests/causal_wal_hardening_tests.rs WAL independent-process fixture execution only. std-env crates/warp-core/tests/external_action_protocol_tests.rs external-action filesystem WAL fixture temp directory selection only. std-fs crates/warp-core/tests/external_action_protocol_tests.rs external-action filesystem WAL fixture I/O only. std-process crates/warp-core/tests/external_action_protocol_tests.rs external-action filesystem WAL fixture temp directory disambiguation only. From fc4c3422dd56888bc3ad179ebb2b5605314ca3b1 Mon Sep 17 00:00:00 2001 From: James Ross Date: Thu, 30 Jul 2026 06:33:19 -0700 Subject: [PATCH 08/11] test: bound persisted writer epoch history --- crates/warp-core/tests/causal_wal_hardening_tests.rs | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/crates/warp-core/tests/causal_wal_hardening_tests.rs b/crates/warp-core/tests/causal_wal_hardening_tests.rs index aa58f3ed..32b2a325 100644 --- a/crates/warp-core/tests/causal_wal_hardening_tests.rs +++ b/crates/warp-core/tests/causal_wal_hardening_tests.rs @@ -1075,6 +1075,7 @@ fn fixed_seed_filesystem_writer_epoch_chain_survives_bounded_reopens() { let root = temp_wal_root("writer-epoch-fixed-seed"); let mut previous_epoch_id = None; + let mut bounded_ledger_len = None; for ordinal in 0..EPOCH_COUNT { let mut store = must_ok(FilesystemWalStore::open(&root, WalSegmentId::from_raw(1))); let label = format!("fixed-seed-{ordinal:02}"); @@ -1083,6 +1084,14 @@ fn fixed_seed_filesystem_writer_epoch_chain_survives_bounded_reopens() { let epoch = must_ok(store.acquire_writer_epoch(request)); must_ok(store.close_epoch(epoch.epoch_id)); previous_epoch_id = Some(epoch.epoch_id); + let ledger_len = must_ok(fs::metadata(root.join("writer-epochs.ecwal"))).len(); + match bounded_ledger_len { + Some(expected) => assert_eq!( + ledger_len, expected, + "writer-epoch ledger must stay bounded after epoch {ordinal}" + ), + None => bounded_ledger_len = Some(ledger_len), + } } assert_eq!(previous_epoch_id, Some(epoch_id_for("fixed-seed-15"))); From ec9ff7c939ed3b2ce8512cb843de8db53a80e8d7 Mon Sep 17 00:00:00 2001 From: James Ross Date: Thu, 30 Jul 2026 06:39:57 -0700 Subject: [PATCH 09/11] fix: bound filesystem writer epoch recovery --- .ban-nondeterminism-allowlist | 5 +- crates/warp-core/src/causal_wal.rs | 95 ++++++++++++++----- crates/warp-core/src/trusted_runtime_host.rs | 15 ++- .../tests/causal_wal_hardening_tests.rs | 38 ++++---- .../tests/external_action_protocol_tests.rs | 22 ++--- .../warp-core/tests/support/child_process.rs | 69 ++++++++++++++ 6 files changed, 187 insertions(+), 57 deletions(-) create mode 100644 crates/warp-core/tests/support/child_process.rs diff --git a/.ban-nondeterminism-allowlist b/.ban-nondeterminism-allowlist index a3be21c7..21005c6e 100644 --- a/.ban-nondeterminism-allowlist +++ b/.ban-nondeterminism-allowlist @@ -13,10 +13,13 @@ std-fs crates/warp-core/src/wsc/view.rs native WSC file-open helper pending extr std-fs crates/warp-core/src/causal_wal_tests.rs WAL filesystem fixture I/O only. std-env crates/warp-core/tests/causal_wal_hardening_tests.rs WAL independent-process fixture environment handoff only. std-fs crates/warp-core/tests/causal_wal_hardening_tests.rs WAL filesystem fixture I/O only. -std-process crates/warp-core/tests/causal_wal_hardening_tests.rs WAL independent-process fixture execution only. std-env crates/warp-core/tests/external_action_protocol_tests.rs external-action filesystem WAL fixture temp directory selection only. std-fs crates/warp-core/tests/external_action_protocol_tests.rs external-action filesystem WAL fixture I/O only. std-process crates/warp-core/tests/external_action_protocol_tests.rs external-action filesystem WAL fixture temp directory disambiguation only. +time-instant crates/warp-core/tests/support/child_process.rs bounded independent-process fixture deadline only. +thread-sleep crates/warp-core/tests/support/child_process.rs bounded independent-process fixture polling only. +std-fs crates/warp-core/tests/support/child_process.rs timed-out fixture cleanup only. +std-process crates/warp-core/tests/support/child_process.rs bounded independent-process fixture execution only. std-env crates/warp-core/tests/bounded_workspace_observation_tests.rs bounded-observation filesystem fixture temp directory selection only. std-fs crates/warp-core/tests/bounded_workspace_observation_tests.rs bounded-observation filesystem fixture I/O only. std-process crates/warp-core/tests/bounded_workspace_observation_tests.rs bounded-observation filesystem fixture temp directory disambiguation only. diff --git a/crates/warp-core/src/causal_wal.rs b/crates/warp-core/src/causal_wal.rs index 70753cf1..0c574e9b 100644 --- a/crates/warp-core/src/causal_wal.rs +++ b/crates/warp-core/src/causal_wal.rs @@ -75,6 +75,7 @@ const WAL_PAYLOAD_DOMAIN: &[u8] = b"echo:causal_wal:payload:v1\0"; const WAL_WRITER_EPOCH_LEDGER_DOMAIN: &[u8] = b"echo:causal_wal:writer_epoch_ledger:v1\0"; const WAL_WRITER_EPOCH_LEDGER_MAGIC: &[u8; 8] = b"EWEP0001"; const WAL_WRITER_EPOCH_LEDGER_VERSION: u16 = 1; +const WAL_WRITER_EPOCH_RETAINED_CLOSED_LIMIT: usize = 1; const WAL_TICK_RECEIPT_MAGIC_V2: &[u8; 8] = b"ETICK002"; const WAL_TICK_RECEIPT_BATCH_MAGIC_V3: &[u8; 8] = b"ETICK003"; const WAL_RECEIPT_CORRELATION_MAGIC_V2: &[u8; 8] = b"ERCOR002"; @@ -5688,7 +5689,29 @@ impl FilesystemWalStore { } fn persist_writer_epoch_ledger(&self) -> Result<(), WalStoreError> { - write_writer_epoch_ledger_atomic(&self.root, &self.writer_epoch_ledger()) + write_writer_epoch_ledger_atomic( + &self.root, + self.active_epoch.as_ref(), + &self.closed_epochs, + &self.epoch_closures, + ) + } + + fn retain_latest_closed_writer_epoch(&mut self) { + if self.closed_epochs.len() <= WAL_WRITER_EPOCH_RETAINED_CLOSED_LIMIT { + return; + } + let Some(latest) = self.closed_epochs.pop() else { + return; + }; + let latest_closure = self + .epoch_closures + .remove(&latest.epoch_id) + .unwrap_or_default(); + self.closed_epochs.clear(); + self.epoch_closures.clear(); + self.epoch_closures.insert(latest.epoch_id, latest_closure); + self.closed_epochs.push(latest); } fn ensure_writer_lock(&self) -> Result<(), WalStoreError> { @@ -5725,6 +5748,7 @@ impl FilesystemWalStore { .entry(recovered_active_epoch.epoch_id) .or_default(); self.closed_epochs.push(recovered_active_epoch); + self.retain_latest_closed_writer_epoch(); if let Err(error) = self.persist_writer_epoch_ledger() { self.install_writer_epoch_ledger(previous_ledger); return Err(error); @@ -6109,6 +6133,7 @@ impl WalStorePort for FilesystemWalStore { } self.epoch_closures.entry(epoch_id).or_default(); self.closed_epochs.push(epoch); + self.retain_latest_closed_writer_epoch(); if let Err(error) = self.persist_writer_epoch_ledger() { self.install_writer_epoch_ledger(previous_ledger); return Err(error); @@ -8192,29 +8217,31 @@ fn validate_writer_epoch_closure(closure: WriterEpochClosure) -> Result<(), WalS Ok(()) } -fn encode_writer_epoch_ledger(ledger: &WriterEpochLedger) -> Vec { +fn encode_writer_epoch_ledger( + active_epoch: Option<&WriterEpoch>, + closed_epochs: &[WriterEpoch], + epoch_closures: &BTreeMap, +) -> Vec { let mut payload = Vec::new(); payload.extend_from_slice(&WAL_WRITER_EPOCH_LEDGER_VERSION.to_le_bytes()); - payload.extend_from_slice(&len_u64(ledger.closed_epochs.len()).to_le_bytes()); - for epoch in &ledger.closed_epochs { + payload.extend_from_slice(&len_u64(closed_epochs.len()).to_le_bytes()); + for epoch in closed_epochs { push_writer_epoch(&mut payload, epoch); push_writer_epoch_closure( &mut payload, - ledger - .epoch_closures + epoch_closures .get(&epoch.epoch_id) .copied() .unwrap_or_default(), ); } - match &ledger.active_epoch { + match active_epoch { Some(epoch) => { payload.push(1); push_writer_epoch(&mut payload, epoch); push_writer_epoch_closure( &mut payload, - ledger - .epoch_closures + epoch_closures .get(&epoch.epoch_id) .copied() .unwrap_or_default(), @@ -8235,19 +8262,24 @@ fn decode_writer_epoch_ledger(payload: &[u8]) -> Result WAL_WRITER_EPOCH_RETAINED_CLOSED_LIMIT { + return Err(WalStoreError::WriterEpochChainGap); + } let mut ledger = WriterEpochLedger::default(); - for _ in 0..closed_len { + for retained_index in 0..closed_len { let epoch = read_writer_epoch(&mut cursor)?; let closure = read_writer_epoch_closure(&mut cursor)?; validate_writer_epoch_closure(closure)?; - let validated = validate_writer_epoch_request( - None, - &ledger.closed_epochs, - &ledger.epoch_closures, - writer_epoch_request_from_epoch(&epoch), - )?; - if validated != epoch { - return Err(WalStoreError::WriterEpochChainGap); + if retained_index > 0 { + let validated = validate_writer_epoch_request( + None, + &ledger.closed_epochs, + &ledger.epoch_closures, + writer_epoch_request_from_epoch(&epoch), + )?; + if validated != epoch { + return Err(WalStoreError::WriterEpochChainGap); + } } ledger.epoch_closures.insert(epoch.epoch_id, closure); ledger.closed_epochs.push(epoch); @@ -8322,8 +8354,12 @@ fn derive_filesystem_writer_epoch_evidence( hasher.finalize().into() } -fn writer_epoch_ledger_file_bytes(ledger: &WriterEpochLedger) -> Vec { - let payload = encode_writer_epoch_ledger(ledger); +fn writer_epoch_ledger_file_bytes( + active_epoch: Option<&WriterEpoch>, + closed_epochs: &[WriterEpoch], + epoch_closures: &BTreeMap, +) -> Vec { + let payload = encode_writer_epoch_ledger(active_epoch, closed_epochs, epoch_closures); let mut bytes = Vec::new(); bytes.extend_from_slice(WAL_WRITER_EPOCH_LEDGER_MAGIC); bytes.extend_from_slice(&len_u64(payload.len()).to_le_bytes()); @@ -8358,11 +8394,13 @@ fn read_writer_epoch_ledger(root: &Path) -> Result, + closed_epochs: &[WriterEpoch], + epoch_closures: &BTreeMap, ) -> Result<(), WalStoreError> { let path = root.join("writer-epochs.ecwal"); let temp = root.join(".writer-epochs.ecwal.tmp"); - let bytes = writer_epoch_ledger_file_bytes(ledger); + let bytes = writer_epoch_ledger_file_bytes(active_epoch, closed_epochs, epoch_closures); { let mut file = File::create(&temp)?; file.write_all(&bytes)?; @@ -8379,6 +8417,16 @@ fn reconcile_writer_epoch_closures( if ledger.active_epoch.is_none() && ledger.closed_epochs.is_empty() && !commits.is_empty() { return Err(WalStoreError::MissingWriterEpochLedger); } + let retained_start_lsn = ledger + .closed_epochs + .first() + .map(|epoch| epoch.started_at_lsn) + .or_else(|| { + ledger + .active_epoch + .as_ref() + .map(|epoch| epoch.started_at_lsn) + }); for commit in commits { let known_epoch = ledger .active_epoch @@ -8389,6 +8437,9 @@ fn reconcile_writer_epoch_closures( .iter() .any(|epoch| epoch.epoch_id == commit.writer_epoch); if !known_epoch { + if retained_start_lsn.is_some_and(|start_lsn| commit.last_lsn < start_lsn) { + continue; + } return Err(WalStoreError::UnknownPreviousWriterEpoch); } let closure = ledger diff --git a/crates/warp-core/src/trusted_runtime_host.rs b/crates/warp-core/src/trusted_runtime_host.rs index 073f39cf..3acc60ce 100644 --- a/crates/warp-core/src/trusted_runtime_host.rs +++ b/crates/warp-core/src/trusted_runtime_host.rs @@ -2132,14 +2132,21 @@ impl TrustedRuntimeHost { if runtime_wal.recover_filesystem_tick_commit_after_error(first_correlation) { continue; } - if !runtime_wal.uses_filesystem_store() { - *runtime_wal = runtime_wal_before - .ok_or(TrustedRuntimeWalError::InMemoryRollbackSnapshotUnavailable)?; - } + let rollback_error = if runtime_wal.uses_filesystem_store() { + None + } else if let Some(snapshot) = runtime_wal_before { + *runtime_wal = snapshot; + None + } else { + Some(TrustedRuntimeWalError::InMemoryRollbackSnapshotUnavailable) + }; self.runtime = runtime_before; self.provenance = provenance_before; self.echo_operation_action_outcomes = action_outcomes_before; self.admitted_echo_operation_actions = admitted_actions_before; + if let Some(rollback_error) = rollback_error { + return Err(rollback_error.into()); + } return Err(error.into()); } } diff --git a/crates/warp-core/tests/causal_wal_hardening_tests.rs b/crates/warp-core/tests/causal_wal_hardening_tests.rs index 32b2a325..8451c5a4 100644 --- a/crates/warp-core/tests/causal_wal_hardening_tests.rs +++ b/crates/warp-core/tests/causal_wal_hardening_tests.rs @@ -8,6 +8,9 @@ clippy::unnecessary_debug_formatting )] +#[path = "support/child_process.rs"] +mod child_process; + use warp_core::causal_wal::{ apply_committed_transaction, audit_wal_release_readiness, build_materialization_outbox_transaction, build_retained_reading_transaction, @@ -50,7 +53,6 @@ use std::fs::{self, OpenOptions}; use std::io::ErrorKind; use std::io::{Read, Write}; use std::path::{Path, PathBuf}; -use std::process::Command; use std::sync::atomic::{AtomicU64, Ordering}; const WAL_SEGMENT_RECORD_MAGIC: &[u8; 8] = b"ECWALR1!"; @@ -1085,12 +1087,14 @@ fn fixed_seed_filesystem_writer_epoch_chain_survives_bounded_reopens() { must_ok(store.close_epoch(epoch.epoch_id)); previous_epoch_id = Some(epoch.epoch_id); let ledger_len = must_ok(fs::metadata(root.join("writer-epochs.ecwal"))).len(); - match bounded_ledger_len { - Some(expected) => assert_eq!( - ledger_len, expected, - "writer-epoch ledger must stay bounded after epoch {ordinal}" - ), - None => bounded_ledger_len = Some(ledger_len), + if ordinal > 0 { + match bounded_ledger_len { + Some(expected) => assert_eq!( + ledger_len, expected, + "writer-epoch ledger must stay bounded after epoch {ordinal}" + ), + None => bounded_ledger_len = Some(ledger_len), + } } } @@ -1137,19 +1141,15 @@ fn filesystem_writer_epoch_chain_crosses_independent_processes() { let root = temp_wal_root("writer-epoch-process"); let executable = must_ok(std::env::current_exe()); for phase in ["first", "second"] { - let status = must_ok( - Command::new(&executable) - .args([ - "--ignored", - "--exact", - "emit_filesystem_writer_epoch_process_step", - "--nocapture", - ]) - .env("ECHO_WRITER_EPOCH_TEST_ROOT", &root) - .env("ECHO_WRITER_EPOCH_TEST_PHASE", phase) - .status(), + child_process::run_child_phase( + &executable, + "emit_filesystem_writer_epoch_process_step", + phase, + "ECHO_WRITER_EPOCH_TEST_ROOT", + root.as_os_str(), + "ECHO_WRITER_EPOCH_TEST_PHASE", + &root, ); - assert!(status.success(), "child phase `{phase}` must succeed"); } must_ok(fs::remove_dir_all(root)); diff --git a/crates/warp-core/tests/external_action_protocol_tests.rs b/crates/warp-core/tests/external_action_protocol_tests.rs index ec20e101..73869325 100644 --- a/crates/warp-core/tests/external_action_protocol_tests.rs +++ b/crates/warp-core/tests/external_action_protocol_tests.rs @@ -4,9 +4,11 @@ #![allow(clippy::panic)] +#[path = "support/child_process.rs"] +mod child_process; + use std::cell::Cell; use std::path::PathBuf; -use std::process::Command; use std::sync::atomic::{AtomicU64, Ordering}; use warp_core::causal_wal::{ @@ -989,17 +991,15 @@ fn filesystem_external_action_lifecycle_crosses_independent_processes() { let wal_dir = TempWalDir::new("independent-processes"); let executable = must_ok(std::env::current_exe()); for phase in ["request", "claim", "settlement", "replay"] { - let status = must_ok( - Command::new(&executable) - .arg("--ignored") - .arg("--exact") - .arg("emit_filesystem_external_action_process_step") - .arg("--nocapture") - .env("ECHO_EXTERNAL_ACTION_PROCESS_ROOT", &wal_dir.0) - .env("ECHO_EXTERNAL_ACTION_PROCESS_PHASE", phase) - .status(), + child_process::run_child_phase( + &executable, + "emit_filesystem_external_action_process_step", + phase, + "ECHO_EXTERNAL_ACTION_PROCESS_ROOT", + wal_dir.0.as_os_str(), + "ECHO_EXTERNAL_ACTION_PROCESS_PHASE", + &wal_dir.0, ); - assert!(status.success(), "child phase {phase} failed: {status}"); } let report = must_ok(recover_filesystem_store( diff --git a/crates/warp-core/tests/support/child_process.rs b/crates/warp-core/tests/support/child_process.rs new file mode 100644 index 00000000..1a4c329c --- /dev/null +++ b/crates/warp-core/tests/support/child_process.rs @@ -0,0 +1,69 @@ +// SPDX-License-Identifier: Apache-2.0 +// © James Ross Ω FLYING•ROBOTS +//! Bounded child-process execution for filesystem recovery fixtures. + +use std::ffi::OsStr; +use std::fs; +use std::path::Path; +use std::process::Command; +use std::thread; +use std::time::{Duration, Instant}; + +const CHILD_PHASE_TIMEOUT: Duration = Duration::from_secs(30); +const CHILD_PHASE_POLL_INTERVAL: Duration = Duration::from_millis(10); + +/// Runs one ignored integration-test entrypoint with a fixed timeout. +pub fn run_child_phase( + executable: &Path, + test_name: &str, + phase: &str, + root_env_key: &str, + root_env_value: &OsStr, + phase_env_key: &str, + cleanup_root: &Path, +) { + let mut child = match Command::new(executable) + .args([ + "--ignored", + "--exact", + test_name, + "--nocapture", + "--test-threads=1", + ]) + .env(root_env_key, root_env_value) + .env(phase_env_key, phase) + .spawn() + { + Ok(child) => child, + Err(error) => { + let _ = fs::remove_dir_all(cleanup_root); + panic!("child phase `{phase}` failed to spawn: {error}"); + } + }; + let deadline = Instant::now() + CHILD_PHASE_TIMEOUT; + loop { + match child.try_wait() { + Ok(Some(status)) if status.success() => return, + Ok(Some(status)) => { + let _ = fs::remove_dir_all(cleanup_root); + panic!("child phase `{phase}` failed: {status}"); + } + Ok(None) if Instant::now() >= deadline => { + let _ = child.kill(); + let _ = child.wait(); + let _ = fs::remove_dir_all(cleanup_root); + panic!( + "child phase `{phase}` exceeded the {}s deadline", + CHILD_PHASE_TIMEOUT.as_secs() + ); + } + Ok(None) => thread::sleep(CHILD_PHASE_POLL_INTERVAL), + Err(error) => { + let _ = child.kill(); + let _ = child.wait(); + let _ = fs::remove_dir_all(cleanup_root); + panic!("child phase `{phase}` wait failed: {error}"); + } + } + } +} From eea3fe0e606c6eb1996d72c9a1c2c322be02bdc3 Mon Sep 17 00:00:00 2001 From: James Ross Date: Thu, 30 Jul 2026 06:40:24 -0700 Subject: [PATCH 10/11] docs: delimit filesystem writer fencing authority --- docs/topics/WAL.md | 19 ++++++++++++++----- 1 file changed, 14 insertions(+), 5 deletions(-) diff --git a/docs/topics/WAL.md b/docs/topics/WAL.md index f24baffa..829624f0 100644 --- a/docs/topics/WAL.md +++ b/docs/topics/WAL.md @@ -198,9 +198,12 @@ Twelfth, strict filesystem WAL writer authority survives process loss as a durable, checksummed epoch ledger rather than process memory. Epoch acquisition is persisted before the store returns authority. Every committed transaction then advances that epoch's final LSN and commit-digest evidence, and explicit -closure persists the complete predecessor link. Reopen restores closed and -active epochs and reconciles a commit that reached its segment sync boundary -before the ledger snapshot. +closure persists the exact predecessor link. The ledger retains the active +epoch and only its latest closed predecessor; the trusted successor identity +recursively commits the preceding epoch identity and final commit while older +transactions remain in the WAL. Reopen cost and ledger size therefore remain +bounded, and reconciliation still admits a commit that reached its segment +sync boundary before the ledger snapshot. The filesystem store holds an operating-system writer lease for the complete active epoch. A second live process cannot append, close, or replace that @@ -210,8 +213,14 @@ successor with a monotonic start LSN, new epoch identity, fencing token, and lease evidence bound to the exact latest predecessor and its final commit. Duplicate identities, stale or missing predecessor links, reused fencing evidence, LSN regression, corrupted ledgers, and commits without their epoch -ledger fail closed before append. This fencing is generic WAL authority; it -contains no application vocabulary or callback. +ledger fail closed before append. + +The operating-system lease is the filesystem adapter's exclusion authority. +The persisted fencing, process, host, and lease fields are deterministic +chain-position markers, not ambient PID or machine measurements and not a +second lock. Their distinct values make epoch transitions attributable and +replayable without sampling time, randomness, or global process state. This +generic WAL authority contains no application vocabulary or callback. ## Boundaries From 12a67f695cc7a6c5efb8ef29519cd9dd90ae2f27 Mon Sep 17 00:00:00 2001 From: James Ross Date: Thu, 30 Jul 2026 06:40:40 -0700 Subject: [PATCH 11/11] docs: record bounded writer epoch ledgers --- CHANGELOG.md | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a9d3b4fc..3f16af86 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,11 +8,14 @@ ### Added - Strict filesystem WAL stores now persist a checksummed writer-epoch ledger - containing active and closed epochs plus exact final LSN and commit-digest - evidence. An OS-backed writer lease refuses overlapping processes before - append; a recovered trusted host closes only an abandoned epoch under that - lease and derives a fresh, monotonically linked successor. Duplicate, stale, - skipped, regressed, or unfenced epoch chains fail closed. Independent-process + containing the active epoch, its exact latest closed predecessor, and final + LSN and commit-digest evidence. Bounded retention keeps ledger writes and + reopen validation independent of lifetime restart count. An OS-backed writer + lease—not the deterministic chain markers stored in the generic fencing, + process, host, and lease fields—refuses overlapping processes before append. + A recovered trusted host closes only an abandoned epoch under that lease and + derives a fresh, monotonically linked successor. Duplicate, stale, skipped, + regressed, or unfenced epoch chains fail closed. Independent-process witnesses carry an external-action request, claim, settlement, and effect-free replay across successive host processes. - Echo now consumes the exact independently verified Edict