From 1fd2e099715344fea3f35c07f23c4de58ef3bdd7 Mon Sep 17 00:00:00 2001 From: Hare Date: Mon, 26 Oct 2026 14:28:37 +0900 Subject: [PATCH] test: cover real partial publication and terminal receipt durability faults --- crates/worker-runtime/src/execution.rs | 87 +- crates/worker-runtime/src/fs_store.rs | 52 + crates/worker-runtime/src/runtime.rs | 60 ++ .../src/runtime/restore_continuation.rs | 3 + crates/worker-runtime/src/worker_backend.rs | 37 + ...orker_backend_restore_publication_tests.rs | 2 + ...rker_backend_restore_registration_tests.rs | 889 ++++++++++++++++++ 7 files changed, 1128 insertions(+), 2 deletions(-) create mode 100644 crates/worker-runtime/src/worker_backend_restore_registration_tests.rs diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index ffa6e807f..f4c7d85e9 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -215,6 +215,38 @@ impl WorkerExecutionResult { } } +// Private one-shot synchronization at real publication boundaries. Disconnect +// releases the waiter during unwinding; timeout bounds a forgotten test owner. +#[cfg(test)] +#[derive(Debug)] +pub(crate) struct RestoreTestGate { + pub(crate) entered: std::sync::mpsc::Sender, + pub(crate) release: std::sync::mpsc::Receiver<()>, +} + +#[cfg(test)] +impl RestoreTestGate { + pub(crate) fn wait(self, evidence: T) { + let _ = self.entered.send(evidence); + assert!( + !matches!( + self.release + .recv_timeout(std::time::Duration::from_secs(30)), + Err(std::sync::mpsc::RecvTimeoutError::Timeout) + ), + "publication test gate timed out" + ); + } +} + +#[cfg(all(test, feature = "ws-server"))] +#[derive(Debug)] +pub(crate) struct RestoreFlushFault { + pub(crate) remaining_successes: usize, + pub(crate) gate: RestoreTestGate, + pub(crate) restored: std::sync::mpsc::Sender>, +} + /// Runtime hooks available to an execution backend for one Worker. #[cfg(feature = "ws-server")] type WorkerObservationPublisher = Arc< @@ -249,6 +281,8 @@ pub struct WorkerExecutionContext { observation_publisher: WorkerObservationPublisher, #[cfg(feature = "ws-server")] publication_state: Arc>, + #[cfg(all(test, feature = "ws-server"))] + flush_fault: Arc>>, } impl WorkerExecutionContext { @@ -262,6 +296,8 @@ impl WorkerExecutionContext { restore_progress: None, restore_repair_authorizer: None, observation_publisher, + #[cfg(test)] + flush_fault: Arc::default(), publication_state: Arc::new(Mutex::new(WorkerExecutionPublicationState::Active)), } } @@ -276,6 +312,8 @@ impl WorkerExecutionContext { restore_progress: None, restore_repair_authorizer: None, observation_publisher, + #[cfg(test)] + flush_fault: Arc::default(), publication_state: Arc::new(Mutex::new(WorkerExecutionPublicationState::Pending( VecDeque::new(), ))), @@ -386,20 +424,65 @@ impl WorkerExecutionContext { let Some(event) = event else { return Ok(()); }; - if let Err(error) = (self.observation_publisher)(self.worker_ref.clone(), event.clone()) - { + #[cfg(test)] + let fault = { + let mut fault = self.flush_fault.lock().unwrap(); + if fault.as_ref().is_some_and(|f| f.remaining_successes == 0) { + fault.take() + } else { + if let Some(fault) = fault.as_mut() { + fault.remaining_successes -= 1; + } + None + } + }; + #[cfg(test)] + let mut restored = None; + #[cfg(test)] + let publication = if let Some(fault) = fault { + restored = Some(fault.restored); + fault.gate.wait(event.clone()); + Err(RuntimeError::InvalidRequest( + "injected buffered publication failure".into(), + )) + } else { + (self.observation_publisher)(self.worker_ref.clone(), event.clone()) + }; + #[cfg(not(test))] + let publication = (self.observation_publisher)(self.worker_ref.clone(), event.clone()); + if let Err(error) = publication { let mut state = self .publication_state .lock() .map_err(|_| RuntimeError::StatePoisoned)?; if let WorkerExecutionPublicationState::Activating(events) = &mut *state { events.push_front(event); + #[cfg(test)] + if let Some(restored) = restored { + let _ = restored.send(events.iter().cloned().collect()); + } } return Err(error); } } } + #[cfg(all(test, feature = "ws-server"))] + pub(crate) fn test_fail_flush(&self, fault: RestoreFlushFault) { + assert!(self.flush_fault.lock().unwrap().replace(fault).is_none()); + } + + #[cfg(all(test, feature = "ws-server"))] + pub(crate) fn test_buffered_events(&self) -> Vec { + match &*self.publication_state.lock().unwrap() { + WorkerExecutionPublicationState::Pending(events) + | WorkerExecutionPublicationState::Activating(events) => { + events.iter().cloned().collect() + } + _ => Vec::new(), + } + } + #[cfg(not(feature = "ws-server"))] pub(crate) fn activate_candidate(&self) -> Result<(), RuntimeError> { Ok(()) diff --git a/crates/worker-runtime/src/fs_store.rs b/crates/worker-runtime/src/fs_store.rs index d2b9c86de..1278f89f7 100644 --- a/crates/worker-runtime/src/fs_store.rs +++ b/crates/worker-runtime/src/fs_store.rs @@ -128,6 +128,13 @@ pub struct FsRuntimeStore { worker_write_fault: Arc, #[cfg(test)] worker_reconcile_sync_fault: Arc, + #[cfg(test)] + worker_reconcile_read_fault: Arc, + #[cfg(test)] + worker_reconcile_counts: Arc<( + std::sync::atomic::AtomicUsize, + std::sync::atomic::AtomicUsize, + )>, } impl PartialEq for FsRuntimeStore { @@ -174,6 +181,20 @@ impl FsRuntimeStore { .store(true, Ordering::Release); } + #[cfg(test)] + pub(crate) fn fail_next_worker_reconcile_read(&self) { + self.worker_reconcile_read_fault + .store(true, Ordering::Release); + } + + #[cfg(test)] + pub(crate) fn test_worker_reconcile_counts(&self) -> (usize, usize) { + ( + self.worker_reconcile_counts.0.load(Ordering::Acquire), + self.worker_reconcile_counts.1.load(Ordering::Acquire), + ) + } + pub(crate) fn open_or_create( root: PathBuf, runtime_id: &str, @@ -235,6 +256,10 @@ impl FsRuntimeStore { worker_write_fault: Arc::new(AtomicU8::new(AtomicWriteFault::None as u8)), #[cfg(test)] worker_reconcile_sync_fault: Default::default(), + #[cfg(test)] + worker_reconcile_read_fault: Default::default(), + #[cfg(test)] + worker_reconcile_counts: Default::default(), }; let state = if snapshot_exists { Some(store.load_runtime_state()?) @@ -298,8 +323,16 @@ impl FsRuntimeStore { &self, worker_ref: &WorkerRef, ) -> Result { + #[cfg(test)] + self.worker_reconcile_counts + .0 + .fetch_add(1, Ordering::AcqRel); self.read_durable_worker_record_with_sync(worker_ref, |file| { #[cfg(test)] + self.worker_reconcile_counts + .1 + .fetch_add(1, Ordering::AcqRel); + #[cfg(test)] if self .worker_reconcile_sync_fault .swap(false, Ordering::AcqRel) @@ -337,6 +370,17 @@ impl FsRuntimeStore { "Worker aggregate reconciliation requires the Runtime store owner lock".to_string(), )); } + #[cfg(test)] + if self + .worker_reconcile_read_fault + .swap(false, Ordering::AcqRel) + { + return Err(RuntimeError::StoreCommitOutcomeUnknown { + operation: "read reconciled Worker aggregate", + path: self.worker_dir(&worker_ref.worker_id).join(WORKER_FILE), + source: std::io::Error::other("injected reconciliation read failure"), + }); + } let (record, file) = self.read_worker_record_file(worker_ref)?; let worker_dir = self.worker_dir(&worker_ref.worker_id); let path = worker_dir.join(WORKER_FILE); @@ -1724,6 +1768,10 @@ fn validate_current_runtime_store(root: &Path) -> Result<(), RuntimeError> { worker_write_fault: Arc::new(AtomicU8::new(AtomicWriteFault::None as u8)), #[cfg(test)] worker_reconcile_sync_fault: Default::default(), + #[cfg(test)] + worker_reconcile_read_fault: Default::default(), + #[cfg(test)] + worker_reconcile_counts: Default::default(), }; store.load_runtime_state().map(|_| ()) } @@ -1735,6 +1783,10 @@ fn validate_completed_migration_staging(root: &Path) -> Result<(), RuntimeError> worker_write_fault: Arc::new(AtomicU8::new(AtomicWriteFault::None as u8)), #[cfg(test)] worker_reconcile_sync_fault: Default::default(), + #[cfg(test)] + worker_reconcile_read_fault: Default::default(), + #[cfg(test)] + worker_reconcile_counts: Default::default(), }; let state = store.load_runtime_state()?; if let Some(worker) = state.workers.values().find(|worker| { diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 4f9141cc1..755509f7b 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -2340,6 +2340,48 @@ impl Runtime { } } + #[cfg(test)] + pub(crate) fn test_terminal_receipt_gate( + &self, + gate: crate::execution::RestoreTestGate, + ) { + assert!( + self.restore_continuations + .terminal_receipt_gate + .lock() + .unwrap() + .replace(gate) + .is_none() + ); + } + + #[cfg(test)] + pub(crate) fn test_pending_restore_context( + &self, + worker_ref: &WorkerRef, + ) -> ( + WorkerLifecycleOperationId, + crate::execution::WorkerExecutionContext, + bool, + ) { + let state = self.lock().unwrap(); + let worker = state.worker(worker_ref).unwrap(); + assert!( + worker + .restore_guard + .owners + .values() + .all(|owner| owner.receipt.is_none()) + ); + ( + worker.pending_restore.unwrap().operation_id, + worker.restore_candidate_context.clone().unwrap(), + state + .unreconciled_worker_writes + .contains(&worker_ref.worker_id), + ) + } + // Seed durable crash checkpoints without launching a second Controller. // Tests then reopen the real store with empty backend resource maps. #[cfg(all(test, feature = "fs-store"))] @@ -2512,6 +2554,24 @@ impl Runtime { if result.state == WorkerRestoreState::ReconciliationRequired { return Ok(()); } + #[cfg(test)] + if result.state == WorkerRestoreState::Accepted { + let gate = self + .restore_continuations + .terminal_receipt_gate + .lock() + .unwrap() + .take(); + if let Some(gate) = gate { + let operation = self + .lock()? + .worker(worker_ref)? + .pending_restore + .expect("terminal receipt follows committed operation") + .operation_id; + gate.wait(operation); + } + } let mut state = self.lock()?; let mut candidate = state.worker(worker_ref)?.clone(); if candidate.restore_guard.active_request_id.is_some() diff --git a/crates/worker-runtime/src/runtime/restore_continuation.rs b/crates/worker-runtime/src/runtime/restore_continuation.rs index ad50b4d2d..ba8e884c7 100644 --- a/crates/worker-runtime/src/runtime/restore_continuation.rs +++ b/crates/worker-runtime/src/runtime/restore_continuation.rs @@ -20,6 +20,9 @@ pub(super) struct RestoreContinuations { owners: Mutex>>, #[cfg(test)] paused: std::sync::atomic::AtomicBool, + #[cfg(test)] + pub(super) terminal_receipt_gate: + Mutex>>, } #[derive(Debug, Default)] diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 80d8118d6..fe7a65e7e 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -555,9 +555,17 @@ pub trait RuntimeWorkerFactory: Send + Sync + 'static { /// `WorkerController`. #[derive(Default)] struct RuntimeWorkerObservationHub { + #[cfg(test)] + registration_gate: Mutex>, workers: Mutex>, } +#[cfg(test)] +struct RestoreHubRegistrationGate { + gate: crate::execution::RestoreTestGate, + fail: bool, +} + #[derive(Clone)] struct RuntimeObservedWorker { workspace_id: Option, @@ -584,6 +592,17 @@ impl RuntimeWorkerObservationHub { sink: handle.sink.clone(), }, ); + drop(workers); + #[cfg(test)] + { + let gate = self.registration_gate.lock().unwrap().take(); + if let Some(gate) = gate { + gate.gate.wait(handle.clone()); + if gate.fail { + return Err("injected post-insert hub registration failure".into()); + } + } + } Ok(()) } @@ -837,6 +856,7 @@ struct FailedRestoreResources { #[cfg(test)] #[derive(Default)] struct RestoreFaults { + effects_calls: std::sync::atomic::AtomicUsize, fail_at: Mutex>, executions: std::sync::atomic::AtomicUsize, cleanup_fail: AtomicBool, @@ -2694,6 +2714,11 @@ impl RuntimeRestoreCandidate for ProfileRestoreCandidate { failure.io_kind = io_kind; failure }; + #[cfg(test)] + self.factory + .restore_faults + .effects_calls + .fetch_add(1, Ordering::SeqCst); prepared.apply_effects().await.map_err(&map_error)?; self.factory .failed_restore_resources @@ -3392,6 +3417,8 @@ struct SubjectControllerReservation { /// Worker execution backend backed by real worker crate Workers. pub struct WorkerRuntimeExecutionBackend { + #[cfg(test)] + bridge_fault: Mutex>>, backend_id: String, factory: Arc, working_directory_materializer: Option>, @@ -3427,6 +3454,8 @@ where .build() .map_err(|err| format!("failed to build worker adapter runtime: {err}"))?; Ok(Self { + #[cfg(test)] + bridge_fault: Mutex::default(), backend_id: DEFAULT_BACKEND_ID.to_string(), factory: Arc::new(factory), working_directory_materializer: None, @@ -4161,6 +4190,14 @@ where .map_err(|error| error.to_string())?; bridge.snapshot_pending = false; } + #[cfg(test)] + { + let fault = self.bridge_fault.lock().unwrap().take(); + if let Some(fault) = fault { + fault.wait(handle.clone()); + return Err("injected bridge task admission failure after snapshot".into()); + } + } let streams = bridge.streams.clone(); let bridge_context = bridge.context.clone(); let bridge_worker_state = worker_state.clone(); diff --git a/crates/worker-runtime/src/worker_backend_restore_publication_tests.rs b/crates/worker-runtime/src/worker_backend_restore_publication_tests.rs index 9a905c4bd..42c1c9153 100644 --- a/crates/worker-runtime/src/worker_backend_restore_publication_tests.rs +++ b/crates/worker-runtime/src/worker_backend_restore_publication_tests.rs @@ -1499,4 +1499,6 @@ mod publication { fixture.assert_released(); fixture.assert_quiet(); } + + include!("worker_backend_restore_registration_tests.rs"); } diff --git a/crates/worker-runtime/src/worker_backend_restore_registration_tests.rs b/crates/worker-runtime/src/worker_backend_restore_registration_tests.rs new file mode 100644 index 000000000..a57d80dcf --- /dev/null +++ b/crates/worker-runtime/src/worker_backend_restore_registration_tests.rs @@ -0,0 +1,889 @@ +// Included by the production Profile publication suite. Each fault is at a +// distinct partial-publication boundary; no factory result is substituted. + +fn publication_gate() -> ( + mpsc::Sender<()>, + mpsc::Receiver, + crate::execution::RestoreTestGate, +) { + let (release, receiver) = mpsc::channel(); + let (entered, evidence) = mpsc::channel(); + ( + release, + evidence, + crate::execution::RestoreTestGate { + entered, + release: receiver, + }, + ) +} + +fn assert_single_publication(fixture: &Fixture, expected: usize) { + let faults = &fixture.backend.factory.restore_faults; + assert_eq!( + faults.executions.load(Ordering::SeqCst), + expected, + "construction reran" + ); + assert_eq!( + faults.effects_calls.load(Ordering::SeqCst), + expected, + "restore effects reran" + ); + assert_eq!( + faults.publication_attempts.load(Ordering::SeqCst), + expected, + "Controller start reran" + ); + fixture.assert_quiet(); +} + +fn assert_real_attach( + fixture: &Fixture, + handle: &WorkerHandle, +) -> crate::execution::WorkerProtocolTransport { + let worker_ref = &fixture.request.worker_ref; + assert!(matches!( + fixture + .backend + .worker_session(crate::execution::WorkerSessionObservationRequest { + worker_ref: worker_ref.clone(), + }), + runtime_api::WorkerSessionAvailability::LiveProtocol + )); + let transport = fixture + .backend + .clone() + .attach_worker_protocol(worker_ref) + .unwrap(); + transport.validate().unwrap(); + let paired = subscribe_worker_protocol_session(handle); + assert_eq!( + serde_json::to_value(&transport.snapshot).unwrap(), + serde_json::to_value(paired.snapshot_event).unwrap() + ); + assert!( + matches!(&transport.snapshot, Event::Snapshot { session, greeting, state, in_flight, .. } + if session == &session_store::public_snapshot::project_current_session_snapshot(&fixture.entries()) + && !greeting.tools.is_empty() && state.state == protocol::WorkerState::Idle && in_flight.is_empty()) + ); + transport +} + +fn observe_publication_markers( + fixture: &Fixture, + observations: tokio::sync::broadcast::Receiver, + markers: &'static [&'static str], +) -> tokio::sync::broadcast::Receiver { + fixture + .backend + .run_on_adapter_runtime(async move { + let mut observations = observations; + let mut counts = vec![0; markers.len()]; + let mut snapshots = 0; + tokio::time::timeout(WAIT, async { + loop { + let observed = observations.recv().await.unwrap(); + if matches!(observed.payload, Event::Snapshot { .. }) { + snapshots += 1; + } + for (index, marker) in markers.iter().enumerate() { + if is_marker(&observed.payload, marker) { + counts[index] += 1; + } + } + if counts.last() == Some(&1) { + break; + } + } + }) + .await + .map_err(|_| format!("lost original subscription markers: {counts:?}"))?; + assert_eq!( + counts, + vec![1; markers.len()], + "lost or duplicated prefix on original receiver" + ); + assert_eq!(snapshots, 1, "initial full snapshot was lost or replayed"); + while let Ok(event) = observations.try_recv() { + assert!( + !markers + .iter() + .any(|marker| is_marker(&event.payload, marker)), + "duplicate after tail fence" + ); + } + Ok(observations) + }) + .unwrap() +} + +fn partial_registration_recovers_same_controller(hub: bool) { + let mut fixture = Fixture::no_model(); + Arc::get_mut(&mut fixture.backend) + .unwrap() + .spawn_restore_timeout = Duration::from_millis(250); + let runtime = stopped_runtime(&fixture); + let pause = runtime.test_pause_restore_continuation(); + let worker_ref = fixture.request.worker_ref.clone(); + let scope = fixture.request.workspace_scope.as_ref().unwrap(); + let request = runtime.test_restore_request(&worker_ref); + let observations = runtime.subscribe_worker_observation().unwrap(); + let (started_release, started_entered) = activation_gate(&fixture, true); + let first = runtime + .restore_worker_operation_scoped(scope, &worker_ref, request.clone()) + .unwrap(); + assert_eq!( + first.state, + server_api::WorkerRestoreState::ReconciliationRequired + ); + let started = started_entered.recv_timeout(WAIT).unwrap(); + let pending = assert_activation_owned(&fixture, &runtime, &request.request_id, &started); + let old = started.handle.as_ref().unwrap().clone(); + let canonical = fixture.canonical_files(); + alert_marker(&old, "registration-original-paired-prefix"); + let (release, entered, fault) = publication_gate(); + if hub { + *fixture + .backend + .factory + .observation_hub + .registration_gate + .lock() + .unwrap() = Some(RestoreHubRegistrationGate { + gate: fault, + fail: true, + }); + } else { + *fixture.backend.bridge_fault.lock().unwrap() = Some(fault); + } + started_release.release(); + drop(pause); + let boundary_handle = entered.recv_timeout(WAIT).unwrap(); + assert!(boundary_handle.same_controller(&old)); + let durable = durable_operation(&runtime, &worker_ref).unwrap(); + assert_eq!(durable.operation_id, started.operation_id); + assert_eq!(durable.commit, crate::runtime::RestoreCommitFact::Committed); + assert!(restore_owner(&runtime, &worker_ref, &request.request_id)["receipt"].is_null()); + if hub { + let (_, shared, _) = fixture + .backend + .factory + .observation_hub + .get(&worker_ref) + .unwrap(); + assert!( + Arc::ptr_eq(&shared, &old.shared_state), + "real hub insert preceded fault" + ); + } else { + assert!( + fixture + .backend + .factory + .observation_hub + .get(&worker_ref) + .is_none() + ); + } + let (operation, context, uncertain) = runtime.test_pending_restore_context(&worker_ref); + assert_eq!(operation, started.operation_id); + assert!(!uncertain); + assert!( + context + .test_buffered_events() + .iter() + .any(|event| matches!(event, Event::Snapshot { .. })) + ); + alert_marker(&old, "registration-during-fault"); + let pause = runtime.test_pause_restore_continuation(); + release.send(()).unwrap(); + // This lock also observes the end of the actual registration attempt, not + // merely the test's entered notification. Its retained scope owns cleanup. + let execution = fixture.backend.workers.lock().unwrap()[&worker_ref].clone(); + assert!(old.same_controller(&execution.handle)); + assert_eq!(execution.restore_operation_id, Some(operation)); + assert_eq!( + execution.tasks.tasks.lock().unwrap().len(), + if hub { 2 } else { 1 } + ); + let bridge = execution.tasks.protocol_bridge.lock().unwrap(); + if hub { + assert!(bridge.is_none()); + } else { + let bridge = bridge.as_ref().unwrap(); + assert!( + !bridge.snapshot_pending, + "snapshot already registered, do not replay it" + ); + let streams = bridge.streams.lock().unwrap(); + assert!( + streams.is_some(), + "original paired streams survived failed task admission" + ); + assert_eq!( + serde_json::to_value(&streams.as_ref().unwrap().snapshot_event).unwrap(), + serde_json::to_value(&started.snapshot).unwrap() + ); + } + drop(bridge); + assert!( + !old.protocol_is_closed(), + "publication failure must not clean a committed Controller" + ); + assert_single_publication(&fixture, 1); + drop(pause); + runtime.test_wait_restore_continuation(&worker_ref, operation, WAIT); + assert!(durable_operation(&runtime, &worker_ref).is_none()); + let owner = restore_owner(&runtime, &worker_ref, &request.request_id); + assert_eq!( + owner["operation_id"], + serde_json::to_value(operation).unwrap() + ); + assert_eq!(owner["receipt"]["state"], "accepted"); + assert!( + fixture + .backend + .pending_restore(&worker_ref) + .unwrap() + .is_none() + ); + assert!(!pending.task_in_flight.load(Ordering::Acquire)); + assert!(pending.activated.lock().unwrap().is_none()); + assert!(pending.result.lock().unwrap().is_none()); + assert!(pending.task.try_lock().unwrap().is_none()); + assert_eq!(fixture.backend.workers.lock().unwrap().len(), 1); + assert!(old.same_controller(&fixture.backend.workers.lock().unwrap()[&worker_ref].handle)); + let transport = assert_real_attach(&fixture, &old); + assert_eq!(fixture.canonical_files(), canonical); + assert_single_publication(&fixture, 1); + alert_marker(&old, "registration-tail-fence"); + let observations = observe_publication_markers( + &fixture, + observations, + &[ + "registration-original-paired-prefix", + "registration-during-fault", + "registration-tail-fence", + ], + ); + assert_eq!( + runtime + .restore_worker_operation_scoped(scope, &worker_ref, request.clone()) + .unwrap() + .state, + server_api::WorkerRestoreState::Accepted + ); + assert_single_publication(&fixture, 1); + // Retain the original transport/handle and scope through Stop/replacement. + runtime + .stop_worker_scoped(scope, &worker_ref, None) + .unwrap(); + assert!(old.protocol_is_closed()); + assert!(transport.validate().is_err()); + let replacement_request = runtime.test_restore_request(&worker_ref); + assert_eq!( + runtime + .restore_worker_operation_scoped(scope, &worker_ref, replacement_request.clone()) + .unwrap() + .state, + server_api::WorkerRestoreState::Accepted + ); + let replacement = fixture.backend.workers.lock().unwrap()[&worker_ref].clone(); + assert!(!old.same_controller(&replacement.handle)); + assert_ne!(replacement.restore_operation_id, Some(operation)); + assert!( + transport + .dispatch(Method::Pause { + command: protocol::WorkerCommandEnvelope::new(801) + }) + .is_err() + ); + assert!( + transport + .dispatch(Method::Shutdown { + command: protocol::WorkerCommandEnvelope::new(802) + }) + .is_err() + ); + for method in [ + Method::Pause { + command: protocol::WorkerCommandEnvelope::new(803), + }, + Method::Shutdown { + command: protocol::WorkerCommandEnvelope::new(804), + }, + ] { + let old_command = old.clone(); + assert!( + fixture + .backend + .run_on_adapter_runtime(async move { + old_command + .send(method) + .await + .map_err(|error| error.to_string()) + }) + .is_err(), + "closed original Controller command must not route to replacement" + ); + } + fixture + .backend + .factory + .deactivate_stopped_controller(&worker_ref, &old) + .unwrap(); + let (_, shared, _) = fixture + .backend + .factory + .observation_hub + .get(&worker_ref) + .unwrap(); + assert!(Arc::ptr_eq(&shared, &replacement.handle.shared_state)); + alert_marker(&old, "registration-stopped-event-must-not-leak"); + alert_marker(&replacement.handle, "registration-replacement-fence"); + fixture + .backend + .run_on_adapter_runtime(async move { + let mut observations = observations; + tokio::time::timeout(WAIT, async { + loop { + let event = observations.recv().await.unwrap(); + assert!(!is_marker( + &event.payload, + "registration-stopped-event-must-not-leak" + )); + if is_marker(&event.payload, "registration-replacement-fence") { + break; + } + } + }) + .await + .map_err(|_| "replacement observation fence missing".to_string())?; + drop(transport); + Ok(()) + }) + .unwrap(); + assert_single_publication(&fixture, 2); + assert_eq!( + runtime + .restore_worker_operation_scoped(scope, &worker_ref, request.clone()) + .unwrap() + .state, + server_api::WorkerRestoreState::Accepted + ); + assert_eq!(restore_owner_count(&runtime, &worker_ref), 2); + let current = fixture.backend.workers.lock().unwrap()[&worker_ref].clone(); + assert!(current.handle.same_controller(&replacement.handle)); + assert_eq!( + current.restore_operation_id, + replacement.restore_operation_id + ); + assert_eq!( + replacement.handle.shared_state.snapshot().state, + protocol::WorkerState::Idle + ); + drop(current); + drop(execution); + drop(replacement); + drop(pending); + runtime + .stop_worker_scoped(scope, &worker_ref, None) + .unwrap(); + fixture.assert_released(); +} + +fn restore_owner_count(runtime: &EmbeddedRuntime, worker_ref: &WorkerRef) -> usize { + let saved = runtime + .test_runtime_store() + .read_worker_record(worker_ref) + .unwrap(); + serde_json::to_value(saved.restore_guard).unwrap()["owners"] + .as_object() + .unwrap() + .len() +} + +#[test] +#[serial_test::serial(worker_allocation)] +fn bridge_partial_snapshot_registration_retains_original_streams_and_autonomously_recovers() { + partial_registration_recovers_same_controller(false); +} + +#[test] +#[serial_test::serial(worker_allocation)] +fn hub_partial_insert_registration_retains_original_controller_and_autonomously_recovers() { + partial_registration_recovers_same_controller(true); +} + +#[test] +#[serial_test::serial(worker_allocation)] +fn context_partial_flush_retains_only_undelivered_queue_and_recovers_without_repost() { + let mut fixture = Fixture::no_model(); + Arc::get_mut(&mut fixture.backend) + .unwrap() + .spawn_restore_timeout = Duration::from_millis(250); + let runtime = stopped_runtime(&fixture); + let pause = runtime.test_pause_restore_continuation(); + let worker_ref = fixture.request.worker_ref.clone(); + let scope = fixture.request.workspace_scope.as_ref().unwrap(); + let request = runtime.test_restore_request(&worker_ref); + let observations = runtime.subscribe_worker_observation().unwrap(); + let (started_release, started_entered) = activation_gate(&fixture, true); + assert_eq!( + runtime + .restore_worker_operation_scoped(scope, &worker_ref, request.clone()) + .unwrap() + .state, + server_api::WorkerRestoreState::ReconciliationRequired + ); + let started = started_entered.recv_timeout(WAIT).unwrap(); + let pending = assert_activation_owned(&fixture, &runtime, &request.request_id, &started); + let old = started.handle.unwrap(); + let (operation, context, _) = runtime.test_pending_restore_context(&worker_ref); + // Stop after the actual bridge and hub registration, without failing that + // boundary, so the flush begins with its real original full snapshot. + let (hub_release, hub_entered, hub_gate) = publication_gate(); + *fixture + .backend + .factory + .observation_hub + .registration_gate + .lock() + .unwrap() = Some(RestoreHubRegistrationGate { + gate: hub_gate, + fail: false, + }); + started_release.release(); + drop(pause); + assert!( + hub_entered + .recv_timeout(WAIT) + .unwrap() + .same_controller(&old) + ); + let buffered_prefix = context.test_buffered_events(); + assert!(matches!( + buffered_prefix.first(), + Some(Event::Snapshot { .. }) + )); + let prefix_length = buffered_prefix.len(); + let marker = |text: &str| Event::Error { + code: protocol::ErrorCode::Internal, + message: text.into(), + }; + context + .publish_protocol_event(marker("flush-delivered")) + .unwrap(); + context + .publish_protocol_event(marker("flush-failed")) + .unwrap(); + context + .publish_protocol_event(marker("flush-pending")) + .unwrap(); + let (release, entered, gate) = publication_gate(); + let (restored, restored_queue) = mpsc::channel(); + context.test_fail_flush(crate::execution::RestoreFlushFault { + remaining_successes: prefix_length + 1, + gate, + restored, + }); + drop(hub_release); + assert!( + matches!(entered.recv_timeout(WAIT).unwrap(), Event::Error { message, .. } if message == "flush-failed") + ); + let partial_execution = fixture.backend.workers.lock().unwrap()[&worker_ref].clone(); + assert!(old.same_controller(&partial_execution.handle)); + assert_eq!(partial_execution.restore_operation_id, Some(operation)); + assert_eq!(partial_execution.tasks.tasks.lock().unwrap().len(), 2); + let (_, shared, _) = fixture + .backend + .factory + .observation_hub + .get(&worker_ref) + .unwrap(); + assert!(Arc::ptr_eq(&shared, &old.shared_state)); + let pause = runtime.test_pause_restore_continuation(); + context + .publish_protocol_event(marker("flush-arrived-during-error")) + .unwrap(); + release.send(()).unwrap(); + let restored = restored_queue.recv_timeout(WAIT).unwrap(); + let undelivered: Vec<_> = restored + .iter() + .filter_map(|event| match event { + Event::Error { message, .. } if message.starts_with("flush-") => Some(message.as_str()), + _ => None, + }) + .collect(); + assert_eq!( + undelivered, + [ + "flush-failed", + "flush-pending", + "flush-arrived-during-error" + ] + ); + assert!( + !restored + .iter() + .any(|event| matches!(event, Event::Snapshot { .. })), + "delivered snapshot must not rejoin the failed suffix" + ); + assert_eq!( + serde_json::to_value(context.test_buffered_events()).unwrap(), + serde_json::to_value(&restored).unwrap() + ); + assert_eq!( + durable_operation(&runtime, &worker_ref) + .unwrap() + .operation_id, + operation + ); + assert!(restore_owner(&runtime, &worker_ref, &request.request_id)["receipt"].is_null()); + // The next terminal gate observes successful retry only after the same + // queued owner has repaired its failed callback and remaining suffix. + let (terminal_release, terminal_entered, gate) = publication_gate(); + runtime.test_terminal_receipt_gate(gate); + drop(pause); + assert_eq!(terminal_entered.recv_timeout(WAIT).unwrap(), operation); + assert!(context.test_buffered_events().is_empty()); + assert!(restore_owner(&runtime, &worker_ref, &request.request_id)["receipt"].is_null()); + assert_eq!( + durable_operation(&runtime, &worker_ref).unwrap().commit, + crate::runtime::RestoreCommitFact::Committed + ); + let execution = fixture.backend.workers.lock().unwrap()[&worker_ref].clone(); + assert!(old.same_controller(&execution.handle)); + assert_eq!(execution.restore_operation_id, Some(operation)); + drop(terminal_release); // RAII disconnect, not a second request. + runtime.test_wait_restore_continuation(&worker_ref, operation, WAIT); + assert_eq!( + restore_owner(&runtime, &worker_ref, &request.request_id)["receipt"]["state"], + "accepted" + ); + assert_real_attach(&fixture, &old); + fixture + .backend + .run_on_adapter_runtime(async move { + let mut observations = observations; + let mut messages = Vec::new(); + let mut snapshots = 0; + while let Ok(event) = observations.try_recv() { + match event.payload { + Event::Error { message, .. } if message.starts_with("flush-") => { + messages.push(message) + } + Event::Snapshot { .. } => snapshots += 1, + _ => {} + } + } + assert_eq!( + messages, + [ + "flush-delivered", + "flush-failed", + "flush-pending", + "flush-arrived-during-error" + ] + ); + assert_eq!(snapshots, 1); + Ok(()) + }) + .unwrap(); + assert_single_publication(&fixture, 1); + drop(partial_execution); + drop(execution); + drop(pending); + runtime + .stop_worker_scoped(scope, &worker_ref, None) + .unwrap(); + fixture.assert_released(); +} + +#[derive(Clone, Copy, Debug)] +enum TerminalFault { + BeforeRename, + AfterRename, + SyncUnknown, + ReadUnknown, +} + +fn published_terminal_receipt_fault(fault: TerminalFault) { + let mut fixture = Fixture::no_model(); + // Reuse the existing ordinary non-network model boundary, not the product + // Codex profile. All fault/convergence assertions precede the sole input. + let script = MockClient::sequential(vec![MockResponse::Complete(simple_text_events())]); + let injected = script.clone(); + let factory = fixture + .backend + .factory + .as_ref() + .clone() + .with_spawn_model_client_factory(move || Box::new(injected.clone())); + fixture.backend = Arc::new(WorkerRuntimeExecutionBackend::new(factory).unwrap()); + let runtime = Arc::new(stopped_runtime(&fixture)); + let pause = runtime.test_pause_restore_continuation(); + let worker_ref = fixture.request.worker_ref.clone(); + let scope = fixture.request.workspace_scope.clone().unwrap(); + let request = runtime.test_restore_request(&worker_ref); + let observations = runtime.subscribe_worker_observation().unwrap(); + let (release, entered, gate) = publication_gate(); + runtime.test_terminal_receipt_gate(gate); + let restoring = { + let runtime = runtime.clone(); + let worker_ref = worker_ref.clone(); + let scope = scope.clone(); + let request = request.clone(); + std::thread::spawn(move || { + runtime.restore_worker_operation_scoped(&scope, &worker_ref, request) + }) + }; + let operation = entered.recv_timeout(WAIT).unwrap(); + // Restore legitimately repairs derived metadata before publication. Receipt + // failures must not rerun those effects; compare from this actual boundary. + let canonical = fixture.canonical_files(); + let execution = fixture.backend.workers.lock().unwrap()[&worker_ref].clone(); + let old = execution.handle.clone(); + assert_eq!(execution.restore_operation_id, Some(operation)); + assert!( + fixture + .backend + .pending_restore(&worker_ref) + .unwrap() + .is_none() + ); + assert_eq!(execution.tasks.tasks.lock().unwrap().len(), 2); + assert!(execution.tasks.protocol_bridge.lock().unwrap().is_none()); + assert_eq!( + durable_operation(&runtime, &worker_ref) + .unwrap() + .operation_id, + operation + ); + assert_eq!( + durable_operation(&runtime, &worker_ref).unwrap().commit, + crate::runtime::RestoreCommitFact::Committed + ); + let owner = restore_owner(&runtime, &worker_ref, &request.request_id); + assert!(owner["receipt"].is_null()); + let (same_operation, context, uncertain) = runtime.test_pending_restore_context(&worker_ref); + assert_eq!(same_operation, operation); + assert!(!uncertain); + assert!(context.test_buffered_events().is_empty()); + let transport = assert_real_attach(&fixture, &old); + assert_single_publication(&fixture, 1); + alert_marker(&old, "terminal-before-receipt"); + let store = runtime.test_runtime_store(); + let reconcile_before = store.test_worker_reconcile_counts(); + store.fail_next_worker_write(if matches!(fault, TerminalFault::BeforeRename) { + AtomicWriteFault::BeforeRename + } else { + AtomicWriteFault::AfterRename + }); + match fault { + TerminalFault::SyncUnknown => store.fail_next_worker_reconcile_sync(), + TerminalFault::ReadUnknown => store.fail_next_worker_reconcile_read(), + _ => {} + } + release.send(()).unwrap(); + let first = restoring.join().unwrap().unwrap(); + let reconcile_after = store.test_worker_reconcile_counts(); + let reconciliation = ( + reconcile_after.0 - reconcile_before.0, + reconcile_after.1 - reconcile_before.1, + ); + assert_eq!( + reconciliation, + match fault { + TerminalFault::BeforeRename => (0, 0), + TerminalFault::AfterRename => (1, 4), // canonical file plus Worker/workers/root directories + TerminalFault::SyncUnknown => (1, 1), + TerminalFault::ReadUnknown => (1, 0), + }, + "{fault:?}: actual canonical read/sync attempts" + ); + assert_eq!( + first.state, + if matches!(fault, TerminalFault::AfterRename) { + server_api::WorkerRestoreState::Accepted + } else { + server_api::WorkerRestoreState::ReconciliationRequired + }, + "{fault:?}: {first:?}" + ); + assert_eq!(fixture.canonical_files(), canonical); + assert!(!old.protocol_is_closed()); + transport.validate().unwrap(); + assert_single_publication(&fixture, 1); + let disk_owner = restore_owner(&runtime, &worker_ref, &request.request_id); + assert_eq!(disk_owner["operation_id"], owner["operation_id"]); + match fault { + TerminalFault::BeforeRename => { + assert!(disk_owner["receipt"].is_null()); + assert!(!runtime.test_pending_restore_context(&worker_ref).2); + } + TerminalFault::AfterRename => { + assert_eq!(disk_owner["receipt"]["state"], "accepted"); + assert!(durable_operation(&runtime, &worker_ref).is_none()); + } + TerminalFault::SyncUnknown | TerminalFault::ReadUnknown => { + assert_eq!( + disk_owner["receipt"]["state"], "accepted", + "visible rename alone is not durable proof" + ); + let (same_operation, _, fenced) = runtime.test_pending_restore_context(&worker_ref); + assert_eq!(same_operation, operation); + assert!( + fenced, + "unknown canonical outcome must retain the memory owner/write fence" + ); + // Keep durability unproved for one competing Stop as well: it must + // fail before any Shutdown, never manufacture a Stopped judgment. + match fault { + TerminalFault::SyncUnknown => store.fail_next_worker_reconcile_sync(), + TerminalFault::ReadUnknown => store.fail_next_worker_reconcile_read(), + _ => unreachable!(), + } + assert!(matches!( + runtime.stop_worker_scoped(&scope, &worker_ref, None), + Err(crate::error::RuntimeError::StoreCommitOutcomeUnknown { .. }) + )); + assert!(!old.protocol_is_closed()); + assert!(!execution.shutdown_requested.load(Ordering::Acquire)); + assert!(runtime.test_pending_restore_context(&worker_ref).2); + assert_single_publication(&fixture, 1); + } + } + // Recovery is autonomous after the failed write / confirmed unknown fence. + // No POST, fresh request, or input is needed to make it converge. + drop(pause); + if !matches!(fault, TerminalFault::AfterRename) { + runtime.test_wait_restore_continuation(&worker_ref, operation, WAIT); + } + assert!(durable_operation(&runtime, &worker_ref).is_none()); + assert_eq!( + restore_owner(&runtime, &worker_ref, &request.request_id)["receipt"]["state"], + "accepted" + ); + let current = fixture.backend.workers.lock().unwrap()[&worker_ref].clone(); + assert!(old.same_controller(¤t.handle)); + assert_eq!(current.restore_operation_id, Some(operation)); + assert_eq!(fixture.backend.workers.lock().unwrap().len(), 1); + assert_real_attach(&fixture, &old); + assert_single_publication(&fixture, 1); + alert_marker(&old, "terminal-after-recovery"); + let _observations = observe_publication_markers( + &fixture, + observations, + &["terminal-before-receipt", "terminal-after-recovery"], + ); + let accepted = runtime + .restore_worker_operation_scoped(&scope, &worker_ref, request.clone()) + .unwrap(); + assert_eq!(accepted.state, server_api::WorkerRestoreState::Accepted); + assert_eq!( + runtime + .restore_worker_operation_scoped(&scope, &worker_ref, request.clone()) + .unwrap(), + accepted + ); + assert_eq!(restore_owner_count(&runtime, &worker_ref), 1); + let mut stale = request; + stale.request_id = "terminal-fresh-stale-intent".into(); + assert!(matches!( + runtime.restore_worker_operation_scoped(&scope, &worker_ref, stale), + Err(crate::error::RuntimeError::RestoreObservationConflict { .. }) + )); + assert_single_publication(&fixture, 1); + assert_eq!( + script.call_count.load(Ordering::SeqCst), + 0, + "receipt recovery cannot trigger input or a model call" + ); + // This transport was paired before the failed receipt write. Exercise its + // normal log-entry receiver after autonomous settlement, not a new attach. + transport + .dispatch(Method::submit_text( + "terminal-ordinary-input", + "terminal-ordinary-input", + )) + .unwrap(); + fixture + .backend + .run_on_adapter_runtime(async move { + let mut transport = transport; + let (mut admitted, mut committed, mut text, mut ended) = (false, false, false, false); + tokio::time::timeout(WAIT, async { + while !(admitted && committed && text && ended) { + match transport.events.recv().await.unwrap() { + Event::SubmissionAccepted { + submission_request_id, + .. + } if submission_request_id == "terminal-ordinary-input" => admitted = true, + Event::UserMessage { segments, .. } => { + assert_eq!( + segments, + vec![protocol::Segment::text("terminal-ordinary-input")] + ); + assert!(!committed, "paired log receiver duplicated input"); + committed = true; + } + Event::TextDone { text: output } => { + assert_eq!(output, "hello from worker"); + text = true; + } + Event::RunEnd { .. } => ended = true, + Event::SubmissionRejected { .. } | Event::Error { .. } => { + panic!("ordinary input failed after receipt recovery") + } + _ => {} + } + } + }) + .await + .map_err(|_| { + "ordinary input failed to settle on original paired transport".to_string() + })?; + drop(transport); + Ok(()) + }) + .unwrap(); + assert_eq!(script.call_count.load(Ordering::SeqCst), 1); + assert_eq!(fixture.entries().iter().filter(|entry| matches!(entry, + LogEntry::AnnotatedUserInput { segments, .. } if segments == &vec![protocol::Segment::text("terminal-ordinary-input")] + )).count(), 1); + assert_single_publication(&fixture, 1); + drop(execution); + drop(current); + runtime + .stop_worker_scoped(&scope, &worker_ref, None) + .unwrap(); + assert!(old.protocol_is_closed()); + fixture.assert_released(); +} + +#[test] +#[serial_test::serial(worker_allocation)] +fn published_controller_terminal_receipt_before_rename_autonomously_settles_without_reconstruction() +{ + published_terminal_receipt_fault(TerminalFault::BeforeRename); +} + +#[test] +#[serial_test::serial(worker_allocation)] +fn published_controller_terminal_receipt_after_rename_is_canonically_reread_and_synced() { + published_terminal_receipt_fault(TerminalFault::AfterRename); +} + +#[test] +#[serial_test::serial(worker_allocation)] +fn published_controller_terminal_receipt_sync_unknown_retains_owner_and_write_fence() { + published_terminal_receipt_fault(TerminalFault::SyncUnknown); +} + +#[test] +#[serial_test::serial(worker_allocation)] +fn published_controller_terminal_receipt_read_unknown_retains_owner_and_write_fence() { + published_terminal_receipt_fault(TerminalFault::ReadUnknown); +}