diff --git a/.github/scripts/pull-request-dashboard/RATIONALE.md b/.github/scripts/pull-request-dashboard/RATIONALE.md index c8f61f96ef6..a70f8e4cbf8 100644 --- a/.github/scripts/pull-request-dashboard/RATIONALE.md +++ b/.github/scripts/pull-request-dashboard/RATIONALE.md @@ -102,9 +102,14 @@ 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. 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 @@ -134,6 +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 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 @@ -685,6 +697,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/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** -> diff --git a/.github/scripts/pull-request-dashboard/dashboard.py b/.github/scripts/pull-request-dashboard/dashboard.py index b42dd8d8f7f..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) @@ -748,11 +758,18 @@ 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, + 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} @@ -783,6 +800,8 @@ def update_dashboard_for_backfill(args: argparse.Namespace, state_dir: Path) -> observed_at, ), state_branch=args.state_branch, + respect_publisher_lock=True, + publisher_lock_wait_seconds=publisher_lock_wait_seconds, ) if status != 0: return status @@ -808,6 +827,8 @@ def save_current_dashboard_state() -> int: "Update dashboard state", 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: @@ -876,6 +897,8 @@ 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, + publisher_lock_wait_seconds=publisher_lock_wait_seconds, ) if status != 0: return status @@ -951,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 b7c64a02b65..55b9c8fd27f 100644 --- a/.github/scripts/pull-request-dashboard/process_queue_batch.py +++ b/.github/scripts/pull-request-dashboard/process_queue_batch.py @@ -9,8 +9,8 @@ import tempfile import threading 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 +22,7 @@ OWNER = "open-telemetry" STATE_BRANCH_PREFIX = "otelbot/pull-request-dashboard-state" MAX_ATTEMPTS = 3 +PUBLISHER_LOCK_RETRY_AFTER_MS = 5 * 60 * 1000 class LeaseMonitor: @@ -91,6 +92,18 @@ 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_publisher_lock(state_branch_name: str, owner: str) -> Iterator[None]: + with state_branch.publisher_lock(state_branch_name, owner, wait_seconds=0): + yield + + def load_claims(path: Path) -> list[Claim]: raw = json.loads(path.read_text(encoding="utf-8")) if not isinstance(raw, list): @@ -122,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 @@ -212,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", @@ -230,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, @@ -261,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 [ @@ -282,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)) @@ -296,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: @@ -311,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)) @@ -321,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)) @@ -370,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, ] @@ -482,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 @@ -518,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: @@ -587,6 +664,7 @@ def record_and_acknowledge(completed: list[dict[str, Any]]) -> None: 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) diff --git a/.github/scripts/pull-request-dashboard/state_branch.py b/.github/scripts/pull-request-dashboard/state_branch.py index 7e012591bb0..a429eabfa61 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_BUSY_STATUS = 75 class PublisherLock(TypedDict): @@ -36,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: @@ -257,6 +262,49 @@ def load_publisher_lock(state_dir: Path) -> PublisherLock | None: return {"owner": owner, "expiresAt": float(expires_at)} +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: + 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() + 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 PublisherLockTimeoutError( + 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, @@ -313,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)) @@ -343,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: @@ -366,6 +421,8 @@ 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, + publisher_lock_wait_seconds: int = DEFAULT_PUBLISHER_LOCK_WAIT_SECONDS, ) -> int: configure_git() checkout_state(state_dir, state_branch, require_existing=False) @@ -373,6 +430,12 @@ 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, + 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 857b7d9810e..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, @@ -20,6 +21,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 +1974,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 +2019,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 +2044,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 +2060,61 @@ 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", + 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", + return_value=0, + ) as push_state_changes, + ): + 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]} @@ -2137,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 7177f2c9abe..c6f8f6df81a 100644 --- a/.github/scripts/pull-request-dashboard/test_process_queue_batch.py +++ b/.github/scripts/pull-request-dashboard/test_process_queue_batch.py @@ -101,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),)), @@ -262,6 +309,129 @@ def report(results: list[dict[str, object]]) -> None: self.assertTrue(slow_observed_report) + 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) + + 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),), + ) + 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 9e8360dbe0c..be7e685b168 100644 --- a/.github/scripts/pull-request-dashboard/test_state_branch.py +++ b/.github/scripts/pull-request-dashboard/test_state_branch.py @@ -119,6 +119,120 @@ 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, + ) + + 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, **_kwargs: 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]: