From 0f98a42ad7f52e6a422ab3156ba94e6f7bd95e93 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Mon, 14 Sep 2026 17:50:12 +0000 Subject: [PATCH] fix: deduplicate queued ingestion runs and admit overdue sources fairly --- .github/workflows/ci.yml | 10 +++ apps/worker/src/repair-queue-cli.js | 37 +++++++++ docs/ingest-queue.md | 44 +++++++++++ packages/db/src/queries.js | 9 ++- packages/queue/src/index.js | 7 +- packages/queue/src/ingest-scheduling.js | 52 ++++++++++++ packages/queue/src/ingest-scheduling.test.js | 83 ++++++++++++++++++++ packages/queue/src/repair-ingest-queue.js | 56 +++++++++++++ packages/queue/src/workers.js | 26 +++--- 9 files changed, 303 insertions(+), 21 deletions(-) create mode 100644 apps/worker/src/repair-queue-cli.js create mode 100644 docs/ingest-queue.md create mode 100644 packages/queue/src/ingest-scheduling.js create mode 100644 packages/queue/src/ingest-scheduling.test.js create mode 100644 packages/queue/src/repair-ingest-queue.js diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d58a190..dea1f25 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/apps/worker/src/repair-queue-cli.js b/apps/worker/src/repair-queue-cli.js new file mode 100644 index 0000000..793e941 --- /dev/null +++ b/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(); +} diff --git a/docs/ingest-queue.md b/docs/ingest-queue.md new file mode 100644 index 0000000..ca8ddab --- /dev/null +++ b/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. diff --git a/packages/db/src/queries.js b/packages/db/src/queries.js index 235b2d0..e4ac333 100644 --- a/packages/db/src/queries.js +++ b/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} `; } diff --git a/packages/queue/src/index.js b/packages/queue/src/index.js index aef347c..eaf68b0 100644 --- a/packages/queue/src/index.js +++ b/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() { diff --git a/packages/queue/src/ingest-scheduling.js b/packages/queue/src/ingest-scheduling.js new file mode 100644 index 0000000..6dbf3b3 --- /dev/null +++ b/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 }; +} diff --git a/packages/queue/src/ingest-scheduling.test.js b/packages/queue/src/ingest-scheduling.test.js new file mode 100644 index 0000000..f73dd05 --- /dev/null +++ b/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); + }); +}); diff --git a/packages/queue/src/repair-ingest-queue.js b/packages/queue/src/repair-ingest-queue.js new file mode 100644 index 0000000..c246a98 --- /dev/null +++ b/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; +} diff --git a/packages/queue/src/workers.js b/packages/queue/src/workers.js index 189f1f5..95a8101 100644 --- a/packages/queue/src/workers.js +++ b/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 -- */