fix(ingestion): R17 — structural unique index for one open job per document

src/app/api/documents/[id]/reindex/route.ts

@@ -6,10 +6,13 @@ import { upsertDocumentEnrichment } from "@/lib/document-enrichment";
 import { upsertDocumentDeepMemory } from "@/lib/deep-memory";
 import { jsonError } from "@/lib/http";
 import {
+  activeIngestionJobColumns,
+  buildActiveJobsSafetyResult,
   checkIngestionMutationSafety,
   hasActiveAgentEnrichmentJob,
   ingestionMutationSafetyPayload,
   ingestionRollbackFenceStamp,
+  type IngestionJobRow,
 } from "@/lib/ingestion-mutation-safety";
 import { consumeApiRateLimit, rateLimitJsonResponse } from "@/lib/api-rate-limit";
 import {
@@ -256,6 +259,26 @@ export async function POST(request: Request, { params }: { params: Promise<{ id:
       .single();
 
     if (jobError) {
+      // R17: a unique index on ingestion_jobs(document_id) where status in
+      // (pending,processing) can reject this insert with 23505 when a
+      // concurrent request won the race between the pre-check above and this
+      // insert. That is the same "already queued" condition the pre-check
+      // reports, so surface it the same way (409, not a raw constraint 500).
+      if (jobError.code === "23505") {
+        const { data: raceJobs, error: raceJobsError } = await supabase
+          .from("ingestion_jobs")
+          .select(activeIngestionJobColumns)
+          .eq("document_id", id)
+          .in("status", ["pending", "processing"]);
+        if (!raceJobsError && (raceJobs?.length ?? 0) > 0) {
+          const safety = buildActiveJobsSafetyResult(
+            raceJobs as IngestionJobRow[],
+            env.WORKER_STALE_AFTER_MINUTES,
+            new Date().toISOString(),
+          );
+          return NextResponse.json(ingestionMutationSafetyPayload(safety), { status: 409 });
+        }
+      }
       const { data: competingJobs, error: competingJobsError } = await supabase
         .from("ingestion_jobs")
         .select("id")

src/app/api/documents/bulk/reindex/route.ts

@@ -6,9 +6,12 @@ import { upsertDocumentEnrichment } from "@/lib/document-enrichment";
 import { env, isDemoMode } from "@/lib/env";
 import { jsonError, PublicApiError } from "@/lib/http";
 import {
+  activeIngestionJobColumns,
+  buildActiveJobsSafetyResult,
   checkIngestionMutationSafety,
   ingestionMutationSafetyPayload,
   ingestionRollbackFenceStamp,
+  type IngestionJobRow,
 } from "@/lib/ingestion-mutation-safety";
 import { consumeApiRateLimit, rateLimitJsonResponse } from "@/lib/api-rate-limit";
 import { invalidateRagCachesForOwner } from "@/lib/rag";
@@ -227,6 +230,27 @@ export async function POST(request: Request) {
           .select("id")
           .single();
         if (jobError) {
+          // R17: same race as the single-document reindex route — a unique
+          // index on ingestion_jobs(document_id) where status in
+          // (pending,processing) can reject this insert with 23505 when a
+          // concurrent request won the race after the pre-check ran. Surface
+          // the friendly "already queued" message instead of the raw
+          // constraint-violation text.
+          if (jobError.code === "23505") {
+            const { data: raceJobs, error: raceJobsError } = await supabase
+              .from("ingestion_jobs")
+              .select(activeIngestionJobColumns)
+              .eq("document_id", document.id)
+              .in("status", ["pending", "processing"]);
+            if (!raceJobsError && (raceJobs?.length ?? 0) > 0) {
+              const safety = buildActiveJobsSafetyResult(
+                raceJobs as IngestionJobRow[],
+                env.WORKER_STALE_AFTER_MINUTES,
+                new Date().toISOString(),
+              );
+              throw new Error(safety.message);
+            }
+          }
           const { data: competingJobs, error: competingJobsError } = await supabase
             .from("ingestion_jobs")
             .select("id")

src/lib/ingestion-mutation-safety.ts

@@ -13,7 +13,7 @@ type AgentEnrichmentJobRow = {
 
 type IngestionJobStatus = "pending" | "processing" | "failed" | string;
 
-type IngestionJobRow = {
+export type IngestionJobRow = {
   id: string;
   document_id: string | null;
   status: IngestionJobStatus | null;
@@ -61,6 +61,9 @@ function isStaleProcessingJob(job: IngestionJobRow, staleAfterMinutes: number, n
   return age !== null && age >= staleAfterMinutes;
 }
 
+export const activeIngestionJobColumns =
+  "id,document_id,status,stage,locked_at,updated_at,error_message,attempt_count,max_attempts";
+
 function activeJobMessage(documentCount: number, staleCount: number) {
   if (staleCount > 0) {
     return staleCount === 1
@@ -72,6 +75,29 @@ function activeJobMessage(documentCount: number, staleCount: number) {
     : "One or more selected documents already have pending or processing indexing work.";
 }
 
+// Shared by checkIngestionMutationSafety's pre-check and the reindex routes'
+// post-insert 23505 handler (R17): the pre-check SELECT and the unique index
+// on (document_id) where status in (pending,processing) guard the same
+// invariant, so a race that slips past the pre-check and hits the index
+// instead should still produce this same 409 shape, not a raw constraint error.
+export function buildActiveJobsSafetyResult(
+  activeJobs: IngestionJobRow[],
+  staleAfterMinutes: number,
+  checkedAt: string,
+  now: Date = new Date(),
+): Extract<IngestionMutationSafetyResult, { ok: false }> {
+  const staleProcessingJobs = activeJobs.filter((job) => isStaleProcessingJob(job, staleAfterMinutes, now.getTime()));
+  return {
+    ok: false,
+    status: 409,
+    checkedAt,
+    reason: staleProcessingJobs.length > 0 ? "stale_processing_jobs" : "active_jobs",
+    message: activeJobMessage(activeJobs.length, staleProcessingJobs.length),
+    activeJobs,
+    staleProcessingJobs,
+  };
+}
+
 // Rollback fence: a timestamptz value unique to this request. Queue-state
 // writes stamp the row with it so the compensating rollback can run as a
 // single conditional UPDATE (`.eq` on the stamp) — atomic server-side. If a
@@ -168,28 +194,17 @@ export async function checkIngestionMutationSafety(args: {
 
   const { data, error } = await args.supabase
     .from("ingestion_jobs")
-    .select("id,document_id,status,stage,locked_at,updated_at,error_message,attempt_count,max_attempts")
+    .select(activeIngestionJobColumns)
     .in("document_id", uniqueDocumentIds)
     .in("status", ["pending", "processing"]);
   if (error) throw new Error(error.message);
 
   const activeJobs = ((data ?? []) as IngestionJobRow[]).filter(
     (job) => job.status === "pending" || job.status === "processing",
   );
-  const staleProcessingJobs = activeJobs.filter((job) =>
-    isStaleProcessingJob(job, args.staleAfterMinutes, (args.now ?? new Date()).getTime()),
-  );
 
   if (activeJobs.length > 0) {
-    return {
-      ok: false,
-      status: 409,
-      checkedAt: health.checkedAt,
-      reason: staleProcessingJobs.length > 0 ? "stale_processing_jobs" : "active_jobs",
-      message: activeJobMessage(uniqueDocumentIds.length, staleProcessingJobs.length),
-      activeJobs,
-      staleProcessingJobs,
-    };
+    return buildActiveJobsSafetyResult(activeJobs, args.staleAfterMinutes, health.checkedAt, args.now);
   }
 
   return {

supabase/drift-manifest.json

@@ -1,8 +1,8 @@
 {
-  "generated_at": "2026-07-08T06:45:01.039Z",
+  "generated_at": "2026-07-08T16:31:51.432Z",
   "generator": "scripts/generate-drift-manifest.ts",
   "postgres_image": "supabase/postgres:17.6.1.127",
-  "schema_sha256": "cb34f64fc531fc042f545475ddaa7cc65edeeae0dfa4df0b8d7d58edb7123adf",
+  "schema_sha256": "3ee18ec46fd0d7f9586858f8b0345355c60dbdc235f0dd4a802c04c06d750f70",
   "replay_seconds": 13,
   "snapshot": {
     "views": [
@@ -4869,6 +4869,12 @@
         "table": "ingestion_jobs",
         "def_hash": "a59d61034f2cb9337c13283574b698d9"
       },
+      {
+        "def": "CREATE UNIQUE INDEX ingestion_jobs_one_open_per_document_uidx ON public.ingestion_jobs USING btree (document_id) WHERE (status = ANY (ARRAY['pending'::text, 'processing'::text]))",
+        "name": "ingestion_jobs_one_open_per_document_uidx",
+        "table": "ingestion_jobs",
+        "def_hash": "2bf0e17b1b46c2b2b225128eaf6dbf0e"
+      },
       {
         "def": "CREATE UNIQUE INDEX ingestion_jobs_pkey ON public.ingestion_jobs USING btree (id)",
         "name": "ingestion_jobs_pkey",

supabase/migrations/20260708160000_ingestion_jobs_one_open_per_document.sql

@@ -0,0 +1,28 @@
+-- R17 (docs/ingestion-concurrency-fix-workorder.md): no constraint currently
+-- prevents multiple open ingestion_jobs rows per document; the reindex
+-- enqueue race (R13) and duplicate reindex POSTs previously relied on
+-- application-side advisory checks (a SELECT immediately before INSERT),
+-- which is a check-then-act race. This index makes the invariant structural.
+--
+-- NOT applied via `supabase db push` — CLI migrations run inside a
+-- transaction and `CONCURRENTLY` cannot run inside one
+-- (docs/supabase-migration-reconciliation.md, "Index replacements" rule).
+-- Operator: apply the statement below manually against live (outside a
+-- transaction, e.g. `supabase db query --linked` or the SQL editor with
+-- autocommit), after confirming `npm run reindex:health` reports
+-- jobs_pending = 0 and jobs_processing = 0 (live had 0/0 as of 2026-07-08,
+-- so creation will not fail on existing duplicates). Then run
+-- `supabase migration repair --linked --status applied 20260708160000` so
+-- history matches effect, per policy.
+--
+-- Paired app change (same release, already shipped in this PR): the
+-- single-document and bulk reindex routes translate a 23505 unique
+-- violation on job insert into the same 409 "already queued" response the
+-- pre-check produces, instead of a raw constraint 500
+-- (src/app/api/documents/[id]/reindex/route.ts,
+-- src/app/api/documents/bulk/reindex/route.ts).
+
+create unique index concurrently if not exists
+  ingestion_jobs_one_open_per_document_uidx
+  on public.ingestion_jobs (document_id)
+  where status in ('pending', 'processing');

supabase/schema.sql

@@ -777,6 +777,15 @@ create index if not exists ingestion_jobs_status_next_run_idx
   where status in ('pending', 'processing', 'failed');
 create index if not exists ingestion_jobs_document_status_idx
   on public.ingestion_jobs(document_id, status, created_at);
+-- R17 (docs/ingestion-concurrency-fix-workorder.md): structural guard against
+-- more than one open job per document. Migration
+-- 20260708160000_ingestion_jobs_one_open_per_document.sql creates the live
+-- equivalent with CONCURRENTLY (manual apply — see that file's header); this
+-- non-concurrent form is for fresh/scratch replay only, where there is no
+-- concurrent traffic to block.
+create unique index if not exists ingestion_jobs_one_open_per_document_uidx
+  on public.ingestion_jobs(document_id)
+  where status in ('pending', 'processing');
 drop index if exists public.ingestion_job_stages_doc_idx;
 create index if not exists ingestion_job_stages_document_started_idx
   on public.ingestion_job_stages(document_id, started_at desc);

tests/ingestion-mutation-safety.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, it } from "vitest";
-import { isActiveAgentEnrichmentJob } from "../src/lib/ingestion-mutation-safety";
+import { buildActiveJobsSafetyResult, isActiveAgentEnrichmentJob } from "../src/lib/ingestion-mutation-safety";
 
 describe("active enrichment-agent detection (R24d)", () => {
   const now = Date.parse("2026-07-07T00:00:00.000Z");
@@ -60,3 +60,43 @@ describe("active enrichment-agent detection (R24d)", () => {
     ).toBe(false);
   });
 });
+
+describe("buildActiveJobsSafetyResult (R17 — reindex route 23505 race handling)", () => {
+  const checkedAt = "2026-07-08T00:00:00.000Z";
+  const staleAfterMinutes = 45;
+  const now = new Date("2026-07-08T00:10:00.000Z");
+
+  const job = (overrides: Partial<Parameters<typeof buildActiveJobsSafetyResult>[0][number]> = {}) => ({
+    id: "job-1",
+    document_id: "doc-1",
+    status: "pending" as const,
+    stage: "queued",
+    locked_at: null,
+    updated_at: "2026-07-08T00:05:00.000Z",
+    error_message: null,
+    attempt_count: 0,
+    max_attempts: 3,
+    ...overrides,
+  });
+
+  it("produces the same 409 'already queued' shape the pre-check produces, for a single fresh job", () => {
+    const result = buildActiveJobsSafetyResult([job()], staleAfterMinutes, checkedAt, now);
+    expect(result.ok).toBe(false);
+    expect(result.status).toBe(409);
+    expect(result.reason).toBe("active_jobs");
+    expect(result.message).toBe("Document already has pending or processing indexing work.");
+    expect(result.activeJobs).toHaveLength(1);
+    expect(result.staleProcessingJobs).toHaveLength(0);
+  });
+
+  it("reports stale_processing_jobs when the race winner's job has an expired lease", () => {
+    const result = buildActiveJobsSafetyResult(
+      [job({ status: "processing", locked_at: "2026-07-07T22:00:00.000Z" })],
+      staleAfterMinutes,
+      checkedAt,
+      now,
+    );
+    expect(result.reason).toBe("stale_processing_jobs");
+    expect(result.staleProcessingJobs).toHaveLength(1);
+  });
+});