fix: resolve queue job visibility against the database clock

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.

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();

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")

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)

packages/db/src/queue.ts

@@ -69,7 +69,7 @@ export class IngestionQueue {
   }
 
   async dequeue(limit = 1): Promise<QueuedJob[]> {
-    const timeout = `${Math.round(this.options.visibilityTimeoutMs / 1000)} seconds`
+    const timeout = `${this.options.visibilityTimeoutMs} milliseconds`
     return this.prisma.$queryRaw<QueuedJob[]>`
       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'
   }