[bugfix]: avoid duplicate worker shutdown cleanup path

Author: Satyam-53

fastvideo/tests/worker/test_multiproc_executor.py

@@ -55,6 +55,16 @@ def shutdown(self):
         return {"status": "shutdown"}
 
 
+class _FailingWorker:
+
+    def __init__(self):
+        self.shutdown_calls = 0
+
+    def shutdown(self):
+        self.shutdown_calls += 1
+        raise RuntimeError("interrupted shutdown")
+
+
 class _GracefulProcess:
 
     def __init__(self, required_timeout: float):
@@ -149,6 +159,29 @@ def test_worker_shutdown_closes_worker_and_compile_pool_once(monkeypatch) -> Non
     assert compile_shutdowns == [True]
 
 
+def test_worker_shutdown_still_idempotent_after_failure(monkeypatch) -> None:
+    worker = _FailingWorker()
+    compile_shutdowns = []
+    proc = WorkerMultiprocProc.__new__(WorkerMultiprocProc)
+    proc.rank = 0
+    proc.worker = worker
+    proc._shutdown_started = False
+    proc._shutdown_complete = False
+    proc._shutdown_response = None
+    monkeypatch.setattr(
+        "fastvideo.worker.multiproc_executor._shutdown_torch_compile_workers",
+        lambda: compile_shutdowns.append(True),
+    )
+
+    first = proc.shutdown()
+    second = proc.shutdown()
+
+    assert first == {"status": "shutdown"}
+    assert second == {"status": "shutdown"}
+    assert worker.shutdown_calls == 1
+    assert compile_shutdowns == [True]
+
+
 def test_compile_worker_cleanup_uses_only_an_already_loaded_inductor(monkeypatch) -> None:
     compile_shutdowns = []
     fake_async_compile = SimpleNamespace(shutdown_compile_workers=lambda: compile_shutdowns.append(True))

fastvideo/worker/multiproc_executor.py

@@ -522,6 +522,7 @@ def __init__(
         self.streaming_input_queue = streaming_input_queue
         self.streaming_output_queue = streaming_output_queue
         self._initial_log_handler = _initial_log_handler
+        self._shutdown_started = False
         self._shutdown_complete = False
         self._shutdown_response: dict[str, Any] | None = None
         wrapper = WorkerWrapperBase(fastvideo_args=fastvideo_args, rpc_rank=rank)
@@ -719,10 +720,26 @@ def shutdown(self) -> dict[str, Any]:
             assert self._shutdown_response is not None
             return self._shutdown_response
 
-        response = self.worker.shutdown()
-        _shutdown_torch_compile_workers()
-        self._shutdown_response = response
-        self._shutdown_complete = True
+        if getattr(self, "_shutdown_started", False):
+            if self._shutdown_response is None:
+                self._shutdown_response = {"status": "shutdown"}
+            return self._shutdown_response
+
+        self._shutdown_started = True
+        response = {"status": "shutdown"}
+        shutdown_succeeded = False
+
+        try:
+            response = self.worker.shutdown()
+            shutdown_succeeded = True
+        except Exception:
+            logger.exception("Worker %d failed to shut down", self.rank)
+        finally:
+            _shutdown_torch_compile_workers()
+            if shutdown_succeeded:
+                self._shutdown_complete = True
+            self._shutdown_response = response
+
         return response
 
     def worker_busy_loop(self) -> None: