From 7f532b0ecdebd90eb82dd436a2b641fa4be67e1d Mon Sep 17 00:00:00 2001 From: Trask Stalnaker Date: Wed, 2 Sep 2026 14:47:01 -0700 Subject: [PATCH 1/3] Prevent dashboard publisher starvation Make dashboard state writers wait for active publisher leases before each CAS attempt, and bound queue-worker lock waits below the job timeout. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 8c4a7acc-3827-4097-9d75-2d054cf9fc3f --- .../pull-request-dashboard/RATIONALE.md | 23 ++- .../pull-request-dashboard/dashboard.py | 4 + .../process_queue_batch.py | 59 +++++--- .../pull-request-dashboard/state_branch.py | 63 +++++++- .../pull-request-dashboard/test_dashboard.py | 23 +++ .../test_process_queue_batch.py | 12 ++ .../test_state_branch.py | 143 ++++++++++++++++++ 7 files changed, 300 insertions(+), 27 deletions(-) diff --git a/.github/scripts/pull-request-dashboard/RATIONALE.md b/.github/scripts/pull-request-dashboard/RATIONALE.md index c8f61f96ef6..1c595cf7245 100644 --- a/.github/scripts/pull-request-dashboard/RATIONALE.md +++ b/.github/scripts/pull-request-dashboard/RATIONALE.md @@ -102,9 +102,15 @@ the implementation understandable and operationally cheap. one even when `cancel-in-progress` is false. Direct and queued publishers also acquire a lease stored on the repository's state branch. This shared lock prevents the drain from overlapping the direct concurrency group while either - path performs external delivery or publishes the issue. Accepted work lives - on the state branch: a targeted publisher limits status-comment and Slack - delivery to its triggering PR. Webhook runs can arrive concurrently for many + path performs external delivery or publishes the issue. Dashboard state + writers treat an active lease as a write barrier. They can calculate updates + concurrently, but wait to commit accepted state until publication finishes. + A writer that races lease acquisition loses its compare-and-swap push and + waits on its next attempt. Queue drains give all publisher-lock waits one + shared deadline, leaving time to return unfinished claims before the job + timeout. Accepted work lives on the state branch: a targeted publisher limits + status-comment and Slack delivery to its triggering PR. Webhook runs arrive + concurrently for many PRs, so allowing each publisher to fan out into repository-wide delivery would create long jobs and put pressure on the GitHub Actions job queue, especially when a new status-comment revision queues every open PR. The @@ -134,6 +140,10 @@ the implementation understandable and operationally cheap. do not contend on the same git ref during scheduled and webhook-driven runs. - Updates use `git push --force-with-lease`, so git refs provide the durable compare-and-swap boundary for concurrent same-repository runs. +- Before each compare-and-swap attempt, dashboard state writers wait for an + active publisher lease to finish. Waiting does not consume a retry. The lease + and compare-and-swap check together prevent a stream of state commits from + starving delivery. - A missing repository state branch is bootstrapped by non-PR backfills. The dashboard state records when every open non-draft PR has been populated at least once. Targeted PR runs, dashboard publishing, status comments, and @@ -685,6 +695,7 @@ the implementation understandable and operationally cheap. durable ledgers; Slack eligibility is reconstructed from accepted dashboard and notification state. - The dashboard issue is rendered from `dashboard-state.json` and the target - repository's current open PR list after delivery. If another update advances - the state branch while a publisher is already working, external views can - briefly lag until the next publisher. + repository's current open PR list after delivery. The state-branch lease keeps + accepted-state commits from advancing while a publisher works. A writer and + publisher can race before the lease commit; compare-and-swap selects one, and + the loser refetches before continuing. diff --git a/.github/scripts/pull-request-dashboard/dashboard.py b/.github/scripts/pull-request-dashboard/dashboard.py index b42dd8d8f7f..cc18df0a741 100644 --- a/.github/scripts/pull-request-dashboard/dashboard.py +++ b/.github/scripts/pull-request-dashboard/dashboard.py @@ -748,6 +748,7 @@ def update_dashboard_for_pr_number(args: argparse.Namespace, state_dir: Path) -> "Update dashboard state", lambda: apply_targeted_dashboard_update(args, update, observed_at), state_branch=args.state_branch, + respect_publisher_lock=True, ) @@ -783,6 +784,7 @@ def update_dashboard_for_backfill(args: argparse.Namespace, state_dir: Path) -> observed_at, ), state_branch=args.state_branch, + respect_publisher_lock=True, ) if status != 0: return status @@ -808,6 +810,7 @@ def save_current_dashboard_state() -> int: "Update dashboard state", save_current_dashboard_state, state_branch=args.state_branch, + respect_publisher_lock=True, ) for pr_summary in selection.selected_prs: @@ -876,6 +879,7 @@ def update_selected_pr(pr_summary: dict[str, Any] = pr_summary) -> int: "Update dashboard state", update_selected_pr, state_branch=args.state_branch, + respect_publisher_lock=True, ) if status != 0: return status diff --git a/.github/scripts/pull-request-dashboard/process_queue_batch.py b/.github/scripts/pull-request-dashboard/process_queue_batch.py index b7c64a02b65..b68e0f967a6 100644 --- a/.github/scripts/pull-request-dashboard/process_queue_batch.py +++ b/.github/scripts/pull-request-dashboard/process_queue_batch.py @@ -8,9 +8,10 @@ import sys import tempfile import threading +import time from collections import defaultdict -from collections.abc import Callable -from contextlib import AbstractContextManager +from collections.abc import Callable, Iterator +from contextlib import AbstractContextManager, contextmanager from dataclasses import dataclass from pathlib import Path from typing import Any @@ -22,6 +23,7 @@ OWNER = "open-telemetry" STATE_BRANCH_PREFIX = "otelbot/pull-request-dashboard-state" MAX_ATTEMPTS = 3 +QUEUE_LOCK_WAIT_BUDGET_SECONDS = 40 * 60 class LeaseMonitor: @@ -91,6 +93,22 @@ class WorkItem: claims: tuple[Claim, ...] +@contextmanager +def queue_lock_wait_deadline( + now: Callable[[], float] = time.time, +) -> Iterator[None]: + name = state_branch.PUBLISHER_LOCK_DEADLINE_ENV + previous = os.environ.get(name) + os.environ[name] = str(int(now()) + QUEUE_LOCK_WAIT_BUDGET_SECONDS) + try: + yield + finally: + if previous is None: + os.environ.pop(name, None) + else: + os.environ[name] = previous + + def load_claims(path: Path) -> list[Claim]: raw = json.loads(path.read_text(encoding="utf-8")) if not isinstance(raw, list): @@ -582,24 +600,25 @@ def record_and_acknowledge(completed: list[dict[str, Any]]) -> None: args.results.write_text(json.dumps(completed, indent=2) + "\n", encoding="utf-8") acknowledge_all(client, args.results, common) - try: - monitor.start() - processor = DashboardBatchProcessor( - args.config, - lease_check=monitor.assert_valid, - publisher_lock_owner=args.worker_id, - ) - work_items, resolved = resolve_work_items(claims, processor.resolve_head) - record_and_acknowledge(resolved) - process_batch( - work_items, - processor.process_repository, - max_repositories=args.max_repositories, - on_results=record_and_acknowledge, - ) - finally: - args.results.write_text(json.dumps(results, indent=2) + "\n", encoding="utf-8") - monitor.close() + with queue_lock_wait_deadline(): + try: + monitor.start() + processor = DashboardBatchProcessor( + args.config, + lease_check=monitor.assert_valid, + publisher_lock_owner=args.worker_id, + ) + work_items, resolved = resolve_work_items(claims, processor.resolve_head) + record_and_acknowledge(resolved) + process_batch( + work_items, + processor.process_repository, + max_repositories=args.max_repositories, + on_results=record_and_acknowledge, + ) + finally: + args.results.write_text(json.dumps(results, indent=2) + "\n", encoding="utf-8") + monitor.close() dead_letters = sum(result["outcome"] == "dead" for result in results) retries = sum(result["outcome"] == "retry" for result in results) print( diff --git a/.github/scripts/pull-request-dashboard/state_branch.py b/.github/scripts/pull-request-dashboard/state_branch.py index 7e012591bb0..8a5010b4343 100644 --- a/.github/scripts/pull-request-dashboard/state_branch.py +++ b/.github/scripts/pull-request-dashboard/state_branch.py @@ -29,6 +29,7 @@ DEFAULT_PUBLISHER_LOCK_LEASE_SECONDS = 60 * 60 DEFAULT_PUBLISHER_LOCK_WAIT_SECONDS = 60 * 60 PUBLISHER_LOCK_POLL_SECONDS = 5 +PUBLISHER_LOCK_DEADLINE_ENV = "PR_DASHBOARD_PUBLISHER_LOCK_DEADLINE" class PublisherLock(TypedDict): @@ -257,6 +258,63 @@ def load_publisher_lock(state_dir: Path) -> PublisherLock | None: return {"owner": owner, "expiresAt": float(expires_at)} +def publisher_lock_wait_deadline(wait_seconds: int, current_time: float) -> float: + if wait_seconds < 0: + raise ValueError("publisher lock wait must be non-negative") + deadline = current_time + wait_seconds + configured = os.environ.get(PUBLISHER_LOCK_DEADLINE_ENV) + if configured is None: + return deadline + try: + configured_deadline = int(configured) + except ValueError as error: + raise ValueError(f"{PUBLISHER_LOCK_DEADLINE_ENV} must be an integer") from error + if configured_deadline < 1: + raise ValueError(f"{PUBLISHER_LOCK_DEADLINE_ENV} must be positive") + return min(deadline, configured_deadline) + + +def wait_for_publisher_unlock( + state_dir: Path, + state_branch: str, + *, + wait_seconds: int = DEFAULT_PUBLISHER_LOCK_WAIT_SECONDS, + now: Callable[[], float] = time.time, + sleep: Callable[[float], None] = time.sleep, +) -> None: + deadline = publisher_lock_wait_deadline(wait_seconds, now()) + announced_owner: str | None = None + while True: + current_time = now() + lock = load_publisher_lock(state_dir) + if lock is None or lock["expiresAt"] <= current_time: + return + if lock["owner"] != announced_owner: + print( + f"dashboard publisher lock {state_branch} is held by {lock['owner']}; waiting", + file=sys.stderr, + ) + announced_owner = lock["owner"] + remaining = deadline - current_time + if remaining <= 0: + raise TimeoutError( + f"timed out waiting for dashboard publisher lock {state_branch} " + f"held by {lock['owner']}" + ) + sleep( + min( + PUBLISHER_LOCK_POLL_SECONDS, + remaining, + lock["expiresAt"] - current_time, + ) + ) + if not reset_state(state_dir, state_branch): + raise RuntimeError( + f"dashboard state branch {state_branch} disappeared while waiting " + "for its publisher lock" + ) + + def commit_publisher_lock( state_dir: Path, state_branch: str, @@ -280,7 +338,7 @@ def acquire_publisher_lock( raise ValueError("publisher lock owner must not be empty") if lease_seconds < 1 or wait_seconds < 0: raise ValueError("publisher lock lease must be positive and wait must be non-negative") - deadline = now() + wait_seconds + deadline = publisher_lock_wait_deadline(wait_seconds, now()) while True: current_time = now() with temporary_state_dir() as state_dir: @@ -366,6 +424,7 @@ def push_state_changes( max_attempts: int = DEFAULT_MAX_ATTEMPTS, add_paths: list[str] | None = None, retry_snapshots: list[tuple[Path, Path]] | None = None, + respect_publisher_lock: bool = False, ) -> int: configure_git() checkout_state(state_dir, state_branch, require_existing=False) @@ -373,6 +432,8 @@ def push_state_changes( snapshots = retry_snapshots or [] for attempt in range(1, max_attempts + 1): + if respect_publisher_lock: + wait_for_publisher_unlock(state_dir, state_branch) status = update_state() if status != 0: return status diff --git a/.github/scripts/pull-request-dashboard/test_dashboard.py b/.github/scripts/pull-request-dashboard/test_dashboard.py index 857b7d9810e..9a0e52575b2 100644 --- a/.github/scripts/pull-request-dashboard/test_dashboard.py +++ b/.github/scripts/pull-request-dashboard/test_dashboard.py @@ -20,6 +20,7 @@ select_backfill_prs, set_backfill_pr_failed, update_dashboard_for_backfill, + update_dashboard_for_pr_number, write_initial_backfill_output, ) from dashboard_state_update import ( @@ -1972,6 +1973,7 @@ def test_failed_pr_does_not_block_later_backfill_progress(self) -> None: current_state = dashboard_state() backfill_state = {"cursor": {}} refreshed_pr_numbers: list[int] = [] + respects_publisher_lock: list[bool] = [] def load_dashboard_state() -> DashboardState: return current_state @@ -2016,6 +2018,7 @@ def save_dashboard_state( return 0 def push_state_changes(_state_dir, _message, update_state, **_kwargs) -> int: + respects_publisher_lock.append(_kwargs.get("respect_publisher_lock", False)) return update_state() with ( @@ -2040,6 +2043,7 @@ def push_state_changes(_state_dir, _message, update_state, **_kwargs) -> int: status = update_dashboard_for_backfill(args, Path("state")) self.assertEqual(refreshed_pr_numbers, [1, 2]) + self.assertEqual(respects_publisher_lock, [True, True]) self.assertEqual(2, accept_update.call_count) record_nudge.assert_called_once_with(2, ANY, ANY, prepare_due=False) self.assertEqual(status, BACKFILL_RECORDED_FAILURE_STATUS) @@ -2055,6 +2059,25 @@ def push_state_changes(_state_dir, _message, update_state, **_kwargs) -> int: self.assertEqual(backfill_state["cursor"], {"last_pr_number": 2}) self.assertEqual(backfill_failed_pr_numbers(backfill_state), {1}) + def test_targeted_state_update_respects_publisher_lock(self) -> None: + args = Namespace(pr_number=1, state_branch="state") + update = object() + + with ( + patch("dashboard.state_branch.configure_git"), + patch("dashboard.state_branch.checkout_state"), + patch("dashboard.state_branch.remove_existing_state_dir"), + patch("dashboard.build_targeted_dashboard_update", return_value=update), + patch( + "dashboard.state_branch.push_state_changes", + return_value=0, + ) as push_state_changes, + ): + status = update_dashboard_for_pr_number(args, Path("state")) + + self.assertEqual(0, status) + self.assertTrue(push_state_changes.call_args.kwargs["respect_publisher_lock"]) + def test_successful_retry_clears_recorded_failure(self) -> None: state = {"failed_pr_numbers": [1, 2]} diff --git a/.github/scripts/pull-request-dashboard/test_process_queue_batch.py b/.github/scripts/pull-request-dashboard/test_process_queue_batch.py index 7177f2c9abe..c048d1a6f5b 100644 --- a/.github/scripts/pull-request-dashboard/test_process_queue_batch.py +++ b/.github/scripts/pull-request-dashboard/test_process_queue_batch.py @@ -2,6 +2,7 @@ from contextlib import contextmanager, nullcontext import json +import os import subprocess import sys import tempfile @@ -262,6 +263,17 @@ def report(results: list[dict[str, object]]) -> None: self.assertTrue(slow_observed_report) + def test_queue_bounds_publisher_lock_waits_and_restores_environment(self) -> None: + name = process_queue_batch.state_branch.PUBLISHER_LOCK_DEADLINE_ENV + + with mock.patch.dict(os.environ, {name: "50"}): + with process_queue_batch.queue_lock_wait_deadline(now=lambda: 100): + self.assertEqual( + str(100 + process_queue_batch.QUEUE_LOCK_WAIT_BUDGET_SECONDS), + os.environ[name], + ) + self.assertEqual("50", os.environ[name]) + def test_delivery_error_still_publishes_committed_active_state(self) -> None: commands: list[str] = [] diff --git a/.github/scripts/pull-request-dashboard/test_state_branch.py b/.github/scripts/pull-request-dashboard/test_state_branch.py index 9e8360dbe0c..d96b85e2d88 100644 --- a/.github/scripts/pull-request-dashboard/test_state_branch.py +++ b/.github/scripts/pull-request-dashboard/test_state_branch.py @@ -119,6 +119,149 @@ def test_active_publisher_lock_times_out( ) +class PublisherWriteBarrierTest(unittest.TestCase): + @patch.object(state_branch, "reset_state", return_value=True) + @patch.object( + state_branch, + "load_publisher_lock", + side_effect=[ + {"owner": "publisher", "expiresAt": 200}, + None, + ], + ) + def test_waits_for_active_publisher_lock( + self, + load_publisher_lock: object, + reset_state: object, + ) -> None: + sleeps: list[float] = [] + + state_branch.wait_for_publisher_unlock( + Path("state"), + "state-branch", + now=lambda: 100, + sleep=sleeps.append, + ) + + self.assertEqual([5], sleeps) + self.assertEqual(2, load_publisher_lock.call_count) + reset_state.assert_called_once_with(Path("state"), "state-branch") + + @patch.object(state_branch, "reset_state") + @patch.object( + state_branch, + "load_publisher_lock", + return_value={"owner": "publisher", "expiresAt": 100}, + ) + def test_expired_publisher_lock_does_not_wait( + self, + _load_publisher_lock: object, + reset_state: object, + ) -> None: + sleeps: list[float] = [] + + state_branch.wait_for_publisher_unlock( + Path("state"), + "state-branch", + now=lambda: 100, + sleep=sleeps.append, + ) + + self.assertEqual([], sleeps) + reset_state.assert_not_called() + + @patch.object( + state_branch, + "load_publisher_lock", + return_value={"owner": "publisher", "expiresAt": 200}, + ) + def test_active_publisher_lock_times_out( + self, + _load_publisher_lock: object, + ) -> None: + with self.assertRaisesRegex( + TimeoutError, + "state-branch held by publisher", + ): + state_branch.wait_for_publisher_unlock( + Path("state"), + "state-branch", + wait_seconds=0, + now=lambda: 100, + ) + + @patch.object(state_branch, "reset_state", return_value=True) + @patch.object( + state_branch, + "load_publisher_lock", + side_effect=[ + {"owner": "publisher", "expiresAt": 200}, + None, + ], + ) + def test_shared_deadline_bounds_publisher_wait( + self, + _load_publisher_lock: object, + _reset_state: object, + ) -> None: + sleeps: list[float] = [] + + with patch.dict( + state_branch.os.environ, + {state_branch.PUBLISHER_LOCK_DEADLINE_ENV: "102"}, + ): + state_branch.wait_for_publisher_unlock( + Path("state"), + "state-branch", + now=lambda: 100, + sleep=sleeps.append, + ) + + self.assertEqual([2], sleeps) + + def test_checks_barrier_before_each_cas_attempt(self) -> None: + lifecycle: list[str] = [] + + def run( + command: list[str], + **_kwargs: object, + ) -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess( + command, + 1 if command[:4] == ["git", "diff", "--cached", "--quiet"] else 0, + ) + + def update_state() -> int: + lifecycle.append("update") + return 0 + + with ( + patch.object(state_branch, "configure_git"), + patch.object(state_branch, "checkout_state"), + patch.object(state_branch, "run", side_effect=run), + patch.object(state_branch, "push_state", side_effect=[False, True]), + patch.object(state_branch, "reset_state", return_value=True), + patch.object(state_branch, "retry_delay_seconds", return_value=0), + patch.object(state_branch.time, "sleep"), + patch.object( + state_branch, + "wait_for_publisher_unlock", + side_effect=lambda *_args: lifecycle.append("wait"), + ) as wait_for_publisher_unlock, + ): + status = state_branch.push_state_changes( + Path("state"), + "Update dashboard state", + update_state, + state_branch="state-branch", + respect_publisher_lock=True, + ) + + self.assertEqual(0, status) + self.assertEqual(["wait", "update", "wait", "update"], lifecycle) + self.assertEqual(2, wait_for_publisher_unlock.call_count) + + class FetchStateBranchTest(unittest.TestCase): @staticmethod def rejected_fetch() -> subprocess.CompletedProcess[str]: From 20ff5158719c28499f20a89a7a1019bb66780848 Mon Sep 17 00:00:00 2001 From: Trask Stalnaker Date: Wed, 2 Sep 2026 15:56:40 -0700 Subject: [PATCH 2/3] Defer queue work during dashboard publication Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 8c4a7acc-3827-4097-9d75-2d054cf9fc3f --- .../pull-request-dashboard/RATIONALE.md | 26 +-- .../pull-request-dashboard/dashboard.py | 33 +++- .../pull-request-dashboard/netlify.toml | 2 +- .../netlify/lib/dashboard-queue.mjs | 8 +- .../process_queue_batch.py | 145 ++++++++++----- .../pull-request-dashboard/state_branch.py | 50 ++--- .../pull-request-dashboard/test_dashboard.py | 66 ++++++- .../test_dashboard_queue.mjs | 76 +++++++- .../test_process_queue_batch.py | 176 +++++++++++++++++- .../test_state_branch.py | 31 +-- 10 files changed, 486 insertions(+), 127 deletions(-) diff --git a/.github/scripts/pull-request-dashboard/RATIONALE.md b/.github/scripts/pull-request-dashboard/RATIONALE.md index 1c595cf7245..a70f8e4cbf8 100644 --- a/.github/scripts/pull-request-dashboard/RATIONALE.md +++ b/.github/scripts/pull-request-dashboard/RATIONALE.md @@ -103,14 +103,13 @@ the implementation understandable and operationally cheap. acquire a lease stored on the repository's state branch. This shared lock prevents the drain from overlapping the direct concurrency group while either path performs external delivery or publishes the issue. Dashboard state - writers treat an active lease as a write barrier. They can calculate updates - concurrently, but wait to commit accepted state until publication finishes. - A writer that races lease acquisition loses its compare-and-swap push and - waits on its next attempt. Queue drains give all publisher-lock waits one - shared deadline, leaving time to return unfinished claims before the job - timeout. Accepted work lives on the state branch: a targeted publisher limits - status-comment and Slack delivery to its triggering PR. Webhook runs arrive - concurrently for many + writers treat an active lease as a write barrier. Direct writers wait for the + publisher to finish. Queue drains do not hold a runner while waiting: they + return the repository's claims to the durable queue with a retry delay. A + writer that races lease acquisition loses its compare-and-swap push and waits + or returns its work on the next attempt. Accepted work lives on the state + branch: a targeted publisher limits status-comment and Slack delivery to its + triggering PR. Webhook runs arrive concurrently for many PRs, so allowing each publisher to fan out into repository-wide delivery would create long jobs and put pressure on the GitHub Actions job queue, especially when a new status-comment revision queues every open PR. The @@ -140,10 +139,13 @@ the implementation understandable and operationally cheap. do not contend on the same git ref during scheduled and webhook-driven runs. - Updates use `git push --force-with-lease`, so git refs provide the durable compare-and-swap boundary for concurrent same-repository runs. -- Before each compare-and-swap attempt, dashboard state writers wait for an - active publisher lease to finish. Waiting does not consume a retry. The lease - and compare-and-swap check together prevent a stream of state commits from - starving delivery. +- Before each compare-and-swap attempt, dashboard state writers respect an + active publisher lease. Direct writers wait without consuming a retry. Queue + drains fail fast and return their claims with a five-minute retry delay. The + delay does not consume the claim's processing-failure budget. The claims become + runnable after that delay; a recovery scan runs every five minutes to start + their next drain. The lease and compare-and-swap check together prevent a + stream of state commits from starving delivery. - A missing repository state branch is bootstrapped by non-PR backfills. The dashboard state records when every open non-draft PR has been populated at least once. Targeted PR runs, dashboard publishing, status comments, and diff --git a/.github/scripts/pull-request-dashboard/dashboard.py b/.github/scripts/pull-request-dashboard/dashboard.py index cc18df0a741..9cdcdad34fc 100644 --- a/.github/scripts/pull-request-dashboard/dashboard.py +++ b/.github/scripts/pull-request-dashboard/dashboard.py @@ -731,10 +731,20 @@ def apply_targeted_dashboard_update( def update_dashboard_for_pr_number(args: argparse.Namespace, state_dir: Path) -> int: if args.pr_number is None: raise RuntimeError("update_dashboard_for_pr_number requires --pr-number") + publisher_lock_wait_seconds = getattr( + args, + "publisher_lock_wait_seconds", + state_branch.DEFAULT_PUBLISHER_LOCK_WAIT_SECONDS, + ) state_branch.configure_git() state_branch.checkout_state(state_dir, args.state_branch, require_existing=False) try: + state_branch.wait_for_publisher_unlock( + state_dir, + args.state_branch, + wait_seconds=publisher_lock_wait_seconds, + ) update = build_targeted_dashboard_update(args) finally: state_branch.remove_existing_state_dir(state_dir) @@ -749,11 +759,17 @@ def update_dashboard_for_pr_number(args: argparse.Namespace, state_dir: Path) -> lambda: apply_targeted_dashboard_update(args, update, observed_at), state_branch=args.state_branch, respect_publisher_lock=True, + publisher_lock_wait_seconds=publisher_lock_wait_seconds, ) def update_dashboard_for_backfill(args: argparse.Namespace, state_dir: Path) -> int: repo = normalize_repo(args.repo) if args.repo else detect_repo() + publisher_lock_wait_seconds = getattr( + args, + "publisher_lock_wait_seconds", + state_branch.DEFAULT_PUBLISHER_LOCK_WAIT_SECONDS, + ) owner, repo_name = repo.split("/", 1) prs = list_open_prs(repo) open_pr_numbers = {p["number"] for p in prs} @@ -785,6 +801,7 @@ def update_dashboard_for_backfill(args: argparse.Namespace, state_dir: Path) -> ), state_branch=args.state_branch, respect_publisher_lock=True, + publisher_lock_wait_seconds=publisher_lock_wait_seconds, ) if status != 0: return status @@ -811,6 +828,7 @@ def save_current_dashboard_state() -> int: save_current_dashboard_state, state_branch=args.state_branch, respect_publisher_lock=True, + publisher_lock_wait_seconds=publisher_lock_wait_seconds, ) for pr_summary in selection.selected_prs: @@ -880,6 +898,7 @@ def update_selected_pr(pr_summary: dict[str, Any] = pr_summary) -> int: update_selected_pr, state_branch=args.state_branch, respect_publisher_lock=True, + publisher_lock_wait_seconds=publisher_lock_wait_seconds, ) if status != 0: return status @@ -955,13 +974,25 @@ def main() -> int: type=Path, help="append initial_backfill_complete to this GitHub Actions output file", ) + parser.add_argument( + "--publisher-lock-wait-seconds", + type=int, + default=state_branch.DEFAULT_PUBLISHER_LOCK_WAIT_SECONDS, + help="maximum time to wait for an active dashboard publisher", + ) args = parser.parse_args() if args.required_approvals < 1: parser.error("--required-approvals must be at least 1") + if args.publisher_lock_wait_seconds < 0: + parser.error("--publisher-lock-wait-seconds must be non-negative") with state_branch.temporary_state_dir() as state_dir: repo_key = repo_state_key(args.repo) if args.repo else repo_state_key(detect_repo()) set_state_dir(state_dir / repo_key) - status = update_dashboard_via_state_branch(args, state_dir) + try: + status = update_dashboard_via_state_branch(args, state_dir) + except state_branch.PublisherLockTimeoutError as error: + print(error, file=sys.stderr) + return state_branch.PUBLISHER_LOCK_BUSY_STATUS if args.github_output and status in (0, BACKFILL_RECORDED_FAILURE_STATUS): write_initial_backfill_output(args.github_output) return status diff --git a/.github/scripts/pull-request-dashboard/netlify.toml b/.github/scripts/pull-request-dashboard/netlify.toml index 6ef70a56f33..572d2cb328f 100644 --- a/.github/scripts/pull-request-dashboard/netlify.toml +++ b/.github/scripts/pull-request-dashboard/netlify.toml @@ -5,4 +5,4 @@ publish = "public" directory = "netlify/functions" [functions."dashboard-queue-recover"] -schedule = "@hourly" +schedule = "*/5 * * * *" diff --git a/.github/scripts/pull-request-dashboard/netlify/lib/dashboard-queue.mjs b/.github/scripts/pull-request-dashboard/netlify/lib/dashboard-queue.mjs index dd620fe71e9..6a1dee3bc62 100644 --- a/.github/scripts/pull-request-dashboard/netlify/lib/dashboard-queue.mjs +++ b/.github/scripts/pull-request-dashboard/netlify/lib/dashboard-queue.mjs @@ -486,8 +486,12 @@ export class DashboardQueue { item.leaseOwner = null; item.leaseExpiresAt = null; item.claimedGeneration = null; - item.attempts = outcome === "retry" && !hasFollowUp ? item.attempts + 1 : 0; - item.notBefore = outcome === "retry" && !hasFollowUp && retryAfterMs > 0 + if (hasFollowUp) { + item.attempts = 0; + } else if (outcome === "retry" && retryAfterMs === 0) { + item.attempts += 1; + } + item.notBefore = outcome === "retry" && retryAfterMs > 0 ? this.#isoAfter(retryAfterMs) : null; const result = { diff --git a/.github/scripts/pull-request-dashboard/process_queue_batch.py b/.github/scripts/pull-request-dashboard/process_queue_batch.py index b68e0f967a6..55b9c8fd27f 100644 --- a/.github/scripts/pull-request-dashboard/process_queue_batch.py +++ b/.github/scripts/pull-request-dashboard/process_queue_batch.py @@ -8,7 +8,6 @@ import sys import tempfile import threading -import time from collections import defaultdict from collections.abc import Callable, Iterator from contextlib import AbstractContextManager, contextmanager @@ -23,7 +22,7 @@ OWNER = "open-telemetry" STATE_BRANCH_PREFIX = "otelbot/pull-request-dashboard-state" MAX_ATTEMPTS = 3 -QUEUE_LOCK_WAIT_BUDGET_SECONDS = 40 * 60 +PUBLISHER_LOCK_RETRY_AFTER_MS = 5 * 60 * 1000 class LeaseMonitor: @@ -93,20 +92,16 @@ class WorkItem: claims: tuple[Claim, ...] +class CommandFailedError(RuntimeError): + def __init__(self, command: list[str], returncode: int) -> None: + super().__init__(f"command failed with exit code {returncode}: {' '.join(command)}") + self.returncode = returncode + + @contextmanager -def queue_lock_wait_deadline( - now: Callable[[], float] = time.time, -) -> Iterator[None]: - name = state_branch.PUBLISHER_LOCK_DEADLINE_ENV - previous = os.environ.get(name) - os.environ[name] = str(int(now()) + QUEUE_LOCK_WAIT_BUDGET_SECONDS) - try: +def queue_publisher_lock(state_branch_name: str, owner: str) -> Iterator[None]: + with state_branch.publisher_lock(state_branch_name, owner, wait_seconds=0): yield - finally: - if previous is None: - os.environ.pop(name, None) - else: - os.environ[name] = previous def load_claims(path: Path) -> list[Claim]: @@ -140,6 +135,9 @@ def resolve_work_items( for claim in claims: try: pr_number = claim.pr_number or resolve_head(claim.repository, claim.head_sha) + except state_branch.PublisherLockTimeoutError as error: + completed.extend(publisher_lock_acknowledgments((claim,), error)) + continue except Exception as error: completed.extend(failure_acknowledgments((claim,), error)) continue @@ -230,6 +228,7 @@ def __init__( } def resolve_head(self, repository: str, head_sha: str) -> int | None: + self._assert_publisher_unlocked(repository) result = self._run( [ "gh", @@ -248,6 +247,21 @@ def resolve_head(self, repository: str, head_sha: str) -> int | None: ) return matches[0] if matches else None + def _assert_publisher_unlocked(self, repository: str) -> None: + state_branch_name = f"{STATE_BRANCH_PREFIX}/{repository}" + with state_branch.temporary_state_dir() as state_dir: + state_branch.configure_git() + state_branch.checkout_state( + state_dir, + state_branch_name, + require_existing=False, + ) + state_branch.wait_for_publisher_unlock( + state_dir, + state_branch_name, + wait_seconds=0, + ) + def process_repository( self, repository: str, @@ -279,13 +293,13 @@ def process_repository( "SLACK_CHANNEL": config.get("slack_channel", ""), "SLACK_USER_MAP_JSON": json.dumps(config.get("slack_user_mapping", {})), } - state_branch = f"{STATE_BRANCH_PREFIX}/{repository}" + state_branch_name = f"{STATE_BRANCH_PREFIX}/{repository}" results: list[dict[str, Any]] = [] ready: list[WorkItem] = [] try: initial_backfill_complete = self._initial_backfill_complete( - repository, state_branch, env + repository, state_branch_name, env ) except Exception as error: return [ @@ -300,10 +314,28 @@ def process_repository( for claim in item.claims ] - for item in items: + for index, item in enumerate(items): try: - self._update_dashboard(repository, item.pr_number, state_branch, config, env) + self._update_dashboard( + repository, + item.pr_number, + state_branch_name, + config, + env, + ) ready.append(item) + except CommandFailedError as error: + if error.returncode == state_branch.PUBLISHER_LOCK_BUSY_STATUS: + deferred = [*ready, *items[index:]] + for deferred_item in deferred: + results.extend( + publisher_lock_acknowledgments( + deferred_item.claims, + error, + ) + ) + return results + results.extend(failure_acknowledgments(item.claims, error)) except Exception as error: results.extend(failure_acknowledgments(item.claims, error)) @@ -314,10 +346,13 @@ def process_repository( successful: list[WorkItem] = [] publish_active = False try: - with self.publisher_lock(state_branch, self.publisher_lock_owner): + with self.publisher_lock(state_branch_name, self.publisher_lock_owner): for item in ready: delivery_active, delivery_error = self._deliver( - repository, item.pr_number, state_branch, env + repository, + item.pr_number, + state_branch_name, + env, ) publish_active = delivery_active or publish_active if delivery_error is not None: @@ -329,7 +364,7 @@ def process_repository( if publish_active: try: - self._publish(repository, state_branch, config, env) + self._publish(repository, state_branch_name, config, env) except Exception as error: for item in successful: locked_results.extend(failure_acknowledgments(item.claims, error)) @@ -339,6 +374,14 @@ def process_repository( locked_results.extend( acknowledgment(claim, "success") for claim in item.claims ) + except state_branch.PublisherLockTimeoutError as error: + for item in ready: + results.extend( + publisher_lock_acknowledgments( + item.claims, + error, + ) + ) except Exception as error: for item in ready: results.extend(failure_acknowledgments(item.claims, error)) @@ -388,6 +431,8 @@ def _update_dashboard( str(pr_number), "--required-approvals", str(config.get("required_approvals", 1)), + "--publisher-lock-wait-seconds", + "0", "--github-output", github_output.name, ] @@ -500,9 +545,7 @@ def _run( if result.stderr: print(result.stderr, end="", file=sys.stderr) if result.returncode != 0: - raise RuntimeError( - f"command failed with exit code {result.returncode}: {' '.join(command)}" - ) + raise CommandFailedError(command, result.returncode) return result @@ -536,6 +579,22 @@ def failure_acknowledgments( ] +def publisher_lock_acknowledgments( + claims: tuple[Claim, ...], + error: Exception, +) -> list[dict[str, Any]]: + message = str(error) + return [ + acknowledgment( + claim, + "retry", + message, + PUBLISHER_LOCK_RETRY_AFTER_MS, + ) + for claim in claims + ] + + def required_string(value: dict[str, Any], name: str) -> str: result = value.get(name) if not isinstance(result, str) or not result: @@ -600,25 +659,25 @@ def record_and_acknowledge(completed: list[dict[str, Any]]) -> None: args.results.write_text(json.dumps(completed, indent=2) + "\n", encoding="utf-8") acknowledge_all(client, args.results, common) - with queue_lock_wait_deadline(): - try: - monitor.start() - processor = DashboardBatchProcessor( - args.config, - lease_check=monitor.assert_valid, - publisher_lock_owner=args.worker_id, - ) - work_items, resolved = resolve_work_items(claims, processor.resolve_head) - record_and_acknowledge(resolved) - process_batch( - work_items, - processor.process_repository, - max_repositories=args.max_repositories, - on_results=record_and_acknowledge, - ) - finally: - args.results.write_text(json.dumps(results, indent=2) + "\n", encoding="utf-8") - monitor.close() + try: + monitor.start() + processor = DashboardBatchProcessor( + args.config, + lease_check=monitor.assert_valid, + publisher_lock=queue_publisher_lock, + publisher_lock_owner=args.worker_id, + ) + work_items, resolved = resolve_work_items(claims, processor.resolve_head) + record_and_acknowledge(resolved) + process_batch( + work_items, + processor.process_repository, + max_repositories=args.max_repositories, + on_results=record_and_acknowledge, + ) + finally: + args.results.write_text(json.dumps(results, indent=2) + "\n", encoding="utf-8") + monitor.close() dead_letters = sum(result["outcome"] == "dead" for result in results) retries = sum(result["outcome"] == "retry" for result in results) print( diff --git a/.github/scripts/pull-request-dashboard/state_branch.py b/.github/scripts/pull-request-dashboard/state_branch.py index 8a5010b4343..a429eabfa61 100644 --- a/.github/scripts/pull-request-dashboard/state_branch.py +++ b/.github/scripts/pull-request-dashboard/state_branch.py @@ -29,7 +29,7 @@ DEFAULT_PUBLISHER_LOCK_LEASE_SECONDS = 60 * 60 DEFAULT_PUBLISHER_LOCK_WAIT_SECONDS = 60 * 60 PUBLISHER_LOCK_POLL_SECONDS = 5 -PUBLISHER_LOCK_DEADLINE_ENV = "PR_DASHBOARD_PUBLISHER_LOCK_DEADLINE" +PUBLISHER_LOCK_BUSY_STATUS = 75 class PublisherLock(TypedDict): @@ -37,6 +37,10 @@ class PublisherLock(TypedDict): expiresAt: float +class PublisherLockTimeoutError(TimeoutError): + pass + + @contextmanager def temporary_state_dir() -> Iterator[Path]: with tempfile.TemporaryDirectory(prefix="pull-request-dashboard-") as temp_root: @@ -258,22 +262,6 @@ def load_publisher_lock(state_dir: Path) -> PublisherLock | None: return {"owner": owner, "expiresAt": float(expires_at)} -def publisher_lock_wait_deadline(wait_seconds: int, current_time: float) -> float: - if wait_seconds < 0: - raise ValueError("publisher lock wait must be non-negative") - deadline = current_time + wait_seconds - configured = os.environ.get(PUBLISHER_LOCK_DEADLINE_ENV) - if configured is None: - return deadline - try: - configured_deadline = int(configured) - except ValueError as error: - raise ValueError(f"{PUBLISHER_LOCK_DEADLINE_ENV} must be an integer") from error - if configured_deadline < 1: - raise ValueError(f"{PUBLISHER_LOCK_DEADLINE_ENV} must be positive") - return min(deadline, configured_deadline) - - def wait_for_publisher_unlock( state_dir: Path, state_branch: str, @@ -282,7 +270,9 @@ def wait_for_publisher_unlock( now: Callable[[], float] = time.time, sleep: Callable[[float], None] = time.sleep, ) -> None: - deadline = publisher_lock_wait_deadline(wait_seconds, now()) + if wait_seconds < 0: + raise ValueError("publisher lock wait must be non-negative") + deadline = now() + wait_seconds announced_owner: str | None = None while True: current_time = now() @@ -297,7 +287,7 @@ def wait_for_publisher_unlock( announced_owner = lock["owner"] remaining = deadline - current_time if remaining <= 0: - raise TimeoutError( + raise PublisherLockTimeoutError( f"timed out waiting for dashboard publisher lock {state_branch} " f"held by {lock['owner']}" ) @@ -338,7 +328,7 @@ def acquire_publisher_lock( raise ValueError("publisher lock owner must not be empty") if lease_seconds < 1 or wait_seconds < 0: raise ValueError("publisher lock lease must be positive and wait must be non-negative") - deadline = publisher_lock_wait_deadline(wait_seconds, now()) + deadline = now() + wait_seconds while True: current_time = now() with temporary_state_dir() as state_dir: @@ -371,7 +361,9 @@ def acquire_publisher_lock( return remaining = deadline - now() if remaining <= 0: - raise TimeoutError(f"timed out waiting for dashboard publisher lock {state_branch}") + raise PublisherLockTimeoutError( + f"timed out waiting for dashboard publisher lock {state_branch}" + ) sleep(min(PUBLISHER_LOCK_POLL_SECONDS, remaining)) @@ -401,8 +393,13 @@ def release_publisher_lock(state_branch: str, owner: str) -> None: @contextmanager -def publisher_lock(state_branch: str, owner: str) -> Iterator[None]: - acquire_publisher_lock(state_branch, owner) +def publisher_lock( + state_branch: str, + owner: str, + *, + wait_seconds: int = DEFAULT_PUBLISHER_LOCK_WAIT_SECONDS, +) -> Iterator[None]: + acquire_publisher_lock(state_branch, owner, wait_seconds=wait_seconds) try: yield finally: @@ -425,6 +422,7 @@ def push_state_changes( add_paths: list[str] | None = None, retry_snapshots: list[tuple[Path, Path]] | None = None, respect_publisher_lock: bool = False, + publisher_lock_wait_seconds: int = DEFAULT_PUBLISHER_LOCK_WAIT_SECONDS, ) -> int: configure_git() checkout_state(state_dir, state_branch, require_existing=False) @@ -433,7 +431,11 @@ def push_state_changes( for attempt in range(1, max_attempts + 1): if respect_publisher_lock: - wait_for_publisher_unlock(state_dir, state_branch) + wait_for_publisher_unlock( + state_dir, + state_branch, + wait_seconds=publisher_lock_wait_seconds, + ) status = update_state() if status != 0: return status diff --git a/.github/scripts/pull-request-dashboard/test_dashboard.py b/.github/scripts/pull-request-dashboard/test_dashboard.py index 9a0e52575b2..0da6cfaf4a7 100644 --- a/.github/scripts/pull-request-dashboard/test_dashboard.py +++ b/.github/scripts/pull-request-dashboard/test_dashboard.py @@ -8,6 +8,7 @@ import unittest from unittest.mock import ANY, Mock, call, patch +import state_branch from copilot_review import set_copilot_review_request_needed from dashboard import ( BACKFILL_RECORDED_FAILURE_STATUS, @@ -2060,13 +2061,18 @@ def push_state_changes(_state_dir, _message, update_state, **_kwargs) -> int: self.assertEqual(backfill_failed_pr_numbers(backfill_state), {1}) def test_targeted_state_update_respects_publisher_lock(self) -> None: - args = Namespace(pr_number=1, state_branch="state") + args = Namespace( + pr_number=1, + state_branch="state", + publisher_lock_wait_seconds=0, + ) update = object() with ( patch("dashboard.state_branch.configure_git"), patch("dashboard.state_branch.checkout_state"), patch("dashboard.state_branch.remove_existing_state_dir"), + patch("dashboard.state_branch.wait_for_publisher_unlock") as wait_for_unlock, patch("dashboard.build_targeted_dashboard_update", return_value=update), patch( "dashboard.state_branch.push_state_changes", @@ -2076,7 +2082,38 @@ def test_targeted_state_update_respects_publisher_lock(self) -> None: status = update_dashboard_for_pr_number(args, Path("state")) self.assertEqual(0, status) + wait_for_unlock.assert_called_once_with( + Path("state"), + "state", + wait_seconds=0, + ) self.assertTrue(push_state_changes.call_args.kwargs["respect_publisher_lock"]) + self.assertEqual( + 0, + push_state_changes.call_args.kwargs["publisher_lock_wait_seconds"], + ) + + def test_publisher_lock_busy_stops_targeted_calculation(self) -> None: + args = Namespace( + pr_number=1, + state_branch="state", + publisher_lock_wait_seconds=0, + ) + + with ( + patch("dashboard.state_branch.configure_git"), + patch("dashboard.state_branch.checkout_state"), + patch("dashboard.state_branch.remove_existing_state_dir"), + patch( + "dashboard.state_branch.wait_for_publisher_unlock", + side_effect=state_branch.PublisherLockTimeoutError("busy"), + ), + patch("dashboard.build_targeted_dashboard_update") as build_update, + self.assertRaises(state_branch.PublisherLockTimeoutError), + ): + update_dashboard_for_pr_number(args, Path("state")) + + build_update.assert_not_called() def test_successful_retry_clears_recorded_failure(self) -> None: state = {"failed_pr_numbers": [1, 2]} @@ -2160,6 +2197,33 @@ def test_emits_initial_backfill_status_only_for_accepted_state_outcomes(self) -> else: write_output.assert_not_called() + def test_main_returns_busy_status_for_publisher_lock_timeout(self) -> None: + with ( + tempfile.TemporaryDirectory() as temp_dir, + patch( + "sys.argv", + [ + "dashboard.py", + "--state-branch", + "state", + "--repo", + "repo", + "--approver-team", + "approvers", + "--publisher-lock-wait-seconds", + "0", + ], + ), + patch("dashboard.state_branch.temporary_state_dir") as temporary_state_dir, + patch( + "dashboard.update_dashboard_via_state_branch", + side_effect=state_branch.PublisherLockTimeoutError("busy"), + ), + ): + temporary_state_dir.return_value.__enter__.return_value = Path(temp_dir) + + self.assertEqual(main(), state_branch.PUBLISHER_LOCK_BUSY_STATUS) + if __name__ == "__main__": unittest.main() \ No newline at end of file diff --git a/.github/scripts/pull-request-dashboard/test_dashboard_queue.mjs b/.github/scripts/pull-request-dashboard/test_dashboard_queue.mjs index 6ea72ff4ca7..574f995ca03 100644 --- a/.github/scripts/pull-request-dashboard/test_dashboard_queue.mjs +++ b/.github/scripts/pull-request-dashboard/test_dashboard_queue.mjs @@ -536,8 +536,8 @@ test("a dead acknowledgment preserves a dirty follow-up generation", async () => assert.equal(stats.deadLetters, 0); }); -test("a retry acknowledgment gives a dirty generation a fresh attempt budget", async () => { - const { queue } = fixture(); +test("a delayed retry gives a dirty generation a fresh attempt budget", async () => { + const { queue, advance } = fixture(); await queue.enqueue({ repository: "example", prNumber: 123, @@ -577,15 +577,83 @@ test("a retry acknowledgment gives a dirty generation a fresh attempt budget", a claimGeneration: secondClaim.claimGeneration, workerId: "worker", outcome: "retry", - retryAfterMs: 10_000, + retryAfterMs: 500, }), { status: "follow_up", attempts: 0 }); - const [followUp] = await queue.claimWave({ + assert.deepEqual(await queue.claimWave({ generation: request.generation, workerId: "worker", + }), []); + assert.deepEqual(await queue.finishDispatcher({ + generation: request.generation, + workerId: "worker", + }), { requested: false }); + advance(500); + const recovery = await queue.recoverExpiredLeases(); + assert.equal(recovery.requested, true); + await queue.activateDispatcher({ + generation: recovery.generation, + workerId: "next-worker", + }); + const [followUp] = await queue.claimWave({ + generation: recovery.generation, + workerId: "next-worker", }); assert.equal(followUp.attempts, 0); }); +test("a delayed retry preserves the existing attempt budget", async () => { + const { queue, advance } = fixture(); + await queue.enqueue({ + repository: "example", + prNumber: 123, + headSha: "", + triggerEvent: "pull_request", + }); + const request = await queue.requestDispatcher("request"); + await queue.activateDispatcher({ + generation: request.generation, + workerId: "worker", + }); + const [firstClaim] = await queue.claimWave({ + generation: request.generation, + workerId: "worker", + }); + assert.deepEqual(await queue.acknowledge({ + itemKey: firstClaim.itemKey, + claimGeneration: firstClaim.claimGeneration, + workerId: "worker", + outcome: "retry", + }), { status: "retry", attempts: 1 }); + const [secondClaim] = await queue.claimWave({ + generation: request.generation, + workerId: "worker", + }); + + assert.deepEqual(await queue.acknowledge({ + itemKey: secondClaim.itemKey, + claimGeneration: secondClaim.claimGeneration, + workerId: "worker", + outcome: "retry", + retryAfterMs: 500, + }), { status: "retry", attempts: 1 }); + assert.deepEqual(await queue.finishDispatcher({ + generation: request.generation, + workerId: "worker", + }), { requested: false }); + advance(500); + const recovery = await queue.recoverExpiredLeases(); + assert.equal(recovery.requested, true); + await queue.activateDispatcher({ + generation: recovery.generation, + workerId: "next-worker", + }); + const [retried] = await queue.claimWave({ + generation: recovery.generation, + workerId: "next-worker", + }); + assert.equal(retried.attempts, 1); +}); + test("expired recovery preserves a dirty generation at the attempt ceiling", async () => { const { queue, advance } = fixture(); await queue.enqueue({ diff --git a/.github/scripts/pull-request-dashboard/test_process_queue_batch.py b/.github/scripts/pull-request-dashboard/test_process_queue_batch.py index c048d1a6f5b..c6f8f6df81a 100644 --- a/.github/scripts/pull-request-dashboard/test_process_queue_batch.py +++ b/.github/scripts/pull-request-dashboard/test_process_queue_batch.py @@ -2,7 +2,6 @@ from contextlib import contextmanager, nullcontext import json -import os import subprocess import sys import tempfile @@ -102,6 +101,53 @@ def fail_resolution(_repository: str, _head_sha: str) -> int | None: self.assertEqual(completed[0]["itemKey"], head.item_key) self.assertEqual(completed[0]["outcome"], "retry") + def test_head_resolution_defers_publisher_lock_contention(self) -> None: + head = Claim( + "example#head:abc", + 1, + "example", + None, + "a" * 40, + process_queue_batch.MAX_ATTEMPTS - 1, + ) + + def locked(_repository: str, _head_sha: str) -> int | None: + raise process_queue_batch.state_branch.PublisherLockTimeoutError( + "publisher lock is busy" + ) + + work, completed = resolve_work_items([head], locked) + + self.assertEqual(work, []) + self.assertEqual(completed[0]["outcome"], "retry") + self.assertEqual( + completed[0]["retryAfterMs"], + process_queue_batch.PUBLISHER_LOCK_RETRY_AFTER_MS, + ) + + def test_head_resolution_checks_publisher_lock_before_github(self) -> None: + with tempfile.TemporaryDirectory() as directory: + config_path = Path(directory) / "repositories.json" + config_path.write_text("[]", encoding="utf-8") + run = mock.Mock() + processor = process_queue_batch.DashboardBatchProcessor( + config_path, + run=run, + ) + with mock.patch.object( + processor, + "_assert_publisher_unlocked", + side_effect=process_queue_batch.state_branch.PublisherLockTimeoutError( + "publisher lock is busy" + ), + ): + with self.assertRaises( + process_queue_batch.state_branch.PublisherLockTimeoutError + ): + processor.resolve_head("example", "a" * 40) + + run.assert_not_called() + def test_prs_are_grouped_sequentially_by_repository(self) -> None: items = [ WorkItem("b", 2, (claim("b#pr:2", "b", pr_number=2),)), @@ -263,16 +309,128 @@ def report(results: list[dict[str, object]]) -> None: self.assertTrue(slow_observed_report) - def test_queue_bounds_publisher_lock_waits_and_restores_environment(self) -> None: - name = process_queue_batch.state_branch.PUBLISHER_LOCK_DEADLINE_ENV + def test_publisher_lock_busy_defers_the_repository_batch(self) -> None: + lifecycle: list[str] = [] + with tempfile.TemporaryDirectory() as directory: + config_path = Path(directory) / "repositories.json" + config_path.write_text( + json.dumps([{"name": "example"}]), + encoding="utf-8", + ) + processor = process_queue_batch.DashboardBatchProcessor(config_path) + items = [ + WorkItem( + "example", + number, + (claim(f"example#pr:{number}", "example", pr_number=number),), + ) + for number in (1, 2, 3) + ] + + def update(_repo, number, *_args) -> None: + lifecycle.append(f"update-{number}") + if number == 2: + raise process_queue_batch.CommandFailedError( + ["dashboard.py"], + process_queue_batch.state_branch.PUBLISHER_LOCK_BUSY_STATUS, + ) + + with ( + mock.patch.object( + processor, + "_initial_backfill_complete", + return_value=True, + ), + mock.patch.object(processor, "_update_dashboard", side_effect=update), + mock.patch.object( + processor, + "_deliver", + side_effect=lambda *_args: lifecycle.append("deliver"), + ), + ): + results = processor.process_repository("example", items) + + self.assertEqual(["update-1", "update-2"], lifecycle) + self.assertEqual( + [result["itemKey"] for result in results], + ["example#pr:1", "example#pr:2", "example#pr:3"], + ) + self.assertTrue(all(result["outcome"] == "retry" for result in results)) + self.assertTrue( + all( + result["retryAfterMs"] + == process_queue_batch.PUBLISHER_LOCK_RETRY_AFTER_MS + for result in results + ) + ) + + def test_publisher_lock_deferral_does_not_dead_letter_at_attempt_limit(self) -> None: + claim = Claim("example#pr:1", 1, "example", 1, "", 2) + + [result] = process_queue_batch.publisher_lock_acknowledgments( + (claim,), + process_queue_batch.state_branch.PublisherLockTimeoutError("busy"), + ) + + self.assertEqual("retry", result["outcome"]) + self.assertEqual( + process_queue_batch.PUBLISHER_LOCK_RETRY_AFTER_MS, + result["retryAfterMs"], + ) + + def test_queue_publisher_lock_does_not_wait(self) -> None: + with mock.patch.object( + process_queue_batch.state_branch, + "publisher_lock", + return_value=nullcontext(), + ) as publisher_lock: + with process_queue_batch.queue_publisher_lock("state", "worker"): + pass + + publisher_lock.assert_called_once_with("state", "worker", wait_seconds=0) - with mock.patch.dict(os.environ, {name: "50"}): - with process_queue_batch.queue_lock_wait_deadline(now=lambda: 100): - self.assertEqual( - str(100 + process_queue_batch.QUEUE_LOCK_WAIT_BUDGET_SECONDS), - os.environ[name], + def test_busy_publisher_defers_updates_that_are_ready_to_deliver(self) -> None: + @contextmanager + def busy_publisher_lock(_branch: str, _owner: str): + raise process_queue_batch.state_branch.PublisherLockTimeoutError("busy") + yield + + with tempfile.TemporaryDirectory() as directory: + config_path = Path(directory) / "repositories.json" + config_path.write_text( + json.dumps([{"name": "example"}]), + encoding="utf-8", + ) + processor = process_queue_batch.DashboardBatchProcessor( + config_path, + publisher_lock=busy_publisher_lock, + ) + items = [ + WorkItem( + "example", + number, + (claim(f"example#pr:{number}", "example", pr_number=number),), ) - self.assertEqual("50", os.environ[name]) + for number in (1, 2) + ] + with ( + mock.patch.object( + processor, + "_initial_backfill_complete", + return_value=True, + ), + mock.patch.object(processor, "_update_dashboard"), + ): + results = processor.process_repository("example", items) + + self.assertTrue(all(result["outcome"] == "retry" for result in results)) + self.assertTrue( + all( + result["retryAfterMs"] + == process_queue_batch.PUBLISHER_LOCK_RETRY_AFTER_MS + for result in results + ) + ) def test_delivery_error_still_publishes_committed_active_state(self) -> None: commands: list[str] = [] diff --git a/.github/scripts/pull-request-dashboard/test_state_branch.py b/.github/scripts/pull-request-dashboard/test_state_branch.py index d96b85e2d88..be7e685b168 100644 --- a/.github/scripts/pull-request-dashboard/test_state_branch.py +++ b/.github/scripts/pull-request-dashboard/test_state_branch.py @@ -190,35 +190,6 @@ def test_active_publisher_lock_times_out( now=lambda: 100, ) - @patch.object(state_branch, "reset_state", return_value=True) - @patch.object( - state_branch, - "load_publisher_lock", - side_effect=[ - {"owner": "publisher", "expiresAt": 200}, - None, - ], - ) - def test_shared_deadline_bounds_publisher_wait( - self, - _load_publisher_lock: object, - _reset_state: object, - ) -> None: - sleeps: list[float] = [] - - with patch.dict( - state_branch.os.environ, - {state_branch.PUBLISHER_LOCK_DEADLINE_ENV: "102"}, - ): - state_branch.wait_for_publisher_unlock( - Path("state"), - "state-branch", - now=lambda: 100, - sleep=sleeps.append, - ) - - self.assertEqual([2], sleeps) - def test_checks_barrier_before_each_cas_attempt(self) -> None: lifecycle: list[str] = [] @@ -246,7 +217,7 @@ def update_state() -> int: patch.object( state_branch, "wait_for_publisher_unlock", - side_effect=lambda *_args: lifecycle.append("wait"), + side_effect=lambda *_args, **_kwargs: lifecycle.append("wait"), ) as wait_for_publisher_unlock, ): status = state_branch.push_state_changes( From d44904bc76899b9392a9c55f738278bf303ee593 Mon Sep 17 00:00:00 2001 From: Trask Stalnaker Date: Fri, 4 Sep 2026 15:27:41 -0700 Subject: [PATCH 3/3] Address review finding: document the five-minute recovery scan Review finding: This PR makes the scheduled dashboard-queue-recover function the normal restart path for work it defers, and raises its frequency from hourly to every five minutes for exactly that reason. RATIONALE.md now says the claims become runnable after the delay and a recovery scan runs every five minutes to start their next drain. WEBHOOK_SETUP.md still tells operators the opposite: it says a new event normally starts the singleton drain immediately and that scheduled recovery is only a failure backstop. After this change a publisher-lock deferral sets notBefore in the future, finishDispatcher reports nothing runnable, and no new event restarts the drain, so the scheduled scan is the ordinary path for that work. Fix: update the WEBHOOK_SETUP.md paragraph so it also names the scheduled scan as the path that restarts deferred queue work, instead of describing it purely as a failure backstop. Analysis: A publisher-lock deferral acknowledges a claim as a retry with a five-minute delay, which sets notBefore in the future. claimWave and hasRunnableItems both skip an item until notBefore passes, so finishDispatcher finds nothing runnable and requests no successor drain. Nothing then restarts that work until either the scheduled dashboard-queue-recover scan sees the item as runnable and requests a dispatcher, or a later webhook event happens to arrive after the delay has elapsed. That is why this PR changed the function's schedule from hourly to every five minutes. WEBHOOK_SETUP.md is the operator-facing setup document, and its claim that scheduled recovery is only a failure backstop was accurate before this change, because production acknowledgments supplied no nonzero retry delay. This PR makes that sentence wrong and leaves an operator with no explanation for the new schedule. The replacement paragraph states the schedule, names both restart paths, and keeps the dead-letter sentence unchanged. It avoids the evaluator's stronger wording that would have dropped the still-true fact that a new event normally starts the drain immediately. Upsides: The setup document now matches the queue's behavior and explains why the recovery function runs every five minutes. An operator reading it will not treat a scan that starts a deferred drain as a sign that something failed. Downsides: No material downside identified. The change is documentation only and touches no file this PR already changes. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../scripts/pull-request-dashboard/WEBHOOK_SETUP.md | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/.github/scripts/pull-request-dashboard/WEBHOOK_SETUP.md b/.github/scripts/pull-request-dashboard/WEBHOOK_SETUP.md index a6fd0b3e586..de4b53f5821 100644 --- a/.github/scripts/pull-request-dashboard/WEBHOOK_SETUP.md +++ b/.github/scripts/pull-request-dashboard/WEBHOOK_SETUP.md @@ -53,11 +53,12 @@ No Netlify runtime token is shared with the drain workflow. The existing `NETLIFY_AUTH_TOKEN` remains limited to deployment and environment configuration. -The `dashboard-queue-recover` scheduled function reclaims expired worker and -dispatcher leases. A new event normally starts the singleton drain immediately; -scheduled recovery is only a failure backstop. An item whose lease expires -repeatedly without an acknowledgment is moved to the shard's dead letters -instead of being requeued forever. +The `dashboard-queue-recover` scheduled function runs every five minutes. It +reclaims expired worker and dispatcher leases, and it starts a drain for claims +whose retry delay has elapsed. A new event normally starts the singleton drain +immediately, and a later event can also pick up a delayed claim once its delay +passes. An item whose lease expires repeatedly without an acknowledgment is +moved to the shard's dead letters instead of being requeued forever. Disable Deploy Previews. PR preview deploys are unused and only add noise to PRs. In Netlify, go to **Project configuration** -> **Build & deploy** ->