fix(ingestion): stop the retry cron double-submitting a scan already in flight

The Image INSERT trigger queues a new image immediately, but ingestImage stamps
scanRequestedAt only once its upload-path submit returns. A cron run landing inside
that window read the NULL as "never submitted" and submitted a second workflow for
the same image.

Measured on prod 2026-09-18: of 41,865 images scanned in 12h, 166 carried two
workflow ids; 161 of those were created within 30s of a cron tick, against a 9.8%
uniform baseline, spread over 112 users.

A Pending image with no scanRequestedAt is now deferred for SUBMIT_IN_FLIGHT_GRACE
minutes. It stays in the JobQueue while deferred — without that it prunes as stale
and an image whose submit died silently would never be re-driven, which is the
trigger's whole purpose. The deferral count is reported alongside waitingForRetry.

Two existing fixtures modelled "new image, never submitted" as createdAt: now, which
the grace defers; aged them past it so each test still exercises its own subject.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

src/server/jobs/__tests__/ingest-images-pending-ageout.test.ts

@@ -27,7 +27,10 @@ const PENDING_TIMEOUT_MIN = 30;
 
 // Fixture rows returned by the image SELECT. Fields mirror IngestImageRow.
 const OLD = new Date(Date.now() - 2 * HOUR); // past the 30-min age-out threshold
-const FRESH = new Date(); // well under the threshold
+// Under the age-out threshold, but past SUBMIT_IN_FLIGHT_GRACE — a just-created row
+// with no `scanRequestedAt` is deferred as "first submit still in flight", which would
+// mask the age-out behaviour under test here.
+const FRESH = new Date(Date.now() - 10 * 60 * 1000);
 const OLD_SCAN = new Date(Date.now() - 2 * HOUR); // past the 60-min Error retry delay
 
 const ROWS = [

src/server/jobs/__tests__/ingest-images-send-resilience.test.ts

@@ -29,7 +29,10 @@ const { mockDbRead, mockDbWrite, mockIngestImage, mockDeleteImages, mockLimitCon
             height: 100,
             prompt: null,
             scanRequestedAt: null,
-            createdAt: new Date(),
+            // Past SUBMIT_IN_FLIGHT_GRACE: a just-created row is deferred as
+            // "upload-path submit still in flight" and never reaches the send loop
+            // this test is about.
+            createdAt: new Date(Date.now() - 10 * 60 * 1000),
             ingestion: 'Pending',
             retryCount: 0,
             failureClass: null,

src/server/jobs/__tests__/ingest-images-submit-in-flight.test.ts

@@ -0,0 +1,144 @@
+import { describe, it, expect, vi, beforeEach } from 'vitest';
+
+// The Image INSERT trigger (`trg_image_scan_queue`) queues a new image the moment the
+// row lands, but `ingestImage` stamps `scanRequestedAt` only when its upload-path
+// submit returns. A cron run landing inside that window read the NULL as "never
+// submitted" and submitted a SECOND workflow for the same image — two scans, two
+// callbacks, twice the orchestrator work. Measured on prod 2026-09-18: of 41,865
+// images scanned in 12h, 166 carried two workflow ids, and 161 of those were created
+// within 30s of a cron tick against a 9.8% uniform baseline.
+//
+// The grace must not swallow the case it sits next to: an image whose submit died
+// without writing the row at all still has to be re-driven, so a skipped row stays in
+// the JobQueue rather than pruning as stale.
+
+const MINUTE = 60 * 1000;
+// SUBMIT_IN_FLIGHT_GRACE in the job under test (minutes). IN_FLIGHT is inside it,
+// SETTLED is past it but still well under the 30-min age-out.
+const IN_FLIGHT = new Date();
+const SETTLED = new Date(Date.now() - 5 * MINUTE);
+
+const ROWS = [
+  // 1: Pending, just created, never stamped -> its first submit may still be running.
+  mkRow({ id: 1, ingestion: 'Pending', createdAt: IN_FLIGHT, scanRequestedAt: null }),
+  // 2: Pending, past the grace, still never stamped -> its submit died silently; the
+  //    cron is the only thing that will ever re-drive it.
+  mkRow({ id: 2, ingestion: 'Pending', createdAt: SETTLED, scanRequestedAt: null }),
+  // 3: Rescan, just created -> a rescan is an explicit request with no submit in
+  //    flight, so the grace must not hold it back.
+  mkRow({ id: 3, ingestion: 'Rescan', createdAt: IN_FLIGHT, scanRequestedAt: null }),
+];
+
+function mkRow(overrides: {
+  id: number;
+  ingestion: string;
+  createdAt: Date;
+  scanRequestedAt: Date | null;
+  retryCount?: number;
+}) {
+  return {
+    url: `img-${overrides.id}`,
+    type: 'image',
+    width: 100,
+    height: 100,
+    prompt: null,
+    retryCount: 0,
+    failureClass: null,
+    isBackfill: false,
+    ...overrides,
+  };
+}
+
+const { execLog, mockIngestImage, mockDeleteImages, mockLimitConcurrency } = vi.hoisted(() => {
+  const execLog: { sql: string; values: unknown[] }[] = [];
+  return {
+    execLog,
+    mockIngestImage: vi.fn(async () => true),
+    mockDeleteImages: vi.fn(async () => undefined),
+    // Sequential for deterministic assertions.
+    mockLimitConcurrency: vi.fn(async (tasks: Array<() => Promise<unknown>>) => {
+      for (const t of tasks) await t();
+    }),
+  };
+});
+
+vi.mock('~/server/services/image.service', () => ({
+  ingestImage: mockIngestImage,
+  deleteImages: mockDeleteImages,
+}));
+vi.mock('~/server/utils/concurrency-helpers', () => ({ limitConcurrency: mockLimitConcurrency }));
+vi.mock('~/env/other', () => ({ isProd: true }));
+vi.mock('~/env/server', () => ({
+  env: {
+    IMAGE_SCANNING_MAX_PER_RUN: 100,
+    IMAGE_SCANNING_RETRY_DELAY: 5,
+    IMAGE_SCANNING_PENDING_TIMEOUT: 30,
+    DATABASE_IS_PROD: true,
+  },
+}));
+
+import { ingestImages } from '~/server/jobs/image-ingestion';
+import { dbMock } from '~/__tests__/mocks/db.mock';
+
+dbMock.dbRead.jobQueue.findMany.mockImplementation(async () =>
+  ROWS.map((row) => ({ entityId: row.id }))
+);
+dbMock.dbWrite.$queryRaw.mockImplementation(async () => ROWS);
+dbMock.dbWrite.$executeRaw.mockImplementation(
+  async (strings: TemplateStringsArray, ...values: unknown[]) => {
+    execLog.push({ sql: strings.join('?'), values });
+    return 0;
+  }
+);
+
+const ctx = {} as Parameters<typeof ingestImages.run>[0];
+async function runJob() {
+  return (await ingestImages.run(ctx).result) as { submitInFlight: number; sent: number };
+}
+
+function sentIds() {
+  return mockIngestImage.mock.calls.map((c) => (c[0] as { image: { id: number } }).image.id);
+}
+function targetIds(call?: { values: unknown[] }): number[] {
+  return (call?.values.find(Array.isArray) as number[] | undefined) ?? [];
+}
+function pruneDelete() {
+  return execLog.find((c) => c.sql.includes('DELETE FROM "JobQueue"'));
+}
+
+beforeEach(() => {
+  execLog.length = 0;
+  dbMock.dbRead.jobQueue.findMany.mockClear();
+  dbMock.dbWrite.$queryRaw.mockClear();
+  dbMock.dbWrite.$executeRaw.mockClear();
+  mockIngestImage.mockClear();
+});
+
+describe('ingest-images submit-in-flight grace', () => {
+  it('does NOT re-submit a just-created Pending image whose first submit has not returned', async () => {
+    const result = await runJob();
+
+    expect(sentIds()).not.toContain(1);
+    expect(result.submitInFlight).toBe(1);
+  });
+
+  it('keeps the skipped image in the JobQueue so a silently-dead submit is still re-driven', async () => {
+    await runJob();
+
+    // Not processed and not waiting on a cooldown, so without the in-flight branch in
+    // `waitingForRetryIds` this row prunes as stale and never scans.
+    expect(targetIds(pruneDelete())).not.toContain(1);
+  });
+
+  it('still submits a Pending image that is past the grace and was never stamped', async () => {
+    await runJob();
+
+    expect(sentIds()).toContain(2);
+  });
+
+  it('does not hold back a Rescan image — the grace is for Pending submits only', async () => {
+    await runJob();
+
+    expect(sentIds()).toContain(3);
+  });
+});

src/server/jobs/image-ingestion.ts

@@ -15,6 +15,12 @@ import { decreaseDate } from '~/utils/date-helpers';
 const IMAGE_SCANNING_ERROR_DELAY = 60 * 1; // 1 hour
 const IMAGE_SCANNING_RETRY_LIMIT = 9;
 
+// Minutes a just-created image counts as "submit in flight" (see the double-submit
+// note below). Must stay above the upload-path submit's own ceiling or the window it
+// closes reopens; overshooting costs one cron tick of delay for an image whose submit
+// died without writing the row at all.
+const SUBMIT_IN_FLIGHT_GRACE = 2;
+
 // Hard per-image backstop. The orchestrator submit is bounded per attempt (~15s), but the
 // Flipt flag read before it and the scanJobs UPDATE after it are not, so this keeps one hung
 // image to one concurrency slot. A timed-out image stays queued for a later run.
@@ -148,9 +154,22 @@ export const ingestImages = createJob('ingest-images', '*/5 * * * *', async (ctx
   const rescanDate = decreaseDate(now, env.IMAGE_SCANNING_RETRY_DELAY, 'minutes');
   const errorRetryDate = decreaseDate(now, IMAGE_SCANNING_ERROR_DELAY, 'minutes').getTime();
 
+  // A NULL `scanRequestedAt` does not yet mean "never submitted". The Image INSERT
+  // trigger queues a new image immediately, but `ingestImage` stamps
+  // `scanRequestedAt` only once its upload-path submit returns — up to ~50s later
+  // (3 attempts x the 15s per-attempt abort). A run landing inside that window reads
+  // the NULL as eligible and submits a SECOND workflow for the same image. Measured
+  // 2026-09-18: of 41,865 images scanned in 12h, 166 had two workflows, and 161 of
+  // those were created within 30s of a cron tick (uniform baseline: 9.8%).
+  const submitInFlightDate = decreaseDate(now, SUBMIT_IN_FLIGHT_GRACE, 'minutes');
+  const isSubmitInFlight = (img: IngestImageRow) =>
+    img.ingestion === 'Pending' && !img.scanRequestedAt && img.createdAt > submitInFlightDate;
+
   const pendingImages = images.filter(
     (img) =>
-      img.ingestion === 'Pending' && (!img.scanRequestedAt || img.scanRequestedAt <= rescanDate)
+      img.ingestion === 'Pending' &&
+      !isSubmitInFlight(img) &&
+      (!img.scanRequestedAt || img.scanRequestedAt <= rescanDate)
   );
 
   // Age-out safety net for never-returning Pending scans.
@@ -244,6 +263,10 @@ export const ingestImages = createJob('ingest-images', '*/5 * * * *', async (ctx
         ) {
           return true;
         }
+        // Skipped this run only because its first submit may still be in flight. It is
+        // neither processed nor waiting on a cooldown, so without this it prunes as
+        // stale — and if that submit died silently, nothing would ever re-drive it.
+        if (isSubmitInFlight(img)) return true;
         // Rescan waiting for the retry delay (and still under the retry cap) - KEEP.
         // Must mirror the rescanImages cooldown above, otherwise a cooled-down
         // Rescan image is neither processed nor waiting and gets wrongly pruned.
@@ -279,8 +302,11 @@ export const ingestImages = createJob('ingest-images', '*/5 * * * *', async (ctx
     return true;
   });
 
+  const submitInFlightCount = images.filter(isSubmitInFlight).length;
+
   console.log({
     pendingImages: pendingImages.length,
+    submitInFlight: submitInFlightCount,
     pendingUserUploads: pendingUserUploads.length,
     pendingBackfill: pendingBackfill.length,
     agedOutPending: agedOutPendingIds.length,
@@ -377,6 +403,7 @@ export const ingestImages = createJob('ingest-images', '*/5 * * * *', async (ctx
   imageIngestCronCounter.inc({ bucket: 'waitingForRetry' }, waitingForRetryIds.size);
   imageIngestCronCounter.inc({ bucket: 'staleRemoved' }, staleIds.length);
   imageIngestCronCounter.inc({ bucket: 'agedOutPending' }, agedOutPendingIds.length);
+  imageIngestCronCounter.inc({ bucket: 'submitInFlight' }, submitInFlightCount);
 
   // Failed sends = images whose submit was attempted and returned/threw failure,
   // across every lane. Images not reached this run (budget/cancel) are neither sent
@@ -400,6 +427,7 @@ export const ingestImages = createJob('ingest-images', '*/5 * * * *', async (ctx
       rescan: rescanImages.length,
       error: errorImages.length,
       agedOutPending: agedOutPendingIds.length,
+      submitInFlight: submitInFlightCount,
       waitingForRetry: waitingForRetryIds.size,
       staleRemoved: staleIds.length,
       failedSends,
@@ -427,6 +455,7 @@ export const ingestImages = createJob('ingest-images', '*/5 * * * *', async (ctx
     sentRescan: sentRescanIds.length,
     sentError: sentErrorIds.length,
     agedOutPending: agedOutPendingIds.length,
+    submitInFlight: submitInFlightCount,
     waitingForRetry: waitingForRetryIds.size,
     staleRemoved: staleIds.length,
     failedSends,