From cfffc80c36d63e8cd46f054d1f29dd27f084f462 Mon Sep 17 00:00:00 2001 From: Hare Date: Thu, 5 Nov 2026 00:57:32 +0900 Subject: [PATCH] fix: retain original controller termination and exact submission evidence --- crates/worker-runtime/src/execution.rs | 18 +- crates/worker-runtime/src/fs_store.rs | 9 +- crates/worker-runtime/src/runtime.rs | 153 ++++++-- .../runtime/execution_termination_tests.rs | 201 ++++++++++ crates/worker-runtime/src/worker_backend.rs | 344 ++++++++++++---- .../submission_termination_tests.rs | 368 ++++++++++++++++++ .../termination_persistence_tests.rs | 157 ++++++++ .../worker-runtime/src/working_directory.rs | 5 + 8 files changed, 1127 insertions(+), 128 deletions(-) create mode 100644 crates/worker-runtime/src/runtime/execution_termination_tests.rs create mode 100644 crates/worker-runtime/src/worker_backend/submission_termination_tests.rs create mode 100644 crates/worker-runtime/src/worker_backend/termination_persistence_tests.rs diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index 969908f2c..d53035cce 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -1157,6 +1157,18 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static { Err("execution backend has no scope-termination receipt reader".into()) } + /// Reconcile canonical execution-owner evidence against current backend + /// ownership. The default cannot certify a receipt lost by this backend. + fn reconcile_termination_receipt( + &self, + worker_ref: &WorkerRef, + saved: &WorkerExecutionTermination, + ) -> Result, String> { + Ok(self + .termination_receipt(worker_ref)? + .filter(|current| current == saved)) + } + fn dispatch_input(&self, worker_ref: &WorkerRef, input: WorkerInput) -> WorkerExecutionResult; fn upload_file( @@ -1458,11 +1470,13 @@ impl WorkerExecutionBackendRef { self.backend.reconcile_submission(worker_ref, input) } - pub(crate) fn termination_receipt( + pub(crate) fn reconcile_termination_receipt( &self, worker_ref: &WorkerRef, + saved: &WorkerExecutionTermination, ) -> Result, String> { - self.backend.termination_receipt(worker_ref) + self.backend + .reconcile_termination_receipt(worker_ref, saved) } pub(crate) fn dispatch_input( diff --git a/crates/worker-runtime/src/fs_store.rs b/crates/worker-runtime/src/fs_store.rs index 28f005b10..e7440139b 100644 --- a/crates/worker-runtime/src/fs_store.rs +++ b/crates/worker-runtime/src/fs_store.rs @@ -43,7 +43,8 @@ use std::sync::atomic::{AtomicU8, AtomicU64, Ordering}; #[path = "test_support.rs"] pub mod test_support; -pub(crate) const SCHEMA_VERSION: u32 = 15; +pub(crate) const SCHEMA_VERSION: u32 = 16; +const SESSION_ARCHIVE_SCHEMA_VERSION: u32 = 15; const RESTORE_REPAIR_SCHEMA_VERSION: u32 = 14; const RESTORE_DECISION_SCHEMA_VERSION: u32 = 13; const RESTORE_ADMISSION_SCHEMA_VERSION: u32 = 12; @@ -912,11 +913,12 @@ fn plan_runtime_store_migration( | RESTORE_ADMISSION_SCHEMA_VERSION | RESTORE_DECISION_SCHEMA_VERSION | RESTORE_REPAIR_SCHEMA_VERSION + | SESSION_ARCHIVE_SCHEMA_VERSION ) { return Err(runtime_store_corrupt( &runtime_path, format!( - "unsupported Runtime store schema version {schema_version}; expected {OLDEST_SCHEMA_VERSION}, {LEGACY_SCHEMA_VERSION}, {PREVIOUS_SCHEMA_VERSION}, {AGGREGATE_SCHEMA_VERSION}, {EXECUTION_SCHEMA_VERSION}, {SOURCE_SCHEMA_VERSION}, {RESTORE_ADMISSION_SCHEMA_VERSION}, {RESTORE_DECISION_SCHEMA_VERSION}, {RESTORE_REPAIR_SCHEMA_VERSION}, or {SCHEMA_VERSION}" + "unsupported Runtime store schema version {schema_version}; expected {OLDEST_SCHEMA_VERSION}, {LEGACY_SCHEMA_VERSION}, {PREVIOUS_SCHEMA_VERSION}, {AGGREGATE_SCHEMA_VERSION}, {EXECUTION_SCHEMA_VERSION}, {SOURCE_SCHEMA_VERSION}, {RESTORE_ADMISSION_SCHEMA_VERSION}, {RESTORE_DECISION_SCHEMA_VERSION}, {RESTORE_REPAIR_SCHEMA_VERSION}, {SESSION_ARCHIVE_SCHEMA_VERSION}, or {SCHEMA_VERSION}" ), )); } @@ -1253,6 +1255,7 @@ fn migrate_worker_record( | RESTORE_ADMISSION_SCHEMA_VERSION | RESTORE_DECISION_SCHEMA_VERSION | RESTORE_REPAIR_SCHEMA_VERSION + | SESSION_ARCHIVE_SCHEMA_VERSION ) { // Admission identity survives migration. Older pending records did not // distinguish a committed decision from preparation or failed cleanup. @@ -1744,6 +1747,7 @@ fn migrate_runtime_document( | RESTORE_ADMISSION_SCHEMA_VERSION | RESTORE_DECISION_SCHEMA_VERSION | RESTORE_REPAIR_SCHEMA_VERSION + | SESSION_ARCHIVE_SCHEMA_VERSION ) { object.remove("workers"); // This legacy cache is not Runtime lifecycle or restore authority. Bundles @@ -2295,6 +2299,7 @@ fn migrate_runtime_store_in_place( | RESTORE_ADMISSION_SCHEMA_VERSION | RESTORE_DECISION_SCHEMA_VERSION | RESTORE_REPAIR_SCHEMA_VERSION + | SESSION_ARCHIVE_SCHEMA_VERSION ) { atomic_write_json( diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 01c8445a6..fb9e8d75e 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -2400,6 +2400,7 @@ impl Runtime { last_settled_status: candidate.status, }); let operation_id = pending.operation_id; + candidate.restore_guard.execution_termination = None; candidate.pending_restore = Some(pending); candidate.restore_guard.rotate(); candidate.restore_guard.active_request_id = Some(request.request_id.clone()); @@ -3265,30 +3266,52 @@ impl Runtime { let _guard = operation_lock .lock() .map_err(|_| RuntimeError::StatePoisoned)?; - let backend = { - let state = self.lock()?; + let (backend, saved) = { + let mut state = self.lock()?; state.ensure_worker_ref(worker_ref)?; - state.worker(worker_ref)?; - state.execution_backend.clone().ok_or_else(|| { + // Unknown save acknowledgements must resolve against the same + // canonical operation before exposing its termination evidence. + state.refresh_uncertain_worker_write(worker_ref)?; + let worker = state.worker(worker_ref)?; + let saved = worker + .restore_guard + .execution_termination + .clone() + .filter(|receipt| { + worker.pending_stop.map_or_else( + || { + worker.status == WorkerStatus::Stopped + && worker.pending_restore.is_none() + }, + |pending| pending.operation_id == receipt.stop_operation_id, + ) + }); + let backend = state.execution_backend.clone().ok_or_else(|| { RuntimeError::WorkerExecutionUnavailable { worker_id: worker_ref.worker_id, message: "runtime has no execution backend".into(), } - })? + })?; + (backend, saved) }; - let receipt = backend.termination_receipt(worker_ref).map_err(|message| { - RuntimeError::WorkerExecutionUnavailable { + let Some(saved) = saved else { + // A backend map or stopped observation alone is not durable evidence. + return Ok(None); + }; + let receipt = backend + .reconcile_termination_receipt(worker_ref, &saved) + .map_err(|message| RuntimeError::WorkerExecutionUnavailable { worker_id: worker_ref.worker_id, message, - } - })?; - if receipt - .as_ref() - .is_some_and(|receipt| &receipt.worker_ref != worker_ref) - { + })?; + if receipt.as_ref().is_some_and(|receipt| { + !receipt.confirms_scope_shutdown(worker_ref, receipt.stop_operation_id) + || &saved != receipt + }) { return Err(RuntimeError::WorkerExecutionUnavailable { worker_id: worker_ref.worker_id, - message: "scope-termination receipt Worker mismatch".into(), + message: "scope-termination receipt does not match canonical operation evidence" + .into(), }); } Ok(receipt) @@ -4354,15 +4377,6 @@ impl Runtime { operation_id, worker_ref: worker_ref.clone(), }); - if !result.is_accepted() { - return Err(RuntimeError::WorkerExecutionRejected { - worker_id: worker_ref.worker_id, - operation: result.operation, - outcome: result.outcome, - message: result.message_or_default(), - result, - }); - } self.commit_stopped_worker(worker_ref, operation_id, result) } @@ -5102,15 +5116,6 @@ impl Runtime { operation_id, worker_ref: worker_ref.clone(), }); - if !result.is_accepted() { - return Err(RuntimeError::WorkerExecutionRejected { - worker_id: worker_ref.worker_id, - operation: result.operation, - outcome: result.outcome, - message: result.message_or_default(), - result, - }); - } self.commit_stopped_worker(worker_ref, operation_id, result)?; Ok(()) } @@ -5144,6 +5149,7 @@ impl Runtime { ))); } candidate.restore_guard.rotate(); + candidate.restore_guard.execution_termination = None; candidate.pending_stop = Some(PendingWorkerStop { operation_id, last_settled_status: candidate.status, @@ -5179,12 +5185,25 @@ impl Runtime { operation_id: WorkerLifecycleOperationId, mut result: WorkerExecutionResult, ) -> Result { - if !result.is_accepted() - || result.operation != WorkerExecutionOperation::Stop - || result.termination.as_ref().is_none_or(|termination| { - !termination.confirms_scope_shutdown(worker_ref, operation_id) + let termination = result + .termination + .as_ref() + .filter(|termination| { + result.operation == WorkerExecutionOperation::Stop + && result.execution_id == Some(termination.execution_id) + && termination.confirms_scope_shutdown(worker_ref, operation_id) }) - { + .cloned(); + if !result.is_accepted() || termination.is_none() { + if let Some(termination) = termination { + // Scope join is independent of an artifact/resource cleanup error. + // Retain the original evidence without settling the pending Stop. + let mut state = self.lock()?; + let mut candidate = state.worker(worker_ref)?.clone(); + candidate.retain_stop_termination(operation_id, termination)?; + state.persist_stop_worker_record(&candidate)?; + state.workers.insert(worker_ref.worker_id, candidate); + } if result.is_accepted() { result.outcome = crate::execution::WorkerExecutionOutcome::Errored; result.message = Some("stop has no matching actual scope termination evidence; reconciliation remains pending".into()); @@ -5212,6 +5231,8 @@ impl Runtime { if let Some(context) = candidate.restore_candidate_context.take() { context.discard_candidate(); } + candidate + .retain_stop_termination(operation_id, termination.expect("validated termination"))?; let (candidate, ack) = candidate.stopped_candidate(operation_id); state.persist_stop_worker_record(&candidate)?; state.workers.insert(worker_ref.worker_id, candidate); @@ -5264,6 +5285,7 @@ impl Runtime { owner.execution_started = true; } candidate.restore_guard.rotate(); + candidate.restore_guard.execution_termination = None; candidate.pending_restore = Some(PendingWorkerRestore { operation_id, mode, @@ -7509,6 +7531,10 @@ pub(crate) struct RestoreGuard { // eviction would forget request-ID conflicts and historical receipts. #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] stop_owners: BTreeMap, + // Original execution-owner evidence, independent of lifecycle/cleanup success. + // Older aggregates lack it; neither migration nor a stopped status creates it. + #[serde(default, skip_serializing_if = "Option::is_none")] + execution_termination: Option, } impl Default for RestoreGuard { @@ -7518,6 +7544,7 @@ impl Default for RestoreGuard { active_request_id: None, owners: BTreeMap::new(), stop_owners: BTreeMap::new(), + execution_termination: None, } } } @@ -7579,6 +7606,29 @@ struct WorkerRecord { } impl WorkerRecord { + fn retain_stop_termination( + &mut self, + operation_id: WorkerLifecycleOperationId, + termination: crate::execution::WorkerExecutionTermination, + ) -> Result<(), RuntimeError> { + if self + .pending_stop + .is_none_or(|pending| pending.operation_id != operation_id) + || !termination.confirms_scope_shutdown(&self.worker_ref, operation_id) + || self + .restore_guard + .execution_termination + .as_ref() + .is_some_and(|saved| saved != &termination) + { + return Err(RuntimeError::InvalidRequest( + "scope termination differs from the canonical Stop operation".into(), + )); + } + self.restore_guard.execution_termination = Some(termination); + Ok(()) + } + // Pure projection shared by terminal commit and stop-owner capacity preflight. // Resource cleanup is deliberately outside this method. fn stopped_candidate( @@ -10329,6 +10379,19 @@ mod tests { result } + fn termination_receipt( + &self, + worker_ref: &WorkerRef, + ) -> Result, String> { + Ok(self + .stop_terminations + .lock() + .unwrap() + .values() + .find(|receipt| &receipt.worker_ref == worker_ref) + .cloned()) + } + fn stop_worker_operation( &self, request: WorkerExecutionStopRequest, @@ -12417,6 +12480,9 @@ mod tests { assert_eq!(ids[0], ids[1]); } + #[cfg(feature = "fs-store")] + include!("runtime/execution_termination_tests.rs"); + #[cfg(feature = "fs-store")] #[test] fn legacy_restore_guard_without_stop_owners_reopens_without_losing_observation() { @@ -16318,9 +16384,16 @@ mod tests { display_name: None, }) .unwrap_err(); - assert!(error.to_string().contains( - "unsupported Runtime store schema version 2; expected 6, 7, 8, 9, 10, 11, 12, 13, 14, or 15" - )); + assert!( + error + .to_string() + .contains("unsupported Runtime store schema version 2") + ); + assert!( + error + .to_string() + .contains(&format!("or {}", crate::fs_store::SCHEMA_VERSION)) + ); let _ = std::fs::remove_dir_all(root); } diff --git a/crates/worker-runtime/src/runtime/execution_termination_tests.rs b/crates/worker-runtime/src/runtime/execution_termination_tests.rs new file mode 100644 index 000000000..02ed8bea3 --- /dev/null +++ b/crates/worker-runtime/src/runtime/execution_termination_tests.rs @@ -0,0 +1,201 @@ +// These catalog tests substitute the execution response. The real Controller +// join and fs-store reopen connection is covered in worker_backend tests. +#[test] +fn stop_termination_unknown_save_reconciles_original_operation_before_and_after_reopen() { + for reconcile_in_process in [true, false] { + let dir = tempfile::tempdir().unwrap(); + let options = crate::fs_store::FsRuntimeStoreOptions::new(dir.path()) + .with_runtime_id("termination-unknown-save"); + let backend = Arc::new(TestExecutionBackend::default()); + let runtime = + Runtime::with_fs_store_and_execution_backend(options.clone(), backend.clone()).unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let worker = runtime + .create_worker(task_request("original receipt")) + .unwrap(); + let operation_id = WorkerLifecycleOperationId::new(); + runtime + .begin_stop_operation(&worker.worker_ref, operation_id) + .unwrap(); + let result = backend.stop_worker_operation(WorkerExecutionStopRequest { + worker_ref: worker.worker_ref.clone(), + operation_id, + }); + let original = result.termination.clone().unwrap(); + runtime_store(&runtime) + .fail_next_worker_write(crate::fs_store::AtomicWriteFault::AfterRename); + assert!(matches!( + runtime.commit_stopped_worker(&worker.worker_ref, operation_id, result), + Err(RuntimeError::StoreCommitOutcomeUnknown { .. }) + )); + if reconcile_in_process { + runtime_store(&runtime).fail_next_worker_reconcile_sync(); + assert!( + runtime.execution_termination(&worker.worker_ref).is_err(), + "unknown durability must not fall back to the backend's in-memory receipt" + ); + assert_eq!( + runtime.execution_termination(&worker.worker_ref).unwrap(), + Some(original.clone()) + ); + assert_eq!( + runtime.worker_detail(&worker.worker_ref).unwrap().status, + WorkerStatus::Stopped + ); + } + assert_eq!( + backend.stop_operation_ids.lock().unwrap().as_slice(), + &[operation_id] + ); + drop(runtime); + drop(backend); + let backend = Arc::new( + crate::worker_backend::WorkerRuntimeExecutionBackend::new( + crate::worker_backend::ProfileRuntimeWorkerFactory::new( + dir.path().join("no-launch"), + ), + ) + .unwrap(), + ); + let reopened = Runtime::with_fs_store_and_execution_backend(options, backend).unwrap(); + assert_eq!( + reopened.execution_termination(&worker.worker_ref).unwrap(), + Some(original) + ); + assert!(!dir.path().join("no-launch").exists()); + } +} + +#[test] +fn stop_termination_rejects_missing_wrong_operation_run_worker_and_feature_failure() { + for invalid in ["missing", "operation", "execution", "worker", "feature"] { + let (runtime, backend) = runtime_and_backend(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let worker = runtime.create_worker(task_request(invalid)).unwrap(); + let operation_id = WorkerLifecycleOperationId::new(); + runtime + .begin_stop_operation(&worker.worker_ref, operation_id) + .unwrap(); + let mut result = backend.stop_worker_operation(WorkerExecutionStopRequest { + worker_ref: worker.worker_ref.clone(), + operation_id, + }); + match invalid { + "missing" => result.termination = None, + "operation" => { + result.termination.as_mut().unwrap().stop_operation_id = + WorkerLifecycleOperationId::new() + } + "execution" => result.execution_id = Some(uuid::Uuid::now_v7()), + "worker" => { + result.termination.as_mut().unwrap().worker_ref = WorkerRef::new(WorkerId::now_v7()) + } + "feature" => result.termination.as_mut().unwrap().feature_failures.push( + crate::execution::WorkerFeatureShutdownFailure { + category: worker::hook::HookErrorCategory::Timeout, + diagnostic: "Feature did not terminate".into(), + }, + ), + _ => unreachable!(), + } + assert!( + runtime + .commit_stopped_worker(&worker.worker_ref, operation_id, result) + .is_err(), + "{invalid}" + ); + assert!( + runtime + .execution_termination(&worker.worker_ref) + .unwrap() + .is_none(), + "{invalid}" + ); + assert_eq!( + runtime + .lock() + .unwrap() + .worker(&worker.worker_ref) + .unwrap() + .pending_stop + .unwrap() + .operation_id, + operation_id + ); + } +} + +#[test] +fn stop_termination_cleanup_error_retains_join_without_settling_stop_or_overwriting_receipt() { + let dir = tempfile::tempdir().unwrap(); + let options = crate::fs_store::FsRuntimeStoreOptions::new(dir.path()) + .with_runtime_id("termination-cleanup"); + let backend = Arc::new(TestExecutionBackend::default()); + let runtime = + Runtime::with_fs_store_and_execution_backend(options.clone(), backend.clone()).unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let worker = runtime + .create_worker(task_request("cleanup failed")) + .unwrap(); + let operation_id = WorkerLifecycleOperationId::new(); + runtime + .begin_stop_operation(&worker.worker_ref, operation_id) + .unwrap(); + let mut result = backend.stop_worker_operation(WorkerExecutionStopRequest { + worker_ref: worker.worker_ref.clone(), + operation_id, + }); + result.outcome = crate::execution::WorkerExecutionOutcome::Errored; + result.message = Some("artifact cleanup failed after scope join".into()); + let original = result.termination.clone().unwrap(); + assert!( + matches!(runtime.commit_stopped_worker(&worker.worker_ref, operation_id, result.clone()), + Err(RuntimeError::WorkerExecutionRejected { result: rejected, .. }) if rejected == result) + ); + assert_eq!( + runtime.execution_termination(&worker.worker_ref).unwrap(), + Some(original.clone()) + ); + let mut changed = result.clone(); + changed + .termination + .as_mut() + .unwrap() + .submissions + .push(crate::execution::WorkerSubmissionAck { + submission_request_id: "not-original".into(), + submission_id: "other".into(), + disposition: protocol::SubmissionDisposition::Started, + }); + assert!(matches!( + runtime.commit_stopped_worker(&worker.worker_ref, operation_id, changed), + Err(RuntimeError::InvalidRequest(_)) + )); + assert_eq!( + runtime.execution_termination(&worker.worker_ref).unwrap(), + Some(original.clone()) + ); + drop(runtime); + let backend = Arc::new( + crate::worker_backend::WorkerRuntimeExecutionBackend::new( + crate::worker_backend::ProfileRuntimeWorkerFactory::new(dir.path().join("no-launch")), + ) + .unwrap(), + ); + let reopened = Runtime::with_fs_store_and_execution_backend(options, backend).unwrap(); + assert_eq!( + reopened.execution_termination(&worker.worker_ref).unwrap(), + Some(original) + ); + assert_eq!( + reopened + .lock() + .unwrap() + .worker(&worker.worker_ref) + .unwrap() + .pending_stop + .unwrap() + .operation_id, + operation_id + ); +} diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 52b2bc5f1..beba4493a 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -158,8 +158,6 @@ use workdir::{ WorkdirSessionHandle, WorkdirSessionRouter, }; -#[cfg(test)] -use worker::WorkerController; use worker::feature::builtin::manage_workdir::WorkspaceAttachedWorkdirSession; use worker::feature::contracts::{ WorkerObservationError, WorkerObservationProvider, WorkerObservationSubject, @@ -3290,6 +3288,9 @@ struct RuntimeExecutionTaskScope { execution_id: uuid::Uuid, shutdown_report: Arc>>>, submissions: Arc>>, + // Dispatch provenance belongs to this original Controller scope, not its + // WorkerRef or the session receipts inherited by a replacement Controller. + submission_attempts: Arc>>, #[cfg(feature = "ws-server")] protocol_bridge: Arc>>, tasks: Arc>>>, @@ -3392,6 +3393,7 @@ impl RuntimeExecutionTaskScope { execution_id: uuid::Uuid::now_v7(), shutdown_report: Arc::default(), submissions: Arc::default(), + submission_attempts: Arc::default(), #[cfg(feature = "ws-server")] protocol_bridge: Arc::default(), tasks: Arc::new(Mutex::new(Vec::new())), @@ -3402,6 +3404,83 @@ impl RuntimeExecutionTaskScope { scope } + fn submission_fingerprint( + input: &[Segment], + source: &protocol::AuthenticatedInputSource, + ) -> Result<[u8; 32], String> { + use sha2::Digest as _; + let payload = serde_json::to_vec(&(input, source)) + .map_err(|_| "cannot bind original Controller submission payload".to_string())?; + Ok(sha2::Sha256::digest(payload).into()) + } + + fn prepare_submission( + &self, + handle: &WorkerHandle, + request_id: &str, + input: &[Segment], + source: &protocol::AuthenticatedInputSource, + ) -> Result<[u8; 32], String> { + let fingerprint = Self::submission_fingerprint(input, source)?; + let mut attempts = self + .submission_attempts + .lock() + .map_err(|_| "original Controller submission ownership is poisoned".to_string())?; + // An inherited receipt can be read or replayed, but neither attributes + // its original acceptance to this new Controller. None alone grants no + // acceptance/replay: bind only the actual dispatch that follows, then + // require the same handle's exact durable receipt before retaining an ack. + if !attempts.contains_key(request_id) + && matches!( + handle.reconcile_submission(request_id, input, source), + Ok(None) + ) + { + attempts.insert(request_id.to_string(), fingerprint); + } + Ok(fingerprint) + } + + fn retain_submission_acceptance( + &self, + ack: &crate::execution::WorkerSubmissionAck, + fingerprint: &[u8; 32], + ) -> Result<(), String> { + let attempts = self + .submission_attempts + .lock() + .map_err(|_| "original Controller submission ownership is poisoned".to_string())?; + if attempts.get(&ack.submission_request_id) != Some(fingerprint) { + return Ok(()); + } + let mut submissions = self + .submissions + .lock() + .map_err(|_| "original Controller submission receipts are poisoned".to_string())?; + if let Some(saved) = submissions + .iter() + .find(|saved| saved.submission_request_id == ack.submission_request_id) + { + if saved != ack { + return Err("original Controller submission acceptance changed".into()); + } + } else { + // A joined scope may already have an original canonical receipt, + // including a pending Stop after resource-cleanup failure. A later + // read must never amend that immutable evidence. + if self + .shutdown_report + .lock() + .map_err(|_| "Controller shutdown report lock is poisoned".to_string())? + .is_some() + { + return Ok(()); + } + submissions.push(ack.clone()); + } + Ok(()) + } + fn has_cleanup_candidates(&self) -> Result { self.cleanup_candidates .lock() @@ -3793,6 +3872,8 @@ struct WorkerAdmission { pub struct WorkerRuntimeExecutionBackend { #[cfg(test)] bridge_fault: Mutex>>, + #[cfg(test)] + lost_submission_ack: Mutex>, backend_id: String, factory: Arc, working_directory_materializer: Option>, @@ -3833,6 +3914,8 @@ where Ok(Self { #[cfg(test)] bridge_fault: Mutex::default(), + #[cfg(test)] + lost_submission_ack: Mutex::default(), backend_id: DEFAULT_BACKEND_ID.to_string(), factory: Arc::new(factory), working_directory_materializer: None, @@ -4550,6 +4633,8 @@ where submission_request_id: String, ) -> WorkerExecutionResult { let request_id = submission_request_id.clone(); + #[cfg(test)] + let lost_ack = self.lost_submission_ack.lock().unwrap().take(); self.run_cancellable_on_adapter_runtime(USER_INPUT_TASK_TIMEOUT, async move { // Subscribe before enqueueing so a fast durable acceptance cannot // race the Runtime acknowledgement. @@ -4567,6 +4652,12 @@ where submission_id, disposition, }) if submission_request_id == request_id => { + #[cfg(test)] + if lost_ack.as_deref() == Some(request_id.as_str()) { + // Lose only the caller's delivery after the real + // Controller emitted its durable acceptance. + continue; + } return Ok((submission_id, disposition)); } Ok(Event::SubmissionRejected { @@ -5134,6 +5225,57 @@ where Ok(()) } + fn read_termination_receipt( + &self, + worker_ref: &WorkerRef, + saved: Option<&crate::execution::WorkerExecutionTermination>, + ) -> Result, String> { + let operation = self.worker_lock(worker_ref)?; + let _guard = operation + .lock() + .map_err(|_| "Worker operation lock poisoned".to_string())?; + if self + .pending_spawns + .lock() + .map_err(|_| "Worker spawn registry poisoned".to_string())? + .contains_key(worker_ref) + || self + .pending_restores + .lock() + .map_err(|_| "Worker restore registry poisoned".to_string())? + .contains_key(worker_ref) + || self + .subject_reservations + .lock() + .map_err(|_| "Subject reservation registry poisoned".to_string())? + .keys() + .any(|(worker, _)| worker == worker_ref) + { + return Ok(None); + } + let current_execution = self + .workers + .lock() + .map_err(|_| "Worker execution registry poisoned".to_string())? + .get(worker_ref) + .map(|execution| execution.tasks.execution_id); + let receipt = self + .completed_stops + .lock() + .map_err(|_| "Worker termination receipts poisoned".to_string())? + .get(worker_ref) + .cloned(); + if let (Some(current), Some(saved)) = (&receipt, saved) + && current != saved + { + return Err("retained scope termination differs from canonical Stop receipt".into()); + } + let receipt = receipt.or_else(|| saved.cloned()); + // The positive receipt comes only from the joined scope, never from + // absence. A Restore's new Controller cannot reuse its predecessor's end. + Ok(receipt.filter(|receipt| current_execution.is_none_or(|id| id == receipt.execution_id))) + } + fn stop_worker_operation_under_lock( &self, request: WorkerExecutionStopRequest, @@ -7320,44 +7462,20 @@ where &self, worker_ref: &WorkerRef, ) -> Result, String> { - let operation = self.worker_lock(worker_ref)?; - let _guard = operation - .lock() - .map_err(|_| "Worker operation lock poisoned".to_string())?; - if self - .pending_spawns - .lock() - .map_err(|_| "Worker spawn registry poisoned".to_string())? - .contains_key(worker_ref) - || self - .pending_restores - .lock() - .map_err(|_| "Worker restore registry poisoned".to_string())? - .contains_key(worker_ref) - || self - .subject_reservations - .lock() - .map_err(|_| "Subject reservation registry poisoned".to_string())? - .keys() - .any(|(worker, _)| worker == worker_ref) - { - return Ok(None); + self.read_termination_receipt(worker_ref, None) + } + + fn reconcile_termination_receipt( + &self, + worker_ref: &WorkerRef, + saved: &crate::execution::WorkerExecutionTermination, + ) -> Result, String> { + if !saved.confirms_scope_shutdown(worker_ref, saved.stop_operation_id) { + return Err( + "canonical scope termination has a Worker mismatch or Feature failure".into(), + ); } - let current_execution = self - .workers - .lock() - .map_err(|_| "Worker execution registry poisoned".to_string())? - .get(worker_ref) - .map(|execution| execution.tasks.execution_id); - let receipt = self - .completed_stops - .lock() - .map_err(|_| "Worker termination receipts poisoned".to_string())? - .get(worker_ref) - .cloned(); - // The positive receipt comes only from the joined scope, never from - // absence. A Restore's new Controller cannot reuse its predecessor's end. - Ok(receipt.filter(|receipt| current_execution.is_none_or(|id| id == receipt.execution_id))) + self.read_termination_receipt(worker_ref, Some(saved)) } fn reconcile_submission( @@ -7443,8 +7561,22 @@ where "exact submission receipt lookup failed; submission acceptance remains unknown", ), }; + if result.is_accepted() + && let Some(ack) = result.submission.as_ref() + && let Err(message) = RuntimeExecutionTaskScope::submission_fingerprint( + &segments, &source, + ) + .and_then(|fingerprint| { + execution + .tasks + .retain_submission_acceptance(ack, &fingerprint) + }) + { + result = unknown(&message); + } // Identify the Controller observed, not the originating run of a receipt - // retained across Controller replacement. This is not termination evidence. + // retained across Controller replacement. Only this scope's original + // dispatch binding above permits adding it to termination evidence. result.execution_id = Some(execution.tasks.execution_id); result } @@ -7469,13 +7601,27 @@ where error.to_string(), ); } - let (worker, worker_state, _workspace_client) = match self.get_execution(worker_ref) { - Ok(execution) => execution, - Err(mut result) => { - result.operation = WorkerExecutionOperation::Input; - return result; + // Pin the actual endpoint and scope while holding the per-Worker lock; + // do not attach a late acceptance through a fresh WorkerRef lookup. + let execution = match self.workers.lock() { + Ok(workers) => match workers.get(worker_ref) { + Some(execution) => execution.clone(), + None => { + return WorkerExecutionResult::rejected( + WorkerExecutionOperation::Input, + "WorkerRef does not reference a live Worker execution", + ); + } + }, + Err(_) => { + return WorkerExecutionResult::errored( + WorkerExecutionOperation::Input, + "worker adapter registry lock is poisoned", + ); } }; + let worker = execution.handle.clone(); + let worker_state = execution.worker_state.clone(); if input.kind == WorkerInputKind::Notify { let notification_request_id = input @@ -7512,6 +7658,7 @@ where ); } + let mut submission_binding = None; let (method, submission_request_id) = match input.kind { WorkerInputKind::User | WorkerInputKind::UserIfIdle => { let Some(submission_id) = input @@ -7526,6 +7673,20 @@ where ); }; let (segments, source) = runtime_submission_payload(&input, &submission_id); + submission_binding = match execution.tasks.prepare_submission( + &worker, + &submission_id, + &segments, + &source, + ) { + Ok(fingerprint) => Some(fingerprint), + Err(message) => { + return WorkerExecutionResult::errored( + WorkerExecutionOperation::Input, + message, + ); + } + }; let method = if input.kind == WorkerInputKind::UserIfIdle { Method::SubmitIfIdle { submission_request_id: submission_id.clone(), @@ -7566,27 +7727,17 @@ where } else { self.send_method(WorkerExecutionOperation::Input, worker, method) }; - if let Some(execution) = self - .workers - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .get(worker_ref) + if result.is_accepted() + && let Some(ack) = result.submission.as_ref() + && Some(&ack.submission_request_id) == submission_request_id.as_ref() + && let Some(fingerprint) = submission_binding + && let Err(message) = execution + .tasks + .retain_submission_acceptance(ack, &fingerprint) { - result.execution_id = Some(execution.tasks.execution_id); - if result.is_accepted() - && let Some(ack) = result.submission.as_ref() - && Some(&ack.submission_request_id) == submission_request_id.as_ref() - { - let mut submissions = execution - .tasks - .submissions - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - if !submissions.contains(ack) { - submissions.push(ack.clone()); - } - } + result = WorkerExecutionResult::errored(WorkerExecutionOperation::Input, message); } + result.execution_id = Some(execution.tasks.execution_id); result } @@ -8004,6 +8155,8 @@ mod tests { include!("subject_delete_backend_tests.rs"); include!("worker_backend/startup_binding_tests.rs"); + include!("worker_backend/termination_persistence_tests.rs"); + include!("worker_backend/submission_termination_tests.rs"); #[cfg(feature = "fs-store")] include!("subject_delete_create_tests.rs"); @@ -11263,20 +11416,11 @@ mod tests { observed_workspace_clients: Arc, bool)>>>, } - #[async_trait] - impl RuntimeWorkerFactory for MockFactory { - async fn prepare_restore( - &self, - request: WorkerExecutionRestoreRequest, - ) -> Result, RuntimeRestoreFailure> { - self.restore_controller(request).await.map(|controller| { - Box::new(TestRestoreCandidate(Some(controller))) as Box - }) - } - - async fn spawn_controller( + impl MockFactory { + async fn spawn_controller_with_saved_source( &self, request: WorkerExecutionSpawnRequest, + save_restore_source: bool, ) -> Result { let manifest = WorkerManifest::from_toml( r#" @@ -11356,7 +11500,7 @@ mod tests { workspace_client.is_available(), )); let scope = Scope::writable(&scope_root).map_err(|err| err.to_string())?; - let worker = Worker::new( + let mut worker = Worker::new( manifest, Engine::<_, agen::state::Mutable, worker::SessionHistoryMetadata>::new_annotated( self.client.clone(), @@ -11368,18 +11512,50 @@ mod tests { ) .await .map_err(|err| err.to_string())?; - let (handle, shutdown_rx) = - WorkerController::spawn_runtime_managed(worker, &self.runtime_base) - .await - .map_err(|err| err.to_string())?; + if save_restore_source { + worker + .engine_mut() + .set_system_prompt("saved original Controller fixture prompt"); + worker + .enable_worker_metadata_write_through() + .map_err(|error| error.to_string())?; + } + let started = worker::bootstrap::start_worker_controller( + worker, + worker::bootstrap::WorkerBootstrapLayout::Direct { + runtime_base: self.runtime_base.clone(), + }, + ) + .await + .map_err(|err| err.to_string())?; Ok(RuntimeWorkerController { - handle, - shutdown: Arc::new(tokio::sync::Mutex::new(Some(shutdown_rx))), - controller_task: tokio::spawn(async {}), + handle: started.handle, + shutdown: Arc::new(tokio::sync::Mutex::new(Some(started.shutdown))), + controller_task: started.controller_task, workspace_client, restore_streams: None, }) } + } + + #[async_trait] + impl RuntimeWorkerFactory for MockFactory { + async fn prepare_restore( + &self, + request: WorkerExecutionRestoreRequest, + ) -> Result, RuntimeRestoreFailure> { + self.restore_controller(request).await.map(|controller| { + Box::new(TestRestoreCandidate(Some(controller))) as Box + }) + } + + async fn spawn_controller( + &self, + request: WorkerExecutionSpawnRequest, + ) -> Result { + self.spawn_controller_with_saved_source(request, false) + .await + } async fn restore_controller( &self, request: WorkerExecutionRestoreRequest, diff --git a/crates/worker-runtime/src/worker_backend/submission_termination_tests.rs b/crates/worker-runtime/src/worker_backend/submission_termination_tests.rs new file mode 100644 index 000000000..4cc505d47 --- /dev/null +++ b/crates/worker-runtime/src/worker_backend/submission_termination_tests.rs @@ -0,0 +1,368 @@ +// Connect real Submit/receipt persistence, original Controller ownership and +// Runtime's canonical stop receipt. Only model I/O and one caller ACK delivery +// are substituted; no receipt, Shutdown report or joined Controller is faked. +#[cfg(feature = "fs-store")] +struct SavedSourceMockFactory(MockFactory); + +#[cfg(feature = "fs-store")] +#[async_trait] +impl RuntimeWorkerFactory for SavedSourceMockFactory { + async fn spawn_controller( + &self, + request: WorkerExecutionSpawnRequest, + ) -> Result { + self.0 + .spawn_controller_with_saved_source(request, true) + .await + } + + async fn prepare_restore( + &self, + _: WorkerExecutionRestoreRequest, + ) -> Result, RuntimeRestoreFailure> { + panic!("this fixture explicitly restores the original saved Session below") + } +} + +#[cfg(feature = "fs-store")] +#[test] +#[serial_test::serial(worker_allocation)] +fn stop_termination_lost_submit_ack_reconciles_only_original_controller_after_timeout() { + let temp = tempfile::tempdir_in(".").unwrap(); + let root = fs::canonicalize(temp.path()).unwrap(); + let socket_base = PathBuf::from(temp.path().file_name().unwrap()).join("c"); + let options = crate::fs_store::FsRuntimeStoreOptions::new(root.join("catalog")) + .with_runtime_id("lost-submit-ack"); + let client = MockClient::new(simple_text_events()); + let calls = client.call_count.clone(); + let new_backend = || { + Arc::new( + WorkerRuntimeExecutionBackend::new(SavedSourceMockFactory(MockFactory { + client: client.clone(), + runtime_base: socket_base.clone(), + cwd: root.clone(), + store_dir: root.join("sessions"), + worker_metadata_dir: root.join("metadata"), + observed_cwds: Arc::default(), + observed_workspace_clients: Arc::default(), + })) + .unwrap(), + ) + }; + let backend = new_backend(); + let runtime = + EmbeddedRuntime::with_fs_store_and_execution_backend(options.clone(), backend.clone()) + .unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let worker = runtime.create_worker(create_request("lost ACK")).unwrap(); + let original_execution = backend.workers.lock().unwrap()[&worker.worker_ref].clone(); + let mut input = WorkerInput::user(" original immutable input "); + input.kind = WorkerInputKind::UserIfIdle; + input.submission_request_id = Some("original-lost-ack".into()); + *backend.lost_submission_ack.lock().unwrap() = input.submission_request_id.clone(); + + // The real Controller accepts and persists; only this caller's matching ACK + // is lost. The production acceptance timer, not a sleep, ends the wait. + let lost = backend.dispatch_input(&worker.worker_ref, input.clone()); + assert_eq!( + lost.outcome, + crate::execution::WorkerExecutionOutcome::Errored + ); + assert!( + lost.message + .as_deref() + .unwrap() + .contains("timed out waiting") + ); + assert!(lost.submission.is_none()); + assert_eq!( + lost.execution_id, + Some(original_execution.tasks.execution_id) + ); + assert!( + original_execution + .tasks + .submissions + .lock() + .unwrap() + .is_empty() + ); + assert_eq!(calls.load(Ordering::SeqCst), 1); + let (segments, source) = runtime_submission_payload(&input, "original-lost-ack"); + let durable = original_execution + .handle + .reconcile_submission("original-lost-ack", &segments, &source) + .unwrap() + .unwrap(); + + let mut wrong = input.clone(); + wrong.segments = Some(vec![Segment::text("different immutable input")]); + assert!( + !backend + .reconcile_submission(&worker.worker_ref, wrong) + .is_accepted() + ); + assert!( + original_execution + .tasks + .submissions + .lock() + .unwrap() + .is_empty() + ); + let found = backend.reconcile_submission(&worker.worker_ref, input.clone()); + assert!(found.is_accepted(), "{found:?}"); + assert_eq!( + found.execution_id, + Some(original_execution.tasks.execution_id) + ); + let ack = found.submission.unwrap(); + assert_eq!(ack.submission_request_id, "original-lost-ack"); + assert_eq!(ack.submission_id, durable.submission_id); + assert_eq!(ack.disposition, durable.disposition); + assert_eq!( + backend + .reconcile_submission(&worker.worker_ref, input.clone()) + .submission, + Some(ack.clone()) + ); + assert_eq!( + original_execution + .tasks + .submissions + .lock() + .unwrap() + .as_slice(), + &[ack.clone()] + ); + assert_eq!( + calls.load(Ordering::SeqCst), + 1, + "reconcile never redispatches" + ); + + // A second real acceptance is deliberately not reconciled until after + // shutdown. It must not amend an already-issued original termination. + let mut late_input = WorkerInput::user("late receipt lookup"); + late_input.submission_request_id = Some("late-lost-ack".into()); + *backend.lost_submission_ack.lock().unwrap() = late_input.submission_request_id.clone(); + let late_lost = backend.dispatch_input(&worker.worker_ref, late_input.clone()); + assert!( + late_lost + .message + .as_deref() + .unwrap() + .contains("timed out waiting") + ); + let (late_segments, late_source) = runtime_submission_payload(&late_input, "late-lost-ack"); + let late_fingerprint = + RuntimeExecutionTaskScope::submission_fingerprint(&late_segments, &late_source).unwrap(); + + runtime.stop_worker(&worker.worker_ref, None).unwrap(); + assert!(original_execution.handle.protocol_is_closed()); + assert!(original_execution.tasks.tasks.lock().unwrap().is_empty()); + let original = runtime + .execution_termination(&worker.worker_ref) + .unwrap() + .unwrap(); + assert_eq!(original.execution_id, original_execution.tasks.execution_id); + assert_eq!(original.submissions, vec![ack.clone()]); + assert!(original.shutdown_command_id.is_some()); + assert!(original.feature_failures.is_empty()); + let late_receipt = original_execution + .handle + .reconcile_submission("late-lost-ack", &late_segments, &late_source) + .unwrap() + .unwrap(); + let late_ack = crate::execution::WorkerSubmissionAck { + submission_request_id: late_receipt.submission_request_id, + submission_id: late_receipt.submission_id, + disposition: late_receipt.disposition, + }; + original_execution + .tasks + .retain_submission_acceptance(&late_ack, &late_fingerprint) + .unwrap(); + assert_eq!( + original_execution + .tasks + .termination(&worker.worker_ref, original.stop_operation_id), + Some(original.clone()), + "a late lookup cannot rewrite the original scope receipt" + ); + assert_eq!( + runtime.execution_termination(&worker.worker_ref).unwrap(), + Some(original.clone()) + ); + drop(runtime); + drop(backend); + + let backend = new_backend(); + let reopened = + EmbeddedRuntime::with_fs_store_and_execution_backend(options, backend.clone()).unwrap(); + assert_eq!( + reopened.execution_termination(&worker.worker_ref).unwrap(), + Some(original.clone()) + ); + + // Restore the same real saved Session into another actual Controller. Its + // receipt reader must see the old ACK, but it cannot claim that acceptance + // for this different scope, even if the caller replays the same Submit. + let restore_root = root.clone(); + let replacement_socket = socket_base.with_file_name("c2"); + let replacement_client = MockClient::new(simple_text_events()); + let replacement_calls = replacement_client.call_count.clone(); + let replacement = backend + .run_on_adapter_runtime(async move { + let store = CombinedStore::new( + FsStore::new(restore_root.join("sessions")).map_err(|error| error.to_string())?, + FsWorkerStore::new(restore_root.join("metadata")) + .map_err(|error| error.to_string())?, + ); + let fallback = WorkerManifest::from_toml( + r#" + [worker] + name = "r" + [model] + scheme = "anthropic" + model_id = "test-model" + auth = { kind = "none" } + [engine] + max_tokens = 100 + [scope] + allow = [] + "#, + ) + .map_err(|error| error.to_string())?; + let prepared = + Worker::prepare_restored_from_worker_metadata_with_context_and_model_client( + "r", + fallback, + store, + PromptCatalogSource::builtins_only(), + WorkerWorkspaceContext::no_workspace(), + WorkerFilesystemAuthority::None, + Some(Box::new(replacement_client)), + ) + .await + .map_err(|error| error.to_string())?; + let restored = prepared + .activate() + .await + .map_err(|error| error.to_string())?; + worker::bootstrap::start_worker_controller( + restored, + WorkerBootstrapLayout::Direct { + runtime_base: replacement_socket, + }, + ) + .await + .map_err(|error| error.to_string()) + }) + .unwrap(); + assert!( + !original_execution + .handle + .same_controller(&replacement.handle) + ); + let read_old = replacement + .handle + .reconcile_submission("original-lost-ack", &segments, &source) + .unwrap() + .unwrap(); + assert_eq!(read_old, durable); + let operation = backend.worker_lock(&worker.worker_ref).unwrap(); + let connected = { + let _guard = operation.lock().unwrap(); + backend.connect_handle( + WorkerExecutionOperation::Restore, + worker.worker_ref.clone(), + test_execution_context(worker.worker_ref.clone()), + replacement.handle, + Arc::new(tokio::sync::Mutex::new(Some(replacement.shutdown))), + replacement.controller_task, + None, + Vec::new(), + None, + None, + ) + }; + assert!( + matches!(connected, WorkerExecutionSpawnResult::Connected { .. }), + "{connected:?}" + ); + let replacement_execution = backend.workers.lock().unwrap()[&worker.worker_ref].clone(); + assert_ne!( + replacement_execution.tasks.execution_id, + original.execution_id + ); + assert!( + reopened + .execution_termination(&worker.worker_ref) + .unwrap() + .is_none() + ); + let read = backend.reconcile_submission(&worker.worker_ref, input.clone()); + assert!(read.is_accepted(), "{read:?}"); + assert_eq!(read.submission, Some(ack.clone())); + assert_eq!( + read.execution_id, + Some(replacement_execution.tasks.execution_id) + ); + assert!( + replacement_execution + .tasks + .submissions + .lock() + .unwrap() + .is_empty() + ); + let replay = backend.dispatch_input(&worker.worker_ref, input); + assert!(replay.is_accepted(), "{replay:?}"); + assert_eq!(replay.submission, Some(ack.clone())); + assert!( + replacement_execution + .tasks + .submission_attempts + .lock() + .unwrap() + .is_empty() + ); + assert!( + replacement_execution + .tasks + .submissions + .lock() + .unwrap() + .is_empty() + ); + assert_eq!(replacement_calls.load(Ordering::SeqCst), 0); + let mut own_input = WorkerInput::user("new Controller's own input"); + own_input.submission_request_id = Some("replacement-own-input".into()); + let own = backend.dispatch_input(&worker.worker_ref, own_input); + assert!(own.is_accepted(), "{own:?}"); + let own_ack = own.submission.unwrap(); + assert_eq!( + own.execution_id, + Some(replacement_execution.tasks.execution_id) + ); + let stopped = backend.stop_worker_operation(WorkerExecutionStopRequest { + worker_ref: worker.worker_ref.clone(), + operation_id: crate::execution::WorkerLifecycleOperationId::new(), + }); + assert!(stopped.is_accepted(), "{stopped:?}"); + let replacement_receipt = stopped.termination.unwrap(); + assert_eq!( + replacement_receipt.execution_id, + replacement_execution.tasks.execution_id + ); + assert_ne!(replacement_receipt.execution_id, original.execution_id); + assert!(!replacement_receipt.submissions.contains(&ack)); + assert_eq!(replacement_receipt.submissions, vec![own_ack]); + assert!(replacement_receipt.shutdown_command_id.is_some()); + assert!(replacement_execution.tasks.tasks.lock().unwrap().is_empty()); + assert!( + reopened.execution_termination(&worker.worker_ref).is_err(), + "replacement cannot overwrite original canonical receipt" + ); +} diff --git a/crates/worker-runtime/src/worker_backend/termination_persistence_tests.rs b/crates/worker-runtime/src/worker_backend/termination_persistence_tests.rs new file mode 100644 index 000000000..b8ba4760b --- /dev/null +++ b/crates/worker-runtime/src/worker_backend/termination_persistence_tests.rs @@ -0,0 +1,157 @@ +#[cfg(feature = "fs-store")] +#[test] +fn stop_termination_real_join_survives_fs_reopen_rejects_replacement_and_legacy_missing_receipt() { + // Keep socket names relative and short. The focused invocation runs from an + // isolated scratch cwd; data paths are still explicit absolute fixture paths. + let temp = tempfile::tempdir_in(".").unwrap(); + let root = fs::canonicalize(temp.path()).unwrap(); + let socket_base = PathBuf::from(temp.path().file_name().unwrap()).join("c"); + let options = crate::fs_store::FsRuntimeStoreOptions::new(root.join("runtime")) + .with_runtime_id("original-scope-receipt"); + let new_backend = || { + Arc::new( + WorkerRuntimeExecutionBackend::new(MockFactory { + client: MockClient::sequential(vec![]), + runtime_base: socket_base.clone(), + cwd: root.clone(), + store_dir: root.join("sessions"), + worker_metadata_dir: root.join("metadata"), + observed_cwds: Arc::new(Mutex::new(Vec::new())), + observed_workspace_clients: Arc::new(Mutex::new(Vec::new())), + }) + .unwrap(), + ) + }; + let backend = new_backend(); + let runtime = + EmbeddedRuntime::with_fs_store_and_execution_backend(options.clone(), backend.clone()) + .unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let worker = runtime + .create_worker(create_request("durable original receipt")) + .unwrap(); + assert!( + runtime + .execution_termination(&worker.worker_ref) + .unwrap() + .is_none() + ); + let mut input = WorkerInput::user("original submission"); + input.submission_request_id = Some("original-request".into()); + let acceptance = runtime + .send_input(&worker.worker_ref, input) + .unwrap() + .submission + .unwrap(); + runtime.stop_worker(&worker.worker_ref, None).unwrap(); + let original = runtime + .execution_termination(&worker.worker_ref) + .unwrap() + .unwrap(); + assert_eq!( + original.scope, + crate::execution::WorkerExecutionTerminationScope::Controller + ); + assert!(original.shutdown_command_id.is_some()); + assert!(original.feature_failures.is_empty()); + assert!(original.submissions.contains(&acceptance)); + assert_eq!( + backend.termination_receipt(&worker.worker_ref).unwrap(), + Some(original.clone()) + ); + assert!(backend.workers.lock().unwrap().is_empty()); + let worker_path = options + .root + .join("workers") + .join(worker.worker_id.to_string()) + .join("worker.json"); + let document: serde_json::Value = + serde_json::from_slice(&fs::read(&worker_path).unwrap()).unwrap(); + assert_eq!( + document["restore_guard"]["execution_termination"], + serde_json::to_value(&original).unwrap() + ); + drop(runtime); + drop(backend); + + let backend = new_backend(); + let reopened = + EmbeddedRuntime::with_fs_store_and_execution_backend(options.clone(), backend.clone()) + .unwrap(); + assert!(backend.completed_stops.lock().unwrap().is_empty()); + assert!(backend.workers.lock().unwrap().is_empty()); + assert_eq!( + reopened.execution_termination(&worker.worker_ref).unwrap(), + Some(original.clone()) + ); + + // Exercise the backend ownership check even if a replacement was installed + // outside Runtime's lifecycle coordinator. Empty old maps cannot bless it. + assert!(matches!( + backend.restore_worker(direct_restore_request(worker.worker_ref.clone())), + WorkerExecutionSpawnResult::Connected { .. } + )); + assert!( + reopened + .execution_termination(&worker.worker_ref) + .unwrap() + .is_none() + ); + let replacement_stop = backend.stop_worker_operation(WorkerExecutionStopRequest { + worker_ref: worker.worker_ref.clone(), + operation_id: crate::execution::WorkerLifecycleOperationId::new(), + }); + assert!(replacement_stop.is_accepted(), "{replacement_stop:?}"); + assert_ne!( + replacement_stop.termination.as_ref().unwrap().execution_id, + original.execution_id + ); + assert!( + reopened.execution_termination(&worker.worker_ref).is_err(), + "a different run/stop-operation receipt must not replace canonical evidence" + ); + drop(reopened); + drop(backend); + + // Released schema 15 never held an original termination. Its migration must + // preserve status/observation but must not reconstruct shutdown from them. + let mut legacy = document; + legacy["schema_version"] = serde_json::json!(15); + legacy["restore_guard"] + .as_object_mut() + .unwrap() + .remove("execution_termination"); + let observation = legacy["restore_guard"]["observation_token"].clone(); + fs::write(&worker_path, serde_json::to_vec_pretty(&legacy).unwrap()).unwrap(); + let runtime_path = options.root.join("runtime.json"); + let mut runtime_document: serde_json::Value = + serde_json::from_slice(&fs::read(&runtime_path).unwrap()).unwrap(); + runtime_document["schema_version"] = serde_json::json!(15); + fs::write( + &runtime_path, + serde_json::to_vec_pretty(&runtime_document).unwrap(), + ) + .unwrap(); + crate::fs_store::FsRuntimeStore::migrate(&options).unwrap(); + let backend = new_backend(); + let reopened = + EmbeddedRuntime::with_fs_store_and_execution_backend(options.clone(), backend).unwrap(); + assert_eq!( + reopened.worker_detail(&worker.worker_ref).unwrap().status, + crate::catalog::WorkerStatus::Stopped + ); + assert!( + reopened + .execution_termination(&worker.worker_ref) + .unwrap() + .is_none() + ); + let migrated: serde_json::Value = + serde_json::from_slice(&fs::read(&worker_path).unwrap()).unwrap(); + assert_eq!(migrated["restore_guard"]["observation_token"], observation); + assert!( + migrated["restore_guard"] + .get("execution_termination") + .is_none() + ); +} diff --git a/crates/worker-runtime/src/working_directory.rs b/crates/worker-runtime/src/working_directory.rs index a8c9388d4..d6ebf4cfb 100644 --- a/crates/worker-runtime/src/working_directory.rs +++ b/crates/worker-runtime/src/working_directory.rs @@ -4518,6 +4518,11 @@ mod tests { let initial_ref = bound.status().summary.creation_ref.expect("creation ref"); git(&bound.root, &["switch", "-c", "observed-branch"]); + git(&bound.root, &["config", "user.name", "Yoi Test"]); + git( + &bound.root, + &["config", "user.email", "test@example.invalid"], + ); fs::write(bound.root.join("observed.txt"), "observed\n").unwrap(); git(&bound.root, &["add", "observed.txt"]); git(&bound.root, &["commit", "-m", "advance workdir"]);