Fix #571: drain dead worker queues before closing them

WorkerSupervisor.recreate_if_dead() swapped in a fresh worker and then
closed the dead worker's result queues without draining them, so extract/
validate work the worker finished just before crashing was silently
dropped (the sole consumer, ModelUpdater, read the fresh worker's empty
queues after the swap).

The supervisor now drains the dead worker's completed/failed queues into a
carry-over buffer before _reap_dead() closes them, and surfaces them on the
next pop_completed()/pop_failed() — exactly once, ahead of the fresh
worker's results. ModelUpdater reads completed/failed through the
supervisor (statuses stay on .worker, display-only).

Hardening from adversarial review:
- Read the live worker before clearing the buffer, so a failed live read
  does not drop the carry-over.
- try/finally guarantees the dead worker's queues are still closed (FDs
  freed) even if draining raises.

Tests: reproduce the drop, exactly-once surfacing, rapid-restart
accumulation, per-queue drain-failure isolation, buffer survival on a
failing live read, and reap-on-drain-failure.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: nitrobass24

src/python/controller/model_updater.py

@@ -78,15 +78,18 @@ def __init__(
         self._move_retry_counts: dict[str, int] = {}
 
     def update(self) -> None:
-        # Grab the latest extract results (shared)
+        # Grab the latest extract results. Completed/failed go through the
+        # supervisor (not .worker) so results buffered from a dead worker during
+        # recreation are surfaced exactly once (#571); statuses are display-only
+        # and read straight off the live worker.
         latest_extract_statuses = self._extract_process.worker.pop_latest_statuses()
-        latest_extracted_results = self._extract_process.worker.pop_completed()
-        latest_failed_extractions = self._extract_process.worker.pop_failed()
+        latest_extracted_results = self._extract_process.pop_completed()
+        latest_failed_extractions = self._extract_process.pop_failed()
 
-        # Grab the latest validate results (shared)
+        # Grab the latest validate results (same #571 contract as extract).
         latest_validate_statuses = self._validate_process.worker.pop_latest_statuses()
-        latest_validated_results = self._validate_process.worker.pop_completed()
-        latest_failed_validations = self._validate_process.worker.pop_failed()
+        latest_validated_results = self._validate_process.pop_completed()
+        latest_failed_validations = self._validate_process.pop_failed()
 
         # Process each pair context's scan results and LFTP status
         for pc in self._pair_contexts:

src/python/controller/worker_supervisor.py

@@ -21,13 +21,27 @@
 
 import logging
 from collections.abc import Callable
-from typing import Any, Generic, TypeVar
+from typing import Any, Generic, Protocol, TypeVar, cast
 
 from common import AppProcess
 
 W = TypeVar("W", bound=AppProcess)
 
 
+class _ResultReader(Protocol):
+    """The result-queue drain surface shared by the Extract/Validate workers.
+
+    These live on the concrete workers rather than the ``AppProcess`` base (the
+    completed/failed result types differ between Extract and Validate), so the
+    supervisor reaches them via a cast when draining a dead worker before its
+    queues are closed (#571).
+    """
+
+    def pop_completed(self) -> list[Any]: ...
+
+    def pop_failed(self) -> list[Any]: ...
+
+
 class WorkerSupervisor(Generic[W]):
     """Owns the current live worker plus a factory that builds a fresh one.
 
@@ -52,6 +66,11 @@ def __init__(self, factory: Callable[[], W], logger: logging.Logger, feature: st
         # to the same cross-process logging queue before it is started.
         self._mp_log_queue: Any = None
         self._mp_log_level: int | None = None
+        # Results drained from a dead worker during recreation, surfaced on the
+        # next pop_completed()/pop_failed() so finished work emitted just before
+        # a crash is not silently dropped when its queues close (#571).
+        self._buffered_completed: list[Any] = []
+        self._buffered_failed: list[Any] = []
 
     @property
     def worker(self) -> W:
@@ -93,18 +112,51 @@ def close_queues(self) -> None:
     def propagate_exception(self) -> None:
         self._worker.propagate_exception()
 
+    def pop_completed(self) -> list[Any]:
+        """Drain completed results, prepended with any buffered from a dead worker.
+
+        Readers must go through the supervisor (not ``self.worker``) so results a
+        worker emitted just before crashing — captured by :meth:`_drain_dead`
+        during recreation — are surfaced exactly once, ahead of the fresh
+        worker's own results (#571). ModelUpdater drains this in the same
+        controller cycle as recreation, so the carry-over is picked up promptly.
+
+        The live worker is read *before* the buffer is cleared so that a failure
+        reading the live worker does not drop the carry-over: the buffer is left
+        intact and surfaces on a later cycle instead of being lost.
+        """
+        fresh = cast(_ResultReader, self._worker).pop_completed()
+        buffered = self._buffered_completed
+        self._buffered_completed = []
+        return buffered + fresh
+
+    def pop_failed(self) -> list[Any]:
+        """Drain failed results, prepended with any buffered from a dead worker (#571).
+
+        See :meth:`pop_completed` for the carry-over contract, including the
+        read-live-before-clearing-the-buffer ordering that keeps the carry-over
+        from being lost if the live read fails.
+        """
+        fresh = cast(_ResultReader, self._worker).pop_failed()
+        buffered = self._buffered_failed
+        self._buffered_failed = []
+        return buffered + fresh
+
     # --- delegated worker methods (dynamic) ---
 
     def __getattr__(self, item: str) -> Any:
         """Delegate the worker's own methods to the current live instance.
 
-        Covers the result-queue readers (``pop_latest_statuses``/
-        ``pop_completed``/``pop_failed``) and the request-dispatch methods
-        (``extract``/``validate``) — the worker surface that differs between
-        ExtractProcess and ValidateProcess and so cannot be typed on a single
-        ``Generic[W]`` supervisor. Callers that need the concrete return types
-        read them through :attr:`worker` (e.g. ``supervisor.worker.pop_completed()``);
-        ModelUpdater does this.
+        Covers the display-only status reader (``pop_latest_statuses``) and the
+        request-dispatch methods (``extract``/``validate``) — the worker surface
+        that differs between ExtractProcess and ValidateProcess and so cannot be
+        typed on a single ``Generic[W]`` supervisor. Callers that need the
+        concrete status type read it through :attr:`worker` (e.g.
+        ``supervisor.worker.pop_latest_statuses()``); ModelUpdater does this. The
+        completed/failed readers are *not* dynamic — they are the explicit
+        :meth:`pop_completed`/:meth:`pop_failed` above so they can carry over a
+        dead worker's buffered results (#571), and ModelUpdater calls those
+        through the supervisor rather than via :attr:`worker`.
 
         Only invoked for attributes not defined on the supervisor itself, so the
         explicit delegations above always win. ``_worker`` is set in
@@ -144,10 +196,37 @@ def recreate_if_dead(self) -> bool:
         # Swap before reaping the old worker so a failure releasing the dead
         # worker's queues still leaves the supervisor pointing at the live one.
         self._worker = fresh
-        self._reap_dead(dead)
+        # Capture any finished results still in the dead worker's queues before
+        # _reap_dead() closes them, so they are not silently dropped (#571). The
+        # finally guarantees the dead worker's queues are still closed (FDs
+        # freed) even in the unlikely event draining itself raises.
+        try:
+            self._drain_dead(dead)
+        finally:
+            self._reap_dead(dead)
         self._logger.info("%s worker process restarted as %s", self._feature, fresh.name)
         return True
 
+    def _drain_dead(self, dead: W) -> None:
+        """Buffer results the dead worker emitted before crashing (#571).
+
+        Runs before :meth:`_reap_dead` closes the dead worker's queues. Each
+        queue is drained in isolation and best-effort: a worker mid-crash may
+        raise on a read, and a drain failure must never block the restart. The
+        results are *extended* onto the buffers (not assigned) so repeated
+        restarts before a consumer drain accumulate rather than clobber. They
+        surface on the next :meth:`pop_completed`/:meth:`pop_failed`.
+        """
+        reader = cast(_ResultReader, dead)
+        try:
+            self._buffered_completed.extend(reader.pop_completed())
+        except Exception:
+            self._logger.debug("Error draining dead %s worker completions during restart", self._feature, exc_info=True)
+        try:
+            self._buffered_failed.extend(reader.pop_failed())
+        except Exception:
+            self._logger.debug("Error draining dead %s worker failures during restart", self._feature, exc_info=True)
+
     def _is_dead(self) -> bool:
         """True if the worker was started and has since exited."""
         try:

src/python/tests/unittests/test_controller/test_worker_supervisor.py

@@ -143,6 +143,100 @@ def test_subsequent_dispatch_goes_to_fresh_worker(self):
         self.replacement.pop_completed.assert_called_once()
         self.initial.extract.assert_not_called()
 
+    def test_recreate_drains_dead_worker_results_before_close(self):
+        # #571: results the dead worker emitted just before crashing must be
+        # surfaced through the supervisor, not silently dropped when its queues
+        # are closed. Buffer them on the dead worker, kill it, recreate.
+        c1, c2 = MagicMock(name="completed-1"), MagicMock(name="completed-2")
+        f1 = MagicMock(name="failed-1")
+        self.initial.pop_completed.return_value = [c1, c2]
+        self.initial.pop_failed.return_value = [f1]
+        self.initial.is_alive.return_value = False
+        self.initial.exitcode = 1
+
+        created = self.sup.recreate_if_dead()
+        self.assertTrue(created)
+
+        # The dead worker was drained before close_queues() closed its FDs.
+        self.initial.close_queues.assert_called_once()
+        # The buffered results are surfaced exactly once via the supervisor,
+        # ahead of the (empty) fresh worker's results.
+        self.assertEqual([c1, c2], self.sup.pop_completed())
+        self.assertEqual([f1], self.sup.pop_failed())
+        # And only once: a second drain yields just the fresh worker's results.
+        self.assertEqual([], self.sup.pop_completed())
+        self.assertEqual([], self.sup.pop_failed())
+
+    def test_rapid_restarts_accumulate_buffered_results(self):
+        # Two deaths before the consumer drains must accumulate, not clobber (#571).
+        w1 = _make_worker("W-1", alive=False, exitcode=1)
+        w1.pop_completed.return_value = ["a"]
+        w2 = _make_worker("W-2", alive=False, exitcode=1)
+        w2.pop_completed.return_value = ["b"]
+        w3 = _make_worker("W-3", alive=True, exitcode=None)
+        sup: WorkerSupervisor = WorkerSupervisor(
+            factory=_FakeFactory([w1, w2, w3]), logger=self.logger, feature="Extract"
+        )
+
+        sup.recreate_if_dead()  # w1 dies -> buffer ["a"], swap to w2
+        sup.recreate_if_dead()  # w2 dies -> buffer ["a", "b"], swap to w3
+
+        self.assertIs(w3, sup.worker)
+        self.assertEqual(["a", "b"], sup.pop_completed())
+
+    def test_drain_failure_is_isolated_and_does_not_block_restart(self):
+        # A worker mid-crash may raise on one queue read; the restart must still
+        # happen and the other queue's results must still be surfaced (#571).
+        self.initial.is_alive.return_value = False
+        self.initial.exitcode = 1
+        self.initial.pop_completed.side_effect = RuntimeError("queue broken")
+        self.initial.pop_failed.return_value = ["f"]
+
+        created = self.sup.recreate_if_dead()
+
+        self.assertTrue(created)
+        self.assertIs(self.replacement, self.sup.worker)
+        self.initial.close_queues.assert_called_once()
+        # The unreadable completed queue is skipped; the readable failed queue
+        # is still surfaced.
+        self.assertEqual([], self.sup.pop_completed())
+        self.assertEqual(["f"], self.sup.pop_failed())
+
+    def test_buffered_results_survive_a_failing_live_read(self):
+        # #571: if reading the fresh worker raises (e.g. it too crashed and its
+        # queue is closed), the carry-over buffer must NOT be cleared/lost — it
+        # has to survive to a later cycle once a healthy worker can be read.
+        self.initial.pop_completed.return_value = ["carried"]
+        self.initial.is_alive.return_value = False
+        self.initial.exitcode = 1
+        self.sup.recreate_if_dead()  # buffers ["carried"], swaps to replacement
+
+        # The fresh worker's own read blows up on the next drain.
+        self.replacement.pop_completed.side_effect = ValueError("Queue is closed")
+        with self.assertRaises(ValueError):
+            self.sup.pop_completed()
+
+        # Buffer was preserved despite the failed read; once the worker recovers,
+        # the carried result is still surfaced exactly once.
+        self.replacement.pop_completed.side_effect = None
+        self.replacement.pop_completed.return_value = ["fresh"]
+        self.assertEqual(["carried", "fresh"], self.sup.pop_completed())
+        self.assertEqual(["fresh"], self.sup.pop_completed())
+
+    def test_reap_runs_even_if_draining_raises(self):
+        # Defense in depth: if draining the dead worker unexpectedly raises (e.g.
+        # the logger itself fails inside _drain_dead), its queues must still be
+        # closed so file descriptors are freed (#571).
+        self.initial.is_alive.return_value = False
+        self.initial.exitcode = 1
+        self.initial.pop_completed.side_effect = RuntimeError("queue broken")
+        self.logger.debug.side_effect = RuntimeError("logging down")
+
+        with self.assertRaises(RuntimeError):
+            self.sup.recreate_if_dead()
+
+        self.initial.close_queues.assert_called_once()
+
     def test_recreate_survives_dead_worker_close_failure(self):
         # Releasing the dead worker's queues must not abort the swap.
         self.initial.is_alive.return_value = False
@@ -230,13 +324,18 @@ def test_pipeline_recreate_rewires_updater_reads(self):
         sup_logger.warning.assert_called()
         sup_logger.info.assert_called()
 
-        # ModelUpdater.update() now drains the FRESH worker's queues, not the
-        # dead one's — the swap re-wired the updater through the shared supervisor.
+        # ModelUpdater.update() reads completed/failed through the shared
+        # supervisor, so the swap re-wires it to the fresh worker; statuses are
+        # read straight off the fresh worker.
         updater.update()
         replacement.pop_latest_statuses.assert_called()
         replacement.pop_completed.assert_called()
         replacement.pop_failed.assert_called()
-        initial.pop_completed.assert_not_called()
+        # The dead worker is drained exactly once during recreation (#571) so its
+        # last-moment results are surfaced rather than dropped, and is not read
+        # again by the updater afterward.
+        initial.pop_completed.assert_called_once()
+        initial.pop_failed.assert_called_once()
 
     def test_pipeline_dispatch_after_recreate_targets_fresh_worker(self):
         logger = MagicMock()