fix(review): harden backlog-convergence-sweep against duplicate work

migrations/0134_pr_last_backlog_convergence_regated_at.sql

@@ -0,0 +1,16 @@
+-- Per-repo draining guard for backlog-convergence-sweep (#4502), mirroring last_regated_at (0062) but scoped
+-- to THIS sweep specifically, not shared with agent-regate-sweep's own marker.
+--
+-- BEFORE: fanOutBacklogConvergenceSweepJobs has no in-flight guard at all -- while a prior cycle's fanned-out
+-- agent-regate-pr jobs (deliveryId "backlog-convergence:...") are still mid-flight, the next 30-min cron tick
+-- re-selects the same candidates and re-enqueues duplicate jobs. A crashed/restarted worker leaves its claimed
+-- row status:"processing" until queueProcessingTimeoutMs() reclaims it (default 30 min) -- almost exactly this
+-- sweep's own cadence -- so a stuck row from one cycle does not stop the next cycle from firing.
+--
+-- AFTER: sweepRepoBacklogConvergence stamps this marker for every candidate AT DISPATCH time (mirroring
+-- #audit-sweep-dispatch-stamp), and fanOutBacklogConvergenceSweepJobs's resolution loop skips a repo whose
+-- freshest stamp is within the sweep's own draining window (isRegateSweepDraining, BACKLOG_CONVERGENCE_SWEEP_FRESHNESS_MS).
+--
+-- gittensory-computed (sweep-written), keyed to the PR, omitted from upsertPullRequestFromGitHub's SET clause so
+-- a later GitHub sync cannot clobber it. Nullable / no default -> backward-compatible.
+ALTER TABLE pull_requests ADD COLUMN last_backlog_convergence_regated_at TEXT;

migrations/0135_backlog_convergence_fanout_marker.sql

@@ -0,0 +1,14 @@
+-- Fan-out dedup marker for backlog-convergence-sweep (#4502), mirroring last_regate_fanout_at (0063) with a
+-- DISTINCT column so the two sweeps' dedup windows never interfere with each other.
+--
+-- BEFORE: fanOutBacklogConvergenceSweepJobs has no atomic dedup -- a burst of fan-out trigger jobs (a
+-- deploy-restart cron catch-up, or triggers queued behind a backlogged consumer and drained together) each
+-- re-enumerate eligible repos and re-enqueue duplicate per-repo sweeps before the (0134) dispatch-stamp
+-- in-flight guard can engage.
+--
+-- AFTER: claimBacklogConvergenceFanoutSlot performs an atomic conditional UPDATE on this singleton column,
+-- mirroring claimRegateFanoutSlot (0063) -- D1 serializes writes, so only ONE concurrent fan-out's UPDATE
+-- matches the "unset or older than the dedup window" predicate and proceeds.
+--
+-- Reuses the global_agent_controls singleton (0059); nullable / no default -> backward-compatible.
+ALTER TABLE global_agent_controls ADD COLUMN last_backlog_convergence_fanout_at TEXT;

src/db/repositories.ts

@@ -2542,6 +2542,23 @@ export async function claimRegateFanoutSlot(env: Env, now: string, windowMs: num
   }
 }
 
+/** Atomic backlog-convergence-sweep fan-out dedup (#4502), mirroring {@link claimRegateFanoutSlot} exactly but on
+ *  a DISTINCT singleton column so the two differently-cadenced sweeps' dedup windows never interfere. */
+export async function claimBacklogConvergenceFanoutSlot(env: Env, now: string, windowMs: number): Promise<boolean> {
+  const threshold = new Date(Date.parse(now) - windowMs).toISOString();
+  try {
+    const result = await env.DB.prepare(
+      "UPDATE global_agent_controls SET last_backlog_convergence_fanout_at = ?1 WHERE id = 'singleton' AND (last_backlog_convergence_fanout_at IS NULL OR last_backlog_convergence_fanout_at < ?2)",
+    )
+      .bind(now, threshold)
+      .run();
+    /* v8 ignore next -- D1 update metadata normally includes changes; the ?? 0 fallback protects driver anomalies. */
+    return Number(result.meta.changes ?? 0) === 1;
+  } catch {
+    return true;
+  }
+}
+
 /** Atomic per-period dedup for the cross-repo maintainer recap digest (#2249): claim `periodKey` (the current
  *  UTC date, "YYYY-MM-DD") as the singleton's last-sent period. Mirrors {@link claimRegateFanoutSlot}: the
  *  conditional UPDATE matches only when the stored period is unset or DIFFERENT from `periodKey`, so a retried
@@ -3779,6 +3796,20 @@ export async function markPullRequestsRegated(env: Env, fullName: string, number
     .where(and(eq(pullRequests.repoFullName, fullName), inArray(pullRequests.number, numbers)));
 }
 
+/** Batch variant of {@link markPullRequestsRegated} for backlog-convergence-sweep (#4502): stamps the SEPARATE
+ *  last_backlog_convergence_regated_at marker at sweep DISPATCH time, mirroring the same "stamp immediately, not
+ *  in the downstream per-PR job" shape so getLatestBacklogConvergenceRegatedAt reflects this sweep before its
+ *  staggered per-PR jobs complete. */
+export async function markPullRequestsBacklogConvergenceRegated(env: Env, fullName: string, numbers: number[]): Promise<void> {
+  if (numbers.length === 0) return;
+  const db = getDb(env.DB);
+  const now = nowIso();
+  await db
+    .update(pullRequests)
+    .set({ lastBacklogConvergenceRegatedAt: now, updatedAt: now })
+    .where(and(eq(pullRequests.repoFullName, fullName), inArray(pullRequests.number, numbers)));
+}
+
 /** In-flight guard input for the re-gate sweep fan-out (#audit-sweep-fanout): the MOST RECENT last_regated_at
  *  across a repo's OPEN PRs (the freshest sweep stamp), or null if none has been swept. fanOutAgentRegateSweepJobs
  *  passes this to isRegateSweepDraining to skip re-arming a repo whose prior sweep is still draining. */
@@ -3793,6 +3824,19 @@ export async function getLatestRegatedAt(env: Env, fullName: string): Promise<st
   return row.latest;
 }
 
+/** In-flight guard input for the backlog-convergence-sweep fan-out (#4502), mirroring {@link getLatestRegatedAt}
+ *  exactly but over the SEPARATE last_backlog_convergence_regated_at marker. */
+export async function getLatestBacklogConvergenceRegatedAt(env: Env, fullName: string): Promise<string | null> {
+  const db = getDb(env.DB);
+  const [row] = await db
+    .select({ latest: sql<string | null>`max(${pullRequests.lastBacklogConvergenceRegatedAt})` })
+    .from(pullRequests)
+    .where(and(eq(pullRequests.repoFullName, fullName), eq(pullRequests.state, "open")));
+  /* v8 ignore next -- max() always returns exactly one row; the empty-array guard only satisfies the destructure type. */
+  if (!row) return null;
+  return row.latest;
+}
+
 export async function getIssue(env: Env, fullName: string, number: number): Promise<IssueRecord | null> {
   const db = getDb(env.DB);
   const [row] = await db.select().from(issues).where(and(eq(issues.repoFullName, fullName), eq(issues.number, number))).limit(1);

src/db/schema.ts

@@ -470,6 +470,11 @@ export const pullRequests = sqliteTable(
     // review WRITE that would bump updated_at is suppressed (dry-run / paused). gittensory-computed (sweep-written),
     // omitted from the GitHub-sync SET clause so a later sync cannot clobber it. (Mirrors approved_head_sha.)
     lastRegatedAt: text("last_regated_at"),
+    // Draining guard for backlog-convergence-sweep (#4502), mirroring lastRegatedAt but scoped to THIS sweep --
+    // stamped at dispatch by sweepRepoBacklogConvergence, read by fanOutBacklogConvergenceSweepJobs to skip a
+    // repo whose prior fan-out is still draining. Kept separate from lastRegatedAt so the two differently-cadenced
+    // sweeps' in-flight signals never conflate. gittensory-computed, omitted from the GitHub-sync SET clause.
+    lastBacklogConvergenceRegatedAt: text("last_backlog_convergence_regated_at"),
     // Public-surface marker: the head SHA at which the public surface (comment/label/check-run) was LAST published.
     // Used for reporting and stale-surface diagnostics, not as a hard sweep skip; GitHub comments/checks can still
     // be stale or partial while this marker matches headSha. gittensory-computed (publish-written), omitted from

src/index.ts

@@ -28,6 +28,12 @@ const app = createApp();
 // per-repo drain guard (getLatestRegatedAt / isRegateSweepDraining) already protects individual repos once that
 // single fan-out runs, so this only needs to stop a SECOND trigger from queuing up behind the first.
 const REGATE_SWEEP_TRIGGER_TYPES = ["agent-regate-sweep"] as const;
+// Same shape as REGATE_SWEEP_TRIGGER_TYPES, scoped to backlog-convergence-sweep's own top-level trigger (#4502):
+// its per-repo draining guard (getLatestBacklogConvergenceRegatedAt / isRegateSweepDraining) already protects
+// individual repos once a fan-out runs, so this only needs to stop a SECOND trigger queuing up behind the first
+// — the gap that let a crashed/restarted worker's stuck "processing" trigger row (reclaimed only after
+// queueProcessingTimeoutMs(), which defaults to this sweep's own 30-min cadence) go unnoticed by the next tick.
+const BACKLOG_CONVERGENCE_SWEEP_TRIGGER_TYPES = ["backlog-convergence-sweep"] as const;
 
 export { RateLimiter };
 
@@ -175,7 +181,17 @@ async function enqueueScheduledJobs(env: Env, controller: ScheduledController):
     // jobs above: it is a backstop for a rare stranding, not the primary convergence path, so it does not need the
     // sweep's ~2-min cadence. Self-host only (mirrors "agent-regate-sweep") — the trigger job itself is maintenance-
     // classified (MAINTENANCE_JOB_TYPES) so it defers under live-work pressure like every other periodic sweep here.
-    if (selfHostedReviews) jobs.push({ type: "backlog-convergence-sweep", requestedBy: "schedule" });
+    if (selfHostedReviews) {
+      const backlogConvergenceTriggerBacklog = queueSnapshotBacklog(queueSnapshot, BACKLOG_CONVERGENCE_SWEEP_TRIGGER_TYPES);
+      if (backlogConvergenceTriggerBacklog > 0) {
+        // A fan-out trigger is already pending/processing (#4502) — skip re-arming so a crashed/restarted worker's
+        // stuck row (reclaimed only after queueProcessingTimeoutMs(), which coincides with this sweep's own 30-min
+        // cadence) cannot go unnoticed by the next tick and duplicate per-repo/per-PR work underneath it.
+        console.log(JSON.stringify({ event: "backlog_convergence_sweep_trigger_backlog_deferred", backlog: backlogConvergenceTriggerBacklog }));
+      } else {
+        jobs.push({ type: "backlog-convergence-sweep", requestedBy: "schedule" });
+      }
+    }
   }
   // Self-heal (flag GITTENSORY_PR_RECONCILIATION). Every 10 minutes — see isReconciliationWindow above.
   // Enqueued ONLY when the flag is ON — flag-OFF (default) this job is never created, so the cron tick does

src/queue/processors.ts

@@ -54,11 +54,14 @@ import {
   getCachedLinkedIssueSatisfaction,
   putCachedLinkedIssueSatisfaction,
   markPullRequestsRegated,
+  markPullRequestsBacklogConvergenceRegated,
   markPullRequestReviewsInvalidated,
   markPullRequestSurfacePublished,
   markPullRequestVisualCaptureSatisfied,
   getLatestRegatedAt,
+  getLatestBacklogConvergenceRegatedAt,
   claimRegateFanoutSlot,
+  claimBacklogConvergenceFanoutSlot,
   recordAgentCommandFeedback,
   recordAuditEvent,
   countRecentAuditEventsForActorAndTarget,
@@ -265,6 +268,7 @@ import {
   ISSUE_WAKE_MAX_PRS,
   MERGE_WAKE_MAX_PRS,
   SWEEP_FANOUT_DEDUP_MS,
+  BACKLOG_CONVERGENCE_SWEEP_FRESHNESS_MS,
   SWEEP_MAX_PRS,
   isRegateSweepDraining,
   selectRegateCandidates,
@@ -2076,15 +2080,26 @@ async function sweepRepoRegate(
 
 // #selfhost-backlog-convergence: the cron (index.ts) enqueues one fan-out trigger periodically; this enqueues a
 // per-repo sweep job for every repo eligible for convergence (the SAME repo selection as the re-gate sweep, so
-// a repo that opted the agent in — or is explicitly convergence-allowlisted — gets both). Deliberately has no
-// fan-out dedup CAS (contrast fanOutAgentRegateSweepJobs): unlike that sweep, this one stamps nothing
-// optimistically, so a second overlapping trigger just re-reads current state and re-enqueues, which coalesces
-// harmlessly into the same pending agent-regate-pr rows (queue-common.ts's job_key coalescing) rather than
-// duplicating work.
+// a repo that opted the agent in — or is explicitly convergence-allowlisted — gets both). #4502: now mirrors
+// fanOutAgentRegateSweepJobs's three-layer anti-duplication shape exactly — an atomic fan-out-slot claim
+// (claimBacklogConvergenceFanoutSlot) collapses a BURST of this trigger, and the per-repo resolution below skips
+// any repo whose prior fan-out is still draining (getLatestBacklogConvergenceRegatedAt / isRegateSweepDraining) —
+// closing the gap where a crashed/restarted worker's stuck "processing" trigger row went unnoticed by the next
+// 30-min tick and re-enqueued duplicate per-repo (and per-PR) jobs underneath the still-in-flight one.
 async function fanOutBacklogConvergenceSweepJobs(
   env: Env,
   requestedBy: "schedule" | "api" | "test",
 ): Promise<void> {
+  const now = nowIso();
+  if (!(await claimBacklogConvergenceFanoutSlot(env, now, SWEEP_FANOUT_DEDUP_MS))) {
+    await recordAuditEvent(env, {
+      eventType: "agent.sweep.backlog_convergence.fanout",
+      outcome: "denied",
+      detail: "backlog-convergence fan-out deduped: another fan-out already claimed this window",
+      metadata: { requestedBy, deduped: true },
+    });
+    return;
+  }
   const repositoriesByKey = new Map((await listRepositories(env)).map((repo) => [repo.fullName.toLowerCase(), repo]));
   const byKey = new Map<string, { fullName: string; installationId?: number }>();
   for (const repo of repositoriesByKey.values())
@@ -2096,12 +2111,39 @@ async function fanOutBacklogConvergenceSweepJobs(
       ...(typeof repo?.installationId === "number" ? { installationId: repo.installationId } : {}),
     });
   }
+  // #4502 (ports #3899): resolve every repo's settings + drain-state CONCURRENTLY (bounded), not one at a time —
+  // mirrors fanOutAgentRegateSweepJobs's own port of this fix, the same "many small per-repo D1/KV reads" shape.
+  const outcomes = await mapWithConcurrencyLimit(
+    [...byKey.values()],
+    SWEEP_FANOUT_RESOLUTION_CONCURRENCY,
+    async (repo): Promise<SweepFanoutResolutionOutcome> => {
+      const repoFullName = repo.fullName;
+      try {
+        const settings = await resolveRepositorySettings(env, repoFullName);
+        if (!(isConvergenceRepoAllowed(env, repoFullName) || isAgentConfigured(settings.autonomy))) return { kind: "ineligible" };
+        if (isRegateSweepDraining(await getLatestBacklogConvergenceRegatedAt(env, repoFullName), now, BACKLOG_CONVERGENCE_SWEEP_FRESHNESS_MS))
+          return { kind: "draining" };
+        return { kind: "configured", repo };
+      } catch (error) {
+        console.error(
+          JSON.stringify({
+            level: "error",
+            event: "backlog_convergence_fanout_repo_check_failed",
+            repository: repoFullName,
+            error: errorMessage(error),
+          }),
+        );
+        return { kind: "errored" };
+      }
+    },
+  );
   const configured: Array<{ fullName: string; installationId?: number }> = [];
-  for (const repo of byKey.values()) {
-    const settings = await resolveRepositorySettings(env, repo.fullName);
-    if (isConvergenceRepoAllowed(env, repo.fullName) || isAgentConfigured(settings.autonomy)) {
-      configured.push(repo);
-    }
+  let skippedDraining = 0;
+  let skippedErrored = 0;
+  for (const outcome of outcomes) {
+    if (outcome.kind === "configured") configured.push(outcome.repo);
+    else if (outcome.kind === "draining") skippedDraining += 1;
+    else if (outcome.kind === "errored") skippedErrored += 1;
   }
   await Promise.all(
     configured.map((repo, index) => {
@@ -2112,15 +2154,25 @@ async function fanOutBacklogConvergenceSweepJobs(
         ...(typeof repo.installationId === "number" ? { installationId: repo.installationId } : {}),
       };
       const delaySeconds = Math.min(index * 10, 600);
-      return delaySeconds > 0
-        ? env.JOBS.send(message, { delaySeconds })
-        : env.JOBS.send(message);
+      const send = delaySeconds > 0 ? env.JOBS.send(message, { delaySeconds }) : env.JOBS.send(message);
+      // #audit-sweep-fanout-isolation (mirrors fanOutAgentRegateSweepJobs): one repo's dispatch failure must not
+      // reject this Promise.all and abort every OTHER repo's already-in-flight send.
+      return send.catch((error) => {
+        console.error(
+          JSON.stringify({
+            level: "error",
+            event: "backlog_convergence_fanout_dispatch_failed",
+            repository: repo.fullName,
+            error: errorMessage(error),
+          }),
+        );
+      });
     }),
   );
   await recordAuditEvent(env, {
     eventType: "agent.sweep.backlog_convergence.fanout",
     outcome: "queued",
-    metadata: { repoCount: configured.length, requestedBy },
+    metadata: { repoCount: configured.length, skippedDraining, skippedErrored, requestedBy },
   });
 }
 
@@ -2215,6 +2267,24 @@ async function sweepRepoBacklogConvergence(
   const openPullRequests = await listOpenPullRequests(env, repoFullName);
   const candidates = selectBacklogConvergenceCandidates({ pulls: openPullRequests });
   if (candidates.length === 0) return;
+  // Stamp the backlog-convergence draining marker for EVERY candidate NOW, at dispatch — not in the downstream
+  // per-PR job (#4502, mirrors #audit-sweep-dispatch-stamp). This makes getLatestBacklogConvergenceRegatedAt
+  // reflect this sweep immediately, so fanOutBacklogConvergenceSweepJobs's in-flight guard skips re-arming this
+  // repo on the next cron tick BEFORE the staggered per-PR re-reviews finish. A plain D1 write → dry-run stays inert.
+  await markPullRequestsBacklogConvergenceRegated(
+    env,
+    repoFullName,
+    candidates.map((pr) => pr.number),
+  ).catch((error) => {
+    console.error(
+      JSON.stringify({
+        level: "warn",
+        event: "backlog_convergence_mark_regated_failed",
+        repository: repoFullName,
+        error: errorMessage(error),
+      }),
+    );
+  });
   await Promise.all(
     candidates.map((pr, index) => {
       const job: JobMessage = {

src/settings/agent-sweep.ts

@@ -49,6 +49,12 @@ export const SWEEP_FRESHNESS_MS = 2 * 60 * 1000;
 // behind a per-PR backlog and drained together).
 export const SWEEP_FANOUT_DEDUP_MS = 90 * 1000;
 
+// Draining window for backlog-convergence-sweep (#4502), the isRegateSweepDraining windowMs for THIS sweep
+// specifically -- distinct from SWEEP_FRESHNESS_MS because this sweep runs every ~30 min (not ~2 min), so its
+// own per-PR dispatch batch can legitimately still be draining minutes after fan-out. Sized to roughly match its
+// own cron cadence, mirroring how SWEEP_FRESHNESS_MS is sized to the ~2-min regate-sweep cadence.
+export const BACKLOG_CONVERGENCE_SWEEP_FRESHNESS_MS = 30 * 60 * 1000;
+
 // Candidate ordering mode (#3815, RepositorySettings["regateSweepOrderMode"]). "staleness" (default) is
 // selectRegateCandidates' original ordering; "oldest-first" is opt-in per repo. See the function doc comment
 // for the convergence-guarantee rationale each preserves.