diff --git a/docs/reindex-runbook.md b/docs/reindex-runbook.md index 08fc0d7ff..95f4b9ecb 100644 --- a/docs/reindex-runbook.md +++ b/docs/reindex-runbook.md @@ -2,13 +2,38 @@ Use this sequence when applying RAG indexing changes to the live `Clinical KB Database` Supabase project. -## Safe sequence +## Consolidated pipeline command + +The quickest way to run a full reindex cycle is the consolidated pipeline command: + +```sh +npm run reindex +``` + +This single command encapsulates the full safe sequence below: it confirms the target project, checks Supabase health, snapshots reindex health, applies ingestion queue recovery (with an interactive confirmation prompt), runs the worker, and repeats until the queue is clear or the round limit is reached. + +Options: + +| Flag | Default | Description | +|---|---|---| +| `--yes` | off | Skip confirmation prompts (non-interactive / CI use) | +| `--max-rounds` | 10 | Maximum worker iterations before stopping | +| `--limit` | 20 | Recovery action limit per round | + +```sh +# Non-interactive, up to 5 worker rounds: +npm run reindex -- --yes --max-rounds 5 +``` + +## Manual safe sequence + +If you prefer to run each step by hand (or need to apply a migration in between): 1. Confirm the target project is `sjrfecxgysukkwxsowpy`. 2. Run `npm run supabase:recovery-status`. 3. Apply any pending local migration only after the probe succeeds. 4. Run `npm run reindex:health`. -5. If stale or failed jobs exist, run `npm run recover:ingestion -- --apply --limit 20`. +5. If stale or failed jobs exist, run `npm run recover:ingestion` and confirm the prompt. 6. Run `npm run worker:once` with the conservative defaults from `.env.example`. 7. Repeat `npm run reindex:health` and `npm run worker:once` until the queue is clear. 8. Run indexing and RAG evals only after documents have adaptive chunks and retrieval synopses. diff --git a/eslint.config.mjs b/eslint.config.mjs index 0f68a1f11..c388f1c69 100644 --- a/eslint.config.mjs +++ b/eslint.config.mjs @@ -18,6 +18,10 @@ const eslintConfig = defineConfig([ ".tmp-visual/**", "sample-documents/**", "scratch/**", + ".claude/**", + "**/.claude/**", + ".tmp-visual/**", + "**/.tmp-visual/**", "next-env.d.ts", ]), ]); diff --git a/package.json b/package.json index 9ee8ce132..398e7abbb 100644 --- a/package.json +++ b/package.json @@ -43,6 +43,7 @@ "check:supabase-project": "tsx scripts/check-supabase-project.ts", "check:indexing": "tsx scripts/check-indexing.ts", "recover:ingestion": "tsx scripts/recover-ingestion-queue.ts", + "reindex": "tsx scripts/reindex.ts", "reindex:health": "tsx scripts/reindex-health.ts", "supabase:recovery-status": "tsx scripts/supabase-recovery-status.ts", "promote:query-misses": "tsx scripts/promote-query-misses.ts", diff --git a/scripts/cli-utils.ts b/scripts/cli-utils.ts new file mode 100644 index 000000000..2d3ff19ab --- /dev/null +++ b/scripts/cli-utils.ts @@ -0,0 +1,20 @@ +import { createInterface } from "node:readline"; + +/** + * Prompts the user with a yes/no question and returns their answer. + * Returns `false` if stdin is not a TTY (e.g. when piped). + */ +export function confirm(question: string): Promise { + if (!process.stdin.isTTY) { + console.log(" Non-interactive input detected; defaulting to No."); + return Promise.resolve(false); + } + + return new Promise((resolve) => { + const rl = createInterface({ input: process.stdin, output: process.stdout }); + rl.question(`${question} (y/N) `, (answer: string) => { + rl.close(); + resolve(answer.trim().toLowerCase() === "y"); + }); + }); +} diff --git a/scripts/recover-ingestion-queue.ts b/scripts/recover-ingestion-queue.ts index ebc6527eb..369bd3d3c 100644 --- a/scripts/recover-ingestion-queue.ts +++ b/scripts/recover-ingestion-queue.ts @@ -1,5 +1,6 @@ import { loadEnvConfig } from "@next/env"; import type { IngestionRecoveryJob } from "@/lib/ingestion-recovery"; +import { confirm } from "./cli-utils"; loadEnvConfig(process.cwd()); @@ -9,6 +10,14 @@ type RecoveryDocument = { chunk_count?: number | null; }; +type RawJobRow = { + id: string; + document_id: string; + status: string | null; + locked_at: string | null; + documents: RecoveryDocument | RecoveryDocument[] | null; +}; + function supabaseStageError(stage: string, error: unknown) { const record = error && typeof error === "object" ? (error as Record) : {}; const message = @@ -35,6 +44,7 @@ function parseArgs(argv: string[]) { }; return { apply: argv.includes("--apply"), + yes: argv.includes("--yes"), staleAfterMinutes: Number.parseInt(valueFor("stale-after-minutes") ?? "", 10), limit: Number.parseInt(valueFor("limit") ?? "", 10), }; @@ -59,7 +69,12 @@ async function main() { : env.WORKER_STALE_AFTER_MINUTES; const limit = Number.isFinite(args.limit) ? args.limit : 20; const supabase = createAdminClient(); + + console.log("=== Ingestion Queue Recovery ==="); + console.log(`Checking Supabase health...`); assertSupabaseHealthy(await probeSupabaseHealth(supabase), "Ingestion queue recovery"); + console.log(" Supabase is healthy.\n"); + const { data, error } = await supabase .from("ingestion_jobs") .select("id,document_id,status,locked_at,documents(status,page_count,chunk_count)") @@ -68,7 +83,7 @@ async function main() { if (error) throw supabaseStageError("load open ingestion jobs", error); - const jobs = (data ?? []).map((job) => ({ + const jobs = (data ?? []).map((job: RawJobRow) => ({ ...job, documents: Array.isArray(job.documents) ? (job.documents[0] as RecoveryDocument | undefined) : job.documents, })) as IngestionRecoveryJob[]; @@ -77,29 +92,43 @@ async function main() { const resetDocumentIds = Array.from( new Set(actions.filter((action) => action.action === "retry").map((action) => action.documentId)), ); + const supersedeCount = actions.filter((action) => action.action === "supersede").length; + const retryCount = actions.filter((action) => action.action === "retry").length; + const remainingCount = Math.max(0, plan.actions.length - actions.length); - console.log( - JSON.stringify( - { - mode: args.apply ? "apply" : "dry-run", - staleAfterMinutes, - limit, - scannedJobs: jobs.length, - resetDocuments: resetDocumentIds.length, - supersedeJobs: actions.filter((action) => action.action === "supersede").length, - retryJobs: actions.filter((action) => action.action === "retry").length, - remainingActions: Math.max(0, plan.actions.length - actions.length), - }, - null, - 2, - ), - ); + console.log(`Stale-after threshold : ${staleAfterMinutes} min`); + console.log(`Action limit : ${limit}`); + console.log(`Scanned jobs : ${jobs.length}`); + console.log(`Documents to reset : ${resetDocumentIds.length}`); + console.log(`Jobs to supersede : ${supersedeCount}`); + console.log(`Jobs to retry : ${retryCount}`); + if (remainingCount > 0) { + console.log(`Remaining (over limit): ${remainingCount}`); + } - if (!args.apply) { - console.log("Dry run only. Re-run with --apply to mutate the ingestion queue."); + if (actions.length === 0) { + console.log("\nNothing to recover. Queue looks healthy."); return; } + let shouldApply = args.apply; + + if (!shouldApply) { + if (args.yes) { + shouldApply = true; + } else { + console.log(""); + shouldApply = await confirm("Apply these changes?"); + } + } + + if (!shouldApply) { + console.log("\nNo changes applied. Re-run with --apply or confirm interactively to mutate the ingestion queue."); + return; + } + + console.log("\nApplying recovery..."); + for (const documentId of resetDocumentIds) { const { error: resetError } = await supabase.rpc("reset_document_index", { p_document_id: documentId }); if (resetError) throw supabaseStageError("reset document index", resetError); @@ -146,6 +175,9 @@ async function main() { } console.log("Ingestion queue recovery applied."); + if (remainingCount > 0) { + console.log(`\n${remainingCount} action(s) remain over the limit. Re-run to process the next batch.`); + } } main().catch((error) => { diff --git a/scripts/reindex.ts b/scripts/reindex.ts new file mode 100644 index 000000000..662cf36b1 --- /dev/null +++ b/scripts/reindex.ts @@ -0,0 +1,410 @@ +/** + * Consolidated reindex pipeline. + * + * Runs the full safe reindex sequence from the runbook in a single command: + * 1. Confirm target Supabase project + * 2. Probe Supabase health + * 3. Snapshot reindex health + * 4. Apply ingestion queue recovery if stale/failed jobs are present + * 5. Run worker:once + * 6. Repeat steps 3–5 until the queue is clear or --max-rounds is reached + * + * Usage: + * npm run reindex # interactive – prompts before each recovery + * npm run reindex -- --yes # non-interactive – auto-confirm every prompt + * npm run reindex -- --max-rounds 5 # limit worker iterations (default: 10) + * npm run reindex -- --limit 50 # recovery action limit per round (default: 20) + */ + +import { spawn } from "node:child_process"; +import { loadEnvConfig } from "@next/env"; +import { confirm } from "./cli-utils"; + +loadEnvConfig(process.cwd()); + +function parseArgs(argv: string[]) { + const valueFor = (name: string) => { + const inline = argv.find((arg) => arg.startsWith(`--${name}=`))?.split("=")[1]; + if (inline) return inline; + const index = argv.indexOf(`--${name}`); + return index >= 0 ? argv[index + 1] : undefined; + }; + return { + yes: argv.includes("--yes"), + maxRounds: Number.parseInt(valueFor("max-rounds") ?? "10", 10), + limit: Number.parseInt(valueFor("limit") ?? "20", 10), + }; +} + +function runWorkerOnce(): Promise { + return new Promise((resolve, reject) => { + const child = spawn("npx", ["tsx", "worker/index.ts", "--once"], { + stdio: "inherit", + shell: false, + }); + child.on("close", (code: number | null) => { + if (code === 0) { + resolve(); + } else { + reject(new Error(`worker:once exited with code ${code}`)); + } + }); + child.on("error", reject); + }); +} + +type CountResult = { + count: number | null; + error: { message?: string } | null; +}; + +async function safeCount(label: string, countPromise: PromiseLike) { + try { + const result = await countPromise; + if (result.error) { + return { label, count: null, error: result.error.message ?? "Supabase query failed." }; + } + return { label, count: result.count ?? 0, error: null }; + } catch (error) { + return { label, count: null, error: error instanceof Error ? error.message : String(error) }; + } +} + +async function main() { + const [ + { env, requireServerEnv }, + { buildIngestionRecoveryPlan, isFreshProcessingJob }, + { hasIncompleteDocumentsWithoutOpenJobs, isReindexQueueClear }, + { createAdminClient }, + { assertSupabaseHealthy, probeSupabaseHealth }, + ] = await Promise.all([ + import("@/lib/env"), + import("@/lib/ingestion-recovery"), + import("@/lib/reindex-pipeline"), + import("@/lib/supabase/admin"), + import("@/lib/supabase/health"), + ]); + requireServerEnv(); + + const args = parseArgs(process.argv.slice(2)); + const staleAfterMinutes = env.WORKER_STALE_AFTER_MINUTES; + const limit = args.limit; + + console.log("=== Reindex Pipeline ===\n"); + + // Step 1 – Confirm target project + const { checkSupabaseProjectConfig, expectedSupabaseProject, formatSupabaseProjectCheck } = await import( + "@/lib/supabase/project" + ); + const projectCheck = checkSupabaseProjectConfig({ + NEXT_PUBLIC_SUPABASE_URL: process.env.NEXT_PUBLIC_SUPABASE_URL, + SUPABASE_PROJECT_REF: process.env.SUPABASE_PROJECT_REF, + SUPABASE_PROJECT_NAME: process.env.SUPABASE_PROJECT_NAME, + }); + if (projectCheck.status === "missing" || projectCheck.status === "mismatch") { + console.error(`Project check failed: ${formatSupabaseProjectCheck(projectCheck)}`); + process.exitCode = 1; + return; + } + console.log(`Project : ${expectedSupabaseProject.name} (${expectedSupabaseProject.ref})`); + + // Step 2 – Supabase health + console.log("\n[Step 1] Supabase health check..."); + const supabase = createAdminClient(); + assertSupabaseHealthy(await probeSupabaseHealth(supabase), "Reindex pipeline"); + console.log(" Supabase is healthy."); + + // Steps 3–5 – Iterative health snapshot → recovery → worker + for (let round = 1; round <= args.maxRounds; round++) { + console.log(`\n[Step 2] Reindex health snapshot (round ${round}/${args.maxRounds})...`); + + const snapshotNow = new Date(); + const staleCutoff = new Date(snapshotNow.getTime() - staleAfterMinutes * 60_000).toISOString(); + const [ + pendingJobs, + processingJobs, + failedJobs, + staleJobs, + queuedDocuments, + processingDocuments, + failedDocuments, + indexedDocuments, + ] = + await Promise.all([ + safeCount( + "jobs_pending", + supabase.from("ingestion_jobs").select("id", { count: "exact", head: true }).eq("status", "pending"), + ), + safeCount( + "jobs_processing", + supabase.from("ingestion_jobs").select("id", { count: "exact", head: true }).eq("status", "processing"), + ), + safeCount( + "jobs_failed", + supabase.from("ingestion_jobs").select("id", { count: "exact", head: true }).eq("status", "failed"), + ), + safeCount( + "jobs_stale", + supabase + .from("ingestion_jobs") + .select("id", { count: "exact", head: true }) + .eq("status", "processing") + .or(`locked_at.is.null,locked_at.lt.${staleCutoff}`), + ), + safeCount( + "documents_queued", + supabase.from("documents").select("id", { count: "exact", head: true }).eq("status", "queued"), + ), + safeCount( + "documents_processing", + supabase.from("documents").select("id", { count: "exact", head: true }).eq("status", "processing"), + ), + safeCount( + "documents_failed", + supabase.from("documents").select("id", { count: "exact", head: true }).eq("status", "failed"), + ), + safeCount( + "documents_indexed", + supabase.from("documents").select("id", { count: "exact", head: true }).eq("status", "indexed"), + ), + ]); + + const countItems = [ + pendingJobs, + processingJobs, + failedJobs, + staleJobs, + queuedDocuments, + processingDocuments, + failedDocuments, + indexedDocuments, + ]; + const countErrors = countItems.filter((item) => item.error); + + console.log(` Documents indexed : ${indexedDocuments.count ?? "-"}`); + console.log(` Documents queued : ${queuedDocuments.count ?? "-"}`); + console.log(` Documents processing: ${processingDocuments.count ?? "-"}`); + console.log(` Documents failed : ${failedDocuments.count ?? "-"}`); + console.log(` Jobs pending : ${pendingJobs.count ?? "-"}`); + console.log(` Jobs processing : ${processingJobs.count ?? "-"}`); + console.log(` Jobs failed : ${failedJobs.count ?? "-"}`); + console.log(` Jobs stale : ${staleJobs.count ?? "-"}`); + + if (countErrors.length > 0) { + console.error("\n Read errors:"); + for (const item of countErrors) { + console.error(` ${item.label}: ${item.error}`); + } + console.error("\n Fix the reported read errors before running workers or evals."); + process.exitCode = 1; + return; + } + + const openJobs = (pendingJobs.count ?? 0) + (processingJobs.count ?? 0) + (failedJobs.count ?? 0); + const needsRecovery = (staleJobs.count ?? 0) > 0 || (failedJobs.count ?? 0) > 0; + let skipWorkerRun = false; + let hasActiveProcessingJobs = !needsRecovery && (processingJobs.count ?? 0) > 0; + const queueSnapshot = { + openJobs, + queuedDocuments: queuedDocuments.count ?? 0, + processingDocuments: processingDocuments.count ?? 0, + failedDocuments: failedDocuments.count ?? 0, + }; + + if (isReindexQueueClear(queueSnapshot)) { + console.log("\nQueue is clear. Reindex pipeline complete."); + return; + } + + if (hasIncompleteDocumentsWithoutOpenJobs(queueSnapshot)) { + console.error( + "\nNo open ingestion jobs remain, but some documents are still processing or failed. Run targeted recovery or manual inspection before declaring the corpus reindexed.", + ); + process.exitCode = 1; + return; + } + + // Step 3 – Recovery (only if stale/failed jobs present) + if (needsRecovery) { + console.log(`\n[Step 3] Queue recovery (stale: ${staleJobs.count}, failed: ${failedJobs.count})...`); + + const { data, error } = await supabase + .from("ingestion_jobs") + .select("id,document_id,status,locked_at,documents(status,page_count,chunk_count)") + .in("status", ["processing", "failed"]) + .order("created_at", { ascending: true }); + + if (error) { + console.error(` Failed to load jobs for recovery: ${error.message}`); + process.exitCode = 1; + return; + } + + type RecoveryDocument = { status?: string | null; page_count?: number | null; chunk_count?: number | null }; + type RawJobRow = { + id: string; + document_id: string; + status: string | null; + locked_at: string | null; + documents: RecoveryDocument | RecoveryDocument[] | null; + }; + const staleAfterCutoff = new Date(staleCutoff); + const jobs = (data ?? []) + .map((job: RawJobRow) => ({ + ...job, + documents: Array.isArray(job.documents) + ? (job.documents[0] as RecoveryDocument | undefined) + : (job.documents as RecoveryDocument | undefined), + })) + .filter((job) => { + if (job.status === "failed") { + return true; + } + if (job.status !== "processing") { + return false; + } + if (job.locked_at === null) { + return true; + } + return new Date(job.locked_at) <= staleAfterCutoff; + }); + hasActiveProcessingJobs = (data ?? []) + .map((job: RawJobRow) => ({ + ...job, + documents: Array.isArray(job.documents) + ? (job.documents[0] as RecoveryDocument | undefined) + : (job.documents as RecoveryDocument | undefined), + })) + .some((job) => isFreshProcessingJob(job, snapshotNow, staleAfterMinutes)); + const plan = buildIngestionRecoveryPlan({ jobs, staleAfterMinutes }); + const actions = plan.actions.slice(0, limit); + const resetDocumentIds = Array.from( + new Set(actions.filter((a) => a.action === "retry").map((a) => a.documentId)), + ); + const supersedeCount = actions.filter((a) => a.action === "supersede").length; + const retryCount = actions.filter((a) => a.action === "retry").length; + + console.log(` Documents to reset: ${resetDocumentIds.length}`); + console.log(` Jobs to supersede : ${supersedeCount}`); + console.log(` Jobs to retry : ${retryCount}`); + + if (actions.length === 0) { + console.log(" Nothing to recover in this round."); + skipWorkerRun = true; + } else { + let shouldApply = args.yes; + if (!shouldApply) { + console.log(""); + shouldApply = await confirm(" Apply recovery changes?"); + } + + if (!shouldApply) { + console.log(" Recovery skipped. Re-checking queue before worker run."); + skipWorkerRun = true; + } else { + for (const documentId of resetDocumentIds) { + const { error: resetError } = await supabase.rpc("reset_document_index", { + p_document_id: documentId, + }); + if (resetError) { + console.error(` Failed to reset document index ${documentId}: ${resetError.message}`); + process.exitCode = 1; + return; + } + const { error: docError } = await supabase + .from("documents") + .update({ status: "queued", error_message: null, page_count: 0, chunk_count: 0, image_count: 0 }) + .eq("id", documentId); + if (docError) { + console.error(` Failed to reset document status ${documentId}: ${docError.message}`); + process.exitCode = 1; + return; + } + } + + for (const action of actions) { + if (action.action === "supersede") { + const { error: supersedeError } = await supabase + .from("ingestion_jobs") + .update({ + status: "completed", + stage: "superseded by successful index", + progress: 100, + error_message: null, + locked_at: null, + locked_by: null, + completed_at: new Date().toISOString(), + }) + .eq("id", action.jobId); + if (supersedeError) { + console.error(` Failed to supersede job ${action.jobId}: ${supersedeError.message}`); + process.exitCode = 1; + return; + } + continue; + } + + const { error: retryError } = await supabase + .from("ingestion_jobs") + .update({ + status: "pending", + stage: "queued after recovery", + progress: 0, + attempt_count: 0, + error_message: null, + locked_at: null, + locked_by: null, + next_run_at: new Date().toISOString(), + completed_at: null, + }) + .eq("id", action.jobId); + if (retryError) { + console.error(` Failed to requeue job ${action.jobId}: ${retryError.message}`); + process.exitCode = 1; + return; + } + } + + console.log(" Recovery applied."); + } + } + } + + // Step 4 – Worker run + if (skipWorkerRun) { + continue; + } + + if (hasActiveProcessingJobs) { + console.log( + "\n Active processing jobs detected; skipping worker run to avoid concurrency spikes.", + ); + continue; + } + + console.log("\n[Step 4] Running worker:once..."); + try { + await runWorkerOnce(); + console.log(" Worker run complete."); + } catch (err) { + console.error(` Worker run failed: ${err instanceof Error ? err.message : String(err)}`); + process.exitCode = 1; + return; + } + } + + console.log(`\nReached maximum rounds (${args.maxRounds}). Queue did not clear; re-run with a higher --max-rounds or inspect stale jobs first.`); + process.exitCode = 1; +} + +main().catch((error) => { + import("@/lib/privacy") + .then(({ safeErrorLogDetails }) => { + console.error("Reindex pipeline failed", safeErrorLogDetails(error)); + process.exitCode = 1; + }) + .catch(() => { + console.error(error instanceof Error ? error.message : error); + process.exitCode = 1; + }); +}); diff --git a/scripts/supabase-recovery-status.ts b/scripts/supabase-recovery-status.ts index 34bf03a03..efbc2bd40 100644 --- a/scripts/supabase-recovery-status.ts +++ b/scripts/supabase-recovery-status.ts @@ -24,22 +24,17 @@ async function main() { ]); requireServerEnv(); const supabase = createAdminClient(); + + console.log("=== Supabase Recovery Status ==="); const health = await probeSupabaseHealth(supabase); if (!health.ok) { + console.log(`Status : UNAVAILABLE`); + console.log(`Checked at: ${health.checkedAt}`); + console.log(`Error : ${health.message}`); console.log( - JSON.stringify( - { - ok: false, - status: "supabase_unavailable", - checkedAt: health.checkedAt, - error: health.message, - recommendation: - "Do not run migrations, imports, workers, recovery mutations, or evals. Retry later or contact Supabase support if this persists for more than 30 minutes with local workers stopped.", - }, - null, - 2, - ), + "\nDo not run migrations, imports, workers, recovery mutations, or evals. " + + "Retry later or contact Supabase support if this persists for more than 30 minutes with local workers stopped.", ); process.exitCode = 1; return; @@ -85,23 +80,27 @@ async function main() { : openJobs === 0 ? "Queue is clear. Run indexing checks or resume imports in small waves." : (staleJobs.count ?? 0) > 0 || (failedJobs.count ?? 0) > 0 - ? "Run npm run recover:ingestion -- --apply --limit 20, then npm run worker:once." + ? "Run npm run recover:ingestion, then npm run worker:once." : "Run npm run worker:once with conservative defaults."; - console.log( - JSON.stringify( - { - ok: errors.length === 0, - status: errors.length ? "read_errors" : "ready", - checkedAt: health.checkedAt, - counts: Object.fromEntries(counts.map((item) => [item.label, item.count])), - errors, - recommendation, - }, - null, - 2, - ), - ); + const status = errors.length ? "READ ERRORS" : "READY"; + console.log(`Status : ${status}`); + console.log(`Checked at: ${health.checkedAt}`); + console.log(""); + console.log("Queue counts:"); + for (const item of counts) { + const value = item.error ? `ERROR: ${item.error}` : String(item.count ?? "-"); + console.log(` ${item.label.padEnd(24)}: ${value}`); + } + + if (errors.length > 0) { + console.log("\nRead errors:"); + for (const item of errors) { + console.log(` ${item.label}: ${item.error}`); + } + } + + console.log(`\nRecommendation: ${recommendation}`); if (errors.length) process.exitCode = 1; } diff --git a/src/lib/ingestion-recovery.ts b/src/lib/ingestion-recovery.ts index 989bc9d40..e2498ffd4 100644 --- a/src/lib/ingestion-recovery.ts +++ b/src/lib/ingestion-recovery.ts @@ -14,11 +14,31 @@ export type IngestionRecoveryAction = | { action: "supersede"; jobId: string; documentId: string } | { action: "retry"; jobId: string; documentId: string; resetDocument: boolean }; +function parseLockedAt(value: string | null | undefined) { + if (!value) return null; + const time = Date.parse(value); + return Number.isFinite(time) ? time : null; +} + export function isStaleProcessingJob(job: IngestionRecoveryJob, now: Date, staleAfterMinutes: number) { if (job.status !== "processing" || !job.locked_at) return false; - const lockedAt = new Date(job.locked_at); - if (Number.isNaN(lockedAt.getTime())) return false; - return lockedAt.getTime() < now.getTime() - staleAfterMinutes * 60_000; + const lockedAt = parseLockedAt(job.locked_at); + if (lockedAt === null) return false; + return lockedAt < now.getTime() - staleAfterMinutes * 60_000; +} + +export function isRecoverableProcessingJob(job: IngestionRecoveryJob, now: Date, staleAfterMinutes: number) { + if (job.status !== "processing") return false; + const lockedAt = parseLockedAt(job.locked_at); + if (lockedAt === null) return true; + return lockedAt < now.getTime() - staleAfterMinutes * 60_000; +} + +export function isFreshProcessingJob(job: IngestionRecoveryJob, now: Date, staleAfterMinutes: number) { + if (job.status !== "processing" || !job.locked_at) return false; + const lockedAt = parseLockedAt(job.locked_at); + if (lockedAt === null) return false; + return lockedAt >= now.getTime() - staleAfterMinutes * 60_000; } export function buildIngestionRecoveryPlan(args: { @@ -35,7 +55,9 @@ export function buildIngestionRecoveryPlan(args: { const chunkCount = Number(job.documents?.chunk_count ?? 0); const isIndexedDocument = documentStatus === "indexed" && chunkCount > 0; const isRecoverableStatus = - job.status === "failed" || job.status === "pending" || isStaleProcessingJob(job, now, args.staleAfterMinutes); + job.status === "failed" || + job.status === "pending" || + isRecoverableProcessingJob(job, now, args.staleAfterMinutes); if (isIndexedDocument && job.status !== "completed") { actions.push({ action: "supersede", jobId: job.id, documentId: job.document_id }); diff --git a/src/lib/reindex-pipeline.ts b/src/lib/reindex-pipeline.ts new file mode 100644 index 000000000..1a10710d6 --- /dev/null +++ b/src/lib/reindex-pipeline.ts @@ -0,0 +1,23 @@ +export type ReindexQueueSnapshot = { + openJobs: number; + queuedDocuments: number; + processingDocuments: number; + failedDocuments: number; +}; + +export function isReindexQueueClear(snapshot: ReindexQueueSnapshot) { + return ( + snapshot.openJobs === 0 && + snapshot.queuedDocuments === 0 && + snapshot.processingDocuments === 0 && + snapshot.failedDocuments === 0 + ); +} + +export function hasIncompleteDocumentsWithoutOpenJobs(snapshot: ReindexQueueSnapshot) { + return ( + snapshot.openJobs === 0 && + snapshot.queuedDocuments === 0 && + (snapshot.processingDocuments > 0 || snapshot.failedDocuments > 0) + ); +} diff --git a/tests/deep-memory.test.ts b/tests/deep-memory.test.ts index b86b98eeb..9f5d2fca2 100644 --- a/tests/deep-memory.test.ts +++ b/tests/deep-memory.test.ts @@ -259,8 +259,11 @@ describe("deep RAG memory indexing", () => { insertedMemoryRows.push(...payload); return Promise.resolve({ data: null, error: null }); }, - select: () => ({ - eq: () => { + select: (columns: string) => ({ + eq: (column: string, value: unknown) => { + void columns; + void column; + void value; if (table === "documents") { return { single: async () => ({ data: { metadata: {} }, error: null }), @@ -278,7 +281,9 @@ describe("deep RAG memory indexing", () => { }, }), update: (payload: Record) => ({ - eq: () => { + eq: (column: string, value: unknown) => { + void column; + void value; updatedRows.set(table, [...(updatedRows.get(table) ?? []), payload]); return Promise.resolve({ data: null, error: null }); }, @@ -319,3 +324,4 @@ describe("deep RAG memory indexing", () => { ); }); }); + diff --git a/tests/ingestion-recovery.test.ts b/tests/ingestion-recovery.test.ts index 95d7e63f1..349713925 100644 --- a/tests/ingestion-recovery.test.ts +++ b/tests/ingestion-recovery.test.ts @@ -1,5 +1,10 @@ import { describe, expect, it } from "vitest"; -import { buildIngestionRecoveryPlan, isStaleProcessingJob } from "../src/lib/ingestion-recovery"; +import { + buildIngestionRecoveryPlan, + isFreshProcessingJob, + isRecoverableProcessingJob, + isStaleProcessingJob, +} from "../src/lib/ingestion-recovery"; describe("ingestion queue recovery planning", () => { const now = new Date("2026-06-15T00:00:00.000Z"); @@ -62,4 +67,35 @@ describe("ingestion queue recovery planning", () => { ), ).toBe(false); }); + + it("retries processing jobs with no lock timestamp", () => { + const plan = buildIngestionRecoveryPlan({ + now, + staleAfterMinutes: 45, + jobs: [ + { + id: "null-lock", + document_id: "doc-null", + status: "processing", + locked_at: null, + documents: { status: "processing", chunk_count: 0 }, + }, + ], + }); + + expect(plan.retryCount).toBe(1); + expect(plan.actions[0]).toMatchObject({ action: "retry", jobId: "null-lock", documentId: "doc-null" }); + }); + + it("treats fresh processing jobs as active but not recoverable", () => { + const job = { + id: "fresh", + document_id: "doc", + status: "processing" as const, + locked_at: "2026-06-14T23:30:00.000Z", + }; + + expect(isRecoverableProcessingJob(job, now, 45)).toBe(false); + expect(isFreshProcessingJob(job, now, 45)).toBe(true); + }); }); diff --git a/tests/reindex-pipeline.test.ts b/tests/reindex-pipeline.test.ts new file mode 100644 index 000000000..b76d2791c --- /dev/null +++ b/tests/reindex-pipeline.test.ts @@ -0,0 +1,45 @@ +import { describe, expect, it } from "vitest"; +import { hasIncompleteDocumentsWithoutOpenJobs, isReindexQueueClear } from "../src/lib/reindex-pipeline"; + +describe("reindex pipeline queue state", () => { + it("does not declare the queue clear while documents are still processing", () => { + expect( + isReindexQueueClear({ + openJobs: 0, + queuedDocuments: 0, + processingDocuments: 1, + failedDocuments: 0, + }), + ).toBe(false); + }); + + it("flags orphaned incomplete documents when no jobs remain", () => { + expect( + hasIncompleteDocumentsWithoutOpenJobs({ + openJobs: 0, + queuedDocuments: 0, + processingDocuments: 1, + failedDocuments: 0, + }), + ).toBe(true); + expect( + hasIncompleteDocumentsWithoutOpenJobs({ + openJobs: 0, + queuedDocuments: 1, + processingDocuments: 0, + failedDocuments: 1, + }), + ).toBe(false); + }); + + it("declares the queue clear only when no open jobs or incomplete documents remain", () => { + expect( + isReindexQueueClear({ + openJobs: 0, + queuedDocuments: 0, + processingDocuments: 0, + failedDocuments: 0, + }), + ).toBe(true); + }); +});