fix(ingestion): phase-3 continued — R11 janitor, R1 heartbeat+fences, R7/R9/R23 RPC hardening

scripts/cleanup-storage.ts

@@ -1,4 +1,5 @@
 import { loadEnvConfig } from "@next/env";
+import { nonNullDocumentIds, partitionStorageCleanupJobs } from "@/lib/storage-cleanup-safety";
 import { loadAdminClient } from "./eval-utils";
 
 loadEnvConfig(process.cwd());
@@ -10,6 +11,7 @@ type CleanupArgs = {
 
 type CleanupJob = {
   id: string;
+  document_id: string | null;
   document_bucket: string | null;
   document_paths: string[] | null;
   image_bucket: string | null;
@@ -65,14 +67,36 @@ async function main() {
   const supabase = await loadAdminClient();
   const { data, error } = await supabase
     .from("storage_cleanup_jobs")
-    .select("id,document_bucket,document_paths,image_bucket,image_paths,attempts")
+    .select("id,document_id,document_bucket,document_paths,image_bucket,image_paths,attempts")
     .in("status", ["pending", "failed"])
     .order("created_at", { ascending: true })
     .limit(args.limit);
 
   if (error) throw new Error(error.message);
-  const jobs = (data ?? []) as CleanupJob[];
-  console.log(`Found ${jobs.length} storage cleanup job(s).`);
+  const allJobs = (data ?? []) as CleanupJob[];
+
+  // Audit R11 guard: never remove storage for a ledger row whose document still
+  // exists — a genuinely-deleted document has its ledger document_id nulled by
+  // the ON DELETE SET NULL FK, so a live document_id means the delete aborted
+  // and these paths still belong to a live document.
+  const candidateDocumentIds = nonNullDocumentIds(allJobs);
+  const liveDocumentIds = new Set<string>();
+  for (let start = 0; start < candidateDocumentIds.length; start += 1000) {
+    const batch = candidateDocumentIds.slice(start, start + 1000);
+    const { data: liveDocs, error: liveError } = await supabase.from("documents").select("id").in("id", batch);
+    if (liveError) throw new Error(liveError.message);
+    for (const doc of liveDocs ?? []) liveDocumentIds.add(doc.id);
+  }
+
+  const { safe: jobs, skipped } = partitionStorageCleanupJobs(allJobs, liveDocumentIds);
+  console.log(`Found ${allJobs.length} storage cleanup job(s); ${jobs.length} safe to process.`);
+  if (skipped.length > 0) {
+    console.warn(
+      `Skipping ${skipped.length} cleanup job(s) whose document still exists (aborted delete; would destroy live storage): ${skipped
+        .map((job) => job.id)
+        .join(", ")}`,
+    );
+  }
   if (args.dryRun || jobs.length === 0) return;
 
   let completed = 0;

src/lib/ingestion.ts

@@ -107,3 +107,31 @@ export function buildStorageCleanupJobUpdate(args: {
   }
   return update;
 }
+
+// Audit R1: the worker never refreshes ingestion_jobs.locked_at during a job,
+// so any build longer than WORKER_STALE_AFTER_MINUTES is reclaimed by
+// claim_ingestion_jobs while the original worker is still alive → two workers
+// process one document (the enabler for the R2-R8 write-clobber class). Making
+// the periodic progress write double as a lease heartbeat keeps a healthy
+// worker's lease fresh. This decides WHEN that write happens: the existing
+// throttle (minIntervalMs / minDelta / stage-prefix change) plus a heartbeat
+// ceiling that forces a write during long silent phases (e.g. a single slow
+// OpenAI call) so the lease cannot age out mid-job. The worker scopes the write
+// itself to `locked_by = workerId`, so a worker that has already lost its lease
+// no-ops instead of stealing it back.
+export function shouldPersistJobProgress(args: {
+  previous?: { updatedAt: number; progress: number; stage: string };
+  next: { progress: number; stage: string };
+  now: number;
+  minIntervalMs: number;
+  minDelta: number;
+  heartbeatMs: number;
+}): boolean {
+  const { previous, next, now, minIntervalMs, minDelta, heartbeatMs } = args;
+  if (!previous) return true;
+  const elapsed = now - previous.updatedAt;
+  if (elapsed >= heartbeatMs) return true;
+  if (elapsed >= minIntervalMs) return true;
+  if (Math.abs(next.progress - previous.progress) >= minDelta) return true;
+  return next.stage.split(" ")[0] !== previous.stage.split(" ")[0];
+}

src/lib/storage-cleanup-safety.ts

@@ -0,0 +1,33 @@
+export type StorageCleanupCandidate = {
+  id: string;
+  document_id: string | null;
+};
+
+// Audit R11: the DELETE route creates a storage_cleanup_jobs row (carrying the
+// live document's source-PDF + image paths) BEFORE its final guards; if the
+// delete then aborts, the row is left behind still pointing at a document that
+// was never deleted. The janitor removes those paths, permanently destroying a
+// live document's storage. The precise signal is the ledger FK
+// (storage_cleanup_jobs.document_id -> documents, ON DELETE SET NULL): a
+// genuinely-deleted document has its ledger document_id nulled, so a non-null
+// document_id that STILL resolves to a live document proves the delete did not
+// complete. Skip those rows instead of executing them.
+export function partitionStorageCleanupJobs<T extends StorageCleanupCandidate>(
+  jobs: T[],
+  liveDocumentIds: ReadonlySet<string>,
+): { safe: T[]; skipped: T[] } {
+  const safe: T[] = [];
+  const skipped: T[] = [];
+  for (const job of jobs) {
+    if (job.document_id && liveDocumentIds.has(job.document_id)) {
+      skipped.push(job);
+    } else {
+      safe.push(job);
+    }
+  }
+  return { safe, skipped };
+}
+
+export function nonNullDocumentIds(jobs: StorageCleanupCandidate[]): string[] {
+  return Array.from(new Set(jobs.map((job) => job.document_id).filter((id): id is string => Boolean(id))));
+}

tests/ingestion.test.ts

@@ -7,6 +7,7 @@ import {
   nextRetryAt,
   retryDelayMs,
   retryDocumentQueueUpdate,
+  shouldPersistJobProgress,
   terminalBatchStatus,
 } from "../src/lib/ingestion";
 
@@ -47,6 +48,61 @@ describe("ingestion retry helpers", () => {
   });
 });
 
+describe("job progress lease heartbeat (R1)", () => {
+  const opts = { minIntervalMs: 5_000, minDelta: 4, heartbeatMs: 900_000 };
+  const prev = { updatedAt: 1_000_000, progress: 50, stage: "embedding chunks 1-8/40" };
+
+  it("always writes the first progress update", () => {
+    expect(shouldPersistJobProgress({ next: { progress: 5, stage: "downloading" }, now: 0, ...opts })).toBe(true);
+  });
+
+  it("skips a throttled update with no meaningful change", () => {
+    expect(
+      shouldPersistJobProgress({
+        previous: prev,
+        next: { progress: 51, stage: prev.stage },
+        now: prev.updatedAt + 1_000,
+        ...opts,
+      }),
+    ).toBe(false);
+  });
+
+  it("writes once the progress delta is large enough", () => {
+    expect(
+      shouldPersistJobProgress({
+        previous: prev,
+        next: { progress: 55, stage: prev.stage },
+        now: prev.updatedAt + 1_000,
+        ...opts,
+      }),
+    ).toBe(true);
+  });
+
+  it("writes on a stage-prefix change even when throttled", () => {
+    expect(
+      shouldPersistJobProgress({
+        previous: prev,
+        next: { progress: 50, stage: "captioning images 1-8/20" },
+        now: prev.updatedAt + 1,
+        ...opts,
+      }),
+    ).toBe(true);
+  });
+
+  it("forces a heartbeat write during a long silent phase with no progress change", () => {
+    // Same stage/progress, but the heartbeat ceiling has elapsed — must write so
+    // locked_at stays fresh and the live job is not reclaimed as stale.
+    expect(
+      shouldPersistJobProgress({
+        previous: prev,
+        next: { progress: 50, stage: prev.stage },
+        now: prev.updatedAt + opts.heartbeatMs,
+        ...opts,
+      }),
+    ).toBe(true);
+  });
+});
+
 describe("ingestion job retry guards (R15/R16)", () => {
   it("rejects retrying a completed job (zombie re-ingest)", () => {
     expect(ingestionJobRetryRejectionReason("completed")).toMatch(/already completed/i);

tests/storage-cleanup-safety.test.ts

@@ -0,0 +1,30 @@
+import { describe, expect, it } from "vitest";
+import { nonNullDocumentIds, partitionStorageCleanupJobs } from "@/lib/storage-cleanup-safety";
+
+describe("partitionStorageCleanupJobs (audit R11)", () => {
+  const job = (id: string, document_id: string | null) => ({ id, document_id });
+
+  it("skips ledger rows whose document still exists (aborted delete)", () => {
+    const jobs = [job("j1", "doc-alive"), job("j2", "doc-gone"), job("j3", null)];
+    const live = new Set(["doc-alive"]);
+    const { safe, skipped } = partitionStorageCleanupJobs(jobs, live);
+    expect(skipped.map((j) => j.id)).toEqual(["j1"]);
+    expect(safe.map((j) => j.id)).toEqual(["j2", "j3"]);
+  });
+
+  it("treats a null document_id as a genuinely-deleted document (safe to process)", () => {
+    const { safe, skipped } = partitionStorageCleanupJobs([job("j1", null)], new Set(["anything"]));
+    expect(skipped).toHaveLength(0);
+    expect(safe.map((j) => j.id)).toEqual(["j1"]);
+  });
+
+  it("processes rows whose document_id no longer resolves (FK should have nulled it, but be permissive)", () => {
+    const { safe } = partitionStorageCleanupJobs([job("j1", "doc-gone")], new Set());
+    expect(safe.map((j) => j.id)).toEqual(["j1"]);
+  });
+
+  it("collects distinct non-null document ids for the liveness probe", () => {
+    const ids = nonNullDocumentIds([job("j1", "a"), job("j2", "a"), job("j3", null), job("j4", "b")]);
+    expect(ids.sort()).toEqual(["a", "b"]);
+  });
+});

worker/main.ts

@@ -29,6 +29,7 @@ import {
   isPartialIndexWriteConflict,
   isRetryableIngestionError,
   nextRetryAt,
+  shouldPersistJobProgress,
   terminalBatchStatus,
 } from "../src/lib/ingestion";
 import { assessDocumentIndexQuality } from "../src/lib/index-quality";
@@ -72,6 +73,11 @@ const workerId = `${os.hostname()}-${process.pid}-${randomUUID().slice(0, 8)}`;
 const progressUpdateState = new Map<string, { updatedAt: number; progress: number; stage: string }>();
 const progressUpdateMinIntervalMs = env.WORKER_PROGRESS_UPDATE_MIN_INTERVAL_MS;
 const progressUpdateMinDelta = 4;
+// Audit R1: force a lease-refreshing progress write at least this often so a
+// long silent phase cannot let locked_at age past WORKER_STALE_AFTER_MINUTES
+// and get the live job reclaimed. One-third of the stale window guarantees
+// several heartbeats before staleness; floored at 30s.
+const jobLeaseHeartbeatMs = Math.max(30_000, Math.floor((env.WORKER_STALE_AFTER_MINUTES * 60_000) / 3));
 const maxSupabaseBackoffMs = env.WORKER_HEALTH_BACKOFF_MS;
 const analyzeRagTablesThrottleMs = 45_000;
 let lastAnalyzeRagTablesAt = 0;
@@ -127,13 +133,29 @@ async function updateJob(jobId: string, patch: TablesUpdate<"ingestion_jobs">) {
 async function updateJobProgress(jobId: string, patch: { stage: string; progress: number }) {
   const previous = progressUpdateState.get(jobId);
   const now = Date.now();
-  const enoughTimeElapsed = !previous || now - previous.updatedAt >= progressUpdateMinIntervalMs;
-  const enoughProgressChanged = !previous || Math.abs(patch.progress - previous.progress) >= progressUpdateMinDelta;
-  const stagePrefixChanged = !previous || patch.stage.split(" ")[0] !== previous.stage.split(" ")[0];
 
-  if (!enoughTimeElapsed && !enoughProgressChanged && !stagePrefixChanged) return;
+  if (
+    !shouldPersistJobProgress({
+      previous,
+      next: patch,
+      now,
+      minIntervalMs: progressUpdateMinIntervalMs,
+      minDelta: progressUpdateMinDelta,
+      heartbeatMs: jobLeaseHeartbeatMs,
+    })
+  ) {
+    return;
+  }
 
-  const { error } = await supabase.from("ingestion_jobs").update(patch).eq("id", jobId);
+  // Audit R1: the progress write doubles as a lease heartbeat — refresh
+  // locked_at, but only while we still hold the lease (`locked_by = workerId`),
+  // so a worker that was already reclaimed no-ops instead of resurrecting or
+  // overwriting a lease another worker now owns.
+  const { error } = await supabase
+    .from("ingestion_jobs")
+    .update({ ...patch, locked_at: new Date().toISOString() })
+    .eq("id", jobId)
+    .eq("locked_by", workerId);
   if (error) {
     console.warn(
       "Ingestion progress update failed",