From b57ce5665f3b747c754d44f487d5a2130e72ac08 Mon Sep 17 00:00:00 2001 From: James Ross Date: Mon, 12 Oct 2026 20:11:37 -0700 Subject: [PATCH 01/10] Acknowledge a batch of submissions after one WAL sync A host that submits many intents in one frame waited for one storage sync per submission. submit_intents_with_runtime_wal_ack accepts the intents in order, appends each one's ordinary acceptance transaction with its commit record unsynced, then syncs the segment once and acknowledges the batch. The bytes on disk are the transactions single submission writes, so recovery is unchanged. A batch is not atomic across a crash: each transaction is atomic through its own commit record, a torn tail is truncated as before, and a surviving prefix recovers as accepted and pending with nothing acknowledged. On an intake or append error the batch returns the error and no handles; earlier submissions stay accepted and are synced, and resubmitting returns the existing submission. Segment rotation syncs deferred commit records first. Tests count commit syncs: three single submissions force three, a batch of five forces one, and a fresh host recovers all eight as accepted and pending. A batch whose first append fails acknowledges nothing and succeeds on retry. Measured in the guarded worker (Linux VM, release build, filesystem WAL on the VM disk, 120 frames, one run, frame p50): 8 actions per frame 6.44 ms -> 3.55 ms; 32 actions per frame 22.99 ms -> 10.69 ms. This is unit 1 of #787. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 2 + crates/warp-core/src/causal_wal.rs | 78 ++++++++++- crates/warp-core/src/trusted_runtime_host.rs | 74 +++++++++++ .../tests/trusted_runtime_host_loop_tests.rs | 121 ++++++++++++++++++ docs/topics/WAL.md | 7 + 5 files changed, 277 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 57ac6bf6..e234dfcc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,8 @@ ### Added +- `submit_intents_with_runtime_wal_ack` accepts several intents and returns after one storage sync has made all of their acceptance transactions durable. Each intent writes the same acceptance transaction as a single submission, so recovery is unchanged. A failed batch acknowledges nothing; submissions before the failure stay accepted and pending, and resubmitting returns the existing submission. + - Native projected compare-and-set invocation encoding retains canonical application input and statically binds the expected attachment-value digest. The native lowerer and independent verifier reconstruct CAS-specific profiles and reject rebound digest paths or under-budget configurations. Generated schema, reproducible components and a retained public Edict compiler build establish package production. The retained compiler package now has scheduler execution, pending/committed Action recovery, stale-digest noncommit recovery and typed refusal witnesses. `EchoOperationInvocationV1::anchored_node_attachment_compare_and_set_with_application_input` and `ExecutableOperationPackageV1::with_compare_and_set_application_result_projection` construct the projected forms without byte editing. - The trusted executable-operation host can retain two native alternative strands, diff --git a/crates/warp-core/src/causal_wal.rs b/crates/warp-core/src/causal_wal.rs index e3b4b41f..a177f265 100644 --- a/crates/warp-core/src/causal_wal.rs +++ b/crates/warp-core/src/causal_wal.rs @@ -5588,6 +5588,13 @@ pub struct FilesystemWalStore { writer_lock: Option, manifests: Vec, sync_evidence: Vec, + /// While set, a commit record is appended without its own storage sync. + defer_commit_sync: bool, + /// Transactions whose commit records are appended and not yet synced. + deferred_commits: Vec, + /// Number of storage syncs that made commit records durable. + #[cfg(any(test, feature = "host_test"))] + commit_sync_count: u64, #[cfg(any(test, feature = "host_test"))] fault_plan: FilesystemWalFaultPlan, } @@ -5638,6 +5645,10 @@ impl FilesystemWalStore { writer_lock: None, manifests: Vec::new(), sync_evidence, + defer_commit_sync: false, + deferred_commits: Vec::new(), + #[cfg(any(test, feature = "host_test"))] + commit_sync_count: 0, #[cfg(any(test, feature = "host_test"))] fault_plan: FilesystemWalFaultPlan::default(), }) @@ -5911,7 +5922,12 @@ impl FilesystemWalStore { )); } let transaction_id = commit.transaction_id; - append_segment_record(&self.segment_path(), DiskWalRecord::Commit(&commit), true)?; + let sync_now = !self.defer_commit_sync; + append_segment_record( + &self.segment_path(), + DiskWalRecord::Commit(&commit), + sync_now, + )?; self.epoch_closures.insert( epoch_id, WriterEpochClosure { @@ -5919,10 +5935,12 @@ impl FilesystemWalStore { final_commit_digest: Some(commit.commit_digest), }, ); - self.sync_evidence.push(FilesystemSyncEvidence::transaction( - FilesystemSyncBoundary::CommitFileSynced, - transaction_id, - )); + self.deferred_commits.push(transaction_id); + if sync_now { + // One sync of the segment file covers this commit record and any + // earlier record that was appended without its own sync. + self.record_commit_records_synced(); + } #[cfg(any(test, feature = "host_test"))] if self .fault_plan @@ -5935,6 +5953,54 @@ impl FilesystemWalStore { Ok(()) } + fn record_commit_records_synced(&mut self) { + #[cfg(any(test, feature = "host_test"))] + { + self.commit_sync_count += 1; + } + for transaction_id in self.deferred_commits.drain(..) { + self.sync_evidence.push(FilesystemSyncEvidence::transaction( + FilesystemSyncBoundary::CommitFileSynced, + transaction_id, + )); + } + } + + /// Starts appending commit records without a storage sync for each one. + /// + /// The caller must not report any transaction appended in this state as + /// durable until [`Self::finish_deferred_commit_sync`] returns `Ok`. + pub fn begin_deferred_commit_sync(&mut self) { + self.defer_commit_sync = true; + } + + /// Ends deferral and makes every commit record appended since + /// [`Self::begin_deferred_commit_sync`] durable with one storage sync. + /// + /// # Errors + /// + /// Returns a store error if the segment cannot be synced. The records stay + /// appended; a later successful commit sync also covers them. + pub fn finish_deferred_commit_sync(&mut self) -> Result<(), WalStoreError> { + self.defer_commit_sync = false; + if self.deferred_commits.is_empty() { + return Ok(()); + } + File::options() + .append(true) + .open(self.segment_path())? + .sync_all()?; + self.record_commit_records_synced(); + Ok(()) + } + + /// Returns how many storage syncs have made commit records durable. + #[cfg(any(test, feature = "host_test"))] + #[must_use] + pub fn commit_sync_count_for_test(&self) -> u64 { + self.commit_sync_count + } + /// Appends an uncommitted frame to simulate a torn transaction tail. pub fn append_uncommitted_frame( &mut self, @@ -5950,6 +6016,8 @@ impl FilesystemWalStore { epoch_id: WriterEpochId, ) -> Result { self.ensure_writer_lock()?; + // A sealed segment must not hold a commit record that was never synced. + self.finish_deferred_commit_sync()?; let active_epoch = self .active_epoch .as_ref() diff --git a/crates/warp-core/src/trusted_runtime_host.rs b/crates/warp-core/src/trusted_runtime_host.rs index dca41fd1..11b2a855 100644 --- a/crates/warp-core/src/trusted_runtime_host.rs +++ b/crates/warp-core/src/trusted_runtime_host.rs @@ -2853,6 +2853,30 @@ impl TrustedRuntimeWal { self.store.is_filesystem() } + fn begin_deferred_commit_sync(&mut self) { + if let TrustedRuntimeWalStore::Filesystem(store) = &mut self.store { + store.begin_deferred_commit_sync(); + } + } + + fn finish_deferred_commit_sync(&mut self) -> Result<(), TrustedRuntimeWalError> { + if let TrustedRuntimeWalStore::Filesystem(store) = &mut self.store { + store.finish_deferred_commit_sync()?; + } + Ok(()) + } + + /// Returns how many storage syncs have made commit records durable, when + /// the adapter is backed by the filesystem store. + #[cfg(any(test, feature = "host_test"))] + #[must_use] + pub fn filesystem_commit_sync_count_for_test(&self) -> Option { + match &self.store { + TrustedRuntimeWalStore::Filesystem(store) => Some(store.commit_sync_count_for_test()), + TrustedRuntimeWalStore::InMemory(_) => None, + } + } + #[cfg(any(test, feature = "host_test"))] fn replace_filesystem_fault_plan_for_test( &mut self, @@ -6177,6 +6201,56 @@ impl TrustedRuntimeApp<'_> { self.submit_intent_with_runtime_wal_ack_inner(envelope, AppIntentAdmission::Ordinary) } + /// Submits several canonical intents and returns only after one storage + /// sync has made all of their acceptance transactions durable. + /// + /// Each envelope is accepted exactly as + /// [`Self::submit_intent_with_runtime_wal_ack`] accepts it, in order, and + /// writes the same acceptance transaction. Only the sync is shared. + /// + /// # Errors + /// + /// Returns the first intake or WAL error, or the error of the final sync. + /// No handle is returned on error, so no submission of the batch is + /// acknowledged. Submissions before the failing one stay accepted and + /// pending, as they would after a process loss between commit and + /// acknowledgement; submitting the same envelope again returns its + /// existing submission. + pub fn submit_intents_with_runtime_wal_ack( + &mut self, + envelopes: Vec, + ) -> Result, TrustedRuntimeHostError> { + let Some(runtime_wal) = self.host.runtime_wal.as_mut() else { + return Err(TrustedRuntimeHostError::RuntimeWalUnavailable); + }; + runtime_wal.begin_deferred_commit_sync(); + let mut handles = Vec::with_capacity(envelopes.len()); + let mut failure = None; + for envelope in envelopes { + match self + .submit_intent_with_runtime_wal_ack_inner(envelope, AppIntentAdmission::Ordinary) + { + Ok(handle) => handles.push(handle), + Err(error) => { + failure = Some(error); + break; + } + } + } + // Always leave deferral, and sync whatever was appended: those records + // are in the log whether or not the batch is acknowledged. + let synced = self + .host + .runtime_wal + .as_mut() + .map_or(Ok(()), TrustedRuntimeWal::finish_deferred_commit_sync); + if let Some(error) = failure { + return Err(error); + } + synced?; + Ok(handles) + } + fn submit_intent_with_runtime_wal_ack_inner( &mut self, envelope: IngressEnvelope, 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 6a180d50..cc64ab49 100644 --- a/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs +++ b/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs @@ -1909,6 +1909,127 @@ fn filesystem_runtime_wal_failure_submission_restores_exact_runtime() { } } +fn distinct_intents(worldline_id: WorldlineId, count: u8) -> Vec { + (0..count) + .map(|index| { + IngressEnvelope::local_intent( + IngressTarget::DefaultWriter { worldline_id }, + make_intent_kind("echo.intent/eint-v1"), + vec![index; 4], + ) + }) + .collect() +} + +#[test] +fn batched_submission_makes_every_acceptance_durable_with_one_sync() { + let wal_root = temp_runtime_wal_dir("batched-submission-one-sync"); + let (runtime, worldline_id) = runtime(); + let mut host = + TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize"); + host.enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&wal_root)) + .expect("host should configure filesystem runtime WAL adapter"); + let syncs = |host: &TrustedRuntimeHost| { + host.runtime_wal() + .expect("runtime WAL should stay configured") + .filesystem_commit_sync_count_for_test() + .expect("filesystem store") + }; + + let single_before = syncs(&host); + let mut envelopes = distinct_intents(worldline_id, 8); + let batch = envelopes.split_off(3); + for envelope in envelopes { + host.app() + .submit_intent_with_runtime_wal_ack(envelope) + .expect("single submission should be acknowledged"); + } + assert_eq!( + syncs(&host) - single_before, + 3, + "each single submission waits for its own sync" + ); + + let batch_before = syncs(&host); + let handles = host + .app() + .submit_intents_with_runtime_wal_ack(batch) + .expect("batched submission should be acknowledged"); + assert_eq!(handles.len(), 5); + assert!(handles.iter().all(|handle| !handle.duplicate)); + assert_eq!( + syncs(&host) - batch_before, + 1, + "a batch waits for one sync, not one per submission" + ); + + let submission_ids = handles + .iter() + .map(|handle| handle.submission_id) + .collect::>(); + drop(host); + let (runtime, _) = crate::runtime(); + let mut reopened = + TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize"); + reopened + .enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&wal_root)) + .expect("a fresh host should recover the batch"); + let recovery = reopened + .runtime_wal() + .expect("runtime WAL should stay configured") + .recover_read_only() + .expect("batched acceptances should recover"); + assert_eq!(recovery.certificate.committed_transactions_replayed, 8); + for submission_id in submission_ids { + assert_eq!( + recovery + .submissions + .get(&submission_id) + .expect("every batched submission should recover") + .posture, + RecoveredSubmissionPosture::AcceptedPending + ); + } + drop(reopened); + fs::remove_dir_all(&wal_root).expect("WAL fixture should be removable"); +} + +#[test] +fn batched_submission_failure_acknowledges_nothing_and_can_be_retried() { + let wal_root = temp_runtime_wal_dir("batched-submission-failure"); + let (runtime, worldline_id) = runtime(); + let mut host = + TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize"); + host.enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&wal_root)) + .expect("host should configure filesystem runtime WAL adapter"); + host.inject_runtime_wal_filesystem_fault_for_test(FilesystemWalFaultPlan::fail_next( + FilesystemWalFaultTarget::AppendFrame, + )) + .expect("host should inject a filesystem failure"); + + host.app() + .submit_intents_with_runtime_wal_ack(distinct_intents(worldline_id, 4)) + .expect_err("a batch whose first append fails must not be acknowledged"); + assert_eq!(host.runtime().witnessed_submission_count(), 0); + + let handles = host + .app() + .submit_intents_with_runtime_wal_ack(distinct_intents(worldline_id, 4)) + .expect("the same batch must be acceptable once the WAL accepts it"); + assert_eq!(handles.len(), 4); + assert!(handles.iter().all(|handle| !handle.duplicate)); + assert_eq!(host.runtime().witnessed_submission_count(), 4); + + let single = host + .app() + .submit_intent_with_runtime_wal_ack(distinct_intents(worldline_id, 1).remove(0)) + .expect("resubmitting a batched envelope returns its submission"); + assert!(single.duplicate); + assert_eq!(single.submission_id, handles[0].submission_id); + drop(host); + fs::remove_dir_all(&wal_root).expect("WAL fixture should be removable"); +} + #[test] fn filesystem_runtime_wal_failure_manifest_publish_reports_store_error() { let wal_root = temp_runtime_wal_dir("failure-manifest-publish"); diff --git a/docs/topics/WAL.md b/docs/topics/WAL.md index d019498d..500f9335 100644 --- a/docs/topics/WAL.md +++ b/docs/topics/WAL.md @@ -25,6 +25,13 @@ after the runtime WAL has committed the submission-acceptance transaction. That acceptance fact recovers as an `AcceptedPending` submission if the scheduler has not yet produced a decided outcome. +`submit_intents_with_runtime_wal_ack(...)` accepts several intents in order +and returns after one storage sync covers all of their acceptance +transactions. Each intent still writes its own acceptance transaction, so +recovery reads a batch exactly as it reads single submissions. A batch is not +atomic across a crash: a prefix can survive, none of it acknowledged, and each +survivor recovers as `AcceptedPending`. + Second, failed WAL acceptance is not allowed to leave half-visible runtime state. If no runtime WAL is configured, the ACK path returns an explicit unavailable error before mutating intake state. If WAL commit fails, Echo rolls From defaece770cd11ed198d385622f8f8a97ad9b4a1 Mon Sep 17 00:00:00 2001 From: James Ross Date: Mon, 12 Oct 2026 20:34:57 -0700 Subject: [PATCH 02/10] Keep batched acceptances non-durable until their sync and fail closed Address independent review of batched submission. An acceptance appended without its own sync was recorded as durable at once, so a scheduler pass could have admitted an Action whose acceptance was not on stable storage. Unsynced acceptances are now tracked separately. They stop a second append of the same submission and are never treated as durable; they are promoted only after the shared sync, or a later synced commit on the same segment, succeeds. Deferral is no longer a store-wide flag that an early return or panic could leave set. The caller chooses per append whether the commit record is synced, and one call syncs every deferred record. Closing an epoch and rotating a segment sync deferred records first. The commit-marker fault point fires only for a commit that was synced, and a new fault target fails the shared sync. A failed shared sync is no longer hidden behind an earlier intake error. It rolls back the in-memory intake of every submission the call accepted and latches the adapter: further WAL work, and reconciliation from a merely readable log, are refused with StorageSyncFailed until the host reopens the WAL. Tests: sync failure restores the exact runtime, refuses later work, and a fresh host recovers what storage holds; a repeated envelope inside a batch appends once; an empty batch forces no sync; deferred commits carry no sync evidence until synced; a synced commit and an epoch close cover them. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 2 +- crates/warp-core/src/causal_wal.rs | 91 +++++--- crates/warp-core/src/trusted_runtime_host.rs | 204 +++++++++++++++--- .../tests/causal_wal_hardening_tests.rs | 82 +++++++ .../tests/trusted_runtime_host_loop_tests.rs | 115 ++++++++++ docs/topics/WAL.md | 14 +- 6 files changed, 449 insertions(+), 59 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e234dfcc..fee6be0c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,7 +7,7 @@ ### Added -- `submit_intents_with_runtime_wal_ack` accepts several intents and returns after one storage sync has made all of their acceptance transactions durable. Each intent writes the same acceptance transaction as a single submission, so recovery is unchanged. A failed batch acknowledges nothing; submissions before the failure stay accepted and pending, and resubmitting returns the existing submission. +- `submit_intents_with_runtime_wal_ack` accepts several intents and returns after one storage sync has made all of their acceptance transactions durable. Each intent writes the same acceptance transaction as a single submission, so recovery is unchanged. No acceptance of a batch counts as durable, or can be scheduled, before that sync succeeds. A failed batch acknowledges nothing. If an intent is refused, the intents before it stay accepted and are synced. If the sync fails, the call's intake is rolled back and the WAL refuses further work with `StorageSyncFailed` until it is reopened. - Native projected compare-and-set invocation encoding retains canonical application input and statically binds the expected attachment-value digest. The native lowerer and independent verifier reconstruct CAS-specific profiles and reject rebound digest paths or under-budget configurations. Generated schema, reproducible components and a retained public Edict compiler build establish package production. The retained compiler package now has scheduler execution, pending/committed Action recovery, stale-digest noncommit recovery and typed refusal witnesses. `EchoOperationInvocationV1::anchored_node_attachment_compare_and_set_with_application_input` and `ExecutableOperationPackageV1::with_compare_and_set_application_result_projection` construct the projected forms without byte editing. diff --git a/crates/warp-core/src/causal_wal.rs b/crates/warp-core/src/causal_wal.rs index a177f265..640fb253 100644 --- a/crates/warp-core/src/causal_wal.rs +++ b/crates/warp-core/src/causal_wal.rs @@ -5517,6 +5517,9 @@ pub enum FilesystemWalFaultTarget { CommitMarkerSynced, /// Fail the next manifest publish before writing manifest material. PublishManifest, + /// Fail the next storage sync that would make deferred commit records + /// durable. + DeferredCommitSync, } /// Host-test-only filesystem WAL fault plan. @@ -5527,6 +5530,7 @@ pub struct FilesystemWalFaultPlan { flush_commit: u32, commit_marker_synced: u32, publish_manifest: u32, + deferred_commit_sync: u32, } #[cfg(any(test, feature = "host_test"))] @@ -5540,24 +5544,35 @@ impl FilesystemWalFaultPlan { flush_commit: 0, commit_marker_synced: 0, publish_manifest: 0, + deferred_commit_sync: 0, }, FilesystemWalFaultTarget::FlushCommit => Self { append_frame: 0, flush_commit: 1, commit_marker_synced: 0, publish_manifest: 0, + deferred_commit_sync: 0, }, FilesystemWalFaultTarget::CommitMarkerSynced => Self { append_frame: 0, flush_commit: 0, commit_marker_synced: 1, publish_manifest: 0, + deferred_commit_sync: 0, }, FilesystemWalFaultTarget::PublishManifest => Self { append_frame: 0, flush_commit: 0, commit_marker_synced: 0, publish_manifest: 1, + deferred_commit_sync: 0, + }, + FilesystemWalFaultTarget::DeferredCommitSync => Self { + append_frame: 0, + flush_commit: 0, + commit_marker_synced: 0, + publish_manifest: 0, + deferred_commit_sync: 1, }, } } @@ -5568,6 +5583,7 @@ impl FilesystemWalFaultPlan { FilesystemWalFaultTarget::FlushCommit => &mut self.flush_commit, FilesystemWalFaultTarget::CommitMarkerSynced => &mut self.commit_marker_synced, FilesystemWalFaultTarget::PublishManifest => &mut self.publish_manifest, + FilesystemWalFaultTarget::DeferredCommitSync => &mut self.deferred_commit_sync, }; if *remaining == 0 { return false; @@ -5588,8 +5604,6 @@ pub struct FilesystemWalStore { writer_lock: Option, manifests: Vec, sync_evidence: Vec, - /// While set, a commit record is appended without its own storage sync. - defer_commit_sync: bool, /// Transactions whose commit records are appended and not yet synced. deferred_commits: Vec, /// Number of storage syncs that made commit records durable. @@ -5645,7 +5659,6 @@ impl FilesystemWalStore { writer_lock: None, manifests: Vec::new(), sync_evidence, - defer_commit_sync: false, deferred_commits: Vec::new(), #[cfg(any(test, feature = "host_test"))] commit_sync_count: 0, @@ -5875,6 +5888,30 @@ impl FilesystemWalStore { pub fn append_transaction( &mut self, transaction: WalCommittedTransaction, + ) -> Result<(), WalStoreError> { + self.append_transaction_with_commit_sync(transaction, true) + } + + /// Appends a transaction without syncing its commit record. + /// + /// The transaction is not durable until [`Self::sync_deferred_commits`] + /// or a later synced commit on the same segment succeeds. The caller must + /// not report it as durable before then. + /// + /// # Errors + /// + /// Returns the same errors as [`Self::append_transaction`]. + pub fn append_transaction_deferring_commit_sync( + &mut self, + transaction: WalCommittedTransaction, + ) -> Result<(), WalStoreError> { + self.append_transaction_with_commit_sync(transaction, false) + } + + fn append_transaction_with_commit_sync( + &mut self, + transaction: WalCommittedTransaction, + sync_now: bool, ) -> Result<(), WalStoreError> { transaction.validate()?; let admission_kernel_capability = transaction.admission_kernel_capability; @@ -5889,6 +5926,7 @@ impl FilesystemWalStore { transaction.commit, admission_kernel_capability, external_action_coordinator_capability, + sync_now, ) } @@ -5898,6 +5936,7 @@ impl FilesystemWalStore { commit: WalTransactionCommit, admission_kernel_capability: Option, external_action_coordinator_capability: Option, + sync_now: bool, ) -> Result<(), WalStoreError> { self.ensure_writer_lock()?; let active_epoch = self @@ -5922,7 +5961,6 @@ impl FilesystemWalStore { )); } let transaction_id = commit.transaction_id; - let sync_now = !self.defer_commit_sync; append_segment_record( &self.segment_path(), DiskWalRecord::Commit(&commit), @@ -5941,10 +5979,13 @@ impl FilesystemWalStore { // earlier record that was appended without its own sync. self.record_commit_records_synced(); } + // The marker-synced fault models a failure after this commit's own + // sync. A deferred commit has no sync of its own to fail after. #[cfg(any(test, feature = "host_test"))] - if self - .fault_plan - .should_fail(FilesystemWalFaultTarget::CommitMarkerSynced) + if sync_now + && self + .fault_plan + .should_fail(FilesystemWalFaultTarget::CommitMarkerSynced) { return Err(WalStoreError::Io( "injected filesystem WAL commit_marker_synced failure".to_owned(), @@ -5966,26 +6007,26 @@ impl FilesystemWalStore { } } - /// Starts appending commit records without a storage sync for each one. - /// - /// The caller must not report any transaction appended in this state as - /// durable until [`Self::finish_deferred_commit_sync`] returns `Ok`. - pub fn begin_deferred_commit_sync(&mut self) { - self.defer_commit_sync = true; - } - - /// Ends deferral and makes every commit record appended since - /// [`Self::begin_deferred_commit_sync`] durable with one storage sync. + /// Makes every commit record that was appended without its own sync + /// durable with one storage sync. /// /// # Errors /// - /// Returns a store error if the segment cannot be synced. The records stay - /// appended; a later successful commit sync also covers them. - pub fn finish_deferred_commit_sync(&mut self) -> Result<(), WalStoreError> { - self.defer_commit_sync = false; + /// Returns a store error if the segment cannot be synced. Nothing appended + /// without a sync may be reported as durable after such an error. + pub fn sync_deferred_commits(&mut self) -> Result<(), WalStoreError> { if self.deferred_commits.is_empty() { return Ok(()); } + #[cfg(any(test, feature = "host_test"))] + if self + .fault_plan + .should_fail(FilesystemWalFaultTarget::DeferredCommitSync) + { + return Err(WalStoreError::Io( + "injected filesystem WAL deferred_commit_sync failure".to_owned(), + )); + } File::options() .append(true) .open(self.segment_path())? @@ -6017,7 +6058,7 @@ impl FilesystemWalStore { ) -> Result { self.ensure_writer_lock()?; // A sealed segment must not hold a commit record that was never synced. - self.finish_deferred_commit_sync()?; + self.sync_deferred_commits()?; let active_epoch = self .active_epoch .as_ref() @@ -6116,7 +6157,7 @@ impl WalStorePort for FilesystemWalStore { epoch_id: WriterEpochId, commit: WalTransactionCommit, ) -> Result<(), WalStoreError> { - self.flush_commit_with_capabilities(epoch_id, commit, None, None) + self.flush_commit_with_capabilities(epoch_id, commit, None, None, true) } fn flush_external_action_commit( @@ -6125,7 +6166,7 @@ impl WalStorePort for FilesystemWalStore { commit: WalTransactionCommit, capability: ExternalActionCoordinatorCapability, ) -> Result<(), WalStoreError> { - self.flush_commit_with_capabilities(epoch_id, commit, None, Some(capability)) + self.flush_commit_with_capabilities(epoch_id, commit, None, Some(capability), true) } fn read_frames(&self) -> Vec { @@ -6208,6 +6249,8 @@ impl WalStorePort for FilesystemWalStore { fn close_epoch(&mut self, epoch_id: WriterEpochId) -> Result<(), WalStoreError> { self.ensure_writer_lock()?; + // An epoch must not close over a commit record that was never synced. + self.sync_deferred_commits()?; let previous_ledger = self.writer_epoch_ledger(); let epoch = self .active_epoch diff --git a/crates/warp-core/src/trusted_runtime_host.rs b/crates/warp-core/src/trusted_runtime_host.rs index 11b2a855..0feb06c4 100644 --- a/crates/warp-core/src/trusted_runtime_host.rs +++ b/crates/warp-core/src/trusted_runtime_host.rs @@ -173,6 +173,11 @@ pub enum TrustedRuntimeWalError { /// An earlier append must be reconciled before another can reach storage. #[error("trusted runtime WAL writer prefix requires reconciliation")] UncertainWriterPrefix, + /// A storage sync failed. Records written before it are not known to be + /// durable, so this adapter refuses further work until the host reopens + /// the WAL and recovers from what storage actually holds. + #[error("trusted runtime WAL storage sync failed; reopen the WAL to recover")] + StorageSyncFailed, /// WAL transaction construction failed before storage append. #[error("trusted runtime WAL transaction build error: {0}")] Build(#[from] WalBuildError), @@ -2518,6 +2523,22 @@ pub struct TrustedRuntimeWal { causal_history_frontier_digest: Hash, causal_anchor_claim_projection: CausalAnchorClaimProjection, durable_submission_acceptances: BTreeMap, + /// Acceptances whose commit records are appended and not yet synced. They + /// stop a second append of the same submission and are never treated as + /// durable. + unsynced_submission_acceptances: BTreeMap, + /// Set when a storage sync fails; see + /// [`TrustedRuntimeWalError::StorageSyncFailed`]. + storage_sync_failed: bool, +} + +/// Whether an appended commit record is synced before the append returns. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum CommitSync { + /// Sync the commit record now. The transaction is durable on return. + Now, + /// Leave the commit record unsynced for a later shared sync. + Deferred, } impl TrustedRuntimeWal { @@ -2587,6 +2608,8 @@ impl TrustedRuntimeWal { causal_history_frontier_digest: recovered_cursor.causal_history_frontier_digest, causal_anchor_claim_projection: recovered_cursor.causal_anchor_claim_projection, durable_submission_acceptances, + unsynced_submission_acceptances: BTreeMap::new(), + storage_sync_failed: false, evidence_catalog: Some(evidence_catalog), evidence_catalog_posture: EvidenceCatalogPosture::Fresh, #[cfg(any(test, feature = "host_test"))] @@ -2653,6 +2676,8 @@ impl TrustedRuntimeWal { 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(), + unsynced_submission_acceptances: self.unsynced_submission_acceptances.clone(), + storage_sync_failed: self.storage_sync_failed, }) } @@ -2853,19 +2878,45 @@ impl TrustedRuntimeWal { self.store.is_filesystem() } - fn begin_deferred_commit_sync(&mut self) { - if let TrustedRuntimeWalStore::Filesystem(store) = &mut self.store { - store.begin_deferred_commit_sync(); - } + /// Returns `true` when an acceptance for this submission is in the log, + /// whether or not its commit record has been synced. Use this only to + /// avoid appending the same acceptance twice. + fn has_recorded_submission_acceptance( + &self, + submission_id: Hash, + canonical_envelope_digest: Hash, + ) -> bool { + self.has_submission_acceptance(submission_id, canonical_envelope_digest) + || self.unsynced_submission_acceptances.get(&submission_id) + == Some(&canonical_envelope_digest) } - fn finish_deferred_commit_sync(&mut self) -> Result<(), TrustedRuntimeWalError> { + /// Makes every transaction appended with [`CommitSync::Deferred`] durable + /// with one storage sync, and only then counts its acceptances as durable. + fn sync_deferred_commits(&mut self) -> Result<(), TrustedRuntimeWalError> { + if self.storage_sync_failed { + return Err(TrustedRuntimeWalError::StorageSyncFailed); + } if let TrustedRuntimeWalStore::Filesystem(store) = &mut self.store { - store.finish_deferred_commit_sync()?; + if let Err(error) = store.sync_deferred_commits() { + self.storage_sync_failed = true; + return Err(error.into()); + } } + self.promote_synced_submission_acceptances(); Ok(()) } + fn promote_synced_submission_acceptances(&mut self) { + let synced = std::mem::take(&mut self.unsynced_submission_acceptances); + self.durable_submission_acceptances.extend(synced); + } + + /// Forgets unsynced acceptances that the host is not going to acknowledge. + fn forget_unsynced_submission_acceptances(&mut self) { + self.unsynced_submission_acceptances.clear(); + } + /// Returns how many storage syncs have made commit records durable, when /// the adapter is backed by the filesystem store. #[cfg(any(test, feature = "host_test"))] @@ -2888,6 +2939,12 @@ impl TrustedRuntimeWal { } fn refresh_cursor_from_store_for_writer(&mut self) -> Result<(), TrustedRuntimeWalError> { + // After a failed storage sync a readable record is not a durable one, + // so the live adapter must not reconcile itself from the log. Only + // reopening the WAL recovers. + if self.storage_sync_failed { + return Err(TrustedRuntimeWalError::StorageSyncFailed); + } self.operation_context_index = None; let report = self.store.recover_for_writer()?; let contexts = observed_context::Contexts::from_recovery(&report)?; @@ -3019,6 +3076,7 @@ impl TrustedRuntimeWal { &mut self, envelope: &IngressEnvelope, handle: IntentSubmissionHandle, + sync: CommitSync, ) -> Result { let record = SubmissionAcceptanceRecord { submission_id: handle.submission_id, @@ -3042,10 +3100,18 @@ impl TrustedRuntimeWal { after_digest: next_submission_frontier, }], )?; - let commit = self.append_transaction(transaction)?; + let commit = self.append_transaction_with(transaction, sync)?; self.submission_frontier_digest = next_submission_frontier; - self.durable_submission_acceptances - .insert(record.submission_id, record.canonical_envelope_digest); + match sync { + CommitSync::Now => { + self.durable_submission_acceptances + .insert(record.submission_id, record.canonical_envelope_digest); + } + CommitSync::Deferred => { + self.unsynced_submission_acceptances + .insert(record.submission_id, record.canonical_envelope_digest); + } + } Ok(commit) } @@ -3350,6 +3416,17 @@ impl TrustedRuntimeWal { &mut self, transaction: WalCommittedTransaction, ) -> Result { + self.append_transaction_with(transaction, CommitSync::Now) + } + + fn append_transaction_with( + &mut self, + transaction: WalCommittedTransaction, + sync: CommitSync, + ) -> Result { + if self.storage_sync_failed { + return Err(TrustedRuntimeWalError::StorageSyncFailed); + } let next_causal_history_frontier = logical_causal_history_frontier_digest( self.causal_history_frontier_digest, transaction.commit.transaction_kind, @@ -3378,7 +3455,17 @@ impl TrustedRuntimeWal { return Err(TrustedRuntimeWalError::UncertainWriterPrefix); } contexts.apply_frames(&frames)?; - self.store.append_transaction(transaction)?; + match sync { + CommitSync::Now => { + self.store.append_transaction(transaction)?; + // A synced commit covers every earlier record in the segment. + self.promote_synced_submission_acceptances(); + } + CommitSync::Deferred => { + self.store + .append_transaction_deferring_commit_sync(transaction)?; + } + } self.operation_context_index = Some((commit.commit_digest, contexts)); self.next_lsn = next_lsn; self.previous_frame_digest = last_frame_digest; @@ -3568,6 +3655,18 @@ impl TrustedRuntimeWalStore { } } + /// Appends without syncing the commit record. The in-memory store has no + /// sync, so it appends as usual. + fn append_transaction_deferring_commit_sync( + &mut self, + transaction: WalCommittedTransaction, + ) -> Result<(), WalStoreError> { + match self { + Self::InMemory(store) => store.append_transaction(transaction), + Self::Filesystem(store) => store.append_transaction_deferring_commit_sync(transaction), + } + } + #[cfg(any(test, feature = "host_test"))] fn append_uncommitted_frame( &mut self, @@ -6198,7 +6297,11 @@ impl TrustedRuntimeApp<'_> { &mut self, envelope: IngressEnvelope, ) -> Result { - self.submit_intent_with_runtime_wal_ack_inner(envelope, AppIntentAdmission::Ordinary) + self.submit_intent_with_runtime_wal_ack_inner( + envelope, + AppIntentAdmission::Ordinary, + CommitSync::Now, + ) } /// Submits several canonical intents and returns only after one storage @@ -6206,48 +6309,81 @@ impl TrustedRuntimeApp<'_> { /// /// Each envelope is accepted exactly as /// [`Self::submit_intent_with_runtime_wal_ack`] accepts it, in order, and - /// writes the same acceptance transaction. Only the sync is shared. + /// writes the same acceptance transaction. Only the sync is shared. No + /// acceptance of the batch counts as durable, and none can be scheduled, + /// before that sync succeeds. /// /// # Errors /// - /// Returns the first intake or WAL error, or the error of the final sync. /// No handle is returned on error, so no submission of the batch is - /// acknowledged. Submissions before the failing one stay accepted and - /// pending, as they would after a process loss between commit and - /// acknowledgement; submitting the same envelope again returns its - /// existing submission. + /// acknowledged. + /// + /// - If an envelope is refused, or its append fails, the envelopes before + /// it are still synced. They stay accepted and durable, as they would + /// after a process loss between commit and acknowledgement; submitting + /// the same envelope again returns its existing submission. The error of + /// the refused envelope is returned. + /// - If the storage sync fails, the in-memory intake of every submission + /// this call accepted is rolled back, the sync error is returned, and + /// the WAL refuses further work with + /// [`TrustedRuntimeWalError::StorageSyncFailed`] until the host reopens + /// it. Storage may still hold some of the records; reopening recovers + /// whatever it holds. pub fn submit_intents_with_runtime_wal_ack( &mut self, envelopes: Vec, ) -> Result, TrustedRuntimeHostError> { - let Some(runtime_wal) = self.host.runtime_wal.as_mut() else { + if self.host.runtime_wal.is_none() { return Err(TrustedRuntimeHostError::RuntimeWalUnavailable); - }; - runtime_wal.begin_deferred_commit_sync(); + } let mut handles = Vec::with_capacity(envelopes.len()); + let mut accepted = Vec::new(); let mut failure = None; for envelope in envelopes { - match self - .submit_intent_with_runtime_wal_ack_inner(envelope, AppIntentAdmission::Ordinary) - { - Ok(handle) => handles.push(handle), + let marker = self.host.runtime.intent_submission_rollback_marker(); + match self.submit_intent_with_runtime_wal_ack_inner( + envelope, + AppIntentAdmission::Ordinary, + CommitSync::Deferred, + ) { + Ok(handle) => { + if !handle.duplicate { + accepted.push((marker, handle)); + } + handles.push(handle); + } Err(error) => { failure = Some(error); break; } } } - // Always leave deferral, and sync whatever was appended: those records - // are in the log whether or not the batch is acknowledged. + // Sync whatever was appended: those records are in the log whether or + // not the batch is acknowledged. let synced = self .host .runtime_wal .as_mut() - .map_or(Ok(()), TrustedRuntimeWal::finish_deferred_commit_sync); + .map_or(Ok(()), TrustedRuntimeWal::sync_deferred_commits); + if let Err(error) = synced { + // Nothing this call appended is known to be durable. Do not leave + // it visible as accepted intake. + for (marker, handle) in accepted.into_iter().rev() { + self.host + .pending_echo_operation_actions + .remove(&handle.submission_id); + self.host + .runtime + .rollback_intent_submission(marker, &handle); + } + if let Some(runtime_wal) = self.host.runtime_wal.as_mut() { + runtime_wal.forget_unsynced_submission_acceptances(); + } + return Err(error.into()); + } if let Some(error) = failure { return Err(error); } - synced?; Ok(handles) } @@ -6255,6 +6391,7 @@ impl TrustedRuntimeApp<'_> { &mut self, envelope: IngressEnvelope, admission: AppIntentAdmission, + sync: CommitSync, ) -> Result { if self.host.runtime_wal.is_none() { return Err(TrustedRuntimeHostError::RuntimeWalUnavailable); @@ -6279,7 +6416,8 @@ impl TrustedRuntimeApp<'_> { return Err(TrustedRuntimeHostError::RuntimeWalUnavailable); }; if handle.duplicate - && runtime_wal.has_submission_acceptance(handle.submission_id, envelope.ingress_id()) + && runtime_wal + .has_recorded_submission_acceptance(handle.submission_id, envelope.ingress_id()) { self.host.track_pending_echo_operation_action_v1( handle.submission_id, @@ -6287,7 +6425,7 @@ impl TrustedRuntimeApp<'_> { ); return Ok(handle); } - if let Err(error) = runtime_wal.record_submission_acceptance(&envelope, handle) { + if let Err(error) = runtime_wal.record_submission_acceptance(&envelope, handle, sync) { if runtime_wal.recover_filesystem_submission_acceptance_after_error( handle.submission_id, envelope.ingress_id(), @@ -6330,7 +6468,11 @@ impl TrustedRuntimeApp<'_> { return Err(TrustedRuntimeHostError::RuntimeWalUnavailable); } let envelope = self.host.resolve_contract_inverse_envelope(&request)?; - self.submit_intent_with_runtime_wal_ack_inner(envelope, AppIntentAdmission::ContractInverse) + self.submit_intent_with_runtime_wal_ack_inner( + envelope, + AppIntentAdmission::ContractInverse, + CommitSync::Now, + ) } /// Recovers the typed causal derivation for one admitted inverse receipt. diff --git a/crates/warp-core/tests/causal_wal_hardening_tests.rs b/crates/warp-core/tests/causal_wal_hardening_tests.rs index a2b4c5ee..bc4d5a65 100644 --- a/crates/warp-core/tests/causal_wal_hardening_tests.rs +++ b/crates/warp-core/tests/causal_wal_hardening_tests.rs @@ -1075,6 +1075,88 @@ fn filesystem_takeover_after_process_loss_derives_final_commit_from_the_segment( must_ok(fs::remove_dir_all(root)); } +fn commit_sync_evidence_count(store: &FilesystemWalStore) -> usize { + store + .sync_evidence() + .iter() + .filter(|entry| entry.boundary == FilesystemSyncBoundary::CommitFileSynced) + .count() +} + +#[test] +fn filesystem_deferred_commits_are_not_evidenced_as_synced_until_one_sync() { + let root = temp_wal_root("deferred-commit-sync"); + 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_deferring_commit_sync(submission_transaction( + "deferred-first", + Lsn::from_raw(0), + )), + ); + must_ok( + store.append_transaction_deferring_commit_sync(submission_transaction( + "deferred-second", + Lsn::from_raw(2), + )), + ); + assert_eq!( + commit_sync_evidence_count(&store), + 0, + "an unsynced commit record must not be evidenced as synced" + ); + + must_ok(store.sync_deferred_commits()); + assert_eq!(commit_sync_evidence_count(&store), 2); + must_ok(store.sync_deferred_commits()); + assert_eq!(commit_sync_evidence_count(&store), 2); + drop(store); + must_ok(fs::remove_dir_all(root)); +} + +#[test] +fn filesystem_synced_commit_and_epoch_close_cover_deferred_commits() { + let root = temp_wal_root("deferred-commit-covered"); + 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_deferring_commit_sync(submission_transaction( + "covered-first", + Lsn::from_raw(0), + )), + ); + must_ok(store.append_transaction(submission_transaction("covered-second", Lsn::from_raw(2)))); + assert_eq!( + commit_sync_evidence_count(&store), + 2, + "a synced commit covers every earlier record in the segment" + ); + + must_ok( + store.append_transaction_deferring_commit_sync(submission_transaction( + "covered-third", + Lsn::from_raw(4), + )), + ); + assert_eq!(commit_sync_evidence_count(&store), 2); + must_ok(store.close_epoch(epoch_id())); + assert_eq!( + commit_sync_evidence_count(&store), + 3, + "an epoch must not close over an unsynced commit record" + ); + drop(store); + + let report = must_ok(recover_filesystem_store( + &root, + RecoveryAccessMode::ReadOnly, + )); + assert_eq!(report.transactions.len(), 3); + must_ok(fs::remove_dir_all(root)); +} + #[test] fn filesystem_takeover_refuses_unreconciled_tail_before_reusing_its_lsn() { let mut fixture = WalHardeningFixture::new("takeover-uncommitted"); 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 cc64ab49..fa24b960 100644 --- a/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs +++ b/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs @@ -2030,6 +2030,121 @@ fn batched_submission_failure_acknowledges_nothing_and_can_be_retried() { fs::remove_dir_all(&wal_root).expect("WAL fixture should be removable"); } +#[test] +fn batched_submission_sync_failure_rolls_back_intake_and_fails_closed() { + let wal_root = temp_runtime_wal_dir("batched-submission-sync-failure"); + let (runtime, worldline_id) = runtime(); + let mut host = + TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize"); + host.enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&wal_root)) + .expect("host should configure filesystem runtime WAL adapter"); + let runtime_before = format!("{:?}", host.runtime()); + host.inject_runtime_wal_filesystem_fault_for_test(FilesystemWalFaultPlan::fail_next( + FilesystemWalFaultTarget::DeferredCommitSync, + )) + .expect("host should inject a sync failure"); + + let error = host + .app() + .submit_intents_with_runtime_wal_ack(distinct_intents(worldline_id, 4)) + .expect_err("a batch whose sync fails must not be acknowledged"); + assert!(matches!( + error, + TrustedRuntimeHostError::Wal(TrustedRuntimeWalError::Store(WalStoreError::Io( + message + ))) if message.contains("injected filesystem WAL deferred_commit_sync failure") + )); + assert_eq!( + format!("{:?}", host.runtime()), + runtime_before, + "a failed sync must not leave the batch visible as accepted intake" + ); + let refused = host + .app() + .submit_intent_with_runtime_wal_ack(distinct_intents(worldline_id, 1).remove(0)) + .expect_err("the WAL must refuse work after a failed sync"); + assert!(matches!( + refused, + TrustedRuntimeHostError::Wal(TrustedRuntimeWalError::StorageSyncFailed) + )); + assert_eq!( + format!("{:?}", host.runtime()), + runtime_before, + "a refused submission must not stay visible either" + ); + assert!( + matches!(host.tick_once(), Ok(steps) if steps.is_empty()), + "nothing from the failed batch may be scheduled" + ); + + // The records were written; only their sync failed. Reopening recovers + // what storage holds, as it would after a process loss. + drop(host); + let (runtime, _) = crate::runtime(); + let mut reopened = + TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize"); + reopened + .enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&wal_root)) + .expect("a fresh host should recover what storage holds"); + let recovery = reopened + .runtime_wal() + .expect("runtime WAL should stay configured") + .recover_read_only() + .expect("the written prefix should recover"); + assert_eq!(recovery.certificate.committed_transactions_replayed, 4); + assert!(recovery.receipts.receipt_by_submission.is_empty()); + drop(reopened); + fs::remove_dir_all(&wal_root).expect("WAL fixture should be removable"); +} + +#[test] +fn batched_submission_handles_duplicates_and_an_empty_batch() { + let wal_root = temp_runtime_wal_dir("batched-submission-duplicates"); + let (runtime, worldline_id) = runtime(); + let mut host = + TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize"); + host.enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&wal_root)) + .expect("host should configure filesystem runtime WAL adapter"); + let syncs = |host: &TrustedRuntimeHost| { + host.runtime_wal() + .expect("runtime WAL should stay configured") + .filesystem_commit_sync_count_for_test() + .expect("filesystem store") + }; + + let before = syncs(&host); + let empty = host + .app() + .submit_intents_with_runtime_wal_ack(Vec::new()) + .expect("an empty batch is acknowledged without work"); + assert!(empty.is_empty()); + assert_eq!(syncs(&host), before, "an empty batch forces no sync"); + + let mut envelopes = distinct_intents(worldline_id, 2); + envelopes.insert(1, envelopes[0].clone()); + let handles = host + .app() + .submit_intents_with_runtime_wal_ack(envelopes) + .expect("a batch that repeats an envelope should be acknowledged"); + assert_eq!(handles.len(), 3); + assert!(!handles[0].duplicate); + assert!(handles[1].duplicate); + assert_eq!(handles[1].submission_id, handles[0].submission_id); + assert!(!handles[2].duplicate); + assert_eq!(syncs(&host) - before, 1); + let runtime_wal = host + .runtime_wal() + .expect("runtime WAL should stay configured"); + assert_eq!( + runtime_wal.submission_acceptance_count(), + 2, + "a repeated envelope must not append a second acceptance" + ); + assert_eq!(runtime_wal.commits().len(), 2); + drop(host); + fs::remove_dir_all(&wal_root).expect("WAL fixture should be removable"); +} + #[test] fn filesystem_runtime_wal_failure_manifest_publish_reports_store_error() { let wal_root = temp_runtime_wal_dir("failure-manifest-publish"); diff --git a/docs/topics/WAL.md b/docs/topics/WAL.md index 500f9335..c9187e56 100644 --- a/docs/topics/WAL.md +++ b/docs/topics/WAL.md @@ -28,9 +28,17 @@ not yet produced a decided outcome. `submit_intents_with_runtime_wal_ack(...)` accepts several intents in order and returns after one storage sync covers all of their acceptance transactions. Each intent still writes its own acceptance transaction, so -recovery reads a batch exactly as it reads single submissions. A batch is not -atomic across a crash: a prefix can survive, none of it acknowledged, and each -survivor recovers as `AcceptedPending`. +recovery reads a batch exactly as it reads single submissions. Until that sync +succeeds, no acceptance of the batch counts as durable and none can be +scheduled. A batch is not atomic across a crash: a prefix can survive, none of +it acknowledged, and each survivor recovers as `AcceptedPending`. + +If an intent of a batch is refused, the intents before it are still synced and +stay accepted; the call returns the refusal and no handles. If the shared sync +fails, Echo rolls back the in-memory intake of every submission that call +accepted, returns the sync error, and refuses further WAL work with +`StorageSyncFailed` until the host reopens the WAL. A record that is readable +after a failed sync is not treated as durable. Second, failed WAL acceptance is not allowed to leave half-visible runtime state. If no runtime WAL is configured, the ACK path returns an explicit From d4f7cc12e10ef34f934f755354855ff953967086 Mon Sep 17 00:00:00 2001 From: James Ross Date: Mon, 12 Oct 2026 21:04:26 -0700 Subject: [PATCH 03/10] Fail closed on any storage error while commit syncs are deferred Address the second independent review of batched submission. When an append failed in the middle of a batch, the host reconciled the adapter from the log, as it does for a single submission. That counted every readable acceptance as durable, including the earlier acceptances of the same batch whose sync had not happened yet. If the shared sync then failed, those acceptances stayed marked durable after their intake was rolled back. An append error in a deferred batch now fails closed in the same way as a failed sync: the adapter latches StorageFailed (renamed from StorageSyncFailed), drops its unsynced acceptances, the in-memory intake of everything the call accepted is rolled back, and nothing is reconciled from the log until the host reopens the WAL. The filesystem store also latches after a failed sync of deferred commit records. It never evidences those records as synced, and refuses to append, sync, rotate or close: a later successful sync cannot vouch for what an earlier sync failed to write. The fault plan can let a number of operations through before failing, so a test can fail the second commit of a batch. Hosted CI now runs the trusted host loop tests and the WAL hardening tests with fault injection enabled; it ran neither before. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/ci.yml | 8 ++ CHANGELOG.md | 2 +- crates/warp-core/src/causal_wal.rs | 63 +++++++++- crates/warp-core/src/trusted_runtime_host.rs | 110 +++++++++++------- .../tests/causal_wal_hardening_tests.rs | 30 +++++ .../tests/trusted_runtime_host_loop_tests.rs | 65 +++++++---- docs/topics/WAL.md | 14 ++- 7 files changed, 216 insertions(+), 76 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6bc9a67c..49f79dc5 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -153,6 +153,10 @@ jobs: run: | cargo clippy -p warp-core --features native_rule_bootstrap,trusted_runtime --test edict_projected_cas_tests -- -D warnings -D missing_docs cargo clippy -p warp-core --features native_rule_bootstrap,trusted_runtime,host_test --test edict_projected_cas_tests -- -D warnings -D missing_docs + - name: cargo clippy (trusted host loop and WAL fault injection) + run: | + cargo clippy -p warp-core --features native_rule_bootstrap,trusted_runtime,host_test --test trusted_runtime_host_loop_tests -- -D warnings -D missing_docs + cargo clippy -p warp-core --features host_test --test causal_wal_hardening_tests -- -D warnings -D missing_docs - name: cargo clippy (compiler-produced pure evaluation) run: cargo clippy -p warp-core --features trusted_runtime --lib --test edict_pure_evaluation_tests --test edict_pure_unsigned_subtraction_tests --test edict_pure_byte_length_tests --test edict_byte_equality_tests --test edict_byte_slice_tests --test edict_node_read_tests --test edict_byte_concat_tests --test edict_source_functions_tests -- -D warnings -D missing_docs @@ -328,6 +332,10 @@ jobs: run: | cargo test -p warp-core --features native_rule_bootstrap,trusted_runtime --test edict_projected_cas_tests cargo test -p warp-core --features native_rule_bootstrap,trusted_runtime,host_test --test edict_projected_cas_tests + - name: cargo test (trusted host loop and WAL fault injection) + run: | + cargo test -p warp-core --features native_rule_bootstrap,trusted_runtime,host_test --test trusted_runtime_host_loop_tests + cargo test -p warp-core --features host_test --test causal_wal_hardening_tests - name: cargo test (compiler-produced pure evaluation) run: | cargo test -p warp-core --features trusted_runtime --test edict_pure_evaluation_tests --test edict_pure_unsigned_subtraction_tests --test edict_pure_byte_length_tests --test edict_byte_equality_tests --test edict_byte_slice_tests --test edict_node_read_tests --test edict_byte_concat_tests --test edict_source_functions_tests diff --git a/CHANGELOG.md b/CHANGELOG.md index fee6be0c..aa5468d3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,7 +7,7 @@ ### Added -- `submit_intents_with_runtime_wal_ack` accepts several intents and returns after one storage sync has made all of their acceptance transactions durable. Each intent writes the same acceptance transaction as a single submission, so recovery is unchanged. No acceptance of a batch counts as durable, or can be scheduled, before that sync succeeds. A failed batch acknowledges nothing. If an intent is refused, the intents before it stay accepted and are synced. If the sync fails, the call's intake is rolled back and the WAL refuses further work with `StorageSyncFailed` until it is reopened. +- `submit_intents_with_runtime_wal_ack` accepts several intents and returns after one storage sync has made all of their acceptance transactions durable. Each intent writes the same acceptance transaction as a single submission, so recovery is unchanged. No acceptance of a batch counts as durable, or can be scheduled, before that sync succeeds. A failed batch acknowledges nothing. If the runtime refuses an intent before anything is written for it, the intents before it stay accepted and are synced. If an append or the sync fails, the call's intake is rolled back and the WAL refuses further work with `StorageFailed` until it is reopened. - Native projected compare-and-set invocation encoding retains canonical application input and statically binds the expected attachment-value digest. The native lowerer and independent verifier reconstruct CAS-specific profiles and reject rebound digest paths or under-budget configurations. Generated schema, reproducible components and a retained public Edict compiler build establish package production. The retained compiler package now has scheduler execution, pending/committed Action recovery, stale-digest noncommit recovery and typed refusal witnesses. `EchoOperationInvocationV1::anchored_node_attachment_compare_and_set_with_application_input` and `ExecutableOperationPackageV1::with_compare_and_set_application_result_projection` construct the projected forms without byte editing. diff --git a/crates/warp-core/src/causal_wal.rs b/crates/warp-core/src/causal_wal.rs index 640fb253..8725c657 100644 --- a/crates/warp-core/src/causal_wal.rs +++ b/crates/warp-core/src/causal_wal.rs @@ -5531,10 +5531,20 @@ pub struct FilesystemWalFaultPlan { commit_marker_synced: u32, publish_manifest: u32, deferred_commit_sync: u32, + /// Operations of the failing target to let through first. + skip: u32, } #[cfg(any(test, feature = "host_test"))] impl FilesystemWalFaultPlan { + /// Lets `skipped` operations of the failing target succeed before the + /// failure. + #[must_use] + pub const fn after(mut self, skipped: u32) -> Self { + self.skip = skipped; + self + } + /// Builds a plan that fails the next operation for `target`. #[must_use] pub const fn fail_next(target: FilesystemWalFaultTarget) -> Self { @@ -5545,6 +5555,7 @@ impl FilesystemWalFaultPlan { commit_marker_synced: 0, publish_manifest: 0, deferred_commit_sync: 0, + skip: 0, }, FilesystemWalFaultTarget::FlushCommit => Self { append_frame: 0, @@ -5552,6 +5563,7 @@ impl FilesystemWalFaultPlan { commit_marker_synced: 0, publish_manifest: 0, deferred_commit_sync: 0, + skip: 0, }, FilesystemWalFaultTarget::CommitMarkerSynced => Self { append_frame: 0, @@ -5559,6 +5571,7 @@ impl FilesystemWalFaultPlan { commit_marker_synced: 1, publish_manifest: 0, deferred_commit_sync: 0, + skip: 0, }, FilesystemWalFaultTarget::PublishManifest => Self { append_frame: 0, @@ -5566,6 +5579,7 @@ impl FilesystemWalFaultPlan { commit_marker_synced: 0, publish_manifest: 1, deferred_commit_sync: 0, + skip: 0, }, FilesystemWalFaultTarget::DeferredCommitSync => Self { append_frame: 0, @@ -5573,6 +5587,7 @@ impl FilesystemWalFaultPlan { commit_marker_synced: 0, publish_manifest: 0, deferred_commit_sync: 1, + skip: 0, }, } } @@ -5588,6 +5603,10 @@ impl FilesystemWalFaultPlan { if *remaining == 0 { return false; } + if self.skip > 0 { + self.skip -= 1; + return false; + } *remaining -= 1; true } @@ -5606,6 +5625,10 @@ pub struct FilesystemWalStore { sync_evidence: Vec, /// Transactions whose commit records are appended and not yet synced. deferred_commits: Vec, + /// Set when a sync of deferred commit records fails. A later sync cannot + /// vouch for records an earlier sync failed to write, so the store then + /// refuses to append, sync, rotate or close until it is reopened. + deferred_commit_sync_failed: bool, /// Number of storage syncs that made commit records durable. #[cfg(any(test, feature = "host_test"))] commit_sync_count: u64, @@ -5660,6 +5683,7 @@ impl FilesystemWalStore { manifests: Vec::new(), sync_evidence, deferred_commits: Vec::new(), + deferred_commit_sync_failed: false, #[cfg(any(test, feature = "host_test"))] commit_sync_count: 0, #[cfg(any(test, feature = "host_test"))] @@ -5913,6 +5937,7 @@ impl FilesystemWalStore { transaction: WalCommittedTransaction, sync_now: bool, ) -> Result<(), WalStoreError> { + self.ensure_deferred_commit_sync_has_not_failed()?; transaction.validate()?; let admission_kernel_capability = transaction.admission_kernel_capability; let external_action_coordinator_capability = @@ -6014,24 +6039,50 @@ impl FilesystemWalStore { /// /// Returns a store error if the segment cannot be synced. Nothing appended /// without a sync may be reported as durable after such an error. + /// + /// If the sync fails, the deferred records are never evidenced as synced + /// and the store refuses further writes until it is reopened. pub fn sync_deferred_commits(&mut self) -> Result<(), WalStoreError> { + self.ensure_deferred_commit_sync_has_not_failed()?; if self.deferred_commits.is_empty() { return Ok(()); } #[cfg(any(test, feature = "host_test"))] - if self + let injected = self .fault_plan - .should_fail(FilesystemWalFaultTarget::DeferredCommitSync) - { - return Err(WalStoreError::Io( + .should_fail(FilesystemWalFaultTarget::DeferredCommitSync); + #[cfg(not(any(test, feature = "host_test")))] + let injected = false; + let synced = if injected { + Err(WalStoreError::Io( "injected filesystem WAL deferred_commit_sync failure".to_owned(), - )); + )) + } else { + self.sync_segment() + }; + if let Err(error) = synced { + self.deferred_commit_sync_failed = true; + self.deferred_commits.clear(); + return Err(error); } + self.record_commit_records_synced(); + Ok(()) + } + + fn sync_segment(&self) -> Result<(), WalStoreError> { File::options() .append(true) .open(self.segment_path())? .sync_all()?; - self.record_commit_records_synced(); + Ok(()) + } + + fn ensure_deferred_commit_sync_has_not_failed(&self) -> Result<(), WalStoreError> { + if self.deferred_commit_sync_failed { + return Err(WalStoreError::Io( + "a deferred commit sync failed; reopen the WAL store".to_owned(), + )); + } Ok(()) } diff --git a/crates/warp-core/src/trusted_runtime_host.rs b/crates/warp-core/src/trusted_runtime_host.rs index 0feb06c4..0760429b 100644 --- a/crates/warp-core/src/trusted_runtime_host.rs +++ b/crates/warp-core/src/trusted_runtime_host.rs @@ -173,11 +173,12 @@ pub enum TrustedRuntimeWalError { /// An earlier append must be reconciled before another can reach storage. #[error("trusted runtime WAL writer prefix requires reconciliation")] UncertainWriterPrefix, - /// A storage sync failed. Records written before it are not known to be - /// durable, so this adapter refuses further work until the host reopens - /// the WAL and recovers from what storage actually holds. - #[error("trusted runtime WAL storage sync failed; reopen the WAL to recover")] - StorageSyncFailed, + /// Storage failed while commit syncs were deferred. Records written before + /// the failure are not known to be durable, so this adapter refuses + /// further work until the host reopens the WAL and recovers from what + /// storage actually holds. + #[error("trusted runtime WAL storage failed; reopen the WAL to recover")] + StorageFailed, /// WAL transaction construction failed before storage append. #[error("trusted runtime WAL transaction build error: {0}")] Build(#[from] WalBuildError), @@ -2527,9 +2528,9 @@ pub struct TrustedRuntimeWal { /// stop a second append of the same submission and are never treated as /// durable. unsynced_submission_acceptances: BTreeMap, - /// Set when a storage sync fails; see - /// [`TrustedRuntimeWalError::StorageSyncFailed`]. - storage_sync_failed: bool, + /// Set when storage fails while commit syncs are deferred; see + /// [`TrustedRuntimeWalError::StorageFailed`]. + storage_failed: bool, } /// Whether an appended commit record is synced before the append returns. @@ -2609,7 +2610,7 @@ impl TrustedRuntimeWal { causal_anchor_claim_projection: recovered_cursor.causal_anchor_claim_projection, durable_submission_acceptances, unsynced_submission_acceptances: BTreeMap::new(), - storage_sync_failed: false, + storage_failed: false, evidence_catalog: Some(evidence_catalog), evidence_catalog_posture: EvidenceCatalogPosture::Fresh, #[cfg(any(test, feature = "host_test"))] @@ -2626,6 +2627,9 @@ impl TrustedRuntimeWal { } /// Returns committed WAL markers recorded by the adapter. + /// + /// This is what the log holds, not what is durable: it includes a commit + /// record whose sync is still deferred or has failed. #[must_use] pub fn commits(&self) -> Vec { self.store.read_commits() @@ -2677,7 +2681,7 @@ impl TrustedRuntimeWal { causal_anchor_claim_projection: self.causal_anchor_claim_projection.clone(), durable_submission_acceptances: self.durable_submission_acceptances.clone(), unsynced_submission_acceptances: self.unsynced_submission_acceptances.clone(), - storage_sync_failed: self.storage_sync_failed, + storage_failed: self.storage_failed, }) } @@ -2894,12 +2898,12 @@ impl TrustedRuntimeWal { /// Makes every transaction appended with [`CommitSync::Deferred`] durable /// with one storage sync, and only then counts its acceptances as durable. fn sync_deferred_commits(&mut self) -> Result<(), TrustedRuntimeWalError> { - if self.storage_sync_failed { - return Err(TrustedRuntimeWalError::StorageSyncFailed); + if self.storage_failed { + return Err(TrustedRuntimeWalError::StorageFailed); } if let TrustedRuntimeWalStore::Filesystem(store) = &mut self.store { if let Err(error) = store.sync_deferred_commits() { - self.storage_sync_failed = true; + self.fail_storage(); return Err(error.into()); } } @@ -2907,16 +2911,25 @@ impl TrustedRuntimeWal { Ok(()) } + /// Records that storage failed while commit syncs were deferred. + /// + /// No unsynced acceptance can become durable after this, and the adapter + /// does not reconcile itself from the log: a record that is readable is + /// not thereby durable. + fn fail_storage(&mut self) { + self.storage_failed = true; + self.unsynced_submission_acceptances.clear(); + } + + fn storage_failed(&self) -> bool { + self.storage_failed + } + fn promote_synced_submission_acceptances(&mut self) { let synced = std::mem::take(&mut self.unsynced_submission_acceptances); self.durable_submission_acceptances.extend(synced); } - /// Forgets unsynced acceptances that the host is not going to acknowledge. - fn forget_unsynced_submission_acceptances(&mut self) { - self.unsynced_submission_acceptances.clear(); - } - /// Returns how many storage syncs have made commit records durable, when /// the adapter is backed by the filesystem store. #[cfg(any(test, feature = "host_test"))] @@ -2942,8 +2955,8 @@ impl TrustedRuntimeWal { // After a failed storage sync a readable record is not a durable one, // so the live adapter must not reconcile itself from the log. Only // reopening the WAL recovers. - if self.storage_sync_failed { - return Err(TrustedRuntimeWalError::StorageSyncFailed); + if self.storage_failed { + return Err(TrustedRuntimeWalError::StorageFailed); } self.operation_context_index = None; let report = self.store.recover_for_writer()?; @@ -3424,8 +3437,8 @@ impl TrustedRuntimeWal { transaction: WalCommittedTransaction, sync: CommitSync, ) -> Result { - if self.storage_sync_failed { - return Err(TrustedRuntimeWalError::StorageSyncFailed); + if self.storage_failed { + return Err(TrustedRuntimeWalError::StorageFailed); } let next_causal_history_frontier = logical_causal_history_frontier_digest( self.causal_history_frontier_digest, @@ -6318,15 +6331,15 @@ impl TrustedRuntimeApp<'_> { /// No handle is returned on error, so no submission of the batch is /// acknowledged. /// - /// - If an envelope is refused, or its append fails, the envelopes before - /// it are still synced. They stay accepted and durable, as they would - /// after a process loss between commit and acknowledgement; submitting - /// the same envelope again returns its existing submission. The error of - /// the refused envelope is returned. - /// - If the storage sync fails, the in-memory intake of every submission - /// this call accepted is rolled back, the sync error is returned, and - /// the WAL refuses further work with - /// [`TrustedRuntimeWalError::StorageSyncFailed`] until the host reopens + /// - If the runtime refuses an envelope before anything is written for it, + /// the envelopes before it are still synced. They stay accepted and + /// durable, as they would after a process loss between commit and + /// acknowledgement; submitting the same envelope again returns its + /// existing submission. The error of the refused envelope is returned. + /// - If an append or the storage sync fails, the in-memory intake of every + /// submission this call accepted is rolled back, the storage error is + /// returned, and the WAL refuses further work with + /// [`TrustedRuntimeWalError::StorageFailed`] until the host reopens /// it. Storage may still hold some of the records; reopening recovers /// whatever it holds. pub fn submit_intents_with_runtime_wal_ack( @@ -6358,14 +6371,18 @@ impl TrustedRuntimeApp<'_> { } } } - // Sync whatever was appended: those records are in the log whether or - // not the batch is acknowledged. - let synced = self - .host - .runtime_wal - .as_mut() - .map_or(Ok(()), TrustedRuntimeWal::sync_deferred_commits); - if let Err(error) = synced { + let Some(runtime_wal) = self.host.runtime_wal.as_mut() else { + return Err(TrustedRuntimeHostError::RuntimeWalUnavailable); + }; + let storage_error = if runtime_wal.storage_failed() { + // An append failed. Its error is the one to report. + failure.take() + } else { + // Sync whatever was appended: those records are in the log + // whether or not the batch is acknowledged. + runtime_wal.sync_deferred_commits().err().map(Into::into) + }; + if let Some(error) = storage_error { // Nothing this call appended is known to be durable. Do not leave // it visible as accepted intake. for (marker, handle) in accepted.into_iter().rev() { @@ -6376,10 +6393,7 @@ impl TrustedRuntimeApp<'_> { .runtime .rollback_intent_submission(marker, &handle); } - if let Some(runtime_wal) = self.host.runtime_wal.as_mut() { - runtime_wal.forget_unsynced_submission_acceptances(); - } - return Err(error.into()); + return Err(error); } if let Some(error) = failure { return Err(error); @@ -6426,6 +6440,16 @@ impl TrustedRuntimeApp<'_> { return Ok(handle); } if let Err(error) = runtime_wal.record_submission_acceptance(&envelope, handle, sync) { + if matches!(sync, CommitSync::Deferred) { + // Earlier records of this batch are appended and not synced. + // Reconciling from the log here would count them, and this + // one, as durable because they are readable. Fail closed. + runtime_wal.fail_storage(); + self.host + .runtime + .rollback_intent_submission(submission_rollback, &handle); + return Err(error.into()); + } if runtime_wal.recover_filesystem_submission_acceptance_after_error( handle.submission_id, envelope.ingress_id(), diff --git a/crates/warp-core/tests/causal_wal_hardening_tests.rs b/crates/warp-core/tests/causal_wal_hardening_tests.rs index bc4d5a65..9a8ced03 100644 --- a/crates/warp-core/tests/causal_wal_hardening_tests.rs +++ b/crates/warp-core/tests/causal_wal_hardening_tests.rs @@ -1115,6 +1115,36 @@ fn filesystem_deferred_commits_are_not_evidenced_as_synced_until_one_sync() { must_ok(fs::remove_dir_all(root)); } +#[cfg(feature = "host_test")] +#[test] +fn filesystem_failed_deferred_sync_is_never_evidenced_and_closes_the_store() { + use warp_core::causal_wal::{FilesystemWalFaultPlan, FilesystemWalFaultTarget}; + + let root = temp_wal_root("deferred-commit-sync-failed"); + 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_deferring_commit_sync(submission_transaction( + "failed-first", + Lsn::from_raw(0), + )), + ); + store.replace_fault_plan_for_test(FilesystemWalFaultPlan::fail_next( + FilesystemWalFaultTarget::DeferredCommitSync, + )); + + assert!(store.sync_deferred_commits().is_err()); + // A later sync cannot vouch for what an earlier sync failed to write. + assert!(store.sync_deferred_commits().is_err()); + assert!(store + .append_transaction(submission_transaction("failed-second", Lsn::from_raw(2))) + .is_err()); + assert!(store.close_epoch(epoch_id()).is_err()); + assert_eq!(commit_sync_evidence_count(&store), 0); + drop(store); + must_ok(fs::remove_dir_all(root)); +} + #[test] fn filesystem_synced_commit_and_epoch_close_cover_deferred_commits() { let root = temp_wal_root("deferred-commit-covered"); 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 fa24b960..9d0e28a9 100644 --- a/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs +++ b/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs @@ -1995,38 +1995,63 @@ fn batched_submission_makes_every_acceptance_durable_with_one_sync() { } #[test] -fn batched_submission_failure_acknowledges_nothing_and_can_be_retried() { +fn batched_submission_append_failure_fails_closed_until_the_wal_is_reopened() { let wal_root = temp_runtime_wal_dir("batched-submission-failure"); let (runtime, worldline_id) = runtime(); let mut host = TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize"); host.enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&wal_root)) .expect("host should configure filesystem runtime WAL adapter"); - host.inject_runtime_wal_filesystem_fault_for_test(FilesystemWalFaultPlan::fail_next( - FilesystemWalFaultTarget::AppendFrame, - )) + let runtime_before = format!("{:?}", host.runtime()); + // Let the first acceptance commit, unsynced, and fail the second. + host.inject_runtime_wal_filesystem_fault_for_test( + FilesystemWalFaultPlan::fail_next(FilesystemWalFaultTarget::FlushCommit).after(1), + ) .expect("host should inject a filesystem failure"); - host.app() - .submit_intents_with_runtime_wal_ack(distinct_intents(worldline_id, 4)) - .expect_err("a batch whose first append fails must not be acknowledged"); - assert_eq!(host.runtime().witnessed_submission_count(), 0); - - let handles = host + let error = host .app() .submit_intents_with_runtime_wal_ack(distinct_intents(worldline_id, 4)) - .expect("the same batch must be acceptable once the WAL accepts it"); - assert_eq!(handles.len(), 4); - assert!(handles.iter().all(|handle| !handle.duplicate)); - assert_eq!(host.runtime().witnessed_submission_count(), 4); - - let single = host + .expect_err("a batch whose append fails must not be acknowledged"); + assert!(matches!( + error, + TrustedRuntimeHostError::Wal(TrustedRuntimeWalError::Store(WalStoreError::Io( + message + ))) if message.contains("injected filesystem WAL flush_commit failure") + )); + assert_eq!( + format!("{:?}", host.runtime()), + runtime_before, + "the acceptance appended before the failure must not stay visible" + ); + // The first acceptance is readable in the log and was never synced. The + // host must not treat it as an existing durable submission. + let refused = host .app() .submit_intent_with_runtime_wal_ack(distinct_intents(worldline_id, 1).remove(0)) - .expect("resubmitting a batched envelope returns its submission"); - assert!(single.duplicate); - assert_eq!(single.submission_id, handles[0].submission_id); + .expect_err("the WAL must refuse work after a failed append in a batch"); + assert!(matches!( + refused, + TrustedRuntimeHostError::Wal(TrustedRuntimeWalError::StorageFailed) + )); + assert_eq!(format!("{:?}", host.runtime()), runtime_before); + drop(host); + let (runtime, _) = crate::runtime(); + let mut reopened = + TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize"); + reopened + .enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&wal_root)) + .expect("a fresh host should recover what storage holds"); + let single = reopened + .app() + .submit_intent_with_runtime_wal_ack(distinct_intents(worldline_id, 1).remove(0)) + .expect("a reopened WAL accepts work again"); + assert!( + single.duplicate, + "recovery reports the acceptance that storage holds" + ); + drop(reopened); fs::remove_dir_all(&wal_root).expect("WAL fixture should be removable"); } @@ -2065,7 +2090,7 @@ fn batched_submission_sync_failure_rolls_back_intake_and_fails_closed() { .expect_err("the WAL must refuse work after a failed sync"); assert!(matches!( refused, - TrustedRuntimeHostError::Wal(TrustedRuntimeWalError::StorageSyncFailed) + TrustedRuntimeHostError::Wal(TrustedRuntimeWalError::StorageFailed) )); assert_eq!( format!("{:?}", host.runtime()), diff --git a/docs/topics/WAL.md b/docs/topics/WAL.md index c9187e56..84907e2a 100644 --- a/docs/topics/WAL.md +++ b/docs/topics/WAL.md @@ -33,12 +33,14 @@ succeeds, no acceptance of the batch counts as durable and none can be scheduled. A batch is not atomic across a crash: a prefix can survive, none of it acknowledged, and each survivor recovers as `AcceptedPending`. -If an intent of a batch is refused, the intents before it are still synced and -stay accepted; the call returns the refusal and no handles. If the shared sync -fails, Echo rolls back the in-memory intake of every submission that call -accepted, returns the sync error, and refuses further WAL work with -`StorageSyncFailed` until the host reopens the WAL. A record that is readable -after a failed sync is not treated as durable. +If the runtime refuses an intent of a batch before anything is written for +it, the intents before it are still synced and stay accepted; the call returns +the refusal and no handles. If an append or the shared sync fails, Echo rolls +back the in-memory intake of every submission that call accepted, returns the +storage error, and refuses further WAL work with `StorageFailed` until the +host reopens the WAL. A record that is readable after a storage failure is not +treated as durable by the running host; reopening recovers what storage +holds. Second, failed WAL acceptance is not allowed to leave half-visible runtime state. If no runtime WAL is configured, the ACK path returns an explicit From df8e5bab6f3c9062dd5b7fd70459e0f1fc1755b6 Mon Sep 17 00:00:00 2001 From: James Ross Date: Mon, 12 Oct 2026 21:15:04 -0700 Subject: [PATCH 04/10] Sync an existing log before a writer trusts it Address review: after a failed sync the failure was remembered only in memory. A host that reopened the WAL read the still-readable records and reported them as durable acceptances without their ever having been synced; a later power loss could lose a submission that had been acknowledged as a duplicate of a durable one. The same held for a writer that had simply stopped between writing a record and syncing it. A filesystem writer that takes the writer lock now syncs every segment file, and the directory that names it, before it reads the log as recovered history. If that sync fails the writer does not take the log over. Tests: a record left unsynced is synced on takeover; an injected sync failure refuses the takeover. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 2 + crates/warp-core/src/causal_wal.rs | 81 +++++++++++++++++++ .../tests/causal_wal_hardening_tests.rs | 44 ++++++++++ .../tests/trusted_runtime_host_loop_tests.rs | 2 +- docs/topics/WAL.md | 5 ++ 5 files changed, 133 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index dd69c861..f0ac53a2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,6 +33,8 @@ ### Fixed +- A filesystem WAL writer that takes over an existing log syncs its segment files before reading them as recovered history. A record that an earlier writer wrote and never synced was readable after reopen and was reported as a durable acceptance; a later power loss could still lose it. + - A worldline's initial boundary is shared instead of copied. It never changes after construction, but every scheduler rollback marker, replay checkpoint and replay base cloned the whole graph it holds. Tick cost on a large graph drops by roughly a fifth; the remaining graph-proportional copies are the live state itself. - Scheduler rollback checkpoints no longer copy a worldline's committed tick history or committed-ingress ledger on every SuperTick. A checkpoint records the history length and the attempt journals the ingress ids it adds; rollback truncates and removes them. The trusted host no longer clones the runtime, provenance and retained Action outcomes before every tick, or the runtime before every WAL-acknowledged submission: a completed scheduler pass returns an undo handle, and Action staging and submissions are journaled. Rollback results are unchanged; tick and submission cost no longer grow with committed history. diff --git a/crates/warp-core/src/causal_wal.rs b/crates/warp-core/src/causal_wal.rs index 8725c657..2cb7f9ee 100644 --- a/crates/warp-core/src/causal_wal.rs +++ b/crates/warp-core/src/causal_wal.rs @@ -5520,6 +5520,9 @@ pub enum FilesystemWalFaultTarget { /// Fail the next storage sync that would make deferred commit records /// durable. DeferredCommitSync, + /// Fail the next sync of existing segments by a writer that takes over a + /// log. + RecoveredSegmentSync, } /// Host-test-only filesystem WAL fault plan. @@ -5531,6 +5534,7 @@ pub struct FilesystemWalFaultPlan { commit_marker_synced: u32, publish_manifest: u32, deferred_commit_sync: u32, + recovered_segment_sync: u32, /// Operations of the failing target to let through first. skip: u32, } @@ -5555,6 +5559,7 @@ impl FilesystemWalFaultPlan { commit_marker_synced: 0, publish_manifest: 0, deferred_commit_sync: 0, + recovered_segment_sync: 0, skip: 0, }, FilesystemWalFaultTarget::FlushCommit => Self { @@ -5563,6 +5568,7 @@ impl FilesystemWalFaultPlan { commit_marker_synced: 0, publish_manifest: 0, deferred_commit_sync: 0, + recovered_segment_sync: 0, skip: 0, }, FilesystemWalFaultTarget::CommitMarkerSynced => Self { @@ -5571,6 +5577,7 @@ impl FilesystemWalFaultPlan { commit_marker_synced: 1, publish_manifest: 0, deferred_commit_sync: 0, + recovered_segment_sync: 0, skip: 0, }, FilesystemWalFaultTarget::PublishManifest => Self { @@ -5579,6 +5586,7 @@ impl FilesystemWalFaultPlan { commit_marker_synced: 0, publish_manifest: 1, deferred_commit_sync: 0, + recovered_segment_sync: 0, skip: 0, }, FilesystemWalFaultTarget::DeferredCommitSync => Self { @@ -5587,6 +5595,16 @@ impl FilesystemWalFaultPlan { commit_marker_synced: 0, publish_manifest: 0, deferred_commit_sync: 1, + recovered_segment_sync: 0, + skip: 0, + }, + FilesystemWalFaultTarget::RecoveredSegmentSync => Self { + append_frame: 0, + flush_commit: 0, + commit_marker_synced: 0, + publish_manifest: 0, + deferred_commit_sync: 0, + recovered_segment_sync: 1, skip: 0, }, } @@ -5599,6 +5617,7 @@ impl FilesystemWalFaultPlan { FilesystemWalFaultTarget::CommitMarkerSynced => &mut self.commit_marker_synced, FilesystemWalFaultTarget::PublishManifest => &mut self.publish_manifest, FilesystemWalFaultTarget::DeferredCommitSync => &mut self.deferred_commit_sync, + FilesystemWalFaultTarget::RecoveredSegmentSync => &mut self.recovered_segment_sync, }; if *remaining == 0 { return false; @@ -5632,6 +5651,7 @@ pub struct FilesystemWalStore { /// Number of storage syncs that made commit records durable. #[cfg(any(test, feature = "host_test"))] commit_sync_count: u64, + recovered_segment_sync_count: u64, #[cfg(any(test, feature = "host_test"))] fault_plan: FilesystemWalFaultPlan, } @@ -5686,6 +5706,7 @@ impl FilesystemWalStore { deferred_commit_sync_failed: false, #[cfg(any(test, feature = "host_test"))] commit_sync_count: 0, + recovered_segment_sync_count: 0, #[cfg(any(test, feature = "host_test"))] fault_plan: FilesystemWalFaultPlan::default(), }) @@ -5793,6 +5814,9 @@ impl FilesystemWalStore { } let writer_lock = acquire_writer_epoch_lock(&self.root)?; self.reload_writer_epoch_ledger()?; + // Make what the earlier writer left readable durable before it is + // read as recovered history. + self.sync_existing_segments()?; let recovery = recover_filesystem_store(&self.root, RecoveryAccessMode::ReadOnly).map_err( |error| match error { WalRecoveryError::Store(error) => error, @@ -6069,6 +6093,59 @@ impl FilesystemWalStore { Ok(()) } + /// Syncs every segment file under the store root, and the directories + /// that name them. + fn sync_existing_segments(&mut self) -> Result<(), WalStoreError> { + #[cfg(any(test, feature = "host_test"))] + if self + .fault_plan + .should_fail(FilesystemWalFaultTarget::RecoveredSegmentSync) + { + return Err(WalStoreError::Io( + "injected filesystem WAL recovered_segment_sync failure".to_owned(), + )); + } + let mut synced = 0_u64; + for scan_root in segment_scan_roots(&self.root) { + if !scan_root.exists() { + continue; + } + let mut paths = Vec::new(); + for entry in fs::read_dir(&scan_root)? { + let path = entry?.path(); + if path + .file_stem() + .and_then(|name| name.to_str()) + .is_some_and(|name| name.starts_with("segment-")) + && path + .extension() + .is_some_and(|extension| extension.eq_ignore_ascii_case("ecwal")) + { + paths.push(path); + } + } + if paths.is_empty() { + continue; + } + paths.sort(); + for path in paths { + File::options().append(true).open(path)?.sync_all()?; + synced += 1; + } + sync_directory_store(&scan_root)?; + } + self.recovered_segment_sync_count += synced; + Ok(()) + } + + /// Returns how many existing segment files this store synced when it took + /// over the log. + #[cfg(any(test, feature = "host_test"))] + #[must_use] + pub fn recovered_segment_sync_count_for_test(&self) -> u64 { + self.recovered_segment_sync_count + } + fn sync_segment(&self) -> Result<(), WalStoreError> { File::options() .append(true) @@ -6153,6 +6230,10 @@ impl WalStorePort for FilesystemWalStore { }; if writer_lock.is_some() { self.reload_writer_epoch_ledger()?; + // An earlier writer may have stopped, or failed a sync, after + // writing records. They are readable. Make them durable before + // this writer treats the readable log as recovered history. + self.sync_existing_segments()?; } let epoch = validate_writer_epoch_request( self.active_epoch.as_ref(), diff --git a/crates/warp-core/tests/causal_wal_hardening_tests.rs b/crates/warp-core/tests/causal_wal_hardening_tests.rs index 9a8ced03..a37b8e2b 100644 --- a/crates/warp-core/tests/causal_wal_hardening_tests.rs +++ b/crates/warp-core/tests/causal_wal_hardening_tests.rs @@ -1145,6 +1145,50 @@ fn filesystem_failed_deferred_sync_is_never_evidenced_and_closes_the_store() { must_ok(fs::remove_dir_all(root)); } +#[cfg(feature = "host_test")] +#[test] +fn filesystem_writer_syncs_an_existing_log_before_trusting_it() { + use warp_core::causal_wal::{FilesystemWalFaultPlan, FilesystemWalFaultTarget}; + let root = temp_wal_root("recovered-segment-sync"); + let mut store = must_ok(FilesystemWalStore::open(&root, WalSegmentId::from_raw(1))); + must_ok(store.acquire_writer_epoch(writer_epoch_request())); + // The record is written and its sync never happens, as when a writer + // stops between the two. + must_ok( + store.append_transaction_deferring_commit_sync(submission_transaction( + "left-unsynced", + Lsn::from_raw(0), + )), + ); + drop(store); + + let mut refused = must_ok(FilesystemWalStore::open_with_fault_plan_for_test( + &root, + WalSegmentId::from_raw(1), + FilesystemWalFaultPlan::fail_next(FilesystemWalFaultTarget::RecoveredSegmentSync), + )); + let error = must_err( + refused.acquire_fresh_writer_epoch(Lsn::from_raw(2)), + "a writer that cannot sync the existing log must not take it over", + ); + assert!(matches!( + error, + WalStoreError::Io(message) if message.contains("recovered_segment_sync") + )); + assert_eq!(refused.recovered_segment_sync_count_for_test(), 0); + drop(refused); + + let mut successor = must_ok(FilesystemWalStore::open(&root, WalSegmentId::from_raw(1))); + must_ok(successor.acquire_fresh_writer_epoch(Lsn::from_raw(2))); + assert_eq!( + successor.recovered_segment_sync_count_for_test(), + 1, + "the unsynced record is made durable before the log is read as history" + ); + drop(successor); + must_ok(fs::remove_dir_all(root)); +} + #[test] fn filesystem_synced_commit_and_epoch_close_cover_deferred_commits() { let root = temp_wal_root("deferred-commit-covered"); 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 9d0e28a9..9fcc1261 100644 --- a/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs +++ b/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs @@ -2049,7 +2049,7 @@ fn batched_submission_append_failure_fails_closed_until_the_wal_is_reopened() { .expect("a reopened WAL accepts work again"); assert!( single.duplicate, - "recovery reports the acceptance that storage holds" + "reopening syncs the log before trusting it, so the acceptance it holds is durable" ); drop(reopened); fs::remove_dir_all(&wal_root).expect("WAL fixture should be removable"); diff --git a/docs/topics/WAL.md b/docs/topics/WAL.md index 84907e2a..f11c2281 100644 --- a/docs/topics/WAL.md +++ b/docs/topics/WAL.md @@ -42,6 +42,11 @@ host reopens the WAL. A record that is readable after a storage failure is not treated as durable by the running host; reopening recovers what storage holds. +A writer that takes over an existing log syncs its segment files before it +reads them as recovered history. An earlier writer may have stopped, or failed +a sync, after writing records; those records are readable and not yet durable. +If that sync fails, the writer does not take the log over. + Second, failed WAL acceptance is not allowed to leave half-visible runtime state. If no runtime WAL is configured, the ACK path returns an explicit unavailable error before mutating intake state. If WAL commit fails, Echo rolls From 0f266786cce61f87b9d23f83af2cac7e9a9fa6f8 Mon Sep 17 00:00:00 2001 From: James Ross Date: Mon, 12 Oct 2026 21:23:41 -0700 Subject: [PATCH 05/10] Refuse every store write after a failed sync; latch only on storage errors Address the third independent review of batched submission. After a failed deferred sync the store refused whole-transaction appends, but the lower-level port that coordinators write through, append_frame and flush_commit, was not guarded. A commit flushed that way would have been evidenced as synced on a store that had just failed a sync. Both are refused now. A deferred batch latched the adapter on any append error, including one raised before anything reached storage. Storage had not failed in that case, and the adapter stayed unusable until reopened. Only a storage error latches now. An append refused before storage rebuilds the writer cursor from the log, and that rebuild no longer counts a readable acceptance as durable while its sync is still owed. The intents before the refusal are synced and stay accepted. Co-Authored-By: Claude Opus 5.5 --- crates/warp-core/src/causal_wal.rs | 2 + crates/warp-core/src/trusted_runtime_host.rs | 71 +++++++++++++++++-- .../tests/causal_wal_hardening_tests.rs | 8 +++ .../tests/trusted_runtime_host_loop_tests.rs | 62 ++++++++++++++++ docs/topics/WAL.md | 17 ++--- 5 files changed, 146 insertions(+), 14 deletions(-) diff --git a/crates/warp-core/src/causal_wal.rs b/crates/warp-core/src/causal_wal.rs index 2cb7f9ee..d3d9cb65 100644 --- a/crates/warp-core/src/causal_wal.rs +++ b/crates/warp-core/src/causal_wal.rs @@ -5987,6 +5987,7 @@ impl FilesystemWalStore { external_action_coordinator_capability: Option, sync_now: bool, ) -> Result<(), WalStoreError> { + self.ensure_deferred_commit_sync_has_not_failed()?; self.ensure_writer_lock()?; let active_epoch = self .active_epoch @@ -6257,6 +6258,7 @@ impl WalStorePort for FilesystemWalStore { epoch_id: WriterEpochId, frame: WalFrame, ) -> Result<(), WalStoreError> { + self.ensure_deferred_commit_sync_has_not_failed()?; self.ensure_writer_lock()?; let active_epoch = self .active_epoch diff --git a/crates/warp-core/src/trusted_runtime_host.rs b/crates/warp-core/src/trusted_runtime_host.rs index 0760429b..5f9b0531 100644 --- a/crates/warp-core/src/trusted_runtime_host.rs +++ b/crates/warp-core/src/trusted_runtime_host.rs @@ -1424,6 +1424,20 @@ impl TrustedRuntimeHost { self.runtime_wal = Some(runtime_wal); } + /// Refuses one runtime WAL append before it reaches storage, after letting + /// `appends_let_through` appends succeed. + #[cfg(any(test, feature = "host_test"))] + pub fn refuse_runtime_wal_append_before_storage_for_test( + &mut self, + appends_let_through: u32, + ) -> Result<(), TrustedRuntimeHostError> { + self.runtime_wal + .as_mut() + .ok_or(TrustedRuntimeHostError::RuntimeWalUnavailable)? + .refuse_append_before_storage_for_test(appends_let_through); + Ok(()) + } + /// Replaces the filesystem WAL fault plan for targeted host tests. #[cfg(any(test, feature = "host_test"))] pub fn inject_runtime_wal_filesystem_fault_for_test( @@ -2504,6 +2518,10 @@ pub struct TrustedRuntimeWal { evidence_catalog_posture: EvidenceCatalogPosture, #[cfg(any(test, feature = "host_test"))] fail_next_evidence_catalog_update: bool, + /// Refuses one append before it reaches storage, after letting this many + /// appends through. + #[cfg(any(test, feature = "host_test"))] + refuse_append_before_storage_after: Option, #[cfg(any(test, feature = "host_test"))] recover_read_only_call_count: std::cell::Cell, writer_epoch: WriterEpochId, @@ -2616,6 +2634,8 @@ impl TrustedRuntimeWal { #[cfg(any(test, feature = "host_test"))] fail_next_evidence_catalog_update: false, #[cfg(any(test, feature = "host_test"))] + refuse_append_before_storage_after: None, + #[cfg(any(test, feature = "host_test"))] recover_read_only_call_count: std::cell::Cell::new(0), }) } @@ -2657,6 +2677,8 @@ impl TrustedRuntimeWal { #[cfg(any(test, feature = "host_test"))] fail_next_evidence_catalog_update: self.fail_next_evidence_catalog_update, #[cfg(any(test, feature = "host_test"))] + refuse_append_before_storage_after: self.refuse_append_before_storage_after, + #[cfg(any(test, feature = "host_test"))] recover_read_only_call_count: std::cell::Cell::new( self.recover_read_only_call_count.get(), ), @@ -2925,6 +2947,15 @@ impl TrustedRuntimeWal { self.storage_failed } + /// Rebuilds the writer cursor after an append that was refused before it + /// reached storage. Unsynced acceptances stay unsynced. + fn restore_cursor_after_refused_append(&mut self) -> Result<(), TrustedRuntimeWalError> { + if !self.uses_filesystem_store() { + return Ok(()); + } + self.refresh_cursor_from_store_for_writer() + } + fn promote_synced_submission_acceptances(&mut self) { let synced = std::mem::take(&mut self.unsynced_submission_acceptances); self.durable_submission_acceptances.extend(synced); @@ -2995,6 +3026,11 @@ impl TrustedRuntimeWal { self.causal_history_frontier_digest = cursor.causal_history_frontier_digest; self.causal_anchor_claim_projection = cursor.causal_anchor_claim_projection; self.durable_submission_acceptances = durable_submission_acceptances; + // The log is read as written, not as synced. An acceptance whose sync + // is still owed stays unsynced however readable its record is. + for submission_id in self.unsynced_submission_acceptances.keys() { + self.durable_submission_acceptances.remove(submission_id); + } self.operation_context_index = Some((self.previous_committed_transaction_digest, contexts)); Ok(()) } @@ -3467,6 +3503,15 @@ impl TrustedRuntimeWal { if prefix != self.previous_committed_transaction_digest { return Err(TrustedRuntimeWalError::UncertainWriterPrefix); } + #[cfg(any(test, feature = "host_test"))] + match self.refuse_append_before_storage_after { + Some(0) => { + self.refuse_append_before_storage_after = None; + return Err(TrustedRuntimeWalError::UncertainWriterPrefix); + } + Some(remaining) => self.refuse_append_before_storage_after = Some(remaining - 1), + None => {} + } contexts.apply_frames(&frames)?; match sync { CommitSync::Now => { @@ -6127,6 +6172,12 @@ impl TrustedRuntimeWal { pub fn fail_next_evidence_catalog_update_for_test(&mut self) { self.fail_next_evidence_catalog_update = true; } + + /// Refuses one append before it reaches storage, after letting + /// `appends_let_through` appends succeed. + pub fn refuse_append_before_storage_for_test(&mut self, appends_let_through: u32) { + self.refuse_append_before_storage_after = Some(appends_let_through); + } } /// App-facing handle for a trusted local runtime host. @@ -6336,9 +6387,11 @@ impl TrustedRuntimeApp<'_> { /// durable, as they would after a process loss between commit and /// acknowledgement; submitting the same envelope again returns its /// existing submission. The error of the refused envelope is returned. - /// - If an append or the storage sync fails, the in-memory intake of every - /// submission this call accepted is rolled back, the storage error is - /// returned, and the WAL refuses further work with + /// The same holds if the WAL refuses an append before it reaches + /// storage. + /// - If storage fails during an append or the sync, the in-memory intake + /// of every submission this call accepted is rolled back, the storage + /// error is returned, and the WAL refuses further work with /// [`TrustedRuntimeWalError::StorageFailed`] until the host reopens /// it. Storage may still hold some of the records; reopening recovers /// whatever it holds. @@ -6442,9 +6495,15 @@ impl TrustedRuntimeApp<'_> { if let Err(error) = runtime_wal.record_submission_acceptance(&envelope, handle, sync) { if matches!(sync, CommitSync::Deferred) { // Earlier records of this batch are appended and not synced. - // Reconciling from the log here would count them, and this - // one, as durable because they are readable. Fail closed. - runtime_wal.fail_storage(); + // A storage error may have left part of this one in the log + // as well, and nothing readable there is durable: fail + // closed. An append refused before it reached storage wrote + // nothing, so only the writer cursor needs rebuilding. + if matches!(error, TrustedRuntimeWalError::Store(_)) + || runtime_wal.restore_cursor_after_refused_append().is_err() + { + runtime_wal.fail_storage(); + } self.host .runtime .rollback_intent_submission(submission_rollback, &handle); diff --git a/crates/warp-core/tests/causal_wal_hardening_tests.rs b/crates/warp-core/tests/causal_wal_hardening_tests.rs index a37b8e2b..d7d29621 100644 --- a/crates/warp-core/tests/causal_wal_hardening_tests.rs +++ b/crates/warp-core/tests/causal_wal_hardening_tests.rs @@ -1140,6 +1140,14 @@ fn filesystem_failed_deferred_sync_is_never_evidenced_and_closes_the_store() { .append_transaction(submission_transaction("failed-second", Lsn::from_raw(2))) .is_err()); assert!(store.close_epoch(epoch_id()).is_err()); + // The lower-level port a coordinator writes through is refused as well. + let later = submission_transaction("failed-third", Lsn::from_raw(2)); + let later_commit = later.commit.clone(); + assert!(later + .frames + .into_iter() + .all(|frame| store.append_frame(epoch_id(), frame).is_err())); + assert!(store.flush_commit(epoch_id(), later_commit).is_err()); assert_eq!(commit_sync_evidence_count(&store), 0); drop(store); must_ok(fs::remove_dir_all(root)); 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 9fcc1261..176d62d8 100644 --- a/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs +++ b/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs @@ -2055,6 +2055,68 @@ fn batched_submission_append_failure_fails_closed_until_the_wal_is_reopened() { fs::remove_dir_all(&wal_root).expect("WAL fixture should be removable"); } +#[test] +fn batched_submission_refused_before_storage_keeps_earlier_intents_and_the_wal() { + let wal_root = temp_runtime_wal_dir("batched-submission-refused"); + let (runtime, worldline_id) = runtime(); + let mut host = + TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize"); + host.enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&wal_root)) + .expect("host should configure filesystem runtime WAL adapter"); + let syncs_before = host + .runtime_wal() + .expect("runtime WAL should stay configured") + .filesystem_commit_sync_count_for_test() + .expect("filesystem store"); + // The first acceptance commits, unsynced. The second is refused before + // anything is written for it. + host.refuse_runtime_wal_append_before_storage_for_test(1) + .expect("host should arm the refusal"); + + let error = host + .app() + .submit_intents_with_runtime_wal_ack(distinct_intents(worldline_id, 3)) + .expect_err("a batch with a refused intent returns no handles"); + assert!(matches!( + error, + TrustedRuntimeHostError::Wal(TrustedRuntimeWalError::UncertainWriterPrefix) + )); + assert_eq!( + host.runtime().witnessed_submission_count(), + 1, + "the intent before the refusal stays accepted" + ); + let runtime_wal = host + .runtime_wal() + .expect("runtime WAL should stay configured"); + assert_eq!( + runtime_wal + .filesystem_commit_sync_count_for_test() + .expect("filesystem store") + - syncs_before, + 1, + "the intent before the refusal is synced" + ); + assert_eq!(runtime_wal.submission_acceptance_count(), 1); + + // Nothing failed in storage, so the WAL keeps working. + let mut intents = distinct_intents(worldline_id, 3); + let first = host + .app() + .submit_intent_with_runtime_wal_ack(intents.remove(0)) + .expect("the accepted intent is returned as its existing submission"); + assert!(first.duplicate); + let rest = host + .app() + .submit_intents_with_runtime_wal_ack(intents) + .expect("the refused intent and its successor are accepted on retry"); + assert_eq!(rest.len(), 2); + assert!(rest.iter().all(|handle| !handle.duplicate)); + assert_eq!(host.runtime().witnessed_submission_count(), 3); + drop(host); + fs::remove_dir_all(&wal_root).expect("WAL fixture should be removable"); +} + #[test] fn batched_submission_sync_failure_rolls_back_intake_and_fails_closed() { let wal_root = temp_runtime_wal_dir("batched-submission-sync-failure"); diff --git a/docs/topics/WAL.md b/docs/topics/WAL.md index f11c2281..e37d7e53 100644 --- a/docs/topics/WAL.md +++ b/docs/topics/WAL.md @@ -33,14 +33,15 @@ succeeds, no acceptance of the batch counts as durable and none can be scheduled. A batch is not atomic across a crash: a prefix can survive, none of it acknowledged, and each survivor recovers as `AcceptedPending`. -If the runtime refuses an intent of a batch before anything is written for -it, the intents before it are still synced and stay accepted; the call returns -the refusal and no handles. If an append or the shared sync fails, Echo rolls -back the in-memory intake of every submission that call accepted, returns the -storage error, and refuses further WAL work with `StorageFailed` until the -host reopens the WAL. A record that is readable after a storage failure is not -treated as durable by the running host; reopening recovers what storage -holds. +If the runtime or the WAL refuses an intent of a batch before anything is +written for it, the intents before it are still synced and stay accepted; the +call returns the refusal and no handles. + +If storage fails during an append or the shared sync, Echo rolls back the +in-memory intake of every submission that call accepted, returns the storage +error, and refuses further WAL work with `StorageFailed` until the host +reopens the WAL. A record that is readable after a storage failure is not +treated as durable by the running host. A writer that takes over an existing log syncs its segment files before it reads them as recovered history. An earlier writer may have stopped, or failed From 1ea3c6a33f5d2b26dc2bfdc0c4eed8a34f70b8de Mon Sep 17 00:00:00 2001 From: James Ross Date: Mon, 12 Oct 2026 21:35:48 -0700 Subject: [PATCH 06/10] Refuse live recovery reads after a storage failure; sync before a manifest Address review. After a storage failure the running adapter still answered recover_read_only and recover_evidence_catalog_read_only from the log. It could report the batch it had just rolled back as accepted and pending, and return a recovery certificate over records it does not know to be durable. Both now refuse with StorageFailed. publish_manifest synced and renamed a manifest without first syncing commit records whose sync was deferred. A power loss could keep a manifest that names a commit the segment lost. Publication now syncs deferred commits first. Co-Authored-By: Claude Opus 5.5 --- crates/warp-core/src/causal_wal.rs | 3 ++ crates/warp-core/src/trusted_runtime_host.rs | 8 ++++ .../tests/causal_wal_hardening_tests.rs | 40 +++++++++++++++++++ .../tests/trusted_runtime_host_loop_tests.rs | 14 +++++++ 4 files changed, 65 insertions(+) diff --git a/crates/warp-core/src/causal_wal.rs b/crates/warp-core/src/causal_wal.rs index d3d9cb65..700bf848 100644 --- a/crates/warp-core/src/causal_wal.rs +++ b/crates/warp-core/src/causal_wal.rs @@ -6361,6 +6361,9 @@ impl WalStorePort for FilesystemWalStore { if active_epoch.epoch_id != epoch_id { return Err(WalStoreError::WriterEpochMismatch); } + // A manifest names committed records. It must not become durable + // before they are. + self.sync_deferred_commits()?; #[cfg(any(test, feature = "host_test"))] if self .fault_plan diff --git a/crates/warp-core/src/trusted_runtime_host.rs b/crates/warp-core/src/trusted_runtime_host.rs index d5056b3b..1dee6912 100644 --- a/crates/warp-core/src/trusted_runtime_host.rs +++ b/crates/warp-core/src/trusted_runtime_host.rs @@ -2737,6 +2737,11 @@ impl TrustedRuntimeWal { /// Recovers submission and receipt indexes from committed WAL transactions /// without scheduler callbacks. pub fn recover_read_only(&self) -> Result { + // After a storage failure the log may hold records that are readable + // and not durable. This adapter does not report them as recovered. + if self.storage_failed { + return Err(TrustedRuntimeWalError::StorageFailed); + } #[cfg(any(test, feature = "host_test"))] self.recover_read_only_call_count .set(self.recover_read_only_call_count.get() + 1); @@ -2821,6 +2826,9 @@ impl TrustedRuntimeWal { pub fn recover_evidence_catalog_read_only( &self, ) -> Result { + if self.storage_failed { + return Err(TrustedRuntimeWalError::StorageFailed); + } let report = self.store.recover_read_only()?; crate::evidence::CausalSegmentCatalog::from_recovery_scan(&report) .map_err(TrustedRuntimeWalError::EvidenceCatalog) diff --git a/crates/warp-core/tests/causal_wal_hardening_tests.rs b/crates/warp-core/tests/causal_wal_hardening_tests.rs index d7d29621..adbdad96 100644 --- a/crates/warp-core/tests/causal_wal_hardening_tests.rs +++ b/crates/warp-core/tests/causal_wal_hardening_tests.rs @@ -1197,6 +1197,46 @@ fn filesystem_writer_syncs_an_existing_log_before_trusting_it() { must_ok(fs::remove_dir_all(root)); } +#[test] +fn filesystem_manifest_publication_syncs_deferred_commits_first() { + let mut fixture = WalHardeningFixture::new("deferred-commit-manifest"); + let transaction = submission_transaction("before-manifest", Lsn::from_raw(0)); + let commit = transaction.commit.clone(); + must_ok( + fixture + .store + .append_transaction_deferring_commit_sync(transaction), + ); + assert_eq!(commit_sync_evidence_count(&fixture.store), 0); + + must_ok(fixture.store.publish_manifest( + epoch_id(), + WalManifest { + manifest_digest: digest("hardening:manifest:deferred"), + last_committed_lsn: Some(commit.last_lsn), + last_commit_digest: Some(commit.commit_digest), + sealed_segment_count: 0, + }, + )); + + let boundaries = fixture + .store + .sync_evidence() + .iter() + .map(|entry| entry.boundary) + .collect::>(); + let commit_synced = boundaries + .iter() + .position(|boundary| *boundary == FilesystemSyncBoundary::CommitFileSynced); + let manifest_synced = boundaries + .iter() + .position(|boundary| *boundary == FilesystemSyncBoundary::ManifestTempFileSynced); + assert!( + matches!((commit_synced, manifest_synced), (Some(commit), Some(manifest)) if commit < manifest), + "a manifest must not become durable before the commit it names: {boundaries:?}" + ); +} + #[test] fn filesystem_synced_commit_and_epoch_close_cover_deferred_commits() { let root = temp_wal_root("deferred-commit-covered"); 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 176d62d8..9168578e 100644 --- a/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs +++ b/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs @@ -2163,6 +2163,20 @@ fn batched_submission_sync_failure_rolls_back_intake_and_fails_closed() { matches!(host.tick_once(), Ok(steps) if steps.is_empty()), "nothing from the failed batch may be scheduled" ); + let failed_wal = host + .runtime_wal() + .expect("runtime WAL should stay configured"); + assert!( + matches!( + failed_wal.recover_read_only(), + Err(TrustedRuntimeWalError::StorageFailed) + ), + "the running host must not report readable records as recovered" + ); + assert!(matches!( + failed_wal.recover_evidence_catalog_read_only(), + Err(TrustedRuntimeWalError::StorageFailed) + )); // The records were written; only their sync failed. Reopening recovers // what storage holds, as it would after a process loss. From 6c3f6fb35473d6f6954192860acd4ec6f2c08d63 Mon Sep 17 00:00:00 2001 From: James Ross Date: Mon, 12 Oct 2026 21:38:53 -0700 Subject: [PATCH 07/10] Sync a log before a host reads it; rebuild the cursor for every store Address the fourth independent review of batched submission. A host opened an existing log, ran recovery and read its acceptances before the writer lock was taken and the segments were synced. The adapter now syncs existing segments when it opens the store, before any recovery read, and does not open a log it cannot sync. An append refused before storage rebuilt the writer cursor only for the filesystem store. The in-memory adapter kept an empty context index and refused every later append while reporting itself healthy. The cursor is now rebuilt for both stores. truncate_tail_after is refused after a failed deferred sync, like the other store writes. Co-Authored-By: Claude Opus 5.5 --- crates/warp-core/src/causal_wal.rs | 13 +++- crates/warp-core/src/trusted_runtime_host.rs | 7 +- .../tests/causal_wal_hardening_tests.rs | 12 ++++ .../tests/trusted_runtime_host_loop_tests.rs | 65 +++++++++++++++++++ docs/topics/WAL.md | 9 +-- 5 files changed, 98 insertions(+), 8 deletions(-) diff --git a/crates/warp-core/src/causal_wal.rs b/crates/warp-core/src/causal_wal.rs index 700bf848..27783eb5 100644 --- a/crates/warp-core/src/causal_wal.rs +++ b/crates/warp-core/src/causal_wal.rs @@ -6096,7 +6096,17 @@ impl FilesystemWalStore { /// Syncs every segment file under the store root, and the directories /// that name them. - fn sync_existing_segments(&mut self) -> Result<(), WalStoreError> { + /// + /// An earlier writer may have stopped, or failed a sync, after writing + /// records. They are readable and not yet durable. Call this before + /// reading an existing log as recovered history; taking the writer lock + /// does it as well. + /// + /// # Errors + /// + /// Returns the storage error if a segment or its directory cannot be + /// synced. The log must then not be treated as recovered. + pub fn sync_existing_segments(&mut self) -> Result<(), WalStoreError> { #[cfg(any(test, feature = "host_test"))] if self .fault_plan @@ -6345,6 +6355,7 @@ impl WalStorePort for FilesystemWalStore { } fn truncate_tail_after(&mut self, after_lsn: Lsn) -> Result<(), WalStoreError> { + self.ensure_deferred_commit_sync_has_not_failed()?; rewrite_filesystem_segments_after_truncation(&self.root, after_lsn) } diff --git a/crates/warp-core/src/trusted_runtime_host.rs b/crates/warp-core/src/trusted_runtime_host.rs index 1dee6912..2a4d7cee 100644 --- a/crates/warp-core/src/trusted_runtime_host.rs +++ b/crates/warp-core/src/trusted_runtime_host.rs @@ -2958,9 +2958,6 @@ impl TrustedRuntimeWal { /// Rebuilds the writer cursor after an append that was refused before it /// reached storage. Unsynced acceptances stay unsynced. fn restore_cursor_after_refused_append(&mut self) -> Result<(), TrustedRuntimeWalError> { - if !self.uses_filesystem_store() { - return Ok(()); - } self.refresh_cursor_from_store_for_writer() } @@ -3635,6 +3632,10 @@ impl TrustedRuntimeWalStore { }; #[cfg(not(any(test, feature = "host_test")))] let store = FilesystemWalStore::open(root, segment_id)?; + let mut store = store; + // Make what an earlier writer left readable durable before + // any of it is read as recovered history. + store.sync_existing_segments()?; Ok(Self::Filesystem(store)) } } diff --git a/crates/warp-core/tests/causal_wal_hardening_tests.rs b/crates/warp-core/tests/causal_wal_hardening_tests.rs index adbdad96..92c72fea 100644 --- a/crates/warp-core/tests/causal_wal_hardening_tests.rs +++ b/crates/warp-core/tests/causal_wal_hardening_tests.rs @@ -1148,6 +1148,18 @@ fn filesystem_failed_deferred_sync_is_never_evidenced_and_closes_the_store() { .into_iter() .all(|frame| store.append_frame(epoch_id(), frame).is_err())); assert!(store.flush_commit(epoch_id(), later_commit).is_err()); + assert!(store.truncate_tail_after(Lsn::from_raw(0)).is_err()); + assert!(store + .publish_manifest( + epoch_id(), + WalManifest { + manifest_digest: digest("hardening:manifest:after-failed-sync"), + last_committed_lsn: None, + last_commit_digest: None, + sealed_segment_count: 0, + }, + ) + .is_err()); assert_eq!(commit_sync_evidence_count(&store), 0); drop(store); must_ok(fs::remove_dir_all(root)); 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 9168578e..3bef7865 100644 --- a/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs +++ b/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs @@ -2117,6 +2117,71 @@ fn batched_submission_refused_before_storage_keeps_earlier_intents_and_the_wal() fs::remove_dir_all(&wal_root).expect("WAL fixture should be removable"); } +#[test] +fn batched_submission_refused_before_storage_keeps_an_in_memory_wal_usable() { + let (runtime, worldline_id) = runtime(); + let mut host = + TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize"); + host.enable_runtime_wal(TrustedRuntimeWalConfig::in_memory()) + .expect("host should configure in-memory runtime WAL adapter"); + host.refuse_runtime_wal_append_before_storage_for_test(1) + .expect("host should arm the refusal"); + + host.app() + .submit_intents_with_runtime_wal_ack(distinct_intents(worldline_id, 3)) + .expect_err("a batch with a refused intent returns no handles"); + assert_eq!(host.runtime().witnessed_submission_count(), 1); + + let mut intents = distinct_intents(worldline_id, 3); + let first = host + .app() + .submit_intent_with_runtime_wal_ack(intents.remove(0)) + .expect("the accepted intent is returned as its existing submission"); + assert!(first.duplicate); + let rest = host + .app() + .submit_intents_with_runtime_wal_ack(intents) + .expect("the writer cursor is rebuilt, so later appends are accepted"); + assert_eq!(rest.len(), 2); + assert!(rest.iter().all(|handle| !handle.duplicate)); +} + +#[test] +fn a_host_does_not_open_a_log_it_cannot_sync() { + let wal_root = temp_runtime_wal_dir("open-sync-failure"); + let (runtime, worldline_id) = runtime(); + let mut host = + TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize"); + host.enable_runtime_wal(TrustedRuntimeWalConfig::filesystem(&wal_root)) + .expect("host should configure filesystem runtime WAL adapter"); + host.app() + .submit_intent_with_runtime_wal_ack(distinct_intents(worldline_id, 1).remove(0)) + .expect("the first host accepts a submission"); + drop(host); + + let (runtime, _) = crate::runtime(); + let mut reopened = + TrustedRuntimeHost::new(runtime, empty_engine()).expect("trusted host should initialize"); + let error = reopened + .enable_runtime_wal( + TrustedRuntimeWalConfig::filesystem_with_fault_plan_for_test( + &wal_root, + FilesystemWalFaultPlan::fail_next(FilesystemWalFaultTarget::RecoveredSegmentSync), + ), + ) + .expect_err("a log that cannot be synced must not be read as recovered history"); + assert!(matches!( + error, + TrustedRuntimeHostError::Wal(TrustedRuntimeWalError::Store(WalStoreError::Io( + message + ))) if message.contains("recovered_segment_sync") + )); + assert!(reopened.runtime_wal().is_none()); + assert_eq!(reopened.runtime().witnessed_submission_count(), 0); + drop(reopened); + fs::remove_dir_all(&wal_root).expect("WAL fixture should be removable"); +} + #[test] fn batched_submission_sync_failure_rolls_back_intake_and_fails_closed() { let wal_root = temp_runtime_wal_dir("batched-submission-sync-failure"); diff --git a/docs/topics/WAL.md b/docs/topics/WAL.md index e37d7e53..3a5d4fcf 100644 --- a/docs/topics/WAL.md +++ b/docs/topics/WAL.md @@ -43,10 +43,11 @@ error, and refuses further WAL work with `StorageFailed` until the host reopens the WAL. A record that is readable after a storage failure is not treated as durable by the running host. -A writer that takes over an existing log syncs its segment files before it -reads them as recovered history. An earlier writer may have stopped, or failed -a sync, after writing records; those records are readable and not yet durable. -If that sync fails, the writer does not take the log over. +A host that opens an existing log, and a writer that takes it over, sync its +segment files before reading them as recovered history. An earlier writer may +have stopped, or failed a sync, after writing records; those records are +readable and not yet durable. If that sync fails, the host does not open the +log and the writer does not take it over. Second, failed WAL acceptance is not allowed to leave half-visible runtime state. If no runtime WAL is configured, the ACK path returns an explicit From 1b37d17f38b8cbd132ebbdf05cca37f048238227 Mon Sep 17 00:00:00 2001 From: James Ross Date: Mon, 12 Oct 2026 21:59:39 -0700 Subject: [PATCH 08/10] Close three remaining gaps around deferred commit records Address review. A synced commit whose sync failed left earlier deferred records pending, and a later successful sync would have evidenced them, although the failed sync had covered them too. The store now latches in that case, as it does after a failed deferred sync. Truncation is refused while a commit sync is owed. It could remove a deferred record, and a later sync would then evidence a record the log no longer holds. After a storage failure the host no longer returns its cached causal frontier as a basis. That frontier can name records that are not durable. Co-Authored-By: Claude Opus 5.5 --- crates/warp-core/src/causal_wal.rs | 43 ++++++++++++---- crates/warp-core/src/trusted_runtime_host.rs | 5 ++ .../tests/causal_wal_hardening_tests.rs | 50 +++++++++++++++++++ .../tests/trusted_runtime_host_loop_tests.rs | 9 ++++ 4 files changed, 96 insertions(+), 11 deletions(-) diff --git a/crates/warp-core/src/causal_wal.rs b/crates/warp-core/src/causal_wal.rs index 27783eb5..998146d4 100644 --- a/crates/warp-core/src/causal_wal.rs +++ b/crates/warp-core/src/causal_wal.rs @@ -6002,20 +6002,33 @@ impl FilesystemWalStore { external_action_coordinator_capability, )?; #[cfg(any(test, feature = "host_test"))] - if self + let injected = self .fault_plan - .should_fail(FilesystemWalFaultTarget::FlushCommit) - { - return Err(WalStoreError::Io( + .should_fail(FilesystemWalFaultTarget::FlushCommit); + #[cfg(not(any(test, feature = "host_test")))] + let injected = false; + let transaction_id = commit.transaction_id; + let flushed = if injected { + Err(WalStoreError::Io( "injected filesystem WAL flush_commit failure".to_owned(), - )); + )) + } else { + append_segment_record( + &self.segment_path(), + DiskWalRecord::Commit(&commit), + sync_now, + ) + }; + if let Err(error) = flushed { + if sync_now && !self.deferred_commits.is_empty() { + // This commit's sync would also have covered the deferred + // records. It failed, so they are in the same position as + // after a failed deferred sync. + self.deferred_commit_sync_failed = true; + self.deferred_commits.clear(); + } + return Err(error); } - let transaction_id = commit.transaction_id; - append_segment_record( - &self.segment_path(), - DiskWalRecord::Commit(&commit), - sync_now, - )?; self.epoch_closures.insert( epoch_id, WriterEpochClosure { @@ -6356,6 +6369,14 @@ impl WalStorePort for FilesystemWalStore { fn truncate_tail_after(&mut self, after_lsn: Lsn) -> Result<(), WalStoreError> { self.ensure_deferred_commit_sync_has_not_failed()?; + if !self.deferred_commits.is_empty() { + // Truncation could remove a record whose sync is still owed, and + // a later sync would then evidence a record the log no longer + // holds. + return Err(WalStoreError::Io( + "deferred commit records must be synced before truncation".to_owned(), + )); + } rewrite_filesystem_segments_after_truncation(&self.root, after_lsn) } diff --git a/crates/warp-core/src/trusted_runtime_host.rs b/crates/warp-core/src/trusted_runtime_host.rs index 597e6c04..d975d6a1 100644 --- a/crates/warp-core/src/trusted_runtime_host.rs +++ b/crates/warp-core/src/trusted_runtime_host.rs @@ -6243,6 +6243,11 @@ impl TrustedRuntimeApp<'_> { .runtime_wal .as_ref() .ok_or(TrustedRuntimeHostError::RuntimeWalUnavailable)?; + // After a storage failure the cached frontier may name records that + // are not durable. It is not a basis. + if runtime_wal.storage_failed() { + return Err(TrustedRuntimeWalError::StorageFailed.into()); + } Ok(runtime_wal.current_causal_anchor_basis()) } diff --git a/crates/warp-core/tests/causal_wal_hardening_tests.rs b/crates/warp-core/tests/causal_wal_hardening_tests.rs index 92c72fea..5f5229a8 100644 --- a/crates/warp-core/tests/causal_wal_hardening_tests.rs +++ b/crates/warp-core/tests/causal_wal_hardening_tests.rs @@ -1209,6 +1209,56 @@ fn filesystem_writer_syncs_an_existing_log_before_trusting_it() { must_ok(fs::remove_dir_all(root)); } +#[cfg(feature = "host_test")] +#[test] +fn filesystem_failed_synced_commit_never_evidences_pending_deferred_commits() { + use warp_core::causal_wal::{FilesystemWalFaultPlan, FilesystemWalFaultTarget}; + let root = temp_wal_root("deferred-commit-covered-by-failed-sync"); + 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_deferring_commit_sync(submission_transaction( + "deferred-then-failed", + Lsn::from_raw(0), + )), + ); + store.replace_fault_plan_for_test(FilesystemWalFaultPlan::fail_next( + FilesystemWalFaultTarget::FlushCommit, + )); + + // The synced commit would have covered the deferred record as well. + assert!(store + .append_transaction(submission_transaction("synced-fails", Lsn::from_raw(2))) + .is_err()); + assert!(store.sync_deferred_commits().is_err()); + assert_eq!(commit_sync_evidence_count(&store), 0); + drop(store); + must_ok(fs::remove_dir_all(root)); +} + +#[test] +fn filesystem_truncation_is_refused_while_a_commit_sync_is_owed() { + let mut fixture = WalHardeningFixture::new("deferred-commit-truncation"); + must_ok( + fixture + .store + .append_transaction_deferring_commit_sync(submission_transaction( + "before-truncation", + Lsn::from_raw(0), + )), + ); + + assert!( + fixture.store.truncate_tail_after(Lsn::from_raw(0)).is_err(), + "truncation must not remove a record whose sync is still owed" + ); + assert_eq!(commit_sync_evidence_count(&fixture.store), 0); + + must_ok(fixture.store.sync_deferred_commits()); + assert_eq!(commit_sync_evidence_count(&fixture.store), 1); + must_ok(fixture.store.truncate_tail_after(Lsn::from_raw(1))); +} + #[test] fn filesystem_manifest_publication_syncs_deferred_commits_first() { let mut fixture = WalHardeningFixture::new("deferred-commit-manifest"); 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 3bef7865..c4db6cf7 100644 --- a/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs +++ b/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs @@ -2242,6 +2242,15 @@ fn batched_submission_sync_failure_rolls_back_intake_and_fails_closed() { failed_wal.recover_evidence_catalog_read_only(), Err(TrustedRuntimeWalError::StorageFailed) )); + assert!( + matches!( + host.app().current_causal_anchor_basis(), + Err(TrustedRuntimeHostError::Wal( + TrustedRuntimeWalError::StorageFailed + )) + ), + "a frontier over records that are not durable is not a basis" + ); // The records were written; only their sync failed. Reopening recovers // what storage holds, as it would after a process loss. From a1903e8b0ab0823f98a2f07c11e1f916f42634ce Mon Sep 17 00:00:00 2001 From: James Ross Date: Mon, 12 Oct 2026 22:13:32 -0700 Subject: [PATCH 09/10] Sync deferred commits before sealing a segment Address review: seal_segment returned a seal covering a transaction whose commit record had not been synced. rotate_segment already synced first; the direct sealing surface did not. Co-Authored-By: Claude Opus 5.5 --- crates/warp-core/src/causal_wal.rs | 3 +++ .../tests/causal_wal_hardening_tests.rs | 26 +++++++++++++++++++ 2 files changed, 29 insertions(+) diff --git a/crates/warp-core/src/causal_wal.rs b/crates/warp-core/src/causal_wal.rs index 998146d4..95772b28 100644 --- a/crates/warp-core/src/causal_wal.rs +++ b/crates/warp-core/src/causal_wal.rs @@ -6357,6 +6357,9 @@ impl WalStorePort for FilesystemWalStore { if active_epoch.epoch_id != epoch_id { return Err(WalStoreError::WriterEpochMismatch); } + // A seal names committed transactions. It must not exist before + // their commit records are durable. + self.sync_deferred_commits()?; let (frames, _, _) = read_segment_file(&segment_path(&self.root, segment_id))?; let frame_refs = frames.iter().collect::>(); let seal = WalSegmentSeal { diff --git a/crates/warp-core/tests/causal_wal_hardening_tests.rs b/crates/warp-core/tests/causal_wal_hardening_tests.rs index 5f5229a8..4f757756 100644 --- a/crates/warp-core/tests/causal_wal_hardening_tests.rs +++ b/crates/warp-core/tests/causal_wal_hardening_tests.rs @@ -1259,6 +1259,32 @@ fn filesystem_truncation_is_refused_while_a_commit_sync_is_owed() { must_ok(fixture.store.truncate_tail_after(Lsn::from_raw(1))); } +#[test] +fn filesystem_segment_seal_syncs_deferred_commits_first() { + let mut fixture = WalHardeningFixture::new("deferred-commit-seal"); + must_ok( + fixture + .store + .append_transaction_deferring_commit_sync(submission_transaction( + "before-seal", + Lsn::from_raw(0), + )), + ); + assert_eq!(commit_sync_evidence_count(&fixture.store), 0); + + must_ok( + fixture + .store + .seal_segment(epoch_id(), WalSegmentId::from_raw(1)), + ); + + assert_eq!( + commit_sync_evidence_count(&fixture.store), + 1, + "a seal must not exist before the commits it covers are durable" + ); +} + #[test] fn filesystem_manifest_publication_syncs_deferred_commits_first() { let mut fixture = WalHardeningFixture::new("deferred-commit-manifest"); From 13f5c832aacf273fd8f247dc2de47b61de0a1ba6 Mon Sep 17 00:00:00 2001 From: James Ross Date: Mon, 12 Oct 2026 22:23:02 -0700 Subject: [PATCH 10/10] Drop the live evidence catalog after a storage failure Address review: the live catalog had already observed transactions whose sync then failed, and kept a Fresh posture, so its accessors still offered exact committed segments for a batch the host had rolled back. On a storage failure the adapter now drops the catalog and reports NeedsRebuild, naming the last commit that a successful sync is known to cover. Co-Authored-By: Claude Opus 5.5 --- crates/warp-core/src/trusted_runtime_host.rs | 32 +++++++++++++------ .../tests/trusted_runtime_host_loop_tests.rs | 8 +++++ 2 files changed, 31 insertions(+), 9 deletions(-) diff --git a/crates/warp-core/src/trusted_runtime_host.rs b/crates/warp-core/src/trusted_runtime_host.rs index 1f4dac6a..7991ff70 100644 --- a/crates/warp-core/src/trusted_runtime_host.rs +++ b/crates/warp-core/src/trusted_runtime_host.rs @@ -2571,6 +2571,8 @@ pub struct TrustedRuntimeWal { /// Set when storage fails while commit syncs are deferred; see /// [`TrustedRuntimeWalError::StorageFailed`]. storage_failed: bool, + /// Digest of the last commit that a successful sync is known to cover. + last_synced_commit_digest: Hash, } /// Whether an appended commit record is synced before the append returns. @@ -2651,6 +2653,9 @@ impl TrustedRuntimeWal { durable_submission_acceptances, unsynced_submission_acceptances: BTreeMap::new(), storage_failed: false, + // Opening the store synced the log, so the recovered prefix is + // durable. + last_synced_commit_digest: recovered_cursor.previous_committed_transaction_digest, evidence_catalog: Some(evidence_catalog), evidence_catalog_posture: EvidenceCatalogPosture::Fresh, #[cfg(any(test, feature = "host_test"))] @@ -2726,6 +2731,7 @@ impl TrustedRuntimeWal { durable_submission_acceptances: self.durable_submission_acceptances.clone(), unsynced_submission_acceptances: self.unsynced_submission_acceptances.clone(), storage_failed: self.storage_failed, + last_synced_commit_digest: self.last_synced_commit_digest, }) } @@ -2971,6 +2977,13 @@ impl TrustedRuntimeWal { fn fail_storage(&mut self) { self.storage_failed = true; self.unsynced_submission_acceptances.clear(); + // The live catalog may describe records that are not durable. It is + // dropped, and the posture names the last commit known to be synced. + self.evidence_catalog = None; + self.evidence_catalog_posture = EvidenceCatalogPosture::NeedsRebuild { + reason: *blake3::hash(b"storage_failed").as_bytes(), + last_good_commit: self.last_synced_commit_digest, + }; } fn storage_failed(&self) -> bool { @@ -2983,9 +2996,11 @@ impl TrustedRuntimeWal { self.refresh_cursor_from_store_for_writer() } + /// Records that a successful sync covers every record appended so far. fn promote_synced_submission_acceptances(&mut self) { let synced = std::mem::take(&mut self.unsynced_submission_acceptances); self.durable_submission_acceptances.extend(synced); + self.last_synced_commit_digest = self.previous_committed_transaction_digest; } /// Returns how many storage syncs have made commit records durable, when @@ -3541,21 +3556,20 @@ impl TrustedRuntimeWal { } contexts.apply_frames(&frames)?; match sync { - CommitSync::Now => { - self.store.append_transaction(transaction)?; - // A synced commit covers every earlier record in the segment. - self.promote_synced_submission_acceptances(); - } - CommitSync::Deferred => { - self.store - .append_transaction_deferring_commit_sync(transaction)?; - } + CommitSync::Now => self.store.append_transaction(transaction)?, + CommitSync::Deferred => self + .store + .append_transaction_deferring_commit_sync(transaction)?, } self.operation_context_index = Some((commit.commit_digest, contexts)); self.next_lsn = next_lsn; self.previous_frame_digest = last_frame_digest; self.previous_committed_transaction_digest = commit.commit_digest; self.causal_history_frontier_digest = next_causal_history_frontier; + if matches!(sync, CommitSync::Now) { + // A synced commit covers every earlier record in the segment. + self.promote_synced_submission_acceptances(); + } self.try_update_evidence_catalog_after_commit(&commit, &frames, last_good_commit); Ok(commit) } 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 c4db6cf7..5c73c1a6 100644 --- a/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs +++ b/crates/warp-core/tests/trusted_runtime_host_loop_tests.rs @@ -2242,6 +2242,14 @@ fn batched_submission_sync_failure_rolls_back_intake_and_fails_closed() { failed_wal.recover_evidence_catalog_read_only(), Err(TrustedRuntimeWalError::StorageFailed) )); + assert!( + failed_wal.evidence_catalog().is_none(), + "the live catalog must not describe records that are not durable" + ); + assert!(matches!( + failed_wal.evidence_catalog_posture(), + warp_core::EvidenceCatalogPosture::NeedsRebuild { .. } + )); assert!( matches!( host.app().current_causal_anchor_basis(),