From 706d3cea3d77f7ebb7aa8bca8ce14173b14c5776 Mon Sep 17 00:00:00 2001 From: zsimovanforgeops Date: Mon, 20 Jul 2026 18:38:03 -0500 Subject: [PATCH] Scope maxWaiting backpressure by data.sourceId (#2970) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The submission-time backpressure cap counts all waiting (name, queue) rows regardless of which source a job targets. On a multi-source brain this makes per-source submissions with maxWaiting: 1 mutually exclusive: while one source's sync sits waiting, every other source's freshness sync coalesces into that row and never runs. The dispatch log shows the starved source 'dispatched' each interval (queue.add returns the other source's waiting row), so the starvation is invisible unless you notice sources.last_sync_at falling behind — we found a secondary source 29 hours stale on a 5-minute freshness interval. Fix: when the submitted data carries a string sourceId, key the advisory lock, the waiting count, and the coalesce target on it. Submissions without sourceId keep the existing single-scope behavior, so single-source brains and non-sync jobs are unchanged. The new test asserts same-source submissions still coalesce while a different source gets its own row and its own cap; it fails on master. Co-authored-by: Forge (Ron) --- src/core/minions/queue.ts | 22 +++++++++++++++++----- test/minions.test.ts | 14 ++++++++++++++ 2 files changed, 31 insertions(+), 5 deletions(-) diff --git a/src/core/minions/queue.ts b/src/core/minions/queue.ts index 83ae17b67..ccf71cd96 100644 --- a/src/core/minions/queue.ts +++ b/src/core/minions/queue.ts @@ -163,24 +163,36 @@ export class MinionQueue { if (opts?.maxWaiting !== undefined) { const maxWaiting = Math.max(1, Math.floor(opts.maxWaiting)); const backpressureQueue = opts?.queue ?? 'default'; + // Multi-source scope: jobs of the same (name, queue) but different + // data.sourceId are independent workstreams (per-source sync/cycle). + // Counting them together made a waiting default-source sync swallow + // every other source's freshness sync — a secondary source sat 29h stale + // while dispatch logs showed its syncs "dispatched" (coalesced into + // the default row). Key the lock and the count on sourceId when the + // submission carries one; NULL keeps legacy single-scope behavior. + const bpSourceId = typeof (data as Record | undefined)?.sourceId === 'string' + ? (data as Record).sourceId as string + : null; await tx.executeRaw( - `SELECT pg_advisory_xact_lock(hashtext('minion_maxwaiting:' || $1 || ':' || $2))`, - [jobName, backpressureQueue] + `SELECT pg_advisory_xact_lock(hashtext('minion_maxwaiting:' || $1 || ':' || $2 || ':' || coalesce($3, '')))`, + [jobName, backpressureQueue, bpSourceId] ); const waitingCountRows = await tx.executeRaw<{ count: string }>( `SELECT count(*)::text AS count FROM minion_jobs - WHERE name = $1 AND queue = $2 AND status = 'waiting'`, - [jobName, backpressureQueue] + WHERE name = $1 AND queue = $2 AND status = 'waiting' + AND ($3::text IS NULL OR data->>'sourceId' IS NOT DISTINCT FROM $3)`, + [jobName, backpressureQueue, bpSourceId] ); const waitingCount = parseInt(waitingCountRows[0]?.count ?? '0', 10); if (waitingCount >= maxWaiting) { const existingWaiting = await tx.executeRaw>( `SELECT * FROM minion_jobs WHERE name = $1 AND queue = $2 AND status = 'waiting' + AND ($3::text IS NULL OR data->>'sourceId' IS NOT DISTINCT FROM $3) ORDER BY created_at DESC, id DESC LIMIT 1`, - [jobName, backpressureQueue] + [jobName, backpressureQueue, bpSourceId] ); if (existingWaiting.length > 0) { const coalesced = rowToMinionJob(existingWaiting[0]); diff --git a/test/minions.test.ts b/test/minions.test.ts index b1d7d6da9..3f6bf3c07 100644 --- a/test/minions.test.ts +++ b/test/minions.test.ts @@ -1967,6 +1967,20 @@ describe('MinionQueue: v0.19.1 maxWaiting — cap correctness + race (D2/H2)', ( expect(b.queue).toBe('shell'); }); + test('cross-source isolation — waiting sync for source A does NOT swallow source B (multi-source regression)', async () => { + // Regression: the cap counted all waiting (name, queue) rows regardless + // of data.sourceId, so a waiting default-source sync coalesced away every + // other source's freshness sync — a secondary source sat 29h stale while + // dispatch logs showed its syncs "dispatched". + const a = await queue.add('srcsync', { sourceId: 'default' }, { maxWaiting: 1 }); + const a2 = await queue.add('srcsync', { sourceId: 'default' }, { maxWaiting: 1 }); + expect(a2.id).toBe(a.id); // same source still coalesces + const b = await queue.add('srcsync', { sourceId: 'projects' }, { maxWaiting: 1 }); + expect(b.id).not.toBe(a.id); // different source MUST get its own row + const b2 = await queue.add('srcsync', { sourceId: 'projects' }, { maxWaiting: 1 }); + expect(b2.id).toBe(b.id); // and its own cap + }); + test('unset maxWaiting — normal submit path, no coalesce, no cap', async () => { const a = await queue.add('uncapped', {}); const b = await queue.add('uncapped', {});