From 100175ffde8c34af995f87655686f459eaa39164 Mon Sep 17 00:00:00 2001 From: laughingman7743 Date: Thu, 15 Oct 2026 00:17:14 +0900 Subject: [PATCH 1/4] Keep a notice observed while deferring a takeover A deferred takeover now carries the interruption notice that the lease watch observed after the attempt checked for one, for example from a request still in flight at that check. work() raises SpotInterruptionError on it instead of relying only on a fresh metadata check, which can fail and let the worker claim another job or retire a host being reclaimed. Fixes #1179 Co-Authored-By: Claude Opus 5.5 --- benchmarks/README.md | 2 +- benchmarks/pyathena_bench/fleet.py | 12 ++++---- benchmarks/tests/test_fleet.py | 46 ++++++++++++++++++++++++++++++ 3 files changed, 54 insertions(+), 6 deletions(-) diff --git a/benchmarks/README.md b/benchmarks/README.md index fb63a0b2d..e6a323fc3 100644 --- a/benchmarks/README.md +++ b/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//results/` 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. diff --git a/benchmarks/pyathena_bench/fleet.py b/benchmarks/pyathena_bench/fleet.py index d76bbff8b..c99ce8b9b 100644 --- a/benchmarks/pyathena_bench/fleet.py +++ b/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 @@ -1106,7 +1107,7 @@ def settle_error(path: Path, exc: Exception) -> str: 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 +1214,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 +1289,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 diff --git a/benchmarks/tests/test_fleet.py b/benchmarks/tests/test_fleet.py index a32c77cb6..9d64fd434 100644 --- a/benchmarks/tests/test_fleet.py +++ b/benchmarks/tests/test_fleet.py @@ -1640,6 +1640,52 @@ 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() + fired = 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 fired.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() + fired.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=0.5, 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): From e8b042ed9fca5bb9a9a206ba27c2533a0ec47a40 Mon Sep 17 00:00:00 2001 From: laughingman7743 Date: Thu, 15 Oct 2026 00:19:22 +0900 Subject: [PATCH 2/4] Read the observed notice once after the watch stops Co-Authored-By: Claude Opus 5.5 --- benchmarks/pyathena_bench/fleet.py | 8 +++++--- benchmarks/tests/test_fleet.py | 4 +--- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/benchmarks/pyathena_bench/fleet.py b/benchmarks/pyathena_bench/fleet.py index c99ce8b9b..068faf09d 100644 --- a/benchmarks/pyathena_bench/fleet.py +++ b/benchmarks/pyathena_bench/fleet.py @@ -1104,17 +1104,19 @@ def settle_error(path: Path, exc: Exception) -> str: break upload_attempt() + # The watch has stopped, so a notice it observed after the check above is final. + noticed = watch.interrupted.is_set() if deferred: # Nothing ran, so the next attempt resumes from the same earlier attempts. queue.end_lease(claim, "released") - return Outcome("deferred", noticed=watch.interrupted.is_set()) + return Outcome("deferred", noticed=noticed) def abandon() -> Outcome: # A claim lost only by its local expiry, for example while a renewal that # S3 accepted stalled, may still be this worker's; releasing it, which # works only then, returns the job to the fleet. queue.end_lease(claim, "released") - return Outcome("lost", tuple(errors), noticed=watch.interrupted.is_set()) + return Outcome("lost", tuple(errors), noticed=noticed) # The conditional renewal proves that this worker still owns the job, so a # former owner never records it, and leaves at least --lease-seconds for @@ -1159,7 +1161,7 @@ def abandon() -> Outcome: ) if recorded is None: return abandon() - return Outcome("done", tuple(errors), noticed=watch.interrupted.is_set()) + return Outcome("done", tuple(errors), noticed=noticed) def work( diff --git a/benchmarks/tests/test_fleet.py b/benchmarks/tests/test_fleet.py index 9d64fd434..8636dac13 100644 --- a/benchmarks/tests/test_fleet.py +++ b/benchmarks/tests/test_fleet.py @@ -1647,7 +1647,6 @@ def test_a_notice_observed_after_a_throttled_takeover_outlasts_a_failed_check(se armed = threading.Event() polling = threading.Event() uploading = threading.Event() - fired = threading.Event() upload_tree = queue.upload_tree def upload(*args, **kwargs): @@ -1661,13 +1660,12 @@ def notice(): # own check fails. if ( not armed.is_set() - or fired.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() - fired.set() return uploading.wait(5) def settle(output): From 1b13ccfc5ed53e0d4a31af581ead68dcbfa79d61 Mon Sep 17 00:00:00 2001 From: laughingman7743 Date: Thu, 15 Oct 2026 00:20:04 +0900 Subject: [PATCH 3/4] Give the deferred-notice test a wider lease Co-Authored-By: Claude Opus 5.5 --- benchmarks/tests/test_fleet.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/benchmarks/tests/test_fleet.py b/benchmarks/tests/test_fleet.py index 8636dac13..f41ffc230 100644 --- a/benchmarks/tests/test_fleet.py +++ b/benchmarks/tests/test_fleet.py @@ -1679,7 +1679,7 @@ def settle(output): 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=0.5, notice=notice) # fmt: skip + 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") From 34e02ac65c0e40ebb8a61da4cb80aca39614b6a5 Mon Sep 17 00:00:00 2001 From: laughingman7743 Date: Thu, 15 Oct 2026 00:25:56 +0900 Subject: [PATCH 4/4] Read the observed notice at each outcome again After a lost claim, JobWatch joins its threads for only 10 seconds, so a notice check can finish after a single early read; each outcome reads the event last, after releasing its lease. Co-Authored-By: Claude Opus 5.5 --- benchmarks/pyathena_bench/fleet.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/benchmarks/pyathena_bench/fleet.py b/benchmarks/pyathena_bench/fleet.py index 068faf09d..bbc4c588b 100644 --- a/benchmarks/pyathena_bench/fleet.py +++ b/benchmarks/pyathena_bench/fleet.py @@ -1104,19 +1104,19 @@ def settle_error(path: Path, exc: Exception) -> str: break upload_attempt() - # The watch has stopped, so a notice it observed after the check above is final. - noticed = watch.interrupted.is_set() + # 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", noticed=noticed) + 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 # S3 accepted stalled, may still be this worker's; releasing it, which # works only then, returns the job to the fleet. queue.end_lease(claim, "released") - return Outcome("lost", tuple(errors), noticed=noticed) + return Outcome("lost", tuple(errors), noticed=watch.interrupted.is_set()) # The conditional renewal proves that this worker still owns the job, so a # former owner never records it, and leaves at least --lease-seconds for @@ -1161,7 +1161,7 @@ def abandon() -> Outcome: ) if recorded is None: return abandon() - return Outcome("done", tuple(errors), noticed=noticed) + return Outcome("done", tuple(errors), noticed=watch.interrupted.is_set()) def work(