Scope maxWaiting backpressure by data.sourceId (#2970)

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) <forge@zsimovan.dev>

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<string, unknown> | undefined)?.sourceId === 'string'
+          ? (data as Record<string, unknown>).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<Record<string, unknown>>(
             `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]);

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', {});