From 2289b60101f38d0503269573a7b32594f6cc9f09 Mon Sep 17 00:00:00 2001 From: Andrii Kohut Date: Tue, 21 Jul 2026 14:08:47 +0200 Subject: [PATCH] fix: resolve queue job visibility against the database clock --- docs/adr/0004-async-ingestion-202-queue.md | 9 ++++++++ .../migration.sql | 2 ++ packages/db/prisma/schema.prisma | 2 +- packages/db/src/__tests__/queue.test.ts | 17 +++++++++++++++ packages/db/src/queue.ts | 21 ++++++++++--------- 5 files changed, 40 insertions(+), 11 deletions(-) create mode 100644 packages/db/prisma/migrations/20260718115718_queue_visibility_uses_db_clock/migration.sql diff --git a/docs/adr/0004-async-ingestion-202-queue.md b/docs/adr/0004-async-ingestion-202-queue.md index cc21816..4a17fbd 100644 --- a/docs/adr/0004-async-ingestion-202-queue.md +++ b/docs/adr/0004-async-ingestion-202-queue.md @@ -22,3 +22,12 @@ The ingestion api does the minimum synchronously: authenticate the project token - At-least-once delivery plus idempotency keys make re-delivery safe; duplicate batches cannot double-count. - Results appear in the dashboard with a processing lag — the UI must communicate freshness honestly. - Queue depth becomes the platform's primary health metric and backpressure signal. + +## Implementation note: job visibility lives in one clock domain + +Job visibility is a comparison between a stored timestamp and "now", so both sides must come from the **database** clock. Two defects were found and fixed here: + +- `visible_at` was populated by the ORM from the **application** clock while `dequeue` compared it against Postgres `now()`. Any positive skew between the app host and the database (NTP drift, containers on different hosts) made freshly enqueued jobs invisible for the duration of that skew. The column is now database-generated, and retry backoff is computed as `now() + interval` in SQL rather than from the application clock. +- The column is `TIMESTAMP(3)` while `now()` is microsecond-precision, so a stored value can **round up** to as much as half a millisecond into the future. `dequeue` therefore compares against `now() + 1 millisecond`, a tolerance equal to the storage granularity. + +Together these made a just-enqueued job invisible in roughly 99% of tight enqueue-then-dequeue cycles on a machine with 1 ms host/container clock skew. In production the effect was masked by the worker's polling interval, which is exactly why it survived so long: the only visible symptom was an occasional flaky test. diff --git a/packages/db/prisma/migrations/20260718115718_queue_visibility_uses_db_clock/migration.sql b/packages/db/prisma/migrations/20260718115718_queue_visibility_uses_db_clock/migration.sql new file mode 100644 index 0000000..ef2bfd0 --- /dev/null +++ b/packages/db/prisma/migrations/20260718115718_queue_visibility_uses_db_clock/migration.sql @@ -0,0 +1,2 @@ +-- AlterTable +ALTER TABLE "ingestion_job" ALTER COLUMN "visible_at" SET DEFAULT now(); diff --git a/packages/db/prisma/schema.prisma b/packages/db/prisma/schema.prisma index c9b2de2..9637dd0 100644 --- a/packages/db/prisma/schema.prisma +++ b/packages/db/prisma/schema.prisma @@ -103,7 +103,7 @@ model IngestionJob { payload Json status IngestionJobStatus @default(pending) attempts Int @default(0) - visibleAt DateTime @default(now()) @map("visible_at") + visibleAt DateTime @default(dbgenerated("now()")) @map("visible_at") lastError String? @map("last_error") createdAt DateTime @default(now()) @map("created_at") updatedAt DateTime @updatedAt @map("updated_at") diff --git a/packages/db/src/__tests__/queue.test.ts b/packages/db/src/__tests__/queue.test.ts index dc678ec..85b4015 100644 --- a/packages/db/src/__tests__/queue.test.ts +++ b/packages/db/src/__tests__/queue.test.ts @@ -79,6 +79,23 @@ describe.skipIf(!hasDb)('IngestionQueue', () => { expect(await queue.depth()).toBe(0) }) + it('makes a freshly enqueued job immediately visible to dequeue', async () => { + const { orgId, projectId } = await seedProject() + const queue = new IngestionQueue(prisma) + + for (let round = 0; round < 25; round += 1) { + await queue.enqueue({ + orgId, + projectId, + idempotencyKey: `immediate-${round}`, + payload: payload('immediate'), + }) + const batch = await queue.dequeue(1) + expect(batch).toHaveLength(1) + await queue.complete(batch[0]!.id) + } + }) + it('counts in-flight jobs in depth so backpressure sees the real backlog', async () => { const { orgId, projectId } = await seedProject() const queue = new IngestionQueue(prisma) diff --git a/packages/db/src/queue.ts b/packages/db/src/queue.ts index 39612a0..6e5f40e 100644 --- a/packages/db/src/queue.ts +++ b/packages/db/src/queue.ts @@ -69,7 +69,7 @@ export class IngestionQueue { } async dequeue(limit = 1): Promise { - const timeout = `${Math.round(this.options.visibilityTimeoutMs / 1000)} seconds` + const timeout = `${this.options.visibilityTimeoutMs} milliseconds` return this.prisma.$queryRaw` UPDATE ingestion_job SET status = 'processing', @@ -79,7 +79,7 @@ export class IngestionQueue { WHERE id IN ( SELECT id FROM ingestion_job WHERE status IN ('pending', 'processing') - AND visible_at <= now() + AND visible_at <= now() + '1 millisecond'::interval ORDER BY visible_at ASC LIMIT ${limit} FOR UPDATE SKIP LOCKED @@ -110,14 +110,15 @@ export class IngestionQueue { return 'dead' } const backoffMs = this.options.baseBackoffMs * 2 ** (job.attempts - 1) - await this.prisma.ingestionJob.update({ - where: { id: jobId }, - data: { - status: 'pending', - lastError: error, - visibleAt: new Date(Date.now() + backoffMs), - }, - }) + const backoff = `${backoffMs} milliseconds` + await this.prisma.$executeRaw` + UPDATE ingestion_job + SET status = 'pending', + last_error = ${error}, + visible_at = now() + ${backoff}::interval, + updated_at = now() + WHERE id = ${jobId}::uuid + ` return 'retry' }