From 17e3d20f60de8a25d877adc06efb59741de3ff89 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Tue, 27 Oct 2026 08:26:17 -0700 Subject: [PATCH 1/3] ci: owned-pool rescue never cancels a started nightly for a slow runner assess() counted only owned in_progress jobs as blockers, so a nightly whose trusted-pool build waited was cancelled while a Blacksmith or Linux sibling ran. For a nightly (reserved) any in_progress job, on any pool and in its runner's setup too, blocks the stuck-job and setup-hold cancels, even at the watch's end; a refusal at the watch limit leaves a running nightly alone. rescue() reads the jobs again right before any cancel, and the cancel's before_send (#26173) drops it once the run moved to another attempt, finished, or has a job running. A nightly found busy is watched on, through the stall hold. Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?id=aid-85ca59f74f02d81734c3 --- scripts/ci/owned_pool_rescue.py | 102 +++++++++++++- tests/test_ci_owned_pool_rescue.py | 213 +++++++++++++++++++++++++++-- 2 files changed, 293 insertions(+), 22 deletions(-) diff --git a/scripts/ci/owned_pool_rescue.py b/scripts/ci/owned_pool_rescue.py index 5fc4645a6cf5..8058ffd59276 100644 --- a/scripts/ci/owned_pool_rescue.py +++ b/scripts/ci/owned_pool_rescue.py @@ -1129,6 +1129,10 @@ def assess(jobs: Sequence[Mapping[str, Any]], *, now: dt.datetime, budget_second # when another side job is still running. Manual reruns stay minis-first # through attempt 2; the bot's next rerun is the overflow. on_mini = [job for job in jobs if job_pool(job) and job.get("status") == "in_progress" and not in_setup(job)] + if reserved: + # A started nightly is never cancelled for a slow runner: any job of it a runner has taken, anywhere + # (a Blacksmith helper build, a Linux sibling, one still in its runner's setup), blocks the cancel. + on_mini = [job for job in jobs if job.get("status") == "in_progress"] # A job that failed on its own (not refused by the fleet) has already made the run red. Cancelling # and re-running it repeats that failure and puts the stuck job at the back of the same queue # (feat-cmux-next, 2026-10-20: 33 of 44 cmux-next pull request runs reached attempt 2 or 3, each @@ -1146,8 +1150,12 @@ def assess(jobs: Sequence[Mapping[str, Any]], *, now: dt.datetime, budget_second # shards that way to a refusal), so a job held in setup waits for them, as a # refusal does, until the watch is about to end. running = [job for job in jobs if job.get("status") == "in_progress" and not in_setup(job)] + if reserved: + # A nightly job held in setup has started too (and so has any sibling in its setup): never cancelled. + running = [job for job in jobs if job.get("status") == "in_progress"] closing = deadline is not None and now >= deadline - dt.timedelta(seconds=END_MARGIN_SECONDS) - if held and not decided and (not running or closing): + # A nightly's running sibling is never cancelled, not even as the watch ends. + if held and not decided and (not running or closing and not reserved): names = ", ".join(sorted(str(job.get("name") or job.get("id")) for job in held)) return Look("rescue", f"{names} waited in {job_pool(held[0])}'s runner setup for capacity for at least " f"{SETUP_WAIT_SECONDS}s") @@ -1169,7 +1177,9 @@ def assess(jobs: Sequence[Mapping[str, Any]], *, now: dt.datetime, budget_second return Look("watch", f"{held_for} waiting for the runner reserved for it; never moved", waiting=True) if stuck: return Look("watch", f"{names} queued on {job_pool(stuck[0])} with no runner, but " - f"{len(on_mini)} job(s) of the run are running on a persistent runner", waiting=True) + f"{len(on_mini)} job(s) of the run are running" + + (" (a started nightly is never cancelled)" if reserved else " on a persistent runner"), + waiting=True) if waiting or settling: return Look("watch", f"{len(waiting)} job(s) waiting for a persistent runner, {len(settling)} in its setup", waiting=True) @@ -2058,6 +2068,14 @@ def watch(api: GitHub, target: Target, *, budget_seconds: int, wait = min(api.pace(wait), max(1.0, (deadline - now()).total_seconds() + 1)) sleep(wait) continue + if look.action == "refused" and not finished and target.nightly and any( + job.get("status") == "in_progress" for job in jobs): + # At the watch's end a refusal in a nightly whose other jobs still run is left as it is: + # cancelling it to re-run would kill started nightly work. The nightly-next tick re-runs a + # failed tip, and the next nightly builds main. + log(f"::warning::look {looks}: {look.reason}; the nightly still runs other jobs at the watch " + "limit, so it is not cancelled") + return "stop", "watch limit reached with a started nightly still running" log(f"look {looks}: {look.reason}") if look.action in ("rescue", "refused"): return look.action, look.reason @@ -2376,6 +2394,53 @@ def pull_moved(api: GitHub, target: Target, sleep: Callable[[float], None], return "" +# rescue()'s answer when a nightly it was about to cancel had a job running: follow() then watches the run again. +NIGHTLY_BUSY = "not rescued: started nightly work is running" + + +class NightlyBusy(Aborted): + """A nightly's job started during a rate-limit wait before a retried cancel (nightly_cancel()).""" + + +def nightly_cancel(api: GitHub, target: Target, sleep: Callable[[float], None], log: Callable[[str], None], + before_cancel: Callable[[], None] | None = None) -> None: + """api.cancel. For a nightly, its guard runs right before the cancel is sent and before every re-send + (GitHub.cancel()'s before_send): it drops the cancel once the run moved to another attempt or finished, and + raises NightlyBusy once a job of it runs. `before_cancel` (rescue()'s own guard) runs first.""" + def check() -> None: + if before_cancel is not None: + before_cancel() + if not target.nightly: + return + run = api.run(target.run_id) or {} + attempt = int(run.get("run_attempt") or 0) + if attempt != target.attempt: + # Someone re-ran it meanwhile: the cancel would hit that newer attempt's running jobs. + raise Aborted(f"nightly run {target.run_id} is at attempt {attempt}, not {target.attempt}; not cancelled") + if run.get("status") == "completed": + raise Aborted(f"nightly run {target.run_id} has finished; nothing to cancel") + busy = started_nightly_work(api, target, sleep, log) + if busy: + raise NightlyBusy(busy) + if before_cancel is None and not target.nightly: + api.cancel(target.run_id) + return + api.cancel(target.run_id, before_send=check) + + +def started_nightly_work(api: GitHub, target: Target, sleep: Callable[[float], None], + log: Callable[[str], None]) -> str: + """For a nightly, the jobs of it running right now (any pool: owned, Blacksmith, Linux), read just before a + cancel: a started nightly is never cancelled for a slow runner or a newer tip. "" when none (or not a nightly). + + assess() checked the same on its look, but a sibling can start between that look and the cancel.""" + if not target.nightly: + return "" + jobs = read(lambda: api.jobs(target.run_id, target.attempt), sleep, log) + running = sorted(str(job.get("name") or job.get("id")) for job in jobs if job.get("status") == "in_progress") + return ", ".join(running) + + def rescue(api: GitHub, target: Target, *, now: Callable[[], dt.datetime], sleep: Callable[[float], None], log: Callable[[str], None], failed_only: bool = False, deadline: dt.datetime | None = None, refused: bool | None = None, refusal: bool = False, @@ -2416,7 +2481,13 @@ def rescue(api: GitHub, target: Target, *, now: Callable[[], dt.datetime], sleep return "not rescued: someone else already re-ran the run" if run.get("status") == "completed": return f"not rescued: {moved}" - api.cancel(target.run_id) + busy = started_nightly_work(api, target, sleep, log) + if busy: + return f"{NIGHTLY_BUSY} ({busy}); {moved}, but a started nightly is never cancelled" + try: + nightly_cancel(api, target, sleep, log) + except NightlyBusy as busy_now: + return f"{NIGHTLY_BUSY} ({busy_now}); {moved}, but a started nightly is never cancelled" return f"cancelled run {target.run_id}, not re-run: {moved}" if moved: return f"not rescued: {moved}" @@ -2452,10 +2523,13 @@ def rescue(api: GitHub, target: Target, *, now: Callable[[], dt.datetime], sleep slot = started_dev_compile(jobs) if slot: return f"not rescued: a DEV feed compile started on {slot}; a started dev build is never cancelled" - if before_cancel is not None: - api.cancel(target.run_id, before_send=before_cancel) - else: - api.cancel(target.run_id) + busy = started_nightly_work(api, target, sleep, log) + if busy: + return f"{NIGHTLY_BUSY} ({busy}); a started nightly is never cancelled" + try: + nightly_cancel(api, target, sleep, log, before_cancel) + except NightlyBusy as busy_now: + return f"{NIGHTLY_BUSY} ({busy_now}); it started just before the cancel, and a started nightly is never cancelled" log(f"cancelled run {target.run_id}") started = now() # Time spent waiting through rate limits (GitHub.request) is not the run's settling time: @@ -2776,6 +2850,20 @@ def capped(deadline: dt.datetime) -> dt.datetime: refused=(outcome == "refused") if target.e2e or target.side else None, refusal=outcome == "refused") log(result) + if result.startswith(NIGHTLY_BUSY): + # A sibling started after the look: watch on, as before, until nothing else runs or the watch ends. + if clock() >= deadline: + return "stopped: watch limit reached with a started nightly still running" + outcome, reason = watch(client, target, budget_seconds=first_budget, now=clock, sleep=sleep, log=log, + deadline=deadline, floor_seconds=effective_seconds, seen=seen) + if outcome == "rescue": + outcome, reason, held = hold_for_destination( + client, target, outcome, reason, budget_seconds=first_budget, floor_seconds=effective_seconds, + deadline=deadline, now=clock, sleep=sleep, log=log, seen=seen) + if outcome not in ("rescue", "refused"): + return f"stopped: {reason}" + (f"; kept on the owned pool: {held}" if held else "") + log(f"{'rescue' if outcome == 'rescue' else 'refused'}: {reason}") + continue # Only an E2E full re-run picks a new runner in this watch. Other CI # re-runs are resumed by the sweeper through owned_reruns(). if not target.e2e or not result.startswith("re-ran every job") or target.attempt + 1 > LAST_OWNED_ATTEMPT: diff --git a/tests/test_ci_owned_pool_rescue.py b/tests/test_ci_owned_pool_rescue.py index 9041d67375ad..43d068d76a12 100644 --- a/tests/test_ci_owned_pool_rescue.py +++ b/tests/test_ci_owned_pool_rescue.py @@ -2170,11 +2170,13 @@ def nightly_event(**overrides): "pull_requests": [], **overrides}) -def nightly_run(*, queued_at=15, started_at=None, refused_at=None, runner_label=None, setup_until=None): +def nightly_run(*, queued_at=15, started_at=None, refused_at=None, runner_label=None, setup_until=None, + helper_done_at=None): """nightly.yml: decide, then the app build on the trusted pool beside a Blacksmith helper build. `runner_label` adds the reserved runner's own label, as nightly.yml's runs-on does; `setup_until` - keeps a started build in its runner's setup until then (glaeda's hook waiting for capacity).""" + keeps a started build in its runner's setup until then (glaeda's hook waiting for capacity). The + helper build runs until `helper_done_at` (never finishing by default).""" labels = [TRUSTED, *([runner_label] if runner_label else [])] def jobs(seconds): @@ -2193,8 +2195,12 @@ def jobs(seconds): app["steps"] = [{"name": "Set up runner", "started_at": stamp(started_at), "status": "in_progress" if seconds < setup_until else "completed", "conclusion": None if seconds < setup_until else "success"}] - return found + [app, job("build-nightly-ghostty-cli-helper", labels=["blacksmith-6vcpu-macos-15"], - status="in_progress", runner="bs")] + helper_done = helper_done_at is not None and seconds >= helper_done_at + helper = job("build-nightly-ghostty-cli-helper", labels=["blacksmith-6vcpu-macos-15"], + status="completed" if helper_done else "in_progress", runner="bs") + if helper_done: + helper["conclusion"] = "success" + return found + [app, helper] return jobs @@ -2279,7 +2285,7 @@ def test_a_newer_next_push_never_cancels_this_one(self): self.assertIsInstance(target, rescue.Target) self.assertEqual(target.push_branch, "nightly-next") clock = Clock() - api = NightlyAPI(clock, nightly_run(refused_at=60), finished=lambda s: s >= 200, + api = NightlyAPI(clock, nightly_run(refused_at=60, helper_done_at=200), finished=lambda s: s >= 200, newer=[RUN_ID + 7], head="b" * 40, pushes=[RUN_ID + 7]) run_main(api, clock, payload=payload, env_extra={"RESCUE_SECONDS": "30"}) self.assertNotIn("cancel", api.calls) @@ -2288,7 +2294,7 @@ def test_a_newer_next_push_never_cancels_this_one(self): def test_next_refusal_moves_failed_jobs_to_blacksmith(self): clock = Clock() - api = NightlyAPI(clock, nightly_run(refused_at=60), finished=lambda s: s >= 200) + api = NightlyAPI(clock, nightly_run(refused_at=60, helper_done_at=200), finished=lambda s: s >= 200) code, summary = run_main(api, clock, payload=nightly_event(head_branch="nightly-next")) self.assertEqual(code, 0) self.assertIn("rerun-failed", api.calls) @@ -2457,8 +2463,8 @@ def test_a_build_held_in_its_reserved_runner_s_setup_is_never_cancelled(self): def test_a_refused_reserved_build_is_still_moved(self): clock = Clock() - api = NightlyAPI(clock, nightly_run(refused_at=60, runner_label="glaeda-runner-cmux22-nightly-next"), - finished=lambda s: s >= 200) + api = NightlyAPI(clock, nightly_run(refused_at=60, runner_label="glaeda-runner-cmux22-nightly-next", + helper_done_at=200), finished=lambda s: s >= 200) _, summary = run_main(api, clock, payload=nightly_event(head_branch="nightly-next")) self.assertEqual(api.calls[-1], "rerun-failed") @@ -2514,38 +2520,215 @@ def test_only_a_nightly_honours_a_runner_label(self): def test_a_superseded_main_commit_is_never_rerun(self): # On main: a newer push run that is already running or done, not pending. clock = Clock() - api = NightlyAPI(clock, nightly_run(refused_at=60), finished=lambda s: s >= 200, + api = NightlyAPI(clock, nightly_run(refused_at=60, helper_done_at=200), finished=lambda s: s >= 200, head="c" * 40, pushes=[RUN_ID + 4]) _, summary = run_main(api, clock, payload=nightly_event()) self.assertNotIn("rerun-failed", api.calls) self.assertIn("main has moved on to cccccccccccc", summary) # A newer push run on the same head (a re-push) does not supersede it. clock = Clock() - api = NightlyAPI(clock, nightly_run(refused_at=60), finished=lambda s: s >= 200, pushes=[RUN_ID + 3]) + api = NightlyAPI(clock, nightly_run(refused_at=60, helper_done_at=200), finished=lambda s: s >= 200, pushes=[RUN_ID + 3]) run_main(api, clock, payload=nightly_event(head_branch="nightly-next")) self.assertEqual(api.calls[-1], "rerun-failed") # A moved head with no newer push run (outside the workflow's paths) does not either. clock = Clock() - api = NightlyAPI(clock, nightly_run(refused_at=60), finished=lambda s: s >= 200, head="b" * 40) + api = NightlyAPI(clock, nightly_run(refused_at=60, helper_done_at=200), finished=lambda s: s >= 200, head="b" * 40) run_main(api, clock, payload=nightly_event()) self.assertEqual(api.calls[-1], "rerun-failed") - def test_a_stuck_build_moves_to_blacksmith_keeping_what_passed(self): + def test_a_stuck_build_never_cancels_a_started_sibling(self): + # The Blacksmith helper build is mid-job: a started nightly is never cancelled for a slow runner, + # not even at the watch's end. clock = Clock() api = NightlyAPI(clock, nightly_run()) code, summary = run_main(api, clock, payload=nightly_event(), env_extra={"RESCUE_SECONDS": "30"}) self.assertEqual(code, 0) + self.assertNotIn("cancel", api.calls) + self.assertNotIn("rerun-failed", api.calls) + self.assertIn("a started nightly is never cancelled", summary) + self.assertIn("watch limit reached", summary) + + def test_a_stuck_build_moves_to_blacksmith_keeping_what_passed_once_nothing_else_runs(self): + clock = Clock() + api = NightlyAPI(clock, nightly_run(helper_done_at=300)) + code, summary = run_main(api, clock, payload=nightly_event(), env_extra={"RESCUE_SECONDS": "30"}) + self.assertEqual(code, 0) self.assertIn("cancel", api.calls) + self.assertGreaterEqual(api.first_cancel, 300) self.assertEqual(api.calls[-1], "rerun-failed") self.assertNotIn("rerun", api.calls) self.assertNotIn("pull", api.calls) self.assertIn(f"queued on {TRUSTED}", summary) + def test_a_held_build_never_cancels_a_started_sibling_at_the_watch_end(self): + looks = [] + jobs = nightly_run(started_at=20, setup_until=10 ** 9)(30 * 60) + for closing in (False, True): + deadline = START + dt.timedelta(seconds=30 * 60 + (30 if closing else 3600)) + looks.append(rescue.assess(jobs, now=START + dt.timedelta(seconds=30 * 60), budget_seconds=30, + deadline=deadline, reserved=True).action) + self.assertEqual(looks, ["watch", "watch"]) + + def test_a_refusal_at_the_watch_end_never_cancels_a_started_nightly(self): + clock = Clock() + api = NightlyAPI(clock, nightly_run(refused_at=60)) + _, summary = run_main(api, clock, payload=nightly_event()) + self.assertNotIn("cancel", api.calls) + self.assertNotIn("rerun-failed", api.calls) + self.assertIn("watch limit reached with a started nightly still running", summary) + + def test_a_sibling_that_starts_after_the_look_is_never_cancelled(self): + # assess() saw nothing else running; by the cancel a Linux job (no owned or Blacksmith label) runs. + for newer, moved in (((), False), ([RUN_ID + 7], True)): + clock = Clock() + linux = job("publish-nightly", labels=["blacksmith-2vcpu-ubuntu-2404"], status="in_progress", + runner="linux") + api = NightlyAPI(clock, lambda seconds: nightly_run(helper_done_at=0)(seconds) + [linux], + newer=newer) + target = rescue.Target(RUN_ID, 1, HEAD, 0, side=True, nightly=True, path=rescue.NIGHTLY_WORKFLOW_PATH) + result = rescue.rescue(api, target, now=clock.now, sleep=clock.sleep, log=lambda _: None, + failed_only=True) + self.assertNotIn("cancel", api.calls, moved) + self.assertTrue(result.startswith(rescue.NIGHTLY_BUSY), result) + self.assertIn("publish-nightly", result) + + def late_sibling_api(self, *, stalled_from_sibling=False, retry=False): + """Nightly: the build stuck on the trusted pool, nothing else running at the look; a Linux sibling starts + right after it (at the first superseded check, which every rescue makes) and runs 540 s.""" + clock = Clock() + state = {"start": None} + lasts = 540 + + def sibling_running(): + return state["start"] is not None and clock.seconds < state["start"] + lasts + + def jobs(seconds): + found = nightly_run(helper_done_at=0)(seconds) + if sibling_running(): + found.append(job("publish-nightly", labels=["blacksmith-2vcpu-ubuntu-2404"], status="in_progress", + runner="linux")) + return found + stalled = blacksmith_snapshot(queued=40, running=0, oldest=30) + free = blacksmith_snapshot() + + class API(NightlyAPI): + def newer_unfinished_runs(self, path, run_id, branch): + if state["start"] is None and not retry: + state["start"] = clock.seconds + return super().newer_unfinished_runs(path, run_id, branch) + + def cancel(self, run_id, before_send=None): + if retry and state["start"] is None: + # The first send is refused for a rate limit; the sibling starts during the wait. + before_send() + state["start"] = clock.seconds + super().cancel(run_id, before_send) + api = API(clock, jobs, workflow="runs-on: blacksmith-6vcpu-macos-26\n", + snapshot=lambda seconds: (stalled if stalled_from_sibling and state["start"] is not None + else free)(seconds)) + target = rescue.Target(RUN_ID, 1, HEAD, 0, side=True, nightly=True, path=rescue.NIGHTLY_WORKFLOW_PATH) + return clock, api, target, state, lasts + + def test_the_watch_resumes_and_moves_the_build_once_the_late_sibling_ends(self): + for retry in (False, True): + clock, api, target, state, lasts = self.late_sibling_api(retry=retry) + lines = [] + result = rescue.follow(api, target, seconds=30, queue_rounds="0", now=clock.now, sleep=clock.sleep, + log=lines.append) + self.assertTrue(any(line.startswith(rescue.NIGHTLY_BUSY) for line in lines), (retry, lines)) + self.assertEqual(result, "done", retry) + # The retried cancel (retry) never went out: the only cancel sent is the one after the sibling ended. + self.assertEqual(api.calls.count("cancel"), 1, retry) + self.assertGreaterEqual(api.first_cancel, state["start"] + lasts) + self.assertEqual(api.calls[-1], "rerun-failed", retry) + + def test_a_resumed_watch_still_holds_off_a_stalled_label(self): + clock, api, target, state, lasts = self.late_sibling_api(stalled_from_sibling=True) + result = rescue.follow(api, target, seconds=30, queue_rounds="0", now=clock.now, sleep=clock.sleep, + log=lambda _: None) + self.assertNotIn("rerun-failed", api.calls) + self.assertIn("kept on the owned pool", result) + + def test_a_sibling_in_its_runner_setup_is_started_work(self): + setup = {**job("publish-nightly", status="in_progress", runner="linux"), + "steps": [{"name": "Set up job", "status": "in_progress"}]} + jobs = nightly_run(helper_done_at=0)(30 * 60) + [setup] + look = rescue.assess(jobs, now=START + dt.timedelta(seconds=30 * 60), budget_seconds=30, reserved=True) + self.assertEqual(look.action, "watch") + api = NightlyAPI(Clock(), lambda seconds: jobs) + target = rescue.Target(RUN_ID, 1, HEAD, 0, side=True, nightly=True, path=rescue.NIGHTLY_WORKFLOW_PATH) + self.assertEqual(rescue.started_nightly_work(api, target, lambda _: None, lambda _: None), "publish-nightly") + # A nightly build held in its own runner's setup is started too: never cancelled, even at the end. + held = nightly_run(started_at=20, setup_until=10 ** 9, helper_done_at=0)(30 * 60) + for closing in (False, True): + deadline = START + dt.timedelta(seconds=30 * 60 + (30 if closing else 3600)) + self.assertEqual(rescue.assess(held, now=START + dt.timedelta(seconds=30 * 60), budget_seconds=30, + deadline=deadline, reserved=True).action, "watch") + + def test_a_retried_cancel_of_a_nightly_reads_its_jobs_again(self): + class Limited(rescue.GitHub): + def __init__(self): + super().__init__("t", "manaflow-ai/cmux") + self.sent = [] + + def request(self, method, path, *, own_token=False, body=None, before_retry=None, before_send=None): + self.sent.append(path) + if before_send is not None: + before_send() # as GitHub.request does before each send + return {} + + def jobs(self, run_id, attempt): + return [job("publish-nightly", status="in_progress", runner="linux")] + api = Limited() + target = rescue.Target(RUN_ID, 1, HEAD, 0, side=True, nightly=True, path=rescue.NIGHTLY_WORKFLOW_PATH) + api.run = lambda run_id: {"run_attempt": 1, "status": "in_progress"} + with self.assertRaises(rescue.NightlyBusy): + rescue.nightly_cancel(api, target, lambda _: None, lambda _: None) + + def test_a_retried_cancel_is_dropped_once_someone_re_ran_the_nightly(self): + # Through GitHub.request's own rate-limit retry: the first cancel gets 429; during the wait another + # watch starts attempt 2, whose jobs run. No second cancel may go out. + import email.message + sent = [] + attempt = {"now": 1} + + class Http(rescue.GitHub): + def __init__(self): + super().__init__("t", "manaflow-ai/cmux") + self.sleep = lambda seconds: attempt.update(now=2) + + def run(self, run_id): + return {"run_attempt": attempt["now"], "status": "in_progress"} + + def jobs(self, run_id, attempt_number): + return [job("decide", status="completed")] # attempt 1's jobs: nothing running there + + def urlopen(request, timeout=None): + sent.append(request.get_method() + " " + request.full_url) + headers = email.message.Message() + headers["Retry-After"] = "1" + raise urllib.error.HTTPError(request.full_url, 429, "rate limited", headers, io.BytesIO(b"{}")) + api = Http() + target = rescue.Target(RUN_ID, 1, HEAD, 0, side=True, nightly=True, path=rescue.NIGHTLY_WORKFLOW_PATH) + with unittest.mock.patch("urllib.request.urlopen", urlopen): + with self.assertRaises(rescue.Aborted) as raised: + rescue.nightly_cancel(api, target, lambda _: None, lambda _: None) + self.assertNotIsInstance(raised.exception, rescue.NightlyBusy) + self.assertIn("attempt 2", str(raised.exception)) + self.assertEqual([line for line in sent if line.endswith("/cancel")], [sent[0]]) + + def test_a_linux_sibling_blocks_the_look_too(self): + jobs = nightly_run(helper_done_at=0)(30 * 60) + [ + job("publish-nightly", labels=["blacksmith-2vcpu-ubuntu-2404"], status="in_progress", runner="linux")] + look = rescue.assess(jobs, now=START + dt.timedelta(seconds=30 * 60), budget_seconds=30, reserved=True) + self.assertEqual(look.action, "watch") + self.assertIn("a started nightly is never cancelled", look.reason) + def test_a_stuck_build_behind_which_a_newer_nightly_waits_is_cancelled_not_rerun(self): # nightly.yml never cancels in progress: the newer run is pending behind # this one, and a re-run would join the group and cancel it. clock = Clock() - api = NightlyAPI(clock, nightly_run(), newer=[RUN_ID + 7]) + api = NightlyAPI(clock, nightly_run(helper_done_at=300), newer=[RUN_ID + 7]) _, summary = run_main(api, clock, payload=nightly_event(), env_extra={"RESCUE_SECONDS": "30"}) self.assertIn("cancel", api.calls) self.assertNotIn("rerun-failed", api.calls) @@ -2553,7 +2736,7 @@ def test_a_stuck_build_behind_which_a_newer_nightly_waits_is_cancelled_not_rerun def test_a_refused_build_is_rerun_once_the_run_finishes(self): clock = Clock() - api = NightlyAPI(clock, nightly_run(refused_at=60), finished=lambda s: s >= 200) + api = NightlyAPI(clock, nightly_run(refused_at=60, helper_done_at=200), finished=lambda s: s >= 200) code, summary = run_main(api, clock, payload=nightly_event()) self.assertEqual(code, 0) self.assertIn("waiting for the rest of the run to finish", summary) @@ -2562,7 +2745,7 @@ def test_a_refused_build_is_rerun_once_the_run_finishes(self): self.assertGreaterEqual(clock.seconds, 200) # A newer nightly run builds main's newer HEAD instead: left as it is. clock = Clock() - api = NightlyAPI(clock, nightly_run(refused_at=60), finished=lambda s: s >= 200, newer=[RUN_ID + 1]) + api = NightlyAPI(clock, nightly_run(refused_at=60, helper_done_at=200), finished=lambda s: s >= 200, newer=[RUN_ID + 1]) _, summary = run_main(api, clock, payload=nightly_event()) self.assertNotIn("rerun-failed", api.calls) self.assertIn("not rescued", summary) From ff6b83cc3e6d5cabb52f224efd781686c9ed2f3f Mon Sep 17 00:00:00 2001 From: Leo Li Date: Tue, 27 Oct 2026 08:56:55 -0700 Subject: [PATCH 2/3] ci: a busy nightly cancel resumes into the signing-stall rescue A NightlyBusy from the signing-stall cancel keeps watching instead of ending the watch, and the ordinary rescue's busy resume goes through the same stall hold and signing-stall loop with the watch's first-seen times. Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?id=aid-85ca59f74f02d81734c3 --- scripts/ci/owned_pool_rescue.py | 64 ++++++++++++++++++------------ tests/test_ci_owned_pool_rescue.py | 43 ++++++++++++++++++++ 2 files changed, 81 insertions(+), 26 deletions(-) diff --git a/scripts/ci/owned_pool_rescue.py b/scripts/ci/owned_pool_rescue.py index 8058ffd59276..ca4256578f13 100644 --- a/scripts/ci/owned_pool_rescue.py +++ b/scripts/ci/owned_pool_rescue.py @@ -2765,27 +2765,39 @@ def capped(deadline: dt.datetime) -> dt.datetime: outcome, reason = watch(client, target, budget_seconds=first_budget, now=clock, sleep=sleep, log=log, deadline=deadline, floor_seconds=effective_seconds, seen=seen, sign_seen=sign_seen, sign_text=sign_text) - held = "" - while True: - if outcome == "rescue" and not target.refusals_only: - outcome, reason, held = hold_for_destination( - client, target, outcome, reason, budget_seconds=first_budget, floor_seconds=effective_seconds, - deadline=deadline, now=clock, sleep=sleep, log=log, seen=seen, sign_seen=sign_seen, - sign_text=sign_text) - if outcome != "stalled": - break - log(f"stalled: {reason}") - result = rescue_stalled_signing(client, target, now=clock, sleep=rescue_sleep, log=log, - deadline=rescue_deadline, sign_seen=sign_seen, - workflow_text=sign_text[0] if sign_text else None) - log(result) - if not result.startswith(SIGN_NOT_YET) or clock() >= deadline: - return "done" - # The run changed since the look (a job started, a sibling came back): watch on, next look decides. - sleep(POLL_SECONDS) - outcome, reason = watch(client, target, budget_seconds=first_budget, now=clock, sleep=sleep, - log=log, deadline=deadline, floor_seconds=effective_seconds, seen=seen, - sign_seen=sign_seen, sign_text=sign_text) + + def settle(outcome: str, reason: str) -> tuple[str, str, str, str]: + """The stall hold and the signing-stall rescue on a look's outcome, watching on as they need: + (outcome, reason, held, done), where `done` is follow()'s answer once one of them ended the watch.""" + held = "" + while True: + if outcome == "rescue" and not target.refusals_only: + outcome, reason, held = hold_for_destination( + client, target, outcome, reason, budget_seconds=first_budget, floor_seconds=effective_seconds, + deadline=deadline, now=clock, sleep=sleep, log=log, seen=seen, sign_seen=sign_seen, + sign_text=sign_text) + if outcome != "stalled": + return outcome, reason, held, "" + log(f"stalled: {reason}") + result = rescue_stalled_signing(client, target, now=clock, sleep=rescue_sleep, log=log, + deadline=rescue_deadline, sign_seen=sign_seen, + workflow_text=sign_text[0] if sign_text else None) + log(result) + # The run changed since the look (a job started, a sibling came back, a nightly job began running + # just before the cancel): watch on, the next look decides. + if not (result.startswith(SIGN_NOT_YET) or result.startswith(NIGHTLY_BUSY)): + return outcome, reason, held, "done" + if clock() >= deadline: + return outcome, reason, held, ("stopped: watch limit reached with a started nightly still running" + if result.startswith(NIGHTLY_BUSY) else "done") + sleep(POLL_SECONDS) + outcome, reason = watch(client, target, budget_seconds=first_budget, now=clock, sleep=sleep, + log=log, deadline=deadline, floor_seconds=effective_seconds, seen=seen, + sign_seen=sign_seen, sign_text=sign_text) + + outcome, reason, held, done = settle(outcome, reason) + if done: + return done if outcome not in ("rescue", "refused"): return f"stopped: {reason}" + (f"; kept on the owned pool: {held}" if held else "") if target.refusals_only and outcome == "rescue": @@ -2855,11 +2867,11 @@ def capped(deadline: dt.datetime) -> dt.datetime: if clock() >= deadline: return "stopped: watch limit reached with a started nightly still running" outcome, reason = watch(client, target, budget_seconds=first_budget, now=clock, sleep=sleep, log=log, - deadline=deadline, floor_seconds=effective_seconds, seen=seen) - if outcome == "rescue": - outcome, reason, held = hold_for_destination( - client, target, outcome, reason, budget_seconds=first_budget, floor_seconds=effective_seconds, - deadline=deadline, now=clock, sleep=sleep, log=log, seen=seen) + deadline=deadline, floor_seconds=effective_seconds, seen=seen, + sign_seen=sign_seen, sign_text=sign_text) + outcome, reason, held, done = settle(outcome, reason) + if done: + return done if outcome not in ("rescue", "refused"): return f"stopped: {reason}" + (f"; kept on the owned pool: {held}" if held else "") log(f"{'rescue' if outcome == 'rescue' else 'refused'}: {reason}") diff --git a/tests/test_ci_owned_pool_rescue.py b/tests/test_ci_owned_pool_rescue.py index 43d068d76a12..bc9697911f6e 100644 --- a/tests/test_ci_owned_pool_rescue.py +++ b/tests/test_ci_owned_pool_rescue.py @@ -3990,6 +3990,49 @@ def once(*args, **kwargs): self.assertIn("not rescued yet: a sibling came back", summary) self.assertEqual(mutations(api), ["cancel", "rerun-failed"]) + def test_a_started_nightly_at_the_signing_cancel_resumes_the_watch(self): + # The cancel guard found a nightly job running (NightlyBusy): the watch goes on and rescues later. + clock = Clock() + api = NightlyAPI(clock, signing_run(), workflow=SIGN_WORKFLOW) + real = rescue.rescue_stalled_signing + answers = [] + + def once(*args, **kwargs): + if not answers: + answers.append(rescue.NIGHTLY_BUSY + " (publish-nightly)") + return answers[0] + return real(*args, **kwargs) + with unittest.mock.patch.object(rescue, "rescue_stalled_signing", once): + _, summary = run_main(api, clock, payload=nightly_event(head_branch="nightly-next")) + self.assertIn(rescue.NIGHTLY_BUSY, summary) + self.assertEqual(mutations(api), ["cancel", "rerun-failed"]) + self.assertIn("its signing jobs take their alternate macOS label", summary) + + def test_a_busy_rescue_resumes_into_the_signing_stall_rescue(self): + # An ordinary rescue found a started nightly at its cancel; the resumed watch then finds the signing jobs + # stalled, with the same first-seen times, and the signing rescue moves them. + clock = Clock() + api = NightlyAPI(clock, signing_run(), workflow=SIGN_WORKFLOW) + real_watch, real_rescue = rescue.watch, rescue.rescue + state = {"watched": False, "busy": False} + + def watch(*args, **kwargs): + if not state["watched"]: + state["watched"] = True + return "refused", "a job was refused" + return real_watch(*args, **kwargs) + + def rescue_once(*args, **kwargs): + if not state["busy"]: + state["busy"] = True + return rescue.NIGHTLY_BUSY + " (publish-nightly)" + return real_rescue(*args, **kwargs) + with unittest.mock.patch.object(rescue, "watch", watch), \ + unittest.mock.patch.object(rescue, "rescue", rescue_once): + _, summary = run_main(api, clock, payload=nightly_event(head_branch="nightly-next")) + self.assertEqual(mutations(api), ["cancel", "rerun-failed"]) + self.assertIn("its signing jobs take their alternate macOS label", summary) + def test_signing_stall_rules(self): now = START + dt.timedelta(seconds=60 + rescue.SIGN_STALL_SECONDS) queued = dict(job("sign-nightly-identity-legacy", labels=[BLACKSMITH], created=60), id=9) From 53a5ab3d521d512a2f13817634c6568405f99bdb Mon Sep 17 00:00:00 2001 From: Leo Li Date: Tue, 27 Oct 2026 09:09:25 -0700 Subject: [PATCH 3/3] ci: the nightly cancel guard reads the jobs before the run, and again after a slow read A jobs read that waited through a rate limit may return the old attempt's jobs after someone re-ran the run; the run read after it shows the new attempt. A run read that waited looks at the jobs again. Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?id=aid-85ca59f74f02d81734c3 --- scripts/ci/owned_pool_rescue.py | 35 ++++++++++++------ tests/test_ci_owned_pool_rescue.py | 58 +++++++++++++++++++++++++++++- 2 files changed, 82 insertions(+), 11 deletions(-) diff --git a/scripts/ci/owned_pool_rescue.py b/scripts/ci/owned_pool_rescue.py index ca4256578f13..0c7cf77dd11f 100644 --- a/scripts/ci/owned_pool_rescue.py +++ b/scripts/ci/owned_pool_rescue.py @@ -2394,6 +2394,10 @@ def pull_moved(api: GitHub, target: Target, sleep: Callable[[float], None], return "" +# Looks nightly_cancel()'s guard makes for a jobs read no rate-limit wait made stale. +NIGHTLY_GUARD_TRIES = 3 +# A run read slower than this (a token mint, a slow API) leaves the jobs read before it stale too. +NIGHTLY_GUARD_FRESH_SECONDS = 15 # rescue()'s answer when a nightly it was about to cancel had a job running: follow() then watches the run again. NIGHTLY_BUSY = "not rescued: started nightly work is running" @@ -2412,16 +2416,27 @@ def check() -> None: before_cancel() if not target.nightly: return - run = api.run(target.run_id) or {} - attempt = int(run.get("run_attempt") or 0) - if attempt != target.attempt: - # Someone re-ran it meanwhile: the cancel would hit that newer attempt's running jobs. - raise Aborted(f"nightly run {target.run_id} is at attempt {attempt}, not {target.attempt}; not cancelled") - if run.get("status") == "completed": - raise Aborted(f"nightly run {target.run_id} has finished; nothing to cancel") - busy = started_nightly_work(api, target, sleep, log) - if busy: - raise NightlyBusy(busy) + for _ in range(NIGHTLY_GUARD_TRIES): + # The jobs first, then the run: a jobs read that waited (a rate limit, a retried error) may return + # the old attempt's jobs after someone re-ran the run, and the run read after it shows that attempt. + busy = started_nightly_work(api, target, sleep, log) + waited, began = getattr(api, "rate_limit_waited", 0.0), time.monotonic() + run = api.run(target.run_id) or {} + attempt = int(run.get("run_attempt") or 0) + if attempt != target.attempt: + # Someone re-ran it meanwhile: the cancel would hit that newer attempt's running jobs. + raise Aborted(f"nightly run {target.run_id} is at attempt {attempt}, not {target.attempt}; " + "not cancelled") + if run.get("status") == "completed": + raise Aborted(f"nightly run {target.run_id} has finished; nothing to cancel") + if busy: + raise NightlyBusy(busy) + if getattr(api, "rate_limit_waited", 0.0) == waited and \ + time.monotonic() - began < NIGHTLY_GUARD_FRESH_SECONDS: + return + # The run read itself waited (a rate limit, a token mint): a job may have started since the jobs + # read. Look again. + raise NightlyBusy(f"slow reads kept the jobs of run {target.run_id} unconfirmed") if before_cancel is None and not target.nightly: api.cancel(target.run_id) return diff --git a/tests/test_ci_owned_pool_rescue.py b/tests/test_ci_owned_pool_rescue.py index bc9697911f6e..3facf1b96dd5 100644 --- a/tests/test_ci_owned_pool_rescue.py +++ b/tests/test_ci_owned_pool_rescue.py @@ -2271,7 +2271,7 @@ def test_a_next_dispatch_and_a_next_rerun_are_watched_through_the_picker_s_marke def test_a_refused_next_dispatch_moves_its_failed_jobs_to_blacksmith(self): clock = Clock() - api = NightlyAPI(clock, nightly_run(refused_at=60), finished=lambda s: s >= 200) + api = NightlyAPI(clock, nightly_run(refused_at=60, helper_done_at=200), finished=lambda s: s >= 200) code, summary = run_main(api, clock, payload=nightly_event(event="workflow_dispatch", head_branch="nightly-next")) self.assertEqual(code, 0) self.assertNotIn("cancel", api.calls) @@ -2717,6 +2717,62 @@ def urlopen(request, timeout=None): self.assertIn("attempt 2", str(raised.exception)) self.assertEqual([line for line in sent if line.endswith("/cancel")], [sent[0]]) + def guard_http(self, limited): + """A real GitHub over a fake urlopen: `limited` is the request path part ("/jobs" or "/runs/") whose first + send gets a 429; during its wait the nightly moves on as `state` says.""" + import email.message + sent, state = [], {"attempt": 1, "running": False, "limited": False} + + class Response(io.BytesIO): + headers = {} + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def urlopen(request, timeout=None): + method, url = request.get_method(), request.full_url + sent.append(f"{method} {url}") + if not state["limited"] and url.split("?")[0].endswith(limited): + state["limited"] = True + headers = email.message.Message() + headers["Retry-After"] = "1" + raise urllib.error.HTTPError(url, 429, "rate limited", headers, io.BytesIO(b"{}")) + if url.endswith("/cancel"): + return Response(b"{}") + if "/jobs" in url: + status = "in_progress" if state["running"] else "completed" + return Response(json.dumps({"jobs": [job("publish-nightly", status=status)]}).encode()) + return Response(json.dumps({"run_attempt": state["attempt"], "status": "in_progress"}).encode()) + + api = rescue.GitHub("t", "manaflow-ai/cmux") + api.read_headers = api.headers # every read on the token that waits through a rate limit + return api, sent, state, unittest.mock.patch("urllib.request.urlopen", urlopen) + + def test_a_rate_limited_jobs_read_never_lets_the_cancel_hit_a_newer_attempt(self): + # The jobs GET waits through a 429; meanwhile attempt 1 ends and attempt 2 starts. The retried read still + # returns attempt 1's finished jobs, so only the run read after it can tell: no cancel goes out. + api, sent, state, patch = self.guard_http("/attempts/1/jobs") + api.sleep = lambda seconds: state.update(attempt=2) + target = rescue.Target(RUN_ID, 1, HEAD, 0, side=True, nightly=True, path=rescue.NIGHTLY_WORKFLOW_PATH) + with patch, self.assertRaises(rescue.Aborted) as raised: + rescue.nightly_cancel(api, target, lambda _: None, lambda _: None) + self.assertNotIsInstance(raised.exception, rescue.NightlyBusy) + self.assertIn("attempt 2", str(raised.exception)) + self.assertFalse([line for line in sent if line.endswith("/cancel")], sent) + + def test_a_rate_limited_run_read_looks_at_the_jobs_again(self): + # The run GET waits through a 429; meanwhile a job of the nightly starts. The guard reads the jobs again. + api, sent, state, patch = self.guard_http(f"/runs/{RUN_ID}") + api.sleep = lambda seconds: state.update(running=True) + target = rescue.Target(RUN_ID, 1, HEAD, 0, side=True, nightly=True, path=rescue.NIGHTLY_WORKFLOW_PATH) + with patch, self.assertRaises(rescue.NightlyBusy) as raised: + rescue.nightly_cancel(api, target, lambda _: None, lambda _: None) + self.assertIn("publish-nightly", str(raised.exception)) + self.assertFalse([line for line in sent if line.endswith("/cancel")], sent) + def test_a_linux_sibling_blocks_the_look_too(self): jobs = nightly_run(helper_done_at=0)(30 * 60) + [ job("publish-nightly", labels=["blacksmith-2vcpu-ubuntu-2404"], status="in_progress", runner="linux")]