fix: retire the exact stopped Controller from runtime observation
This commit is contained in:
@@ -539,6 +539,16 @@ pub trait RuntimeWorkerFactory: Send + Sync + 'static {
|
|||||||
) -> Result<(), String> {
|
) -> Result<(), String> {
|
||||||
Ok(())
|
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
|
/// Production factory that resolves a normal Worker profile and spawns it under
|
||||||
@@ -577,6 +587,21 @@ impl RuntimeWorkerObservationHub {
|
|||||||
Ok(())
|
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(
|
fn get(
|
||||||
&self,
|
&self,
|
||||||
worker_ref: &WorkerRef,
|
worker_ref: &WorkerRef,
|
||||||
@@ -2560,6 +2585,14 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
|
|||||||
self.observation_hub
|
self.observation_hub
|
||||||
.register(worker_ref.clone(), workspace_id.map(str::to_string), handle)
|
.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<WorkerSessionStore, WorkerAggregateStore>;
|
type ProfileRestoreStore = CombinedStore<WorkerSessionStore, WorkerAggregateStore>;
|
||||||
@@ -4463,6 +4496,14 @@ where
|
|||||||
if let Err(message) = self.cleanup_execution_resources(&execution) {
|
if let Err(message) = self.cleanup_execution_resources(&execution) {
|
||||||
return WorkerExecutionResult::errored(WorkerExecutionOperation::Stop, message);
|
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() {
|
match self.workers.lock() {
|
||||||
Ok(mut workers) => {
|
Ok(mut workers) => {
|
||||||
|
|||||||
Reference in New Issue
Block a user