From a10875d68e8cf554e8a7916511adc78d12303d9f Mon Sep 17 00:00:00 2001 From: Hare Date: Mon, 26 Oct 2026 13:36:37 +0900 Subject: [PATCH] fix: retire the exact stopped Controller from runtime observation --- crates/worker-runtime/src/worker_backend.rs | 41 +++++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 7e1288701..4d879bb71 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -539,6 +539,16 @@ pub trait RuntimeWorkerFactory: Send + Sync + 'static { ) -> Result<(), String> { Ok(()) } + + /// Retire observation of this exact Controller after confirmed cleanup. + /// Called under the same per-Worker lock as publication and replacement. + fn deactivate_stopped_controller( + &self, + _worker_ref: &WorkerRef, + _handle: &WorkerHandle, + ) -> Result<(), String> { + Ok(()) + } } /// Production factory that resolves a normal Worker profile and spawns it under @@ -577,6 +587,21 @@ impl RuntimeWorkerObservationHub { Ok(()) } + fn unregister(&self, worker_ref: &WorkerRef, handle: &WorkerHandle) -> Result<(), String> { + let mut workers = self + .workers + .lock() + .map_err(|_| "Worker observation registry lock is poisoned".to_string())?; + if workers.get(worker_ref).is_some_and(|entry| { + entry + .shared_state + .ptr_eq(&Arc::downgrade(&handle.shared_state)) + }) { + workers.remove(worker_ref); + } + Ok(()) + } + fn get( &self, worker_ref: &WorkerRef, @@ -2560,6 +2585,14 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { self.observation_hub .register(worker_ref.clone(), workspace_id.map(str::to_string), handle) } + + fn deactivate_stopped_controller( + &self, + worker_ref: &WorkerRef, + handle: &WorkerHandle, + ) -> Result<(), String> { + self.observation_hub.unregister(worker_ref, handle) + } } type ProfileRestoreStore = CombinedStore; @@ -4463,6 +4496,14 @@ where if let Err(message) = self.cleanup_execution_resources(&execution) { return WorkerExecutionResult::errored(WorkerExecutionOperation::Stop, message); } + // External consumers may retain a closed handle indefinitely. Weak + // shared-state liveness is not evidence of a published execution. + if let Err(message) = self + .factory + .deactivate_stopped_controller(worker_ref, &execution.handle) + { + return WorkerExecutionResult::errored(WorkerExecutionOperation::Stop, message); + } match self.workers.lock() { Ok(mut workers) => {