From c15e2b6b1212b1620c548aca81d971b29ad8d1e9 Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Fri, 3 Jul 2026 19:01:33 -0700 Subject: [PATCH] fix(queue): drain contributor PR reviews oldest first --- src/queue/processors.ts | 16 +++- src/selfhost/pg-queue.ts | 83 +++++++++++++---- src/selfhost/queue-common.ts | 25 +++++ src/selfhost/sqlite-queue.ts | 84 +++++++++++++---- src/types.ts | 2 + test/unit/queue.test.ts | 41 +++++---- test/unit/selfhost-pg-queue.test.ts | 117 ++++++++++++++++++++++++ test/unit/selfhost-queue-common.test.ts | 37 ++++++++ test/unit/selfhost-sqlite-queue.test.ts | 103 ++++++++++++++++++++- 9 files changed, 447 insertions(+), 61 deletions(-) diff --git a/src/queue/processors.ts b/src/queue/processors.ts index 7b74852f62..efcaf50d1a 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -910,6 +910,7 @@ export async function processJob(env: Env, message: JobMessage): Promise { message.installationId, message.deliveryId, message.force, + message.prCreatedAt, ); return; case "run-agent": @@ -1551,6 +1552,7 @@ async function sweepRepoRegate( repoFullName, prNumber: pr.number, installationId: sweepInstallationId, + ...(pr.createdAt ? { prCreatedAt: pr.createdAt } : {}), }; const delaySeconds = Math.min(index * 10, 600); await (delaySeconds > 0 @@ -1669,6 +1671,7 @@ async function sweepRepoBacklogConvergence( repoFullName, prNumber: pr.number, installationId: sweepInstallationId, + ...(pr.createdAt ? { prCreatedAt: pr.createdAt } : {}), }; const delaySeconds = Math.min(index * 10, 600); return delaySeconds > 0 @@ -1707,6 +1710,7 @@ async function regatePullRequest( installationId: number, deliveryId: string, force?: boolean, + prCreatedAt?: string | null, ): Promise { // Reserve installation rate-limit headroom (#audit-rate-headroom): all repos share ONE GitHub App installation // = ONE REST bucket, so when the shared budget is low, DEFER this re-review until the reset instead of @@ -1729,6 +1733,7 @@ async function regatePullRequest( repoFullName, prNumber, installationId, + ...(prCreatedAt ? { prCreatedAt } : {}), ...(force ? { force: true } : {}), }, { delaySeconds: delayUntil(rateResetAt) }, @@ -3098,6 +3103,7 @@ async function scheduleTrailingIssueLinkedReReview( installationId: number, repoFullName: string, prNumber: number, + prCreatedAt?: string | null, ): Promise { const key = `issue-link-trailing:${repoFullName.toLowerCase()}#${prNumber}`; // Check-then-claim, but the CLAIM only happens after the send actually succeeds (#2371 follow-up): claiming @@ -3114,6 +3120,7 @@ async function scheduleTrailingIssueLinkedReReview( repoFullName, prNumber, installationId, + ...(prCreatedAt ? { prCreatedAt } : {}), }, { delaySeconds: CI_COALESCE_WINDOW_SECONDS }, ); @@ -3488,10 +3495,11 @@ async function maybeReReviewOnLinkedIssueChange( // Issue-side label/assignment changes can flip linked-issue hard-rule verdicts from mergeable to close. // Queue every linked open PR (bounded only by listOpenPullRequests' repo-wide DB limit) so the tail cannot // retain a stale passing gate until the scheduled sweep happens to reach it. - const linkingPrNumbers = openPullRequests + const linkingPrs = openPullRequests .filter((pr) => pr.linkedIssues.includes(issueNumber)) - .map((pr) => pr.number); - for (const [index, prNumber] of linkingPrNumbers.entries()) { + .map((pr) => ({ number: pr.number, createdAt: pr.createdAt ?? null })); + for (const [index, pr] of linkingPrs.entries()) { + const prNumber = pr.number; if (await issueLinkedPrReReviewCoalesced(env, repoFullName, prNumber)) { await scheduleTrailingIssueLinkedReReview( env, @@ -3499,6 +3507,7 @@ async function maybeReReviewOnLinkedIssueChange( installationId, repoFullName, prNumber, + pr.createdAt, ); continue; } @@ -3508,6 +3517,7 @@ async function maybeReReviewOnLinkedIssueChange( repoFullName, prNumber, installationId, + ...(pr.createdAt ? { prCreatedAt: pr.createdAt } : {}), }; const delaySeconds = Math.min(index * 10, 600); await (delaySeconds > 0 diff --git a/src/selfhost/pg-queue.ts b/src/selfhost/pg-queue.ts index 5bcac4e0cc..f2af40d446 100644 --- a/src/selfhost/pg-queue.ts +++ b/src/selfhost/pg-queue.ts @@ -24,6 +24,7 @@ import { jobCoalesceMergeKeyPrefix, jobCoalesceMergedPayload, jobCoalesceSupersededKeyPrefix, + jobClaimSortKey, jobPriority, parsePositiveIntEnv, queueBackgroundConcurrency, @@ -168,13 +169,16 @@ CREATE TABLE IF NOT EXISTS ${TABLE} ( created_at BIGINT NOT NULL, last_error TEXT, priority INTEGER NOT NULL DEFAULT 0, - job_key TEXT + job_key TEXT, + claim_sort_key BIGINT NOT NULL DEFAULT 0 ); ALTER TABLE ${TABLE} ADD COLUMN IF NOT EXISTS priority INTEGER NOT NULL DEFAULT 0; ALTER TABLE ${TABLE} ADD COLUMN IF NOT EXISTS job_key TEXT; +ALTER TABLE ${TABLE} ADD COLUMN IF NOT EXISTS claim_sort_key BIGINT NOT NULL DEFAULT 0; ALTER TABLE ${TABLE} ADD COLUMN IF NOT EXISTS is_maintenance INTEGER NOT NULL DEFAULT 0; ALTER TABLE ${TABLE} ADD COLUMN IF NOT EXISTS foreground_lane TEXT; -CREATE INDEX IF NOT EXISTS ${TABLE}_claim ON ${TABLE}(status, run_after, priority); +DROP INDEX IF EXISTS ${TABLE}_claim; +CREATE INDEX IF NOT EXISTS ${TABLE}_claim ON ${TABLE}(status, priority, claim_sort_key, run_after); CREATE INDEX IF NOT EXISTS ${TABLE}_pending_job_key ON ${TABLE}(job_key, status); CREATE INDEX IF NOT EXISTS ${TABLE}_lane_claim ON ${TABLE}(status, foreground_lane, run_after); CREATE TABLE IF NOT EXISTS ${STATS_TABLE} ( @@ -289,6 +293,14 @@ export function createPgQueue( count: keyBackfilled, }), ); + const sortKeysBackfilled = await backfillJobClaimSortKeys(); + if (sortKeysBackfilled) + console.log( + JSON.stringify({ + event: "selfhost_queue_claim_sort_keys_backfilled", + count: sortKeysBackfilled, + }), + ); const maintenanceFlagsBackfilled = await backfillJobMaintenanceFlags(); if (maintenanceFlagsBackfilled) console.log( @@ -361,6 +373,23 @@ export function createPgQueue( return changed; } + async function backfillJobClaimSortKeys(): Promise { + const res = await pool.query( + `SELECT id, payload, run_after, claim_sort_key FROM ${TABLE} WHERE status IN ('pending', 'processing')`, + ); + let changed = 0; + for (const row of res.rows as Array<{ id: string; payload: string; run_after: number | string; claim_sort_key: number | string }>) { + const sortKey = jobClaimSortKey(row.payload, Number(row.run_after)); + if (sortKey === Number(row.claim_sort_key)) continue; + await pool.query(`UPDATE ${TABLE} SET claim_sort_key=$1 WHERE id=$2`, [ + sortKey, + row.id, + ]); + changed += 1; + } + return changed; + } + async function backfillJobMaintenanceFlags(): Promise { const res = await pool.query( `SELECT id, payload, is_maintenance FROM ${TABLE} WHERE status IN ('pending', 'processing')`, @@ -675,6 +704,7 @@ export function createPgQueue( const key = jobCoalesceKey(payload); const lane = foregroundLaneForJob(message.type, payload); const runAfter = now + delaySeconds * 1000; + const claimSortKey = jobClaimSortKey(payload, runAfter); const absorbedByKey = jobCoalesceAbsorbedByKey(payload); if (absorbedByKey) { const existingFull = ( @@ -715,9 +745,11 @@ export function createPgQueue( // race lets the normal supersede/coalesce/insert path below handle this job instead. const merged = await pool.query( `UPDATE ${TABLE} - SET payload=$1, run_after=GREATEST(run_after, $2), created_at=$3, priority=GREATEST(priority, $4), job_key=$5, last_error=NULL + SET payload=$1, run_after=GREATEST(run_after, $2), created_at=$3, priority=GREATEST(priority, $4), job_key=$5, + claim_sort_key=CASE WHEN claim_sort_key>0 THEN LEAST(claim_sort_key, $8) ELSE $8 END, + last_error=NULL WHERE id=$6 AND status='pending' AND job_key=$7`, - [mergedPayload, runAfter, now, priority, mergedKey, mergeCandidate.id, mergeCandidate.job_key], + [mergedPayload, runAfter, now, priority, mergedKey, mergeCandidate.id, mergeCandidate.job_key, claimSortKey], ); if (merged.rowCount) { await recordQueueMetric("gittensory_jobs_coalesced_total"); @@ -745,9 +777,10 @@ export function createPgQueue( // (4h default) can keep re-arming the clock forever, and sustained pressure defers the job indefinitely. await pool.query( `UPDATE ${TABLE} - SET payload=$1, run_after=GREATEST(run_after, $2), priority=GREATEST(priority, $3), job_key=$4, foreground_lane=$5, last_error=NULL + SET payload=$1, run_after=GREATEST(run_after, $2), priority=GREATEST(priority, $3), job_key=$4, + foreground_lane=$5, claim_sort_key=CASE WHEN claim_sort_key>0 THEN LEAST(claim_sort_key, $7) ELSE $7 END, last_error=NULL WHERE id=$6`, - [payload, runAfter, priority, key, lane, existing.id], + [payload, runAfter, priority, key, lane, existing.id, claimSortKey], ); await pool.query( `DELETE FROM ${TABLE} @@ -771,9 +804,10 @@ export function createPgQueue( // maintenance trickle clock reflects genuine wait time, not the most recent re-request. await pool.query( `UPDATE ${TABLE} - SET payload=$1, run_after=GREATEST(run_after, $2), priority=GREATEST(priority, $3), foreground_lane=$4, last_error=NULL + SET payload=$1, run_after=GREATEST(run_after, $2), priority=GREATEST(priority, $3), + foreground_lane=$4, claim_sort_key=CASE WHEN claim_sort_key>0 THEN LEAST(claim_sort_key, $6) ELSE $6 END, last_error=NULL WHERE id=$5`, - [payload, runAfter, priority, lane, existing.id], + [payload, runAfter, priority, lane, existing.id, claimSortKey], ); await recordQueueMetric("gittensory_jobs_coalesced_total"); kickOne(); @@ -781,8 +815,8 @@ export function createPgQueue( } } await pool.query( - `INSERT INTO ${TABLE} (payload, status, attempts, run_after, created_at, priority, job_key, is_maintenance, foreground_lane) VALUES ($1,'pending',0,$2,$3,$4,$5,$6,$7)`, - [payload, runAfter, now, priority, key, isMaintenanceJobType(message.type) ? 1 : 0, lane], + `INSERT INTO ${TABLE} (payload, status, attempts, run_after, created_at, priority, job_key, is_maintenance, foreground_lane, claim_sort_key) VALUES ($1,'pending',0,$2,$3,$4,$5,$6,$7,$8)`, + [payload, runAfter, now, priority, key, isMaintenanceJobType(message.type) ? 1 : 0, lane, claimSortKey], ); await recordQueueMetric("gittensory_jobs_enqueued_total"); kickOne(); @@ -790,13 +824,13 @@ export function createPgQueue( async function claimNext(): Promise { const now = Date.now(); - const foreground = (await claimNextForegroundLane(now)) ?? (await claimNextWhere(now, "priority >= $2")); + const foreground = (await claimNextForegroundLane(now)) ?? (await claimNextWhere(now, "candidate.priority >= $2")); if (foreground) return foreground; if (activeBackground >= backgroundConcurrency) return null; activeBackground++; let background: JobRow | null; try { - background = await claimNextWhere(now, "priority < $2"); + background = await claimNextWhere(now, "candidate.priority < $2"); } catch (error) { // Release the reserved background slot if the claim query itself throws (a dropped connection / lock // timeout — the exact raw pool failures pump() below is documented to catch). claimNext() runs OUTSIDE @@ -830,7 +864,7 @@ export function createPgQueue( const sequence = fairness ? Number(fairness.claim_sequence) : 0; const lane: ForegroundLane = nextForegroundLane(sequence); if (lane === "fresh") { - const freshRow = await claimNextWhere(now, "priority >= $2", { sql: "foreground_lane='fresh'", params: [] }); + const freshRow = await claimNextWhere(now, "candidate.priority >= $2", { sql: "candidate.foreground_lane='fresh'", params: [] }); if (freshRow) incr("gittensory_jobs_claimed_by_lane_total", { lane: "fresh" }); return freshRow; } @@ -847,8 +881,8 @@ export function createPgQueue( ); const repo = pickBacklogRepo(candidates, fairness?.last_backlog_repo ?? null); if (!repo) return null; - const row = await claimNextWhere(now, "priority >= $2", { - sql: "foreground_lane='backlog' AND job_key LIKE $3", + const row = await claimNextWhere(now, "candidate.priority >= $2", { + sql: "candidate.foreground_lane='backlog' AND candidate.job_key LIKE $3", params: [`agent-regate-pr:${repo}#%`], }); if (row) { @@ -865,13 +899,24 @@ export function createPgQueue( ): Promise { const extraSql = extra ? ` AND ${extra.sql}` : ""; // Atomic, multi-instance-safe: lock + claim one due job, skipping rows another instance already locked. + // The advisory lock closes the same-job-key sibling race: a second worker can SKIP LOCKED past the row + // this statement is updating, but it cannot claim another pending row with the same semantic job key. const res = await pool.query( `UPDATE ${TABLE} SET status='processing', run_after=$1 WHERE id = ( - SELECT id - FROM ${TABLE} - WHERE status='pending' AND run_after<=$1 AND ${priorityPredicate}${extraSql} - ORDER BY priority DESC, run_after, id + SELECT candidate.id + FROM ${TABLE} AS candidate + WHERE candidate.status='pending' AND candidate.run_after<=$1 AND ${priorityPredicate}${extraSql} + AND ( + candidate.job_key IS NULL OR ( + pg_try_advisory_xact_lock(hashtextextended(candidate.job_key, 0)) + AND NOT EXISTS ( + SELECT 1 FROM ${TABLE} AS processing + WHERE processing.status='processing' AND processing.job_key=candidate.job_key + ) + ) + ) + ORDER BY candidate.priority DESC, candidate.claim_sort_key, candidate.run_after, candidate.id FOR UPDATE SKIP LOCKED LIMIT 1 ) diff --git a/src/selfhost/queue-common.ts b/src/selfhost/queue-common.ts index b4fbac68cd..93cb191598 100644 --- a/src/selfhost/queue-common.ts +++ b/src/selfhost/queue-common.ts @@ -663,6 +663,7 @@ type CoalesceMessage = { requestedBy?: unknown; repoFullName?: unknown; prNumber?: unknown; + prCreatedAt?: unknown; attempt?: unknown; force?: unknown; mode?: unknown; @@ -698,6 +699,30 @@ function ragIndexRepoKeyPrefix(repo: string): string { return keyOf("rag-index-repo", repo, ""); } +const LEGACY_AGENT_REGATE_SORT_BASE_MS = Date.parse("2000-01-01T00:00:00.000Z"); + +export function jobClaimSortKey(payload: string, fallbackMs: number): number { + const message = parseCoalesceMessage(payload); + if (message?.type === "agent-regate-pr") { + const createdAtMs = normalizedTimeMs(message.prCreatedAt); + if (createdAtMs !== null) return createdAtMs; + const pr = normalizedNumber(message.prNumber); + if (pr !== null) return LEGACY_AGENT_REGATE_SORT_BASE_MS + pr; + } + return normalizedSortNumber(fallbackMs); +} + +function normalizedTimeMs(value: unknown): number | null { + if (typeof value !== "string" || value.trim() === "") return null; + const parsed = Date.parse(value); + return Number.isFinite(parsed) ? parsed : null; +} + +function normalizedSortNumber(value: unknown): number { + const parsed = typeof value === "number" ? value : typeof value === "string" ? Number(value) : Number.NaN; + return Number.isFinite(parsed) ? Math.max(0, Math.floor(parsed)) : 0; +} + export function jobCoalesceSupersededKeyPrefix(payload: string): string | null { const message = parseCoalesceMessage(payload); if (message?.type !== "rag-index-repo") return null; diff --git a/src/selfhost/sqlite-queue.ts b/src/selfhost/sqlite-queue.ts index 3a114ba849..8ee09dc2dd 100644 --- a/src/selfhost/sqlite-queue.ts +++ b/src/selfhost/sqlite-queue.ts @@ -25,6 +25,7 @@ import { jobCoalesceMergeKeyPrefix, jobCoalesceMergedPayload, jobCoalesceSupersededKeyPrefix, + jobClaimSortKey, jobPriority, parsePositiveIntEnv, queueBackgroundConcurrency, @@ -84,7 +85,8 @@ CREATE TABLE IF NOT EXISTS ${TABLE} ( created_at INTEGER NOT NULL, last_error TEXT, priority INTEGER NOT NULL DEFAULT 0, - job_key TEXT + job_key TEXT, + claim_sort_key INTEGER NOT NULL DEFAULT 0 );`; const STATS_DDL = ` CREATE TABLE IF NOT EXISTS ${STATS_TABLE} ( @@ -99,7 +101,7 @@ CREATE TABLE IF NOT EXISTS ${FAIRNESS_TABLE} ( );`; const CLAIM_INDEX_DDL = ` DROP INDEX IF EXISTS ${TABLE}_claim; -CREATE INDEX ${TABLE}_claim ON ${TABLE}(status, run_after, priority);`; +CREATE INDEX ${TABLE}_claim ON ${TABLE}(status, priority, claim_sort_key, run_after);`; const JOB_KEY_INDEX_DDL = ` CREATE INDEX IF NOT EXISTS ${TABLE}_pending_job_key ON ${TABLE}(job_key, status);`; const LANE_INDEX_DDL = ` @@ -191,6 +193,11 @@ export function createSqliteQueue( } catch { /* column already present */ } + try { + driver.exec(`ALTER TABLE ${TABLE} ADD COLUMN claim_sort_key INTEGER NOT NULL DEFAULT 0`); + } catch { + /* column already present */ + } try { driver.exec(`ALTER TABLE ${TABLE} ADD COLUMN is_maintenance INTEGER NOT NULL DEFAULT 0`); } catch { @@ -222,6 +229,14 @@ export function createSqliteQueue( count: keyBackfilled, }), ); + const sortKeysBackfilled = backfillJobClaimSortKeys(driver); + if (sortKeysBackfilled) + console.log( + JSON.stringify({ + event: "selfhost_queue_claim_sort_keys_backfilled", + count: sortKeysBackfilled, + }), + ); const maintenanceFlagsBackfilled = backfillJobMaintenanceFlags(driver); if (maintenanceFlagsBackfilled) console.log( @@ -403,6 +418,7 @@ export function createSqliteQueue( const key = jobCoalesceKey(payload); const lane = foregroundLaneForJob(message.type, payload); const runAfter = now + delaySeconds * 1000; + const claimSortKey = jobClaimSortKey(payload, runAfter); const absorbedByKey = jobCoalesceAbsorbedByKey(payload); if (absorbedByKey) { const existingFull = driver.query( @@ -437,9 +453,11 @@ export function createSqliteQueue( const mergedKey = jobCoalesceKey(mergedPayload); driver.query( `UPDATE ${TABLE} - SET payload=?, run_after=max(run_after, ?), created_at=?, priority=max(priority, ?), job_key=?, last_error=NULL + SET payload=?, run_after=max(run_after, ?), created_at=?, priority=max(priority, ?), job_key=?, + claim_sort_key=CASE WHEN claim_sort_key>0 THEN min(claim_sort_key, ?) ELSE ? END, + last_error=NULL WHERE id=?`, - [mergedPayload, runAfter, now, priority, mergedKey, mergeCandidate.id], + [mergedPayload, runAfter, now, priority, mergedKey, claimSortKey, claimSortKey, mergeCandidate.id], ); recordQueueMetric(driver, "gittensory_jobs_coalesced_total"); kickOne(); @@ -464,9 +482,11 @@ export function createSqliteQueue( // (4h default) can keep re-arming the clock forever, and sustained pressure defers the job indefinitely. driver.query( `UPDATE ${TABLE} - SET payload=?, run_after=max(run_after, ?), priority=max(priority, ?), job_key=?, foreground_lane=?, last_error=NULL + SET payload=?, run_after=max(run_after, ?), priority=max(priority, ?), job_key=?, foreground_lane=?, + claim_sort_key=CASE WHEN claim_sort_key>0 THEN min(claim_sort_key, ?) ELSE ? END, + last_error=NULL WHERE id=?`, - [payload, runAfter, priority, key, lane, existing.id], + [payload, runAfter, priority, key, lane, claimSortKey, claimSortKey, existing.id], ); driver.query( `DELETE FROM ${TABLE} @@ -488,9 +508,11 @@ export function createSqliteQueue( // maintenance trickle clock reflects genuine wait time, not the most recent re-request. driver.query( `UPDATE ${TABLE} - SET payload=?, run_after=max(run_after, ?), priority=max(priority, ?), foreground_lane=?, last_error=NULL + SET payload=?, run_after=max(run_after, ?), priority=max(priority, ?), foreground_lane=?, + claim_sort_key=CASE WHEN claim_sort_key>0 THEN min(claim_sort_key, ?) ELSE ? END, + last_error=NULL WHERE id=?`, - [payload, runAfter, priority, lane, existing.id], + [payload, runAfter, priority, lane, claimSortKey, claimSortKey, existing.id], ); recordQueueMetric(driver, "gittensory_jobs_coalesced_total"); kickOne(); @@ -498,8 +520,8 @@ export function createSqliteQueue( } } driver.query( - `INSERT INTO ${TABLE} (payload, status, attempts, run_after, created_at, priority, job_key, is_maintenance, foreground_lane) VALUES (?, 'pending', 0, ?, ?, ?, ?, ?, ?)`, - [payload, runAfter, now, priority, key, isMaintenanceJobType(message.type) ? 1 : 0, lane], + `INSERT INTO ${TABLE} (payload, status, attempts, run_after, created_at, priority, job_key, is_maintenance, foreground_lane, claim_sort_key) VALUES (?, 'pending', 0, ?, ?, ?, ?, ?, ?, ?)`, + [payload, runAfter, now, priority, key, isMaintenanceJobType(message.type) ? 1 : 0, lane, claimSortKey], ); recordQueueMetric(driver, "gittensory_jobs_enqueued_total"); kickOne(); @@ -507,13 +529,13 @@ export function createSqliteQueue( function claimNext(): JobRow | null { const now = Date.now(); - const foreground = claimNextForegroundLane(now) ?? claimNextWhere(now, "priority>=?"); + const foreground = claimNextForegroundLane(now) ?? claimNextWhere(now, "candidate.priority>=?"); if (foreground) return foreground; if (activeBackground >= backgroundConcurrency) return null; activeBackground++; let background: JobRow | null; try { - background = claimNextWhere(now, "priority=?", { sql: "foreground_lane='fresh'", params: [] }); + const freshRow = claimNextWhere(now, "candidate.priority>=?", { sql: "candidate.foreground_lane='fresh'", params: [] }); if (freshRow) incr("gittensory_jobs_claimed_by_lane_total", { lane: "fresh" }); return freshRow; } @@ -560,8 +582,8 @@ export function createSqliteQueue( ); const repo = pickBacklogRepo(candidates, fairness?.last_backlog_repo ?? null); if (!repo) return null; - const row = claimNextWhere(now, "priority>=?", { - sql: "foreground_lane='backlog' AND job_key LIKE ?", + const row = claimNextWhere(now, "candidate.priority>=?", { + sql: "candidate.foreground_lane='backlog' AND candidate.job_key LIKE ?", params: [`agent-regate-pr:${repo}#%`], }); if (row) { @@ -608,10 +630,16 @@ export function createSqliteQueue( ): JobRow | null { const extraSql = extra ? ` AND ${extra.sql}` : ""; const { rows } = driver.query( - `SELECT id, payload, attempts, job_key, priority, created_at - FROM ${TABLE} - WHERE status='pending' AND run_after<=? AND ${priorityPredicate}${extraSql} - ORDER BY priority DESC, run_after, id + `SELECT candidate.id, candidate.payload, candidate.attempts, candidate.job_key, candidate.priority, candidate.created_at + FROM ${TABLE} AS candidate + WHERE candidate.status='pending' AND candidate.run_after<=? AND ${priorityPredicate}${extraSql} + AND ( + candidate.job_key IS NULL OR NOT EXISTS ( + SELECT 1 FROM ${TABLE} AS processing + WHERE processing.status='processing' AND processing.job_key=candidate.job_key + ) + ) + ORDER BY candidate.priority DESC, candidate.claim_sort_key, candidate.run_after, candidate.id LIMIT 1`, [now, FOREGROUND_QUEUE_PRIORITY_FLOOR, ...(extra?.params ?? [])], ); @@ -1075,6 +1103,24 @@ function backfillJobKeys(driver: SqliteDriver): number { return changed; } +function backfillJobClaimSortKeys(driver: SqliteDriver): number { + const { rows } = driver.query( + `SELECT id, payload, run_after, claim_sort_key FROM ${TABLE} WHERE status IN ('pending', 'processing')`, + [], + ); + let changed = 0; + for (const row of rows as Array<{ id: number; payload: string; run_after: number; claim_sort_key: number }>) { + const sortKey = jobClaimSortKey(row.payload, row.run_after); + if (sortKey === Number(row.claim_sort_key)) continue; + driver.query(`UPDATE ${TABLE} SET claim_sort_key=? WHERE id=?`, [ + sortKey, + row.id, + ]); + changed += 1; + } + return changed; +} + function backfillJobMaintenanceFlags(driver: SqliteDriver): number { const { rows } = driver.query( `SELECT id, payload, is_maintenance FROM ${TABLE} WHERE status IN ('pending', 'processing')`, diff --git a/src/types.ts b/src/types.ts index 75125ecb10..cf1fd77efb 100644 --- a/src/types.ts +++ b/src/types.ts @@ -40,6 +40,8 @@ export type JobMessage = repoFullName: string; prNumber: number; installationId: number; + /** Original GitHub PR creation time. Durable self-host queues use this to drain contributor PR work oldest-first. */ + prCreatedAt?: string | null | undefined; // #regate-churn (req 8): an explicit manual re-gate request — bypasses the AI review cache and the // bounded non-cacheable-reuse cooldown so it always pays for a fresh opinion. No current scheduled or // webhook-driven caller sets this; it exists so a manual trigger has a supported way to force a fresh diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index 2bceb04cd2..83efc80687 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -1825,7 +1825,7 @@ describe("queue processors", () => { await upsertRepositoryFromGitHub(env, { name: "agent-repo", full_name: "owner/agent-repo", private: false, owner: { login: "owner" } }, 9001); await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo", autonomy: { merge: "auto" }, aiReviewMode: "off", gatePack: "oss-anti-slop", gateCheckMode: "enabled", checkRunMode: "off", commentMode: "off", publicSurface: "off" }); // Links issue #1 — the issue the "labeled" event below fires on. - await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 7, title: "Linking PR", state: "open", user: { login: "contributor" }, head: { sha: "a7" }, labels: [], body: "Closes #1" }); + await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 7, title: "Linking PR", state: "open", user: { login: "contributor" }, head: { sha: "a7" }, labels: [], body: "Closes #1", created_at: "2026-07-03T10:00:00.000Z" }); let fetchCount = 0; vi.stubGlobal("fetch", async (input: RequestInfo | URL) => { fetchCount += 1; @@ -1856,7 +1856,7 @@ describe("queue processors", () => { // staleness-ordered sweep. expect(fetchCount).toBe(0); expect(sent).toEqual([ - { message: expect.objectContaining({ type: "agent-regate-pr", repoFullName: "owner/agent-repo", prNumber: 7, installationId: 9001 }) }, + { message: expect.objectContaining({ type: "agent-regate-pr", repoFullName: "owner/agent-repo", prNumber: 7, installationId: 9001, prCreatedAt: "2026-07-03T10:00:00.000Z" }) }, ]); const webhookRow = await env.DB.prepare("select status from webhook_events where delivery_id = ?").bind("issue-label-wake").first<{ status: string }>(); expect(webhookRow?.status).toBe("processed"); @@ -1898,7 +1898,7 @@ describe("queue processors", () => { await upsertInstallation(env, { action: "created", installation: { id: 9001, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } }); await upsertRepositoryFromGitHub(env, { name: "agent-repo", full_name: "owner/agent-repo", private: false, owner: { login: "owner" } }, 9001); await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo", autonomy: { merge: "auto" }, aiReviewMode: "off", gatePack: "oss-anti-slop", gateCheckMode: "enabled", checkRunMode: "off", commentMode: "off", publicSurface: "off" }); - await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 7, title: "Linking PR", state: "open", user: { login: "contributor" }, head: { sha: "a7" }, labels: [], body: "Closes #1" }); + await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 7, title: "Linking PR", state: "open", user: { login: "contributor" }, head: { sha: "a7" }, labels: [], body: "Closes #1", created_at: "2026-07-03T10:00:00.000Z" }); let checkRunsFetched = false; vi.stubGlobal("fetch", async (input: RequestInfo | URL) => { const url = input.toString(); @@ -1968,7 +1968,7 @@ describe("queue processors", () => { await upsertInstallation(env, { action: "created", installation: { id: 9001, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } }); await upsertRepositoryFromGitHub(env, { name: "agent-repo", full_name: "owner/agent-repo", private: false, owner: { login: "owner" } }, 9001); await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo", autonomy: { merge: "auto" }, aiReviewMode: "off", gatePack: "oss-anti-slop", gateCheckMode: "enabled", checkRunMode: "off", commentMode: "off", publicSurface: "off" }); - await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 7, title: "Linking PR", state: "open", user: { login: "contributor" }, head: { sha: "a7" }, labels: [], body: "Closes #1" }); + await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 7, title: "Linking PR", state: "open", user: { login: "contributor" }, head: { sha: "a7" }, labels: [], body: "Closes #1", created_at: "2026-07-03T10:00:00.000Z" }); // A CI completion for this exact PR claimed the CI-completion window moments earlier — a wholly separate // trigger from the issue-side label change below. await env.SELFHOST_TRANSIENT_CACHE?.set("ci-coalesce:owner/agent-repo#7", "1", 60); @@ -2020,7 +2020,7 @@ describe("queue processors", () => { await upsertInstallation(env, { action: "created", installation: { id: 9001, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } }); await upsertRepositoryFromGitHub(env, { name: "agent-repo", full_name: "owner/agent-repo", private: false, owner: { login: "owner" } }, 9001); await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo", autonomy: { merge: "auto" }, aiReviewMode: "off", gatePack: "oss-anti-slop", gateCheckMode: "enabled", checkRunMode: "off", commentMode: "off", publicSurface: "off" }); - await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 7, title: "Linking PR", state: "open", user: { login: "contributor" }, head: { sha: "a7" }, labels: [], body: "Closes #1" }); + await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 7, title: "Linking PR", state: "open", user: { login: "contributor" }, head: { sha: "a7" }, labels: [], body: "Closes #1", created_at: "2026-07-03T10:00:00.000Z" }); let fetchCallCount = 0; vi.stubGlobal("fetch", async (input: RequestInfo | URL) => { fetchCallCount += 1; @@ -2058,9 +2058,9 @@ describe("queue processors", () => { // Second signal within the window coalesces — no GitHub interaction and one trailing job for the latest state. expect(fetchCallCount).toBe(fetchCallCountAfterFirst); expect(sent).toEqual([ - { message: expect.objectContaining({ type: "agent-regate-pr", repoFullName: "owner/agent-repo", prNumber: 7, installationId: 9001 }) }, + { message: expect.objectContaining({ type: "agent-regate-pr", repoFullName: "owner/agent-repo", prNumber: 7, installationId: 9001, prCreatedAt: "2026-07-03T10:00:00.000Z" }) }, { - message: expect.objectContaining({ type: "agent-regate-pr", repoFullName: "owner/agent-repo", prNumber: 7, installationId: 9001 }), + message: expect.objectContaining({ type: "agent-regate-pr", repoFullName: "owner/agent-repo", prNumber: 7, installationId: 9001, prCreatedAt: "2026-07-03T10:00:00.000Z" }), options: { delaySeconds: 60 }, }, ]); @@ -2170,6 +2170,8 @@ describe("queue processors", () => { options: { delaySeconds: 60 }, }, ]); + const trailingReReview = sent[1]!; + expect((trailingReReview.message as Extract).prCreatedAt).toBeUndefined(); await processJob(env, event("issue-add-then-remove-3", "labeled")); // A THIRD coalesced event in the same window must not schedule a second, redundant trailing job. @@ -2829,7 +2831,7 @@ describe("queue processors", () => { expect(aiCalls).toBe(firstRunAiCalls * 2); // a real state change bypasses the cooldown immediately, regardless of age }); - it("#8: a rate-limit-deferred re-enqueue of a forced re-gate carries the force flag forward", async () => { + it("#8: a rate-limit-deferred re-enqueue of a forced re-gate carries the force flag and PR age forward", async () => { const env = createTestEnv({ GITHUB_APP_PRIVATE_KEY: await generatePrivateKeyPem() }); await seedRegateChurnRepo(env); await upsertPullRequestFromGitHub(env, "JSONbored/gittensory", { number: 68, title: "Clean PR", state: "open", user: { login: "contributor" }, head: { sha: "a68" }, labels: [], body: "Closes #1" }); @@ -2841,10 +2843,10 @@ describe("queue processors", () => { return send(message, options); }) as typeof env.JOBS.send; - await processJob(env, { type: "agent-regate-pr", deliveryId: "rate-limited-force", repoFullName: "JSONbored/gittensory", prNumber: 68, installationId: 123, force: true }); + await processJob(env, { type: "agent-regate-pr", deliveryId: "rate-limited-force", repoFullName: "JSONbored/gittensory", prNumber: 68, installationId: 123, force: true, prCreatedAt: "2026-07-03T10:00:00.000Z" }); rateLimitSpy.mockRestore(); - expect(enqueued).toMatchObject({ type: "agent-regate-pr", prNumber: 68, force: true }); + expect(enqueued).toMatchObject({ type: "agent-regate-pr", prNumber: 68, force: true, prCreatedAt: "2026-07-03T10:00:00.000Z" }); }); it("#8: a manual force re-gate bypasses the cache and cooldown, always paying for a fresh AI opinion", async () => { @@ -19320,28 +19322,35 @@ describe("backlog-convergence sweep (#selfhost-backlog-convergence)", () => { await upsertInstallation(env, { action: "created", installation: { id: 9505, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } }); await upsertRepositoryFromGitHub(env, { name: "agent-repo", full_name: "owner/agent-repo", private: false, owner: { login: "owner" } }, 9505); await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo", autonomy: { merge: "auto" } }); - // #7 never had its surface published; #8 was published at an OLDER head than its current one; #9 is fully converged. - await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 7, title: "Never published", state: "open", user: { login: "contributor" }, head: { sha: "a7" }, labels: [], body: "x" }); - await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 8, title: "Stale surface", state: "open", user: { login: "contributor" }, head: { sha: "b8" }, labels: [], body: "x" }); + // #7 never had its surface published; #8 was published at an OLDER head than its current one; #9 is fully converged; + // #10 is a legacy/sparse row with no GitHub created_at, and still needs a re-gate without PR-age metadata. + await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 7, title: "Never published", state: "open", user: { login: "contributor" }, head: { sha: "a7" }, labels: [], body: "x", created_at: "2026-07-03T10:00:00.000Z" }); + await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 8, title: "Stale surface", state: "open", user: { login: "contributor" }, head: { sha: "b8" }, labels: [], body: "x", created_at: "2026-07-03T11:00:00.000Z" }); await repositoriesModule.markPullRequestSurfacePublished(env, "owner/agent-repo", 8, "old-b8"); await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 9, title: "Converged", state: "open", user: { login: "contributor" }, head: { sha: "a9" }, labels: [], body: "x" }); await repositoriesModule.markPullRequestSurfacePublished(env, "owner/agent-repo", 9, "a9"); + await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 10, title: "Sparse legacy row", state: "open", user: { login: "contributor" }, head: { sha: "a10" }, labels: [], body: "x" }); await processJob(env, { type: "backlog-convergence-sweep", requestedBy: "schedule", repoFullName: "owner/agent-repo" }); const fanned = sent.filter((job): job is Extract => job.type === "agent-regate-pr"); - expect(fanned.map((job) => job.prNumber).sort()).toEqual([7, 8]); + expect(fanned.map((job) => job.prNumber).sort((a, b) => a - b)).toEqual([7, 8, 10]); for (const job of fanned) { expect(job.deliveryId).toBe(`backlog-convergence:owner/agent-repo#${job.prNumber}`); expect(job.installationId).toBe(9505); } + expect(Object.fromEntries(fanned.map((job) => [job.prNumber, job.prCreatedAt]))).toEqual({ + 7: "2026-07-03T10:00:00.000Z", + 8: "2026-07-03T11:00:00.000Z", + 10: undefined, + }); const audit = await env.DB.prepare("select outcome, detail, metadata_json from audit_events where event_type = ?") .bind("agent.sweep.backlog_convergence") .first<{ outcome: string; detail: string; metadata_json: string }>(); expect(audit?.outcome).toBe("completed"); const meta = JSON.parse(audit?.metadata_json ?? "{}"); - expect(meta).toMatchObject({ repoFullName: "owner/agent-repo", openCount: 3, examined: 2 }); - expect(meta.candidatePulls.sort()).toEqual([7, 8]); + expect(meta).toMatchObject({ repoFullName: "owner/agent-repo", openCount: 4, examined: 3 }); + expect(meta.candidatePulls.sort((a: number, b: number) => a - b)).toEqual([7, 8, 10]); }); }); diff --git a/test/unit/selfhost-pg-queue.test.ts b/test/unit/selfhost-pg-queue.test.ts index a5daeb1233..084c6967ae 100644 --- a/test/unit/selfhost-pg-queue.test.ts +++ b/test/unit/selfhost-pg-queue.test.ts @@ -290,6 +290,7 @@ describe("createPgQueue (durable #977)", () => { it("init() skips already-normalized priority and job-key rows", async () => { const priorityUpdateSql = "UPDATE _selfhost_jobs SET priority=$1"; const jobKeyUpdateSql = "UPDATE _selfhost_jobs SET job_key=$1"; + const claimSortUpdateSql = "UPDATE _selfhost_jobs SET claim_sort_key=$1"; const fn = vi.fn().mockImplementation(async (sql: unknown, params?: unknown[]) => { const q = String(sql); if (q.includes("SELECT id, payload, priority")) { @@ -324,6 +325,24 @@ describe("createPgQueue (durable #977)", () => { rowCount: 2, }; } + if (q.includes("SELECT id, payload, run_after, claim_sort_key")) { + return { + rows: [ + { + id: "sorted", + payload: JSON.stringify({ + type: "agent-regate-pr", + deliveryId: "backlog-convergence:owner/repo#7", + repoFullName: "owner/repo", + prNumber: 7, + }), + run_after: 999, + claim_sort_key: Date.parse("2000-01-01T00:00:00.000Z") + 7, + }, + ], + rowCount: 1, + }; + } if (q.includes("WHERE status='processing'")) return { rows: [], rowCount: 0 }; if (q.includes("WHERE status='pending' AND run_after<=$1")) return { rows: [], rowCount: 0 }; return { rows: [], rowCount: 0 }; @@ -340,6 +359,61 @@ describe("createPgQueue (durable #977)", () => { expect.stringContaining(jobKeyUpdateSql), expect.anything(), ); + expect(fn).not.toHaveBeenCalledWith( + expect.stringContaining(claimSortUpdateSql), + expect.anything(), + ); + }); + + it("init() backfills stale PR claim-sort keys while leaving already-normalized rows untouched", async () => { + const updates: unknown[][] = []; + const fn = vi.fn().mockImplementation(async (sql: unknown, params?: unknown[]) => { + const q = String(sql); + if (q.includes("SELECT id, payload, priority")) return { rows: [], rowCount: 0 }; + if (q.includes("SELECT id, payload, job_key") && q.includes("status IN")) return { rows: [], rowCount: 0 }; + if (q.includes("SELECT id, payload, run_after, claim_sort_key")) { + return { + rows: [ + { + id: "stale", + payload: JSON.stringify({ + type: "agent-regate-pr", + deliveryId: "backlog-convergence:owner/repo#12", + repoFullName: "owner/repo", + prNumber: 12, + prCreatedAt: "2026-07-03T12:00:00.000Z", + }), + run_after: "999", + claim_sort_key: 0, + }, + { + id: "fresh", + payload: JSON.stringify({ + type: "agent-regate-pr", + deliveryId: "backlog-convergence:owner/repo#13", + repoFullName: "owner/repo", + prNumber: 13, + }), + run_after: "999", + claim_sort_key: Date.parse("2000-01-01T00:00:00.000Z") + 13, + }, + ], + rowCount: 2, + }; + } + if (q.includes("UPDATE _selfhost_jobs SET claim_sort_key=$1")) { + updates.push(params ?? []); + return { rows: [], rowCount: 1 }; + } + if (q.includes("WHERE status='processing'")) return { rows: [], rowCount: 0 }; + if (q.includes("WHERE status='pending' AND run_after<=$1")) return { rows: [], rowCount: 0 }; + return { rows: [], rowCount: 0 }; + }); + const q = createPgQueue({ query: fn } as unknown as Pool, async () => undefined); + + await q.init(); + + expect(updates).toEqual([[Date.parse("2026-07-03T12:00:00.000Z"), "stale"]]); }); it("init() backfills job keys, recovers crashed jobs, and spreads due startup backlog", async () => { @@ -934,6 +1008,48 @@ describe("createPgQueue (durable #977)", () => { expect(claimSql[1]).toContain("priority < $2"); }); + it("stores a PR-created claim sort key for per-PR re-gate jobs", async () => { + const m = makePool(); + const q = createPgQueue(m.pool, async () => undefined); + await q.init(); + + await q.binding.send({ + type: "agent-regate-pr", + deliveryId: "backlog-convergence:jsonbored/gittensory#10", + repoFullName: "jsonbored/gittensory", + prNumber: 10, + installationId: 123, + prCreatedAt: "2026-07-03T10:00:00.000Z", + }); + + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("claim_sort_key) VALUES"), + expect.arrayContaining([Date.parse("2026-07-03T10:00:00.000Z")]), + ); + }); + + it("REGRESSION: claim SQL sorts by PR claim_sort_key and locks job_key siblings before claiming", async () => { + const m = makePool(); + const seen: string[] = []; + m.enqueueJob("1", regateJob(123, 10), 0, "agent-regate-pr:jsonbored/gittensory#10"); + const q = createPgQueue(m.pool, async (message) => void seen.push(typeOf(message)), { + concurrency: 1, + maxRetries: 1, + backoffMs: () => 0, + }); + await q.init(); + + await q.drain(); + + const claimSql = vi.mocked(m.pool.query).mock.calls + .map((call) => String(call[0])) + .find((sql) => sql.includes("FOR UPDATE SKIP LOCKED") && sql.includes("candidate.claim_sort_key")); + expect(claimSql).toContain("ORDER BY candidate.priority DESC, candidate.claim_sort_key, candidate.run_after, candidate.id"); + expect(claimSql).toContain("pg_try_advisory_xact_lock(hashtextextended(candidate.job_key, 0))"); + expect(claimSql).toContain("processing.status='processing' AND processing.job_key=candidate.job_key"); + expect(seen).toEqual(["agent-regate-pr"]); + }); + it("processes a background-lane job when foreground work is empty", async () => { const m = makePool(); m.enqueueResult({ rows: [], rowCount: 0 }); @@ -1146,6 +1262,7 @@ describe("createPgQueue (durable #977)", () => { m.fn.mockResolvedValueOnce({ rows: [], rowCount: 0 }); // fairness singleton INSERT m.fn.mockResolvedValueOnce({ rows: [], rowCount: 0 }); // priority backfill SELECT m.fn.mockResolvedValueOnce({ rows: [], rowCount: 0 }); // job-key backfill SELECT + m.fn.mockResolvedValueOnce({ rows: [], rowCount: 0 }); // claim-sort-key backfill SELECT m.fn.mockResolvedValueOnce({ rows: [], rowCount: 0 }); // maintenance-flags backfill SELECT m.fn.mockResolvedValueOnce({ rows: [{ id: "legacy", payload: JSON.stringify({ type: "agent-regate-pr", deliveryId: "backlog-convergence:owner/repo#1" }), foreground_lane: null }], diff --git a/test/unit/selfhost-queue-common.test.ts b/test/unit/selfhost-queue-common.test.ts index bbc39a73ac..e0ec30ce0e 100644 --- a/test/unit/selfhost-queue-common.test.ts +++ b/test/unit/selfhost-queue-common.test.ts @@ -22,6 +22,7 @@ import { jobCoalesceMergeKeyPrefix, jobCoalesceMergedPayload, jobCoalesceSupersededKeyPrefix, + jobClaimSortKey, jobPriority, matchesGitHubRateLimitAdmissionTarget, nonConsumingRetryDelayMs, @@ -876,6 +877,42 @@ describe("self-host queue common helpers", () => { ).toBeNull(); }); + it("orders per-PR re-gate jobs by GitHub PR creation time, with a deterministic legacy fallback", () => { + expect( + jobClaimSortKey( + payload({ + type: "agent-regate-pr", + repoFullName: "owner/repo", + prNumber: 42, + prCreatedAt: "2026-07-03T12:34:56.000Z", + }), + 999, + ), + ).toBe(Date.parse("2026-07-03T12:34:56.000Z")); + expect( + jobClaimSortKey( + payload({ type: "agent-regate-pr", repoFullName: "owner/repo", prNumber: 42 }), + 999, + ), + ).toBe(Date.parse("2000-01-01T00:00:00.000Z") + 42); + expect( + jobClaimSortKey( + payload({ + type: "agent-regate-pr", + repoFullName: "owner/repo", + prNumber: 43, + prCreatedAt: "not-a-date", + }), + 999, + ), + ).toBe(Date.parse("2000-01-01T00:00:00.000Z") + 43); + expect(jobClaimSortKey(payload({ type: "agent-regate-pr", repoFullName: "owner/repo", prNumber: "x" }), "567.8" as unknown as number)).toBe(567); + expect(jobClaimSortKey(payload({ type: "rag-index-repo" }), 1234.9)).toBe(1234); + expect(jobClaimSortKey(payload({ type: "rag-index-repo" }), -5)).toBe(0); + expect(jobClaimSortKey(payload({ type: "rag-index-repo" }), {} as unknown as number)).toBe(0); + expect(jobClaimSortKey("not-json", Number.NaN)).toBe(0); + }); + it("coalesces the event-driven jobs by their stable per-invocation id — and only true duplicates (#1942)", () => { // A DUPLICATE re-enqueue of the SAME job (same id — e.g. a webhook redelivery) coalesces. expect(jobCoalesceKey(payload({ type: "run-agent", requestedBy: "github_comment", runId: "run-abc123" }))).toBe("run-agent:run-abc123"); diff --git a/test/unit/selfhost-sqlite-queue.test.ts b/test/unit/selfhost-sqlite-queue.test.ts index 117bab759a..2aab26e202 100644 --- a/test/unit/selfhost-sqlite-queue.test.ts +++ b/test/unit/selfhost-sqlite-queue.test.ts @@ -1308,7 +1308,7 @@ describe("createSqliteQueue (durable #980)", () => { driver.query("PRAGMA index_info(_selfhost_jobs_claim)", []).rows.map( (row) => (row as { name: string }).name, ), - ).toEqual(["status", "run_after", "priority"]); + ).toEqual(["status", "priority", "claim_sort_key", "run_after"]); }); it("rebuilds an old claim index so priority participates in future claims", async () => { @@ -1332,7 +1332,7 @@ describe("createSqliteQueue (durable #980)", () => { driver.query("PRAGMA index_info(_selfhost_jobs_claim)", []).rows.map( (row) => (row as { name: string }).name, ), - ).toEqual(["status", "run_after", "priority"]); + ).toEqual(["status", "priority", "claim_sort_key", "run_after"]); }); it("claims webhook work before regate work, and regate work before earlier background jobs", async () => { @@ -1367,13 +1367,14 @@ describe("createSqliteQueue (durable #980)", () => { }); describe("claim-time backlog-vs-fresh-intake fairness (#selfhost-backlog-convergence)", () => { - const backlogJob = (repo: string, prNumber: number): JobMessage => + const backlogJob = (repo: string, prNumber: number, prCreatedAt?: string): JobMessage => ({ type: "agent-regate-pr", deliveryId: `backlog-convergence:${repo}#${prNumber}`, repoFullName: repo, prNumber, installationId: 1, + ...(prCreatedAt ? { prCreatedAt } : {}), }) as unknown as JobMessage; it("prefers 3 backlog-lane claims for every 1 fresh-intake claim (default ratio), deterministically", async () => { @@ -1454,6 +1455,100 @@ describe("createSqliteQueue (durable #980)", () => { ]); }); + it("REGRESSION: drains one repo's backlog-convergence PR jobs by original PR age, not enqueue timing", async () => { + const driver = makeDriver(); + const seen: string[] = []; + const q = createSqliteQueue(driver, async (m) => void seen.push((m as unknown as { deliveryId: string }).deliveryId), { concurrency: 1 }); + await q.binding.send(backlogJob("owner/repo", 12, "2026-07-03T12:00:00.000Z"), { delaySeconds: 60 }); + await q.binding.send(backlogJob("owner/repo", 10, "2026-07-03T10:00:00.000Z"), { delaySeconds: 60 }); + await q.binding.send(backlogJob("owner/repo", 11, "2026-07-03T11:00:00.000Z"), { delaySeconds: 60 }); + driver.query("UPDATE _selfhost_jobs SET run_after=0", []); + + await q.drain(); + + expect(seen).toEqual([ + "backlog-convergence:owner/repo#10", + "backlog-convergence:owner/repo#11", + "backlog-convergence:owner/repo#12", + ]); + }); + + it("backfills stale claim-sort keys and skips rows that already match the derived key", async () => { + const driver = makeDriver(); + createSqliteQueue(driver, async () => undefined, { concurrency: 1 }); // create the durable schema first + const stale = backlogJob("owner/repo", 12, "2026-07-03T12:00:00.000Z"); + const normalized = backlogJob("owner/repo", 13, "2026-07-03T13:00:00.000Z"); + driver.query( + "INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key, foreground_lane, claim_sort_key) VALUES (?, 'pending', 0, 0, ?, 9, ?, 'backlog', 0)", + [ + JSON.stringify(stale), + 1, + "agent-regate-pr:owner/repo#12", + ], + ); + driver.query( + "INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key, foreground_lane, claim_sort_key) VALUES (?, 'pending', 0, 0, ?, 9, ?, 'backlog', ?)", + [ + JSON.stringify(normalized), + 1, + "agent-regate-pr:owner/repo#13", + Date.parse("2026-07-03T13:00:00.000Z"), + ], + ); + + createSqliteQueue(driver, async () => undefined, { concurrency: 1 }); // startup backfill pass + + const rows = driver.query("SELECT job_key, claim_sort_key FROM _selfhost_jobs ORDER BY job_key", []).rows as Array<{ job_key: string; claim_sort_key: number }>; + expect(rows).toEqual([ + { job_key: "agent-regate-pr:owner/repo#12", claim_sort_key: Date.parse("2026-07-03T12:00:00.000Z") }, + { job_key: "agent-regate-pr:owner/repo#13", claim_sort_key: Date.parse("2026-07-03T13:00:00.000Z") }, + ]); + }); + + it("REGRESSION: does not claim a duplicate PR re-gate while the same job_key is already processing", async () => { + const driver = makeDriver(); + const seen: string[] = []; + const q = createSqliteQueue(driver, async (m) => void seen.push((m as unknown as { deliveryId: string }).deliveryId), { concurrency: 1 }); + const now = Date.now(); + driver.query( + "INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key, foreground_lane, claim_sort_key) VALUES (?, 'processing', 0, ?, ?, 9, ?, 'backlog', ?)", + [ + JSON.stringify(backlogJob("owner/repo", 10, "2026-07-03T10:00:00.000Z")), + now, + now, + "agent-regate-pr:owner/repo#10", + Date.parse("2026-07-03T10:00:00.000Z"), + ], + ); + driver.query( + "INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key, foreground_lane, claim_sort_key) VALUES (?, 'pending', 0, 0, ?, 9, ?, 'backlog', ?)", + [ + JSON.stringify(backlogJob("owner/repo", 10, "2026-07-03T10:00:00.000Z")), + now, + "agent-regate-pr:owner/repo#10", + Date.parse("2026-07-03T10:00:00.000Z"), + ], + ); + driver.query( + "INSERT INTO _selfhost_jobs (payload, status, attempts, run_after, created_at, priority, job_key, foreground_lane, claim_sort_key) VALUES (?, 'pending', 0, 0, ?, 9, ?, 'backlog', ?)", + [ + JSON.stringify(backlogJob("owner/repo", 11, "2026-07-03T11:00:00.000Z")), + now, + "agent-regate-pr:owner/repo#11", + Date.parse("2026-07-03T11:00:00.000Z"), + ], + ); + + await q.drain(); + + expect(seen).toEqual(["backlog-convergence:owner/repo#11"]); + const duplicate = driver.query( + "SELECT status FROM _selfhost_jobs WHERE job_key=? ORDER BY id DESC LIMIT 1", + ["agent-regate-pr:owner/repo#10"], + ).rows[0] as { status: string }; + expect(duplicate.status).toBe("pending"); + }); + it("defaults the claim sequence to 0 when the fairness singleton row is missing (defensive, never crashes)", async () => { const driver = makeDriver(); const seen: string[] = []; @@ -2665,7 +2760,7 @@ describe("createSqliteQueue (durable #980)", () => { // Only claimNextWhere's SELECT starts with this exact column list — spreadDueJobsOnStartup (which already // ran during construction above) selects a different column set, so this doesn't clobber setup. vi.spyOn(driver, "query").mockImplementation((sql: string, params: unknown[]) => { - if (sql.includes("SELECT id, payload, attempts, job_key, priority")) throw new Error("database is locked"); + if (sql.includes("SELECT candidate.id, candidate.payload, candidate.attempts")) throw new Error("database is locked"); return realQuery(sql, params); }); const errorSpy = vi.spyOn(console, "error").mockImplementation(() => undefined);