Fix duplicate ingestion jobs starving newly enabled sources

.github/workflows/ci.yml

@@ -7,6 +7,15 @@ jobs:
   test:
     runs-on: ubuntu-latest
     services:
+      redis:
+        image: redis:7-alpine
+        ports:
+          - 6379:6379
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 10
       postgres:
         image: postgres:17-alpine
         env:
@@ -30,4 +39,5 @@ jobs:
         env:
           DATABASE_URL: postgres://unused:unused@localhost:5432/unused
           MIGRATION_TEST_DATABASE_URL: postgres://postgres@localhost:5432/postgres
+          REDIS_TEST_URL: redis://localhost:6379
           SITE_URL: http://localhost:3000

apps/worker/src/repair-queue-cli.js

@@ -0,0 +1,37 @@
+import { closeQueues, queues } from '@nichedb/queue';
+import { compactPendingRuns } from '../../../packages/queue/src/repair-ingest-queue.js';
+
+// Read-only by default. --apply saves a backup before consolidating requests;
+// the backup contains queue metadata only, never ingested records or secrets.
+const apply = process.argv.includes('--apply');
+try {
+  const jobs = await queues.run.getJobs(
+    ['waiting', 'delayed', 'prioritized', 'paused'],
+    0,
+    -1,
+    true,
+  );
+  const plan = await compactPendingRuns(queues.run, jobs);
+  console.log(JSON.stringify({ apply, ...plan }));
+  if (apply) {
+    const backup = `/tmp/nichedb-ingest-queue-${Date.now()}.json`;
+    await Bun.write(
+      backup,
+      JSON.stringify(
+        jobs.filter(Boolean).map((j) => ({
+          id: j.id,
+          name: j.name,
+          data: j.data,
+          opts: j.opts,
+          timestamp: j.timestamp,
+        })),
+      ),
+    );
+    console.log(`Backup: ${backup}`);
+    console.log(
+      JSON.stringify(await compactPendingRuns(queues.run, jobs, { apply, log: console.log })),
+    );
+  }
+} finally {
+  await closeQueues();
+}

docs/ingest-queue.md

@@ -0,0 +1,44 @@
+# Ingestion queue recovery
+
+The scheduler formerly assigned a different job ID every minute while a source
+waited to run. Waiting did not advance its database schedule, so a slow queue
+accumulated repeated requests for the same sources. On September 14, 2026 the
+production queue held over 74,000 waiting jobs; 38 new California police sources
+had never reached a worker.
+
+Runs now use BullMQ simple deduplication with a source ID key. The key remains
+through waiting and execution, then releases on success or failure. A separate
+unique job ID retains history without suppressing the next scheduled run.
+Scheduled and manually requested jobs use the same deduplication key.
+
+The scheduler pages past sources already represented in Redis. It admits at
+most 50 new runs per tick rather than repeatedly selecting only the first 50
+overdue sources. Paging is ordered by next-run time and source ID. This bounds
+duplicate work; execution still depends on worker capacity and upstream speed.
+
+For queues created before this change, inspect the backlog with:
+
+```sh
+bun apps/worker/src/repair-queue-cli.js
+```
+
+Apply only after the deduplicating scheduler is deployed:
+
+```sh
+bun apps/worker/src/repair-queue-cli.js --apply
+```
+
+The repair writes a timestamped JSON backup of pending job metadata to `/tmp`,
+then creates or locates one deduplicated replacement per source **before**
+removing old requests. It preserves manual force flags. Active jobs, other job
+types, source configuration, ingested records, and completed history remain
+untouched. Removal uses BullMQ's API, which rejects jobs that become active
+during the repair. Run the inspection again afterward to verify the backlog.
+The backup is local to the container and should be retained externally if
+long-term audit storage is needed.
+
+Tests exercise real Redis: concurrent requests while waiting and active,
+rescheduling after success and failure, paging past queued sources, and safe
+consolidation of legacy jobs. CI provides an isolated Redis service; locally set
+`REDIS_TEST_URL` to a test Redis instance. Test queues use unique namespaces and
+remove only their own keys.

packages/db/src/queries.js

@@ -391,7 +391,12 @@ export async function requestRun(id) {
  * The sources whose turn it is. Ordered by how overdue, so a starved one is
  * served first after an outage.
  */
-export async function dueSources({ limit = 20, force = false, runningMinutes = 30 } = {}) {
+export async function dueSources({
+  limit = 20,
+  offset = 0,
+  force = false,
+  runningMinutes = 30,
+} = {}) {
   /*
    * A source whose run is still in flight is not due, whatever its clock says.
    *
@@ -413,7 +418,7 @@ export async function dueSources({ limit = 20, force = false, runningMinutes = 3
         where r.source_id = s.id and r.status = 'running'
           and r.started_at > now() - (${`${runningMinutes} minutes`})::interval
       )
-    order by next_run_at limit ${limit}
+    order by next_run_at, s.id limit ${limit} offset ${offset}
   `;
 }
 

packages/queue/src/index.js

@@ -1,6 +1,7 @@
 import { config } from '@nichedb/config';
 import { Queue } from 'bullmq';
 import IORedis from 'ioredis';
+import { enqueueSourceRun } from './ingest-scheduling.js';
 
 /**
  * BullMQ needs `maxRetriesPerRequest: null` on the connection it blocks on.
@@ -89,11 +90,7 @@ export async function installSchedules({ log = console.log } = {}) {
 
 /** Ask for one source to run now, from a button or the API. */
 export async function enqueueRun(sourceId, { force = false } = {}) {
-  return queues.run.add(
-    'run',
-    { sourceId, force },
-    { jobId: `run-${sourceId}-${minuteStamp()}`, attempts: 1 },
-  );
+  return enqueueSourceRun(queues.run, sourceId, { force });
 }
 
 export async function closeQueues() {

packages/queue/src/ingest-scheduling.js

@@ -0,0 +1,52 @@
+import { randomUUID } from 'node:crypto';
+
+export const sourceDeduplicationId = (sourceId) => `source-${sourceId}`;
+
+/** One waiting or active run per source, across ticks and manual requests.
+ * A fresh job ID lets a completed source run again even while its history is
+ * retained. BullMQ releases the deduplication key on completion or failure.
+ */
+export function enqueueSourceRun(queue, sourceId, { force = false } = {}) {
+  return queue.add(
+    'run',
+    { sourceId, force },
+    {
+      jobId: `run-${sourceId}-${randomUUID()}`,
+      deduplication: { id: sourceDeduplicationId(sourceId) },
+      attempts: 1,
+    },
+  );
+}
+
+/** Pending sources stay due in SQL until execution starts. Look past them
+ * instead of spending every tick on the same first page of queued sources.
+ */
+export async function scheduleDueRuns({
+  queue,
+  readDue,
+  force = false,
+  runningMinutes,
+  limit = 50,
+  pageSize = 100,
+  deadline = Date.now() + 15000,
+}) {
+  let offset = 0;
+  let added = 0;
+  let scanned = 0;
+  while (added < limit && Date.now() < deadline) {
+    const due = await readDue({ limit: pageSize, offset, force, runningMinutes });
+    if (!due.length) break;
+    const pending = await Promise.all(
+      due.map((s) => queue.getDeduplicationJobId(sourceDeduplicationId(s.id))),
+    );
+    for (let i = 0; i < due.length && added < limit; i++) {
+      scanned++;
+      if (pending[i]) continue;
+      await enqueueSourceRun(queue, due[i].id, { force });
+      added++;
+    }
+    offset += due.length;
+    if (due.length < pageSize) break;
+  }
+  return { added, scanned };
+}

packages/queue/src/ingest-scheduling.test.js

@@ -0,0 +1,83 @@
+import { afterEach, beforeEach, describe, expect, test } from 'bun:test';
+import { randomUUID } from 'node:crypto';
+import { Queue, Worker } from 'bullmq';
+import IORedis from 'ioredis';
+import { enqueueSourceRun, scheduleDueRuns, sourceDeduplicationId } from './ingest-scheduling.js';
+import { compactPendingRuns } from './repair-ingest-queue.js';
+
+// CI supplies a Redis service. Every test owns and deletes a separate namespace.
+const redisUrl = process.env.REDIS_TEST_URL;
+describe.skipIf(!redisUrl)('ingestion scheduling with Redis', () => {
+  let queue, connection, worker;
+  beforeEach(async () => {
+    connection = new IORedis(redisUrl, { maxRetriesPerRequest: null });
+    queue = new Queue(`test-ingest-${randomUUID()}`, { connection });
+    await queue.waitUntilReady();
+  });
+  afterEach(async () => {
+    if (worker) await worker.close();
+    worker = null;
+    await queue.obliterate({ force: true });
+    await queue.close();
+    await connection.quit();
+  });
+  test('concurrent requests share one waiting/active run; retained completion allows a fresh run', async () => {
+    const requests = await Promise.all(
+      Array.from({ length: 30 }, () => enqueueSourceRun(queue, 42)),
+    );
+    expect(new Set(requests.map((j) => j.id)).size).toBe(1);
+    expect(await queue.getWaitingCount()).toBe(1);
+    worker = new Worker(queue.name, null, { connection, autorun: false });
+    const active = await worker.getNextJob('test-token', { block: false });
+    expect((await enqueueSourceRun(queue, 42, { force: true })).id).toBe(active.id);
+    expect(await queue.getWaitingCount()).toBe(0);
+    await active.moveToCompleted('ok', 'test-token', false);
+    const next = await enqueueSourceRun(queue, 42);
+    expect(next.id).not.toBe(active.id);
+    expect(await queue.getCompletedCount()).toBe(1);
+    expect(await queue.getWaitingCount()).toBe(1);
+  });
+  test('failure releases the source for a later retry', async () => {
+    await enqueueSourceRun(queue, 77);
+    worker = new Worker(queue.name, null, { connection, autorun: false });
+    const active = await worker.getNextJob('test-token', { block: false });
+    await active.moveToFailed(new Error('upstream unavailable'), 'test-token', false);
+    expect(await queue.getDeduplicationJobId(sourceDeduplicationId(77))).toBeNull();
+    expect((await enqueueSourceRun(queue, 77)).id).not.toBe(active.id);
+    expect(await queue.getFailedCount()).toBe(1);
+  });
+  test('a full page of queued sources cannot starve new sources on later pages', async () => {
+    const sources = Array.from({ length: 123 }, (_, i) => ({ id: i + 1 }));
+    await Promise.all(sources.slice(0, 120).map((s) => enqueueSourceRun(queue, s.id)));
+    const readDue = async ({ limit, offset }) => sources.slice(offset, offset + limit);
+    expect(await scheduleDueRuns({ queue, readDue, limit: 2, pageSize: 50 })).toEqual({
+      added: 2,
+      scanned: 122,
+    });
+    expect(await queue.getWaitingCount()).toBe(122);
+    expect((await scheduleDueRuns({ queue, readDue, limit: 2, pageSize: 50 })).added).toBe(1);
+    expect((await scheduleDueRuns({ queue, readDue, limit: 2, pageSize: 50 })).added).toBe(0);
+    expect(await queue.getWaitingCount()).toBe(123);
+  });
+  test('repair keeps one request per source, preserves force, and leaves active/foreign jobs alone', async () => {
+    await queue.add('run', { sourceId: 1 }, { jobId: 'old-one' });
+    await queue.add('run', { sourceId: 1, force: true }, { jobId: 'old-two' });
+    await queue.add('run', { sourceId: 2 }, { jobId: 'old-three' });
+    await queue.add('unrelated', { sourceId: 9 }, { jobId: 'foreign' });
+    const jobs = await queue.getJobs(['waiting'], 0, -1, true);
+    expect((await compactPendingRuns(queue, jobs)).duplicates).toBe(1);
+    expect(await queue.getWaitingCount()).toBe(4);
+    worker = new Worker(queue.name, null, { connection, autorun: false });
+    const active = await worker.getNextJob('test-token', { block: false });
+    const result = await compactPendingRuns(queue, jobs, { apply: true });
+    expect(result.skipped).toBe(1);
+    expect(await active.getState()).toBe('active');
+    expect(await queue.getJob('foreign')).toBeTruthy();
+    const id = await queue.getDeduplicationJobId(sourceDeduplicationId(1));
+    expect((await queue.getJob(id)).data.force).toBe(true);
+    await active.moveToCompleted('ok', 'test-token', false);
+    expect(
+      (await compactPendingRuns(queue, await queue.getJobs(['waiting']), { apply: true })).removed,
+    ).toBe(0);
+  });
+});

packages/queue/src/repair-ingest-queue.js

@@ -0,0 +1,56 @@
+import { enqueueSourceRun } from './ingest-scheduling.js';
+
+/** Consolidate a snapshot of pending runs through BullMQ's own APIs.
+ * Create/locate the replacement before removing old requests. Active jobs and
+ * jobs with another purpose are never removed. No source or item data changes.
+ */
+export async function compactPendingRuns(queue, jobs, { apply = false, log = () => {} } = {}) {
+  const groups = new Map();
+  for (const job of jobs) {
+    if (job?.name !== 'run' || !/^\d+$/.test(String(job.data?.sourceId))) continue;
+    const key = String(job.data.sourceId);
+    if (!groups.has(key)) groups.set(key, []);
+    groups.get(key).push(job);
+  }
+  const result = { jobs: jobs.length, sources: groups.size, duplicates: 0, removed: 0, skipped: 0 };
+  for (const group of groups.values()) result.duplicates += group.length - 1;
+  if (!apply) return result;
+  for (const [sourceId, group] of groups) {
+    let stillPending = false;
+    for (const job of group) {
+      if (['waiting', 'delayed', 'prioritized', 'paused'].includes(await job.getState())) {
+        stillPending = true;
+        break;
+      }
+    }
+    if (!stillPending) {
+      result.skipped += group.length;
+      continue;
+    }
+    const force = group.some((job) => job.data.force === true);
+    const replacement = await enqueueSourceRun(queue, sourceId, { force });
+    const kept = await queue.getJob(replacement.id);
+    if (!kept)
+      throw new Error(`Replacement disappeared for source ${sourceId}; stopped before deletion`);
+    if (force && !kept.data.force) await kept.updateData({ ...kept.data, force: true });
+    for (const job of group) {
+      if (job.id === replacement.id) continue;
+      if (!['waiting', 'delayed', 'prioritized', 'paused'].includes(await job.getState())) {
+        result.skipped++;
+        continue;
+      }
+      try {
+        // remove() also refuses a job that became active after getState().
+        await job.remove();
+        result.removed++;
+      } catch (error) {
+        if ((await job.getState()) === 'active') result.skipped++;
+        else throw error;
+      }
+    }
+    log(
+      `source ${sourceId}: kept ${replacement.id}; removed ${result.removed} old requests so far`,
+    );
+  }
+  return result;
+}

packages/queue/src/workers.js

@@ -5,7 +5,8 @@ import * as q from '@nichedb/db/queries';
 import { sendEmail, sendPush } from '@nichedb/notify';
 import { buildEvent, sendWebhook } from '@profullstack/autoblog';
 import { Worker } from 'bullmq';
-import { connection, minuteStamp, QUEUES, queues } from './index.js';
+import { connection, QUEUES, queues } from './index.js';
+import { scheduleDueRuns } from './ingest-scheduling.js';
 
 const log = (...a) => console.log('[worker]', ...a);
 
@@ -16,25 +17,22 @@ const longestRunMs = () =>
   Math.max(config.ingest.runDeadlineMs, ...ADAPTERS.map((a) => a.budgetMs ?? 0));
 
 /**
- * Which sources are due? One `run` job each, with a per-minute id so a tick
- * that fires twice cannot double-run a source. startRun pushes next_run_at
- * forward as the job begins, so a long run is not re-enqueued by the next tick.
+ * Which sources are due? Deduplicate through the entire wait and execution,
+ * and page past queued sources so new sources can enter a busy queue.
  */
 async function runTick(job) {
   // One window for both: a run younger than this is in flight and must not be
   // enqueued again; one older than it was just marked abandoned and may be.
   const runningMinutes = Math.ceil(longestRunMs() / 60_000) + 10;
   await q.reapStaleRuns({ minutes: runningMinutes });
-  const due = await q.dueSources({ limit: 50, force: Boolean(job.data?.force), runningMinutes });
-  for (const s of due) {
-    await queues.run.add(
-      'run',
-      { sourceId: s.id },
-      { jobId: `run-${s.id}-${minuteStamp()}`, attempts: 1 },
-    );
-  }
-  if (due.length) log(`tick: ${due.length} source(s) due`);
-  return { due: due.length };
+  const result = await scheduleDueRuns({
+    queue: queues.run,
+    readDue: q.dueSources,
+    force: Boolean(job.data?.force),
+    runningMinutes,
+  });
+  if (result.added) log(`tick: queued ${result.added} source(s); checked ${result.scanned}`);
+  return { due: result.added };
 }
 
 /* -------------------------------------------------------------------- scan -- */