diff --git a/.ban-nondeterminism-allowlist b/.ban-nondeterminism-allowlist index 2be9b786..21005c6e 100644 --- a/.ban-nondeterminism-allowlist +++ b/.ban-nondeterminism-allowlist @@ -11,10 +11,15 @@ 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-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/CHANGELOG.md b/CHANGELOG.md index 9a0a187c..3f16af86 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,17 @@ ### Added +- Strict filesystem WAL stores now persist a checksummed writer-epoch ledger + 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 `workspace.patch.applyValidated@1` request through a capability-rooted single-file patch adapter. The request binds the prior bounded-observation diff --git a/crates/warp-core/src/causal_wal.rs b/crates/warp-core/src/causal_wal.rs index f5a24081..0c574e9b 100644 --- a/crates/warp-core/src/causal_wal.rs +++ b/crates/warp-core/src/causal_wal.rs @@ -72,6 +72,10 @@ 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_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"; @@ -843,6 +847,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 +1561,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 +1586,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 +5572,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 +5618,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 +5667,168 @@ 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.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> { + 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); + 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); + } + } + + 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 +5869,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 +5900,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 +5931,7 @@ impl FilesystemWalStore { &mut self, epoch_id: WriterEpochId, ) -> Result { + self.ensure_writer_lock()?; let active_epoch = self .active_epoch .as_ref() @@ -5775,6 +5968,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 +5983,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 +5998,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 +6092,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 +6121,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 +6133,12 @@ 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); + } + self.writer_lock = None; Ok(()) } } @@ -7941,6 +8159,321 @@ 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( + 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(closed_epochs.len()).to_le_bytes()); + for epoch in closed_epochs { + push_writer_epoch(&mut payload, epoch); + push_writer_epoch_closure( + &mut payload, + epoch_closures + .get(&epoch.epoch_id) + .copied() + .unwrap_or_default(), + ); + } + match active_epoch { + Some(epoch) => { + payload.push(1); + push_writer_epoch(&mut payload, epoch); + push_writer_epoch_closure( + &mut payload, + 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)?; + if closed_len > WAL_WRITER_EPOCH_RETAINED_CLOSED_LIMIT { + return Err(WalStoreError::WriterEpochChainGap); + } + let mut ledger = WriterEpochLedger::default(); + 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)?; + 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); + } + 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( + 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()); + 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, + active_epoch: Option<&WriterEpoch>, + 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(active_epoch, closed_epochs, epoch_closures); + { + 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); + } + 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 + .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 { + if retained_start_lsn.is_some_and(|start_lsn| commit.last_lsn < start_lsn) { + continue; + } + 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 +9892,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..3acc60ce 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 { @@ -2129,13 +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; - } + 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()); } } @@ -2417,7 +2428,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 +2488,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 +2553,47 @@ 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(), + }) + } + + /// 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"))] @@ -3305,7 +3349,7 @@ impl TrustedRuntimeWal { } } -#[derive(Clone, Debug)] +#[derive(Debug)] enum TrustedRuntimeWalStore { InMemory(InMemoryWalStore), Filesystem(FilesystemWalStore), @@ -3353,6 +3397,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 3e17d7c0..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, @@ -881,6 +884,277 @@ 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 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 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(); + 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; + 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}"); + 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); + let ledger_len = must_ok(fs::metadata(root.join("writer-epochs.ecwal"))).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), + } + } + } + + 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"] { + 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, + ); + } + + must_ok(fs::remove_dir_all(root)); +} + #[test] fn interleaved_transactions_rejected() { let first = submission_transaction("interleaved:first", Lsn::from_raw(0)); diff --git a/crates/warp-core/tests/external_action_protocol_tests.rs b/crates/warp-core/tests/external_action_protocol_tests.rs index 8dda756c..73869325 100644 --- a/crates/warp-core/tests/external_action_protocol_tests.rs +++ b/crates/warp-core/tests/external_action_protocol_tests.rs @@ -4,6 +4,9 @@ #![allow(clippy::panic)] +#[path = "support/child_process.rs"] +mod child_process; + use std::cell::Cell; use std::path::PathBuf; use std::sync::atomic::{AtomicU64, Ordering}; @@ -103,6 +106,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 +911,112 @@ 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"] { + 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, + ); + } + + 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"); 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}"); + } + } + } +} 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); diff --git a/docs/topics/WAL.md b/docs/topics/WAL.md index ee8dce68..829624f0 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,34 @@ 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 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 +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. + +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 The WAL belongs to the trusted runtime host. Application-facing code can submit @@ -391,6 +419,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