Keep a notice observed while deferring a fleet takeover

benchmarks/README.md

@@ -491,7 +491,7 @@ Each worker checks instance metadata for a Spot interruption notice (`spot/insta
 On a notice, the worker interrupts the running trial with SIGINT, which cancels the trial's reported queries, uploads the attempt, releases its claim as `interrupted` and its slots, and exits without terminating the host.
 A separate thread watches for the notice, so slow S3 requests do not delay it.
 The group replaces the reclaimed instance, and the new host's worker takes over the interrupted claim as the next attempt: it downloads the earlier attempts, quiesces their trials as after a failed job, and runs its own warmups and then only the measured repetitions that they did not record as successful, through `run --resume`.
-If Athena still throttles a request after its retries while the worker quiesces an earlier attempt, the worker releases the claim without running or recording the job and continues with the next pending job; a later attempt on any host quiesces the earlier attempts again, and the worker keeps its host while the job is pending.
+If Athena still throttles a request after its retries while the worker quiesces an earlier attempt, the worker releases the claim without running or recording the job and continues with the next pending job, or exits after an interruption notice, including one observed while it quiesced; a later attempt on any host quiesces the earlier attempts again, and the worker keeps its host while the job is pending.
 The job's summary merges the attempts and leaves a failed or incomplete trial whose repetition succeeded in a later attempt out of `failed_trials`, noting its reason as `Superseded: ...`, so it reports the trials of an uninterrupted run without hiding failures.
 `report results/recovered/fleet/<name>/results/<job>` produces the same merged summary from the recovered attempts.
 Rebalance recommendations are not acted on; an interruption does not always follow them, and the notice leaves enough time to stop a trial.

benchmarks/pyathena_bench/fleet.py

@@ -980,8 +980,9 @@ class Outcome:
     worker took over the job's claim or a slot, and ``deferred`` when Athena
     throttled the quiescing of an earlier attempt, so the job was released
     without running. ``errors`` lists queries that could not be confirmed as
-    stopped. ``noticed`` tells whether an interruption notice was observed
-    after the run, while its queries were quiesced.
+    stopped. ``noticed`` tells whether the watch observed an interruption
+    notice after the attempt checked for one, for example while queries were
+    quiesced.
     """
 
     state: str
@@ -1103,10 +1104,12 @@ def settle_error(path: Path, exc: Exception) -> str:
                     break
         upload_attempt()
 
+    # Each outcome reads the watch's notice last: after a lost claim, the watch
+    # is joined for only 10 seconds, and a notice check may still finish.
     if deferred:
         # Nothing ran, so the next attempt resumes from the same earlier attempts.
         queue.end_lease(claim, "released")
-        return Outcome("deferred")
+        return Outcome("deferred", noticed=watch.interrupted.is_set())
 
     def abandon() -> Outcome:
         # A claim lost only by its local expiry, for example while a renewal that
@@ -1213,7 +1216,8 @@ def work(
             records are uploaded and its claim is released as interrupted, or
             once the attempt of a failed job ends, if the notice arrived while
             its queries were quiesced or is found after they could not be
-            confirmed as stopped.
+            confirmed as stopped, or once a deferred takeover released its
+            claim.
         RuntimeError: If an attempt stopped by a lost claim could not be quiesced;
             the worker's host is then kept.
     """
@@ -1287,7 +1291,7 @@ def settle(output: Path) -> list[str]:
                     _logger.warning(
                         f"Deferred {job['id']}: Athena throttled quiescing its earlier attempts"
                     )
-                    if check_notice(notice):
+                    if outcome.noticed or check_notice(notice):
                         raise SpotInterruptionError(completed)
                     # A later pass, or another worker, retries the job; try the next one.
                     continue

benchmarks/tests/test_fleet.py

@@ -1640,6 +1640,50 @@ def settle(output):
         assert queue.read_lease(queue.key("claims", "a")).takeable(time.time())
         assert "b" not in queue.names("claims")
 
+    def test_a_notice_observed_after_a_throttled_takeover_outlasts_a_failed_check(self, tmp_path):
+        _, queue = queue_with([job("a"), job("b")], tmp_path)
+        queue.acquire(queue.key("claims", "a"), {"host": "dead", "job": "a"}, -1)
+        queue.put(queue.key("results", "a", "attempt-1", "events.jsonl"), {})
+        armed = threading.Event()
+        polling = threading.Event()
+        uploading = threading.Event()
+        upload_tree = queue.upload_tree
+
+        def upload(*args, **kwargs):
+            uploading.set()
+            return upload_tree(*args, **kwargs)
+
+        queue.upload_tree = upload
+
+        def notice():
+            # Only one request of the watch thread sees the notice; the worker's
+            # own check fails.
+            if (
+                not armed.is_set()
+                or polling.is_set()
+                or threading.current_thread() is threading.main_thread()
+            ):
+                return False
+            # The request is still in flight when the attempt checks for a notice.
+            polling.set()
+            return uploading.wait(5)
+
+        def settle(output):
+            if armed.is_set():
+                return []
+            armed.set()
+            assert polling.wait(5)
+            raise throttling("ListQueryExecutions")
+
+        # Without the observed notice, the worker would run "b" and then "a".
+        with pytest.raises(SpotInterruptionError) as raised:
+            work(queue, tmp_path / "w", api_slots=1, poll_seconds=0,
+                 runner=recording_runner([], trials={"a": [ok_trial(0)], "b": [ok_trial(0)]}),
+                 settle=settle, lease_seconds=2, notice=notice)  # fmt: skip
+        assert raised.value.ran == []
+        assert queue.read_lease(queue.key("claims", "a")).takeable(time.time())
+        assert "b" not in queue.names("claims")
+
 
 class TestWorkerCommand:
     def test_worker_uses_the_results_location_without_the_stack(self, tmp_path, monkeypatch):