From 612f4bb9dffc1e9a96ff767b11453c7fe1743bb7 Mon Sep 17 00:00:00 2001 From: BigSimmo <87357024+BigSimmo@users.noreply.github.com> Date: Wed, 8 Jul 2026 13:33:30 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix(storage):=20R11=20janitor-side=20guard?= =?UTF-8?q?=20=E2=80=94=20never=20delete=20a=20live=20document's=20storage?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Defense-in-depth complement to the DELETE-route path-clearing already landed in PR #346. scripts/cleanup-storage.ts now skips any storage_cleanup_jobs row whose document_id still resolves to a live document: the ledger FK is ON DELETE SET NULL, so a genuinely-deleted document has its ledger document_id nulled — a non-null id that still resolves means the delete aborted and the paths still belong to a live document. Protects pre-#346 poisoned rows and any abort path that still leaves populated paths. Pure partition logic extracted to src/lib/storage-cleanup-safety.ts with tests. Co-Authored-By: Claude Fable 5 --- scripts/cleanup-storage.ts | 30 ++++++++++++++++++++++--- src/lib/storage-cleanup-safety.ts | 33 ++++++++++++++++++++++++++++ tests/storage-cleanup-safety.test.ts | 30 +++++++++++++++++++++++++ 3 files changed, 90 insertions(+), 3 deletions(-) create mode 100644 src/lib/storage-cleanup-safety.ts create mode 100644 tests/storage-cleanup-safety.test.ts diff --git a/scripts/cleanup-storage.ts b/scripts/cleanup-storage.ts index 27389bc13..c252e8784 100644 --- a/scripts/cleanup-storage.ts +++ b/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(); + 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; diff --git a/src/lib/storage-cleanup-safety.ts b/src/lib/storage-cleanup-safety.ts new file mode 100644 index 000000000..25662befc --- /dev/null +++ b/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( + jobs: T[], + liveDocumentIds: ReadonlySet, +): { 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)))); +} diff --git a/tests/storage-cleanup-safety.test.ts b/tests/storage-cleanup-safety.test.ts new file mode 100644 index 000000000..2a8caecc6 --- /dev/null +++ b/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"]); + }); +}); From 79501bf75c47f679d2f4e91aeb69e87bd31bcd31 Mon Sep 17 00:00:00 2001 From: BigSimmo <87357024+BigSimmo@users.noreply.github.com> Date: Wed, 8 Jul 2026 13:33:30 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix(worker):=20R1=20lease=20heartbeat=20?= =?UTF-8?q?=E2=80=94=20stop=20false-reclaiming=20live=20long=20jobs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The worker never refreshed ingestion_jobs.locked_at during a job, so any build longer than WORKER_STALE_AFTER_MINUTES (45m) was reclaimed by claim_ingestion_jobs while the original worker was still alive — two workers on one document, the enabler (R1) for the R2-R8 write-clobber class. updateJobProgress now doubles as a lease heartbeat: shouldPersistJobProgress adds a heartbeat ceiling (1/3 of the stale window) forcing a write during long silent phases, and the write refreshes locked_at scoped to locked_by=workerId so a worker that already lost its lease no-ops instead of stealing it back. Additive and lease-scoped; the residual loser-write race is closed by the RPC locked_by fences (separate migration). Co-Authored-By: Claude Fable 5 --- src/lib/ingestion.ts | 28 +++++++++++++++++++++ tests/ingestion.test.ts | 56 +++++++++++++++++++++++++++++++++++++++++ worker/main.ts | 32 +++++++++++++++++++---- 3 files changed, 111 insertions(+), 5 deletions(-) diff --git a/src/lib/ingestion.ts b/src/lib/ingestion.ts index 132d715b9..b0e987c71 100644 --- a/src/lib/ingestion.ts +++ b/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]; +} diff --git a/tests/ingestion.test.ts b/tests/ingestion.test.ts index 9742793b2..249520ea6 100644 --- a/tests/ingestion.test.ts +++ b/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); diff --git a/worker/main.ts b/worker/main.ts index 8074d38ca..60e24d548 100644 --- a/worker/main.ts +++ 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(); 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",