Merge pull request #101 from randileeharper/fix/playback-pool-shutdown

fix(playback): drain in-flight snapshot tasks on close

Author: randileeharper

tests/test_service.py

@@ -1664,6 +1664,68 @@ def test_playback_snapshot_reuses_instance_thread_pool(service) -> None:
     assert service._playback_ctrl._executor is executor_before
 
 
+def test_playback_close_drains_in_flight_snapshot(service) -> None:
+    # close() must wait for in-flight playback_snapshot() fan-out tasks to
+    # finish before returning, instead of orphaning worker threads with
+    # shutdown(wait=False) (issue #89).
+    import threading as _threading
+    from concurrent.futures import ThreadPoolExecutor
+
+    ctrl = service._playback_ctrl
+    rpc = service._rpc
+
+    # Gate that keeps the first RPC call blocked until we release it.
+    gate = _threading.Event()
+    call_started = _threading.Event()
+
+    original_get = rpc.playback_get
+
+    def gated_playback_get(path: str):
+        if path == "/is-playing" and not gate.is_set():
+            call_started.set()
+            gate.wait(timeout=10)
+        return original_get(path)
+
+    rpc.playback_get = gated_playback_get
+
+    # Start a snapshot in a background thread. It will block on the gate.
+    snapshot_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="test-snapshot")
+    snapshot_future = snapshot_executor.submit(ctrl.playback_snapshot)
+
+    assert call_started.wait(timeout=5), "gated RPC call should have started"
+
+    # Now call close() while the snapshot fan-out is in flight. It should
+    # not return until the in-flight tasks complete (after we release the
+    # gate). We verify ordering: close() blocks, then we release the gate,
+    # then close() returns.
+    close_done = _threading.Event()
+
+    def call_close():
+        ctrl.close()
+        close_done.set()
+
+    close_thread = _threading.Thread(target=call_close, name="test-close")
+    close_thread.start()
+
+    # close() should be blocked waiting for the in-flight future.
+    assert not close_done.wait(timeout=0.3), "close() should not return while tasks are in flight"
+
+    # Release the gate so the in-flight snapshot can complete.
+    gate.set()
+
+    # close() should now return promptly (within the drain timeout).
+    assert close_done.wait(timeout=6), "close() should return after in-flight tasks drain"
+    close_thread.join(timeout=5)
+
+    # The background snapshot should also have completed successfully.
+    snapshot_result = snapshot_future.result(timeout=5)
+    assert snapshot_result["status"] == "ok"
+    snapshot_executor.shutdown(wait=True)
+
+    # Restore the stub so the fixture teardown doesn't hit the gated version.
+    rpc.playback_get = original_get
+
+
 def test_handle_text_request_includes_raw_output_when_enabled(settings, service, tmp_path) -> None:
     class RawStubResolver:
         def resolve(self, text: str, service) -> ResolvedAction:

vesper/playback_controller.py

@@ -18,7 +18,8 @@
 
 from __future__ import annotations
 
-from concurrent.futures import ThreadPoolExecutor
+import threading
+from concurrent.futures import Future, ThreadPoolExecutor
 from typing import Any, Protocol
 
 from .catalog import (
@@ -69,6 +70,10 @@ class PlaybackController:
     # every call to this hot path. See #67.
     _POOL_WORKERS = 14
 
+    # Per-future timeout when draining in-flight snapshots in close(). Bounded
+    # so a hung RPC can't block close() indefinitely (issue #89).
+    _CLOSE_DRAIN_TIMEOUT_SECONDS = 5.0
+
     def __init__(
         self,
         host: PlaybackHost,
@@ -84,10 +89,30 @@ def __init__(
         self._executor = ThreadPoolExecutor(
             max_workers=self._POOL_WORKERS, thread_name_prefix="playback-snapshot"
         )
+        # In-flight snapshot futures, guarded by _pending_lock so close() can
+        # drain a consistent set while playback_snapshot() submits new work.
+        # See #89.
+        self._pending_futures: set[Future[Any]] = set()
+        self._pending_lock = threading.Lock()
 
     def close(self) -> None:
-        """Release the shared snapshot thread pool. Idempotent."""
-        self._executor.shutdown(wait=False)
+        """Release the shared snapshot thread pool. Idempotent.
+
+        Signals the pool to stop accepting work, then waits a bounded time for
+        any in-flight playback_snapshot() fan-out to finish so worker threads
+        are not orphaned and can't raise at interpreter shutdown (issue #89).
+        """
+        # Stop accepting new submissions first. cancel_futures=True (3.9+)
+        # drops not-yet-started tasks; started ones still run to completion.
+        self._executor.shutdown(wait=False, cancel_futures=True)
+        with self._pending_lock:
+            pending = list(self._pending_futures)
+        # Drain in-flight tasks with a short bounded timeout per future. If a
+        # snapshot RPC hangs, we don't block close() indefinitely — the worker
+        # is a daemon thread and won't block interpreter exit, but we give
+        # well-behaved tasks a chance to complete.
+        for future in pending:
+            future.result(timeout=self._CLOSE_DRAIN_TIMEOUT_SECONDS)
 
     def status(self) -> dict[str, Any]:
         playback = self.playback_snapshot()
@@ -152,11 +177,19 @@ def playback_snapshot(self) -> dict[str, Any]:
         # Fan out the 7 playback reads on the reused instance pool rather than
         # creating a ThreadPoolExecutor per call. Each future's result() is
         # awaited below, so all submissions complete before we return. See #67.
+        # Futures are registered in _pending_futures so close() can drain them
+        # if it runs while this fan-out is in flight (issue #89).
         futures = {
             name: self._executor.submit(self._rpc.playback_get, path)
             for name, path in snapshot_paths.items()
         }
-        payloads = {name: future.result() for name, future in futures.items()}
+        with self._pending_lock:
+            self._pending_futures.update(futures.values())
+        try:
+            payloads = {name: future.result() for name, future in futures.items()}
+        finally:
+            with self._pending_lock:
+                self._pending_futures.difference_update(futures.values())
         is_playing_payload = payloads["is_playing"]
         now_playing_payload = payloads["now_playing"]
         volume_payload = payloads["volume"]