diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index fb9e8d75e..d27d165f2 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -3266,6 +3266,14 @@ impl Runtime { let _guard = operation_lock .lock() .map_err(|_| RuntimeError::StatePoisoned)?; + self.execution_termination_under_lock(worker_ref) + } + + // Caller holds the Runtime Worker operation lock, including deletion callers. + fn execution_termination_under_lock( + &self, + worker_ref: &WorkerRef, + ) -> Result, RuntimeError> { let (backend, saved) = { let mut state = self.lock()?; state.ensure_worker_ref(worker_ref)?; @@ -4630,7 +4638,7 @@ impl Runtime { if let Some(scope) = scope { self.ensure_worker_in_workspace(scope, worker_ref)?; } - let backend = { + let (backend, has_termination) = { let state = self.lock()?; state.ensure_running()?; state.ensure_worker_ref(worker_ref)?; @@ -4648,9 +4656,22 @@ impl Runtime { worker_ref.worker_id ))); } - state.execution_backend.clone() + ( + state.execution_backend.clone(), + worker.restore_guard.execution_termination.is_some(), + ) }; - if let Some(backend) = backend { + if has_termination { + // Consume the canonical Stop's exact receipt. A new Stop could + // target a replacement Controller or discard a later candidate; + // neither is authorized by the original scope's termination. + if self.execution_termination_under_lock(worker_ref)?.is_none() { + return Err(RuntimeError::WorkerExecutionUnavailable { + worker_id: worker_ref.worker_id, + message: "canonical Stop receipt no longer matches the execution owner; deletion remains pending".into(), + }); + } + } else if let Some(backend) = backend { let result = backend.stop_worker(worker_ref); if !result.is_accepted() { return Err(RuntimeError::WorkerExecutionRejected { diff --git a/crates/worker-runtime/src/subject_delete_backend_tests.rs b/crates/worker-runtime/src/subject_delete_backend_tests.rs index c41830d5b..7ca9d05d8 100644 --- a/crates/worker-runtime/src/subject_delete_backend_tests.rs +++ b/crates/worker-runtime/src/subject_delete_backend_tests.rs @@ -196,8 +196,21 @@ fn subject_delete_unregistered_profile_controller_retries_unknown_shutdown_witho backend.restore_worker(request.clone()), WorkerExecutionSpawnResult::Rejected(_) )); + let stop_operation_id = backend.admissions.lock().unwrap()[&request.worker_ref] + .deletion_stop_operation_id + .expect("the original rejected Stop keeps its identity for retry"); backend.subject_delete_worker(&deletion).unwrap(); + let original = backend + .termination_receipt(&request.worker_ref) + .unwrap() + .unwrap(); + assert_eq!(original.stop_operation_id, stop_operation_id); + assert_eq!(original.execution_id, execution.tasks.execution_id); backend.subject_delete_worker(&deletion).unwrap(); + assert_eq!( + backend.termination_receipt(&request.worker_ref).unwrap(), + Some(original) + ); assert!(execution.handle.protocol_is_closed()); assert!(execution.tasks.tasks.lock().unwrap().is_empty()); assert!(backend.workers.lock().unwrap().is_empty()); @@ -283,3 +296,111 @@ fn subject_delete_pending_restore_fences_late_controller_and_never_restarts_it() WorkerExecutionSpawnResult::Rejected(_) )); } + +#[test] +fn subject_delete_without_original_owner_retains_child_identity_but_never_issues_receipt() { + let root = tempfile::tempdir().unwrap(); + let backend = WorkerRuntimeExecutionBackend::new(MockFactory { + client: MockClient::new(vec![]), + runtime_base: root.path().join("must-not-launch"), + cwd: root.path().to_path_buf(), + store_dir: root.path().join("sessions"), + worker_metadata_dir: root.path().join("metadata"), + observed_cwds: Arc::default(), + observed_workspace_clients: Arc::default(), + }) + .unwrap(); + let worker_ref = WorkerRef::new(WorkerId::now_v7()); + let deletion = runtime_api::SubjectDeleteWorkerRequest { + workspace_id: "ws-test".into(), + worker_id: worker_ref.worker_id, + create_fingerprint: "not-admitted".into(), + operation_id: "delete-without-owner".into(), + }; + assert!( + backend + .subject_delete_worker(&deletion) + .unwrap_err() + .contains("map absence") + ); + let child = backend.admissions.lock().unwrap()[&worker_ref] + .deletion_stop_operation_id + .unwrap(); + assert!( + backend + .subject_delete_worker(&deletion) + .unwrap_err() + .contains("map absence") + ); + assert_eq!( + backend.admissions.lock().unwrap()[&worker_ref].deletion_stop_operation_id, + Some(child) + ); + assert!(backend.completed_stops.lock().unwrap().is_empty()); + assert!(backend.termination_receipt(&worker_ref).unwrap().is_none()); + assert!(!root.path().join("must-not-launch").exists()); +} + +#[test] +#[cfg(feature = "fs-store")] +fn subject_delete_consumes_prior_canonical_stop_without_retagging_original_scope() { + let temp = tempfile::tempdir_in(".").unwrap(); + let root = fs::canonicalize(temp.path()).unwrap(); + let backend = Arc::new( + WorkerRuntimeExecutionBackend::new(MockFactory { + client: MockClient::new(vec![]), + runtime_base: PathBuf::from(temp.path().file_name().unwrap()).join("c"), + 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 runtime = EmbeddedRuntime::with_fs_store_and_execution_backend( + crate::fs_store::FsRuntimeStoreOptions::new(root.join("catalog")) + .with_runtime_id("delete-prior-stop"), + backend.clone(), + ) + .unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let scope = crate::runtime::RuntimeWorkspaceScope::new("ws-test", "server-test"); + let mut create = create_request("prior Stop Subject deletion"); + create.memory_settings = Some(manifest::WorkspaceMemorySettingsSnapshot { + workspace_id: scope.workspace_id.clone(), + language: "English".into(), + }); + create.workspace_api = Some(WorkspaceApiRef { + workspace_id: scope.workspace_id.clone(), + base_url: "https://workspace.invalid".into(), + }); + let worker = runtime + .create_worker_scoped(&scope, create.clone()) + .unwrap(); + runtime + .stop_worker_scoped(&scope, &worker.worker_ref, None) + .unwrap(); + let original = runtime + .execution_termination(&worker.worker_ref) + .unwrap() + .unwrap(); + let deletion = runtime_api::SubjectDeleteWorkerRequest { + workspace_id: scope.workspace_id.clone(), + worker_id: create.worker_id, + create_fingerprint: create.create_fingerprint, + operation_id: "delete-prior-stop".into(), + }; + runtime + .subject_delete_worker_scoped(&scope, deletion.clone()) + .unwrap(); + backend.subject_delete_worker(&deletion).unwrap(); + assert_eq!( + backend.termination_receipt(&worker.worker_ref).unwrap(), + Some(original.clone()) + ); + assert_eq!( + backend.admissions.lock().unwrap()[&worker.worker_ref].deletion_stop_operation_id, + Some(original.stop_operation_id) + ); +} diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 40d47e8ba..e9f9c8a5c 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -3903,6 +3903,9 @@ struct WorkerAdmission { workspace_id: Option, create_fingerprint: String, deletion: Option, + // Stable child Stop for this exact deletion. This is an operation identity, + // not termination proof; only the original joined owner can issue a receipt. + deletion_stop_operation_id: Option, } /// Worker execution backend backed by real worker crate Workers. @@ -4007,6 +4010,7 @@ where workspace_id: workspace.map(str::to_string), create_fingerprint: fingerprint.into(), deletion: None, + deletion_stop_operation_id: None, }, ); } @@ -5825,18 +5829,42 @@ where let lock = self.worker_lock(&worker)?; let _operation = lock.lock().map_err(|_| "worker operation lock poisoned")?; self.validate_subject_delete_identity(request)?; - self.admissions - .lock() - .map_err(|_| "worker admission lock poisoned")? - .insert( - worker.clone(), - WorkerAdmission { + let operation_id = { + let mut admissions = self + .admissions + .lock() + .map_err(|_| "worker admission lock poisoned")?; + let admission = admissions + .entry(worker.clone()) + .or_insert_with(|| WorkerAdmission { workspace_id: Some(request.workspace_id.clone()), create_fingerprint: request.create_fingerprint.clone(), - deletion: Some(request.clone()), - }, - ); - let result = self.stop_worker_under_lock(&worker, None); + deletion: None, + deletion_stop_operation_id: None, + }); + admission.deletion = Some(request.clone()); + match admission.deletion_stop_operation_id { + Some(operation_id) => operation_id, + None => { + // Reuse a completed original Stop, never retag its receipt + // as a newly generated deletion child. The Stop consumer + // below rejects replay if any replacement owner is present. + let operation_id = self + .completed_stops + .lock() + .map_err(|_| "stop completion receipts lock poisoned")? + .get(&worker) + .map(|receipt| receipt.stop_operation_id) + .unwrap_or_default(); + admission.deletion_stop_operation_id = Some(operation_id); + operation_id + } + } + }; + let result = self.stop_worker_operation_under_lock(WorkerExecutionStopRequest { + worker_ref: worker.clone(), + operation_id, + }); if !result.is_accepted() { return Err(result.message_or_default()); } @@ -8737,12 +8765,17 @@ mod tests { }) .await .unwrap(); - if let Some(receiver) = controller.shutdown.lock().await.take() { - tokio::time::timeout(Duration::from_secs(5), receiver) - .await - .unwrap() - .unwrap(); - } + let receiver = controller + .shutdown + .lock() + .await + .take() + .expect("fixture must retain the original Controller shutdown report"); + let report = tokio::time::timeout(Duration::from_secs(5), receiver) + .await + .unwrap() + .unwrap(); + assert!(report.feature_failures.is_empty(), "{report:?}"); tokio::time::timeout(Duration::from_secs(5), controller.controller_task) .await .unwrap() @@ -9795,11 +9828,22 @@ mod tests { .contains("still present"), "closed Controller is still residual execution" ); - assert!(backend.stop_worker(&request.worker_ref).is_accepted()); + let stop = WorkerExecutionStopRequest { + worker_ref: request.worker_ref.clone(), + operation_id: crate::execution::WorkerLifecycleOperationId::new(), + }; + let original = backend.stop_worker_operation(stop.clone()); + assert!(original.is_accepted(), "{original:?}"); + let receipt = original.termination.as_ref().unwrap(); + assert_eq!(receipt.execution_id, execution.tasks.execution_id); assert!(backend.workers.lock().unwrap().is_empty()); backend.recover_subject_create(&recovery).unwrap(); assert!(execution.tasks.tasks.lock().unwrap().is_empty()); - assert!(backend.stop_worker(&request.worker_ref).is_accepted()); + assert_eq!(backend.stop_worker_operation(stop), original); + assert!( + !backend.stop_worker(&request.worker_ref).is_accepted(), + "identity-less absence is not a replay" + ); } #[test] @@ -11263,6 +11307,8 @@ mod tests { struct PendingCreateFactory { started: mpsc::Sender<()>, dropped: mpsc::Sender<()>, + release: Arc, + inner: MockFactory, } struct FactoryFutureDrop(mpsc::Sender<()>); @@ -11286,11 +11332,12 @@ mod tests { async fn spawn_controller( &self, - _request: WorkerExecutionSpawnRequest, + request: WorkerExecutionSpawnRequest, ) -> Result { let _drop = FactoryFutureDrop(self.dropped.clone()); self.started.send(()).unwrap(); - std::future::pending().await + self.release.notified().await; + self.inner.spawn_controller(request).await } async fn restore_controller( @@ -11303,15 +11350,26 @@ mod tests { #[cfg(feature = "fs-store")] #[test] - fn create_timeout_cancels_factory_and_removes_persisted_worker() { + fn create_timeout_retains_factory_and_persisted_worker_until_original_controller_is_joined() { let root = tempfile::tempdir().unwrap(); let runtime_store_dir = root.path().join("runtime"); let (started_tx, started) = mpsc::channel(); let (dropped_tx, dropped) = mpsc::channel(); + let release = Arc::new(tokio::sync::Notify::new()); let backend = Arc::new( WorkerRuntimeExecutionBackend::new(PendingCreateFactory { started: started_tx, dropped: dropped_tx, + release: release.clone(), + inner: MockFactory { + client: MockClient::new(vec![]), + runtime_base: root.path().join("control"), + cwd: root.path().to_path_buf(), + store_dir: root.path().join("sessions"), + worker_metadata_dir: root.path().join("metadata"), + observed_cwds: Arc::default(), + observed_workspace_clients: Arc::default(), + }, }) .unwrap() .with_spawn_restore_timeout(Duration::from_millis(20)), @@ -11339,23 +11397,46 @@ mod tests { .unwrap_err(); let error = create.join().unwrap().unwrap_err(); + let worker_ref = WorkerRef::new(worker_id); assert!(matches!( delete_error, - crate::error::RuntimeError::WorkerNotFound { worker_id: missing } if missing == worker_id + crate::error::RuntimeError::InvalidRequest(_) )); - assert!(error.to_string().contains("was cancelled")); + assert!(error.to_string().contains("pending"), "{error}"); + assert!(matches!(dropped.try_recv(), Err(mpsc::TryRecvError::Empty))); + assert!(backend.pending_spawn_exists(&worker_ref).unwrap()); + assert!( + runtime + .execution_termination(&worker_ref) + .unwrap() + .is_none() + ); + let worker_path = runtime_store_dir + .join("workers") + .join(worker_id.to_string()); + assert!( + worker_path.exists(), + "timeout must retain the unresolved Worker" + ); + assert_eq!(runtime.list_workers().unwrap().len(), 1); + + // Release the original factory, not a replacement. Stop must transfer + // its actual late Controller into the owned scope and join/report it. + release.notify_one(); + runtime.stop_worker(&worker_ref, None).unwrap(); dropped .try_recv() - .expect("cancelled factory future must be dropped before create returns"); - assert!(runtime.list_workers().unwrap().is_empty()); - assert!(backend.workers.lock().unwrap().is_empty()); - assert!( - !runtime_store_dir - .join("workers") - .join(worker_id.to_string()) - .exists(), - "failed create must remove its persisted Worker aggregate" + .expect("factory completed normally before Stop settled"); + assert!(!backend.pending_spawn_exists(&worker_ref).unwrap()); + let original = runtime.execution_termination(&worker_ref).unwrap().unwrap(); + assert_eq!( + original.scope, + crate::execution::WorkerExecutionTerminationScope::Controller ); + assert!(original.shutdown_command_id.is_some()); + assert!(backend.workers.lock().unwrap().is_empty()); + runtime.delete_worker(&worker_ref).unwrap(); + assert!(!worker_path.exists()); } // Generic controller tests explicitly substitute the Host's non-Subject @@ -11922,7 +12003,13 @@ mod tests { &self, _request: WorkerExecutionSpawnRequest, ) -> Result { - Err("spawn failed".into()) + // This fixture returns before constructing a Worker, Controller or + // background resource. Unlike a string failure, these are known + // owner facts and allow only the adapter-created workdir rollback. + Err(RuntimeRestoreFailure::refused( + RuntimeRestoreStage::Preflight, + "spawn failed", + )) } async fn restore_controller( @@ -14104,7 +14191,19 @@ mod tests { fs::read_to_string(checkout.join("README.md")).unwrap(), "owner recovery in progress" ); - assert!(backend.stop_worker(&request.worker_ref).is_accepted()); + let stop = WorkerExecutionStopRequest { + worker_ref: request.worker_ref.clone(), + operation_id: crate::execution::WorkerLifecycleOperationId::new(), + }; + let result = backend.stop_worker_operation(stop.clone()); + assert!(result.is_accepted(), "{result:?}"); + assert!( + result + .termination + .as_ref() + .unwrap() + .confirms_scope_shutdown(&stop.worker_ref, stop.operation_id) + ); } #[cfg(target_os = "linux")] @@ -14176,7 +14275,19 @@ mod tests { before, "damaged checkout must be retained" ); - assert!(backend.stop_worker(&request.worker_ref).is_accepted()); + let stop = WorkerExecutionStopRequest { + worker_ref: request.worker_ref.clone(), + operation_id: crate::execution::WorkerLifecycleOperationId::new(), + }; + let result = backend.stop_worker_operation(stop.clone()); + assert!(result.is_accepted(), "{result:?}"); + assert!( + result + .termination + .as_ref() + .unwrap() + .confirms_scope_shutdown(&stop.worker_ref, stop.operation_id) + ); println!( "damage={damage}; reason={:?}; classification={:?}; cwd_bind_count={}; retained_files={}; retained_data_unchanged=true; stopped=true", error.failure_reason(), diff --git a/crates/worker-runtime/src/worker_backend/startup_failure.rs b/crates/worker-runtime/src/worker_backend/startup_failure.rs index 1b01db12a..9264ec9b8 100644 --- a/crates/worker-runtime/src/worker_backend/startup_failure.rs +++ b/crates/worker-runtime/src/worker_backend/startup_failure.rs @@ -253,9 +253,15 @@ mod tests { #[test] fn typed_factory_failure_survives_adapter_and_protocol_without_changing_disposition() { let backend = WorkerRuntimeExecutionBackend::new(TypedFailingFactory).unwrap(); - let WorkerExecutionSpawnResult::Errored(result) = backend.spawn_worker(request()) else { - panic!("cause attachment must not reclassify the existing spawn result") + let request = request(); + let worker_ref = request.worker_ref.clone(); + let WorkerExecutionSpawnResult::ReconciliationRequired { result, .. } = + backend.spawn_worker(request) + else { + panic!("unproven resources and failed cleanup must retain the original factory owner") }; + assert!(backend.pending_spawn_exists(&worker_ref).unwrap()); + assert!(backend.termination_receipt(&worker_ref).unwrap().is_none()); assert_primary(&result); let transferred: WorkerExecutionResult = serde_json::from_slice(&serde_json::to_vec(&result).unwrap()).unwrap(); diff --git a/crates/worker-runtime/src/worker_backend/termination_persistence_tests.rs b/crates/worker-runtime/src/worker_backend/termination_persistence_tests.rs index b8ba4760b..b841f8cef 100644 --- a/crates/worker-runtime/src/worker_backend/termination_persistence_tests.rs +++ b/crates/worker-runtime/src/worker_backend/termination_persistence_tests.rs @@ -97,6 +97,15 @@ fn stop_termination_real_join_survives_fs_reopen_rejects_replacement_and_legacy_ .unwrap() .is_none() ); + assert!(reopened.delete_worker(&worker.worker_ref).is_err()); + assert!(worker_path.exists()); + assert!( + backend + .pending_restore(&worker.worker_ref) + .unwrap() + .is_some(), + "deletion must not stop/discard a replacement candidate with the old receipt" + ); let replacement_stop = backend.stop_worker_operation(WorkerExecutionStopRequest { worker_ref: worker.worker_ref.clone(), operation_id: crate::execution::WorkerLifecycleOperationId::new(), @@ -110,6 +119,8 @@ fn stop_termination_real_join_survives_fs_reopen_rejects_replacement_and_legacy_ reopened.execution_termination(&worker.worker_ref).is_err(), "a different run/stop-operation receipt must not replace canonical evidence" ); + assert!(reopened.delete_worker(&worker.worker_ref).is_err()); + assert!(worker_path.exists()); drop(reopened); drop(backend); @@ -155,3 +166,97 @@ fn stop_termination_real_join_survives_fs_reopen_rejects_replacement_and_legacy_ .is_none() ); } + +#[cfg(feature = "fs-store")] +#[test] +fn delete_after_reopen_consumes_original_stop_without_starting_or_restopping_controller() { + let temp = tempfile::tempdir_in(".").unwrap(); + let root = fs::canonicalize(temp.path()).unwrap(); + let options = crate::fs_store::FsRuntimeStoreOptions::new(root.join("runtime")) + .with_runtime_id("delete-original-stop"); + let backend = Arc::new( + WorkerRuntimeExecutionBackend::new(MockFactory { + client: MockClient::new(vec![]), + runtime_base: PathBuf::from(temp.path().file_name().unwrap()).join("c"), + 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 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("delete original stop")) + .unwrap(); + runtime.stop_worker(&worker.worker_ref, None).unwrap(); + let original = runtime + .execution_termination(&worker.worker_ref) + .unwrap() + .unwrap(); + // Inject mismatched observations, never positive termination evidence. + // Deletion must keep the canonical original and refuse every substitution. + for mismatch in ["worker", "execution", "operation"] { + let mut observed = original.clone(); + match mismatch { + "worker" => observed.worker_ref = WorkerRef::new(WorkerId::now_v7()), + "execution" => observed.execution_id = uuid::Uuid::now_v7(), + "operation" => { + observed.stop_operation_id = crate::execution::WorkerLifecycleOperationId::new() + } + _ => unreachable!(), + } + backend + .completed_stops + .lock() + .unwrap() + .insert(worker.worker_ref.clone(), observed); + assert!( + runtime.delete_worker(&worker.worker_ref).is_err(), + "{mismatch}" + ); + assert!( + options + .root + .join("workers") + .join(worker.worker_id.to_string()) + .exists() + ); + } + backend + .completed_stops + .lock() + .unwrap() + .insert(worker.worker_ref.clone(), original.clone()); + drop(runtime); + drop(backend); + let no_launch = root.join("must-not-launch"); + let backend = Arc::new( + WorkerRuntimeExecutionBackend::new(ProfileRuntimeWorkerFactory::new(&no_launch)).unwrap(), + ); + let reopened = + EmbeddedRuntime::with_fs_store_and_execution_backend(options.clone(), backend.clone()) + .unwrap(); + assert!(backend.completed_stops.lock().unwrap().is_empty()); + assert_eq!( + reopened.execution_termination(&worker.worker_ref).unwrap(), + Some(original) + ); + reopened.delete_worker(&worker.worker_ref).unwrap(); + assert!( + !options + .root + .join("workers") + .join(worker.worker_id.to_string()) + .exists() + ); + assert!( + backend.completed_stops.lock().unwrap().is_empty(), + "no new Stop was issued" + ); + assert!(!no_launch.exists()); +} diff --git a/crates/worker-runtime/src/worker_backend_restore_candidate_tests.rs b/crates/worker-runtime/src/worker_backend_restore_candidate_tests.rs index 934703368..752b59044 100644 --- a/crates/worker-runtime/src/worker_backend_restore_candidate_tests.rs +++ b/crates/worker-runtime/src/worker_backend_restore_candidate_tests.rs @@ -1246,11 +1246,18 @@ mod restore_candidate { .is_some(), "retain the real prepared Worker's close-only resource owner" ); + let stop = WorkerExecutionStopRequest { + worker_ref: fixture.request.worker_ref.clone(), + operation_id: crate::execution::WorkerLifecycleOperationId::new(), + }; + let result = fixture.backend.stop_worker_operation(stop.clone()); + assert!(result.is_accepted(), "{result:?}"); assert!( - fixture - .backend - .stop_worker(&fixture.request.worker_ref) - .is_accepted() + result + .termination + .as_ref() + .unwrap() + .confirms_scope_shutdown(&stop.worker_ref, stop.operation_id) ); fixture.assert_released(); @@ -1285,12 +1292,10 @@ mod restore_candidate { assert_eq!(opens.load(Ordering::SeqCst), 0); fixture.assert_quiet(); assert_eq!(fixture.canonical_files(), before); - assert!( - fixture - .backend - .stop_worker(&fixture.request.worker_ref) - .is_accepted() - ); + let replay = fixture.backend.stop_worker_operation(stop); + assert!(replay.is_accepted(), "{replay:?}"); + assert_eq!(replay.termination, result.termination); + assert_eq!(replay.execution_id, result.execution_id); assert_eq!(fixture.canonical_files(), before); }