Prevent dashboard publisher starvation

.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.

.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** ->

.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

.github/scripts/pull-request-dashboard/netlify.toml

@@ -5,4 +5,4 @@ publish = "public"
 directory = "netlify/functions"
 
 [functions."dashboard-queue-recover"]
-schedule = "@hourly"
+schedule = "*/5 * * * *"

.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 = {

.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)