diff --git a/src/selfhost/pg-queue.ts b/src/selfhost/pg-queue.ts index f2461bb725..bf63dc614d 100644 --- a/src/selfhost/pg-queue.ts +++ b/src/selfhost/pg-queue.ts @@ -900,21 +900,25 @@ export function createPgQueue( // pending maintenance need must coalesce into the existing row without resetting how long that need has // genuinely been outstanding -- otherwise a re-enqueue cadence shorter than the trickle's maxDeferAgeMs // (4h default) can keep re-arming the clock forever, and sustained pressure defers the job indefinitely. - await pool.query( + // Guarded by status='pending' so a concurrent claim between the SELECT and here loses cleanly (rowCount + // 0) instead of silently overwriting a processing row -- mirrors the merge path above (gate finding). + const superseded = await pool.query( `UPDATE ${TABLE} 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`, + WHERE id=$6 AND status='pending'`, [payload, runAfter, priority, key, lane, existing.id, claimSortKey], ); - await pool.query( - `DELETE FROM ${TABLE} - WHERE status='pending' AND id<>$1 AND job_key IS NOT NULL AND left(job_key, $2)=$3`, - [existing.id, supersededKeyPrefix.length, supersededKeyPrefix], - ); - await recordQueueMetric("gittensory_jobs_coalesced_total"); - kickOne(); - return; + if (superseded.rowCount) { + await pool.query( + `DELETE FROM ${TABLE} + WHERE status='pending' AND id<>$1 AND job_key IS NOT NULL AND left(job_key, $2)=$3`, + [existing.id, supersededKeyPrefix.length, supersededKeyPrefix], + ); + await recordQueueMetric("gittensory_jobs_coalesced_total"); + kickOne(); + return; + } } } if (key) { @@ -927,16 +931,18 @@ export function createPgQueue( if (existing) { // See the supersededKeyPrefix branch above: created_at is preserved across a coalesced re-enqueue so the // maintenance trickle clock reflects genuine wait time, not the most recent re-request. - await pool.query( + const coalesced = await pool.query( `UPDATE ${TABLE} 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`, + WHERE id=$5 AND status='pending'`, [payload, runAfter, priority, lane, existing.id, claimSortKey], ); - await recordQueueMetric("gittensory_jobs_coalesced_total"); - kickOne(); - return; + if (coalesced.rowCount) { + await recordQueueMetric("gittensory_jobs_coalesced_total"); + kickOne(); + return; + } } } await pool.query( diff --git a/test/unit/selfhost-pg-queue.test.ts b/test/unit/selfhost-pg-queue.test.ts index 12f9753bca..7e75eab821 100644 --- a/test/unit/selfhost-pg-queue.test.ts +++ b/test/unit/selfhost-pg-queue.test.ts @@ -65,6 +65,8 @@ interface MockPool { /** Pre-load a job to be returned by the next RETURNING claim query. */ enqueueJob(id: string, payload: object, attempts?: number, jobKey?: string | null): void; setDeferUpdateRowCount(rowCount: number): void; + /** Default rowCount for coalesce/supersede/merge payload UPDATEs (defaults to 1 = won the race). */ + setCoalesceUpdateRowCount(rowCount: number): void; /** Queues per-call rowCounts for the "AND status='dead'" revive UPDATE, one entry consumed per call in order * (default 1 when the queue is empty) — lets a test simulate an overlapping reviver already winning the race * on a specific row (rowCount 0) while another succeeds (rowCount 1). */ @@ -112,6 +114,7 @@ interface MockPool { function makePool(): MockPool { const results: Partial[] = []; let deferUpdateRowCount = 1; + let coalesceUpdateRowCount = 1; const reviveUpdateRowCounts: number[] = []; let rateLimitRows: Array<{ admission_key?: string | null; repo_full_name?: string | null; remaining: number | string | null; reset_at: string | null; observed_at?: string | null }> = []; let pressureLive: { cnt: number; oldest: number | null; runnableCnt?: number; oldestRunnable?: number | null } = { @@ -194,6 +197,9 @@ function makePool(): MockPool { if (q.includes("SET status='pending', run_after=GREATEST")) { return { rows: [], rowCount: deferUpdateRowCount }; } + if (q.includes("SET payload=$1, run_after=GREATEST") && q.includes("WHERE id=")) { + return { rows: [], rowCount: coalesceUpdateRowCount }; + } if (q.includes("SET status='pending', run_after=$1, last_error=NULL")) { const rowCount = reviveUpdateRowCounts.length > 0 ? (reviveUpdateRowCounts.shift() ?? 1) : 1; return { rows: [], rowCount }; @@ -230,6 +236,9 @@ function makePool(): MockPool { setDeferUpdateRowCount(rowCount) { deferUpdateRowCount = rowCount; }, + setCoalesceUpdateRowCount(rowCount) { + coalesceUpdateRowCount = rowCount; + }, setReviveUpdateRowCounts(rowCounts) { reviveUpdateRowCounts.length = 0; reviveUpdateRowCounts.push(...rowCounts); @@ -615,14 +624,8 @@ describe("createPgQueue (durable #977)", () => { ["rag-index-repo:jsonbored/gittensory:".length, "rag-index-repo:jsonbored/gittensory:"], ); expect(m.pool.query).toHaveBeenCalledWith( - expect.stringContaining("SET payload=$1, run_after=GREATEST"), - expect.arrayContaining([ - expect.stringContaining('"requestedBy":"schedule"'), - expect.any(Number), - 0, - "rag-index-repo:jsonbored/gittensory:full", - "existing-incremental", - ]), + expect.stringContaining("WHERE id=$6 AND status='pending'"), + expect.anything(), ); expect(m.pool.query).toHaveBeenCalledWith( expect.stringContaining("DELETE FROM _selfhost_jobs"), @@ -698,6 +701,49 @@ describe("createPgQueue (durable #977)", () => { ); }); + it("REGRESSION (gate finding): a lost supersede race falls through to insert instead of deleting sibling pending rows", async () => { + const m = makePool(); + const q = createPgQueue(m.pool, async () => undefined); + await q.init(); + m.fn.mockResolvedValueOnce({ rows: [], rowCount: 0 }); // absorbedByKey check + m.fn.mockResolvedValueOnce({ rows: [{ id: "existing-incremental" }], rowCount: 1 }); // supersede lookup + m.fn.mockResolvedValueOnce({ rows: [], rowCount: 0 }); // guarded supersede UPDATE loses the race + + await q.binding.send({ + type: "rag-index-repo", + requestedBy: "schedule", + repoFullName: "JSONbored/gittensory", + }); + + expect(m.pool.query).not.toHaveBeenCalledWith( + expect.stringContaining("DELETE FROM _selfhost_jobs"), + expect.anything(), + ); + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("INSERT INTO"), + expect.arrayContaining([expect.stringContaining('"requestedBy":"schedule"')]), + ); + }); + + it("REGRESSION (gate finding): a lost simple-coalesce race falls through to insert instead of silently overwriting", async () => { + const m = makePool(); + const q = createPgQueue(m.pool, async () => undefined); + await q.init(); + m.fn.mockResolvedValueOnce({ rows: [{ id: "existing" }], rowCount: 1 }); // keyed lookup + m.setCoalesceUpdateRowCount(0); + + await q.binding.send(ciWebhook("ci-2", "check_run"), { delaySeconds: 1 }); + + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("WHERE id=$5 AND status='pending'"), + expect.anything(), + ); + expect(m.pool.query).toHaveBeenCalledWith( + expect.stringContaining("INSERT INTO"), + expect.arrayContaining([expect.stringContaining('"deliveryId":"ci-2"')]), + ); + }); + it("does not merge an incremental into an already-pending FULL job for that repo", async () => { const m = makePool(); const q = createPgQueue(m.pool, async () => undefined);