From e6ac5106231cb45a66f3a9282dbfa0380cfbe1d0 Mon Sep 17 00:00:00 2001 From: BigSimmo <87357024+BigSimmo@users.noreply.github.com> Date: Mon, 29 Jun 2026 08:33:28 +0800 Subject: [PATCH] feat: add atomic reindex staged generation recovery --- package.json | 1 + .../cleanup-abandoned-reindex-generations.ts | 108 ++++++++ src/lib/reindex-pipeline.ts | 31 +++ ..._abandoned_reindex_generation_recovery.sql | 234 ++++++++++++++++++ supabase/schema.sql | 234 ++++++++++++++++++ tests/reindex-pipeline.test.ts | 36 +++ tests/supabase-schema.test.ts | 23 ++ 7 files changed, 667 insertions(+) create mode 100644 scripts/cleanup-abandoned-reindex-generations.ts create mode 100644 supabase/migrations/20260629000000_abandoned_reindex_generation_recovery.sql diff --git a/package.json b/package.json index 65b80a2195..722da7b60c 100644 --- a/package.json +++ b/package.json @@ -47,6 +47,7 @@ "recover:ingestion": "tsx scripts/recover-ingestion-queue.ts", "reindex": "tsx scripts/reindex.ts", "reindex:health": "tsx scripts/reindex-health.ts", + "reindex:cleanup-staged": "tsx scripts/cleanup-abandoned-reindex-generations.ts", "supabase:recovery-status": "tsx scripts/supabase-recovery-status.ts", "promote:query-misses": "tsx scripts/promote-query-misses.ts", "eval:rag": "tsx scripts/eval-rag.ts", diff --git a/scripts/cleanup-abandoned-reindex-generations.ts b/scripts/cleanup-abandoned-reindex-generations.ts new file mode 100644 index 0000000000..a50f779587 --- /dev/null +++ b/scripts/cleanup-abandoned-reindex-generations.ts @@ -0,0 +1,108 @@ +import { loadEnvConfig } from "@next/env"; +import { + abandonedReindexGenerationTotal, + hasAbandonedReindexGenerations, + type AbandonedReindexGenerationCounts, +} from "@/lib/reindex-pipeline"; +import { safeErrorLogDetails } from "@/lib/privacy"; +import { assertSupabaseHealthy, probeSupabaseHealth } from "@/lib/supabase/health"; +import { confirm } from "./cli-utils"; + +loadEnvConfig(process.cwd()); + +type CleanupResult = { + ok?: boolean; + dry_run?: boolean; + document_count?: number; + document_ids?: string[]; + counts?: AbandonedReindexGenerationCounts; +}; + +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 { + apply: argv.includes("--apply"), + yes: argv.includes("--yes"), + documentId: valueFor("document-id") ?? null, + limit: Number.parseInt(valueFor("limit") ?? "", 10), + }; +} + +function formatCounts(counts: AbandonedReindexGenerationCounts) { + const entries = Object.entries(counts).sort(([a], [b]) => a.localeCompare(b)); + if (entries.length === 0) return " none"; + return entries.map(([table, count]) => ` ${table.padEnd(26)}: ${count}`).join("\n"); +} + +async function main() { + const [{ env, requireServerEnv }, { createAdminClient }] = await Promise.all([ + import("@/lib/env"), + import("@/lib/supabase/admin"), + ]); + requireServerEnv(); + + const args = parseArgs(process.argv.slice(2)); + const limit = Number.isFinite(args.limit) ? args.limit : 100; + const supabase = createAdminClient(); + + console.log("=== Abandoned Reindex Generation Cleanup ==="); + console.log(`Supabase project: ${env.SUPABASE_PROJECT_NAME ?? "unknown"} (${env.SUPABASE_PROJECT_REF ?? "unknown"})`); + console.log(`Mode : ${args.apply ? "apply" : "dry-run"}`); + console.log(`Document filter : ${args.documentId ?? "all eligible documents"}`); + console.log(`Document limit : ${limit}`); + console.log(""); + + assertSupabaseHealthy(await probeSupabaseHealth(supabase), "Abandoned reindex generation cleanup"); + + const { data, error } = await supabase.rpc("cleanup_abandoned_document_index_generations", { + p_document_id: args.documentId, + p_limit: limit, + p_dry_run: true, + }); + if (error) throw new Error(error.message); + + const result = (data ?? {}) as CleanupResult; + const counts = result.counts ?? {}; + const total = abandonedReindexGenerationTotal(counts); + console.log(`Eligible documents: ${result.document_count ?? 0}`); + console.log(`Artifact rows : ${total}`); + console.log("Rows by table:"); + console.log(formatCounts(counts)); + + if (!hasAbandonedReindexGenerations(counts)) { + console.log("\nNo abandoned staged generation rows found."); + return; + } + + if (!args.apply) { + console.log("\nDry run only. Re-run with --apply to delete these abandoned staged rows."); + return; + } + + if (!args.yes) { + const shouldApply = await confirm("Delete the abandoned staged generation rows listed above?"); + if (!shouldApply) { + console.log("\nNo changes applied."); + return; + } + } + + const applied = await supabase.rpc("cleanup_abandoned_document_index_generations", { + p_document_id: args.documentId, + p_limit: limit, + p_dry_run: false, + }); + if (applied.error) throw new Error(applied.error.message); + + console.log("\nAbandoned staged generation cleanup applied."); +} + +main().catch((error) => { + console.error("Abandoned reindex generation cleanup failed", safeErrorLogDetails(error)); + process.exitCode = 1; +}); diff --git a/src/lib/reindex-pipeline.ts b/src/lib/reindex-pipeline.ts index f533434530..62171eb3f8 100644 --- a/src/lib/reindex-pipeline.ts +++ b/src/lib/reindex-pipeline.ts @@ -5,6 +5,16 @@ export type ReindexQueueSnapshot = { failedDocuments: number; }; +export type AbandonedReindexGenerationCounts = { + document_chunks?: number; + document_images?: number; + document_table_facts?: number; + document_embedding_fields?: number; + document_index_units?: number; + document_memory_cards?: number; + document_sections?: number; +}; + export function isReindexQueueClear(snapshot: ReindexQueueSnapshot) { return ( snapshot.openJobs === 0 && @@ -45,3 +55,24 @@ export function isCommittedGenerationMetadata(args: { if (!rowGeneration) return true; return Boolean(args.committedGeneration) && rowGeneration === args.committedGeneration; } + +export function isAbandonedStagedGeneration(args: { + rowMetadata?: unknown; + rowGenerationId?: string | null; + committedGeneration?: string | null; +}) { + const rowGeneration = + typeof args.rowGenerationId === "string" && args.rowGenerationId.trim() + ? args.rowGenerationId.trim() + : committedIndexGeneration(args.rowMetadata); + if (!rowGeneration) return false; + return rowGeneration !== (args.committedGeneration ?? null); +} + +export function abandonedReindexGenerationTotal(counts: AbandonedReindexGenerationCounts) { + return Object.values(counts).reduce((total, value) => total + (Number.isFinite(value) ? Number(value) : 0), 0); +} + +export function hasAbandonedReindexGenerations(counts: AbandonedReindexGenerationCounts) { + return abandonedReindexGenerationTotal(counts) > 0; +} diff --git a/supabase/migrations/20260629000000_abandoned_reindex_generation_recovery.sql b/supabase/migrations/20260629000000_abandoned_reindex_generation_recovery.sql new file mode 100644 index 0000000000..163870d508 --- /dev/null +++ b/supabase/migrations/20260629000000_abandoned_reindex_generation_recovery.sql @@ -0,0 +1,234 @@ +create or replace function public.cleanup_abandoned_document_index_generations( + p_document_id uuid default null, + p_limit integer default 100, + p_dry_run boolean default true +) +returns jsonb +language plpgsql +set search_path = public, extensions, pg_temp +as $$ +declare + target_document_ids uuid[] := '{}'::uuid[]; + chunk_count integer := 0; + image_count integer := 0; + table_fact_count integer := 0; + embedding_field_count integer := 0; + index_unit_count integer := 0; + memory_card_count integer := 0; + section_count integer := 0; +begin + perform set_config('statement_timeout', '180000', true); + + with candidate_documents as ( + select distinct document_id + from ( + select c.document_id + from public.document_chunks c + join public.documents d on d.id = c.document_id + where (p_document_id is null or c.document_id = p_document_id) + and c.index_generation_id is not null + and c.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and not exists ( + select 1 from public.ingestion_jobs j + where j.document_id = c.document_id + and j.status in ('pending', 'processing') + ) + union all + select a.document_id + from public.document_images a + join public.documents d on d.id = a.document_id + where (p_document_id is null or a.document_id = p_document_id) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and not exists ( + select 1 from public.ingestion_jobs j + where j.document_id = a.document_id + and j.status in ('pending', 'processing') + ) + union all + select a.document_id + from public.document_table_facts a + join public.documents d on d.id = a.document_id + where (p_document_id is null or a.document_id = p_document_id) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and not exists ( + select 1 from public.ingestion_jobs j + where j.document_id = a.document_id + and j.status in ('pending', 'processing') + ) + union all + select a.document_id + from public.document_embedding_fields a + join public.documents d on d.id = a.document_id + where (p_document_id is null or a.document_id = p_document_id) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and not exists ( + select 1 from public.ingestion_jobs j + where j.document_id = a.document_id + and j.status in ('pending', 'processing') + ) + union all + select a.document_id + from public.document_index_units a + join public.documents d on d.id = a.document_id + where (p_document_id is null or a.document_id = p_document_id) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and not exists ( + select 1 from public.ingestion_jobs j + where j.document_id = a.document_id + and j.status in ('pending', 'processing') + ) + union all + select a.document_id + from public.document_memory_cards a + join public.documents d on d.id = a.document_id + where (p_document_id is null or a.document_id = p_document_id) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and not exists ( + select 1 from public.ingestion_jobs j + where j.document_id = a.document_id + and j.status in ('pending', 'processing') + ) + union all + select a.document_id + from public.document_sections a + join public.documents d on d.id = a.document_id + where (p_document_id is null or a.document_id = p_document_id) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and not exists ( + select 1 from public.ingestion_jobs j + where j.document_id = a.document_id + and j.status in ('pending', 'processing') + ) + ) candidates + limit least(greatest(coalesce(p_limit, 100), 1), 1000) + ) + select coalesce(array_agg(document_id), '{}'::uuid[]) + into target_document_ids + from candidate_documents; + + select count(*) into chunk_count + from public.document_chunks c + join public.documents d on d.id = c.document_id + where c.document_id = any(target_document_ids) + and c.index_generation_id is not null + and c.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + select count(*) into image_count + from public.document_images a + join public.documents d on d.id = a.document_id + where a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + select count(*) into table_fact_count + from public.document_table_facts a + join public.documents d on d.id = a.document_id + where a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + select count(*) into embedding_field_count + from public.document_embedding_fields a + join public.documents d on d.id = a.document_id + where a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + select count(*) into index_unit_count + from public.document_index_units a + join public.documents d on d.id = a.document_id + where a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + select count(*) into memory_card_count + from public.document_memory_cards a + join public.documents d on d.id = a.document_id + where a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + select count(*) into section_count + from public.document_sections a + join public.documents d on d.id = a.document_id + where a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + if not coalesce(p_dry_run, true) then + delete from public.document_chunks c + using public.documents d + where d.id = c.document_id + and c.document_id = any(target_document_ids) + and c.index_generation_id is not null + and c.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + delete from public.document_images a + using public.documents d + where d.id = a.document_id + and a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + delete from public.document_table_facts a + using public.documents d + where d.id = a.document_id + and a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + delete from public.document_embedding_fields a + using public.documents d + where d.id = a.document_id + and a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + delete from public.document_index_units a + using public.documents d + where d.id = a.document_id + and a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + delete from public.document_memory_cards a + using public.documents d + where d.id = a.document_id + and a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + delete from public.document_sections a + using public.documents d + where d.id = a.document_id + and a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + end if; + + return jsonb_build_object( + 'ok', true, + 'dry_run', coalesce(p_dry_run, true), + 'document_count', coalesce(array_length(target_document_ids, 1), 0), + 'document_ids', to_jsonb(target_document_ids), + 'counts', jsonb_build_object( + 'document_chunks', chunk_count, + 'document_images', image_count, + 'document_table_facts', table_fact_count, + 'document_embedding_fields', embedding_field_count, + 'document_index_units', index_unit_count, + 'document_memory_cards', memory_card_count, + 'document_sections', section_count + ) + ); +end; +$$; + +revoke execute on function public.cleanup_abandoned_document_index_generations(uuid, integer, boolean) from public, anon, authenticated; +grant execute on function public.cleanup_abandoned_document_index_generations(uuid, integer, boolean) to service_role; diff --git a/supabase/schema.sql b/supabase/schema.sql index 7882364edb..f397060912 100644 --- a/supabase/schema.sql +++ b/supabase/schema.sql @@ -1205,6 +1205,238 @@ begin end; $$; +create or replace function public.cleanup_abandoned_document_index_generations( + p_document_id uuid default null, + p_limit integer default 100, + p_dry_run boolean default true +) +returns jsonb +language plpgsql +set search_path = public, extensions, pg_temp +as $$ +declare + target_document_ids uuid[] := '{}'::uuid[]; + chunk_count integer := 0; + image_count integer := 0; + table_fact_count integer := 0; + embedding_field_count integer := 0; + index_unit_count integer := 0; + memory_card_count integer := 0; + section_count integer := 0; +begin + perform set_config('statement_timeout', '180000', true); + + with candidate_documents as ( + select distinct document_id + from ( + select c.document_id + from public.document_chunks c + join public.documents d on d.id = c.document_id + where (p_document_id is null or c.document_id = p_document_id) + and c.index_generation_id is not null + and c.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and not exists ( + select 1 from public.ingestion_jobs j + where j.document_id = c.document_id + and j.status in ('pending', 'processing') + ) + union all + select a.document_id + from public.document_images a + join public.documents d on d.id = a.document_id + where (p_document_id is null or a.document_id = p_document_id) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and not exists ( + select 1 from public.ingestion_jobs j + where j.document_id = a.document_id + and j.status in ('pending', 'processing') + ) + union all + select a.document_id + from public.document_table_facts a + join public.documents d on d.id = a.document_id + where (p_document_id is null or a.document_id = p_document_id) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and not exists ( + select 1 from public.ingestion_jobs j + where j.document_id = a.document_id + and j.status in ('pending', 'processing') + ) + union all + select a.document_id + from public.document_embedding_fields a + join public.documents d on d.id = a.document_id + where (p_document_id is null or a.document_id = p_document_id) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and not exists ( + select 1 from public.ingestion_jobs j + where j.document_id = a.document_id + and j.status in ('pending', 'processing') + ) + union all + select a.document_id + from public.document_index_units a + join public.documents d on d.id = a.document_id + where (p_document_id is null or a.document_id = p_document_id) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and not exists ( + select 1 from public.ingestion_jobs j + where j.document_id = a.document_id + and j.status in ('pending', 'processing') + ) + union all + select a.document_id + from public.document_memory_cards a + join public.documents d on d.id = a.document_id + where (p_document_id is null or a.document_id = p_document_id) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and not exists ( + select 1 from public.ingestion_jobs j + where j.document_id = a.document_id + and j.status in ('pending', 'processing') + ) + union all + select a.document_id + from public.document_sections a + join public.documents d on d.id = a.document_id + where (p_document_id is null or a.document_id = p_document_id) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', '') + and not exists ( + select 1 from public.ingestion_jobs j + where j.document_id = a.document_id + and j.status in ('pending', 'processing') + ) + ) candidates + limit least(greatest(coalesce(p_limit, 100), 1), 1000) + ) + select coalesce(array_agg(document_id), '{}'::uuid[]) + into target_document_ids + from candidate_documents; + + select count(*) into chunk_count + from public.document_chunks c + join public.documents d on d.id = c.document_id + where c.document_id = any(target_document_ids) + and c.index_generation_id is not null + and c.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + select count(*) into image_count + from public.document_images a + join public.documents d on d.id = a.document_id + where a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + select count(*) into table_fact_count + from public.document_table_facts a + join public.documents d on d.id = a.document_id + where a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + select count(*) into embedding_field_count + from public.document_embedding_fields a + join public.documents d on d.id = a.document_id + where a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + select count(*) into index_unit_count + from public.document_index_units a + join public.documents d on d.id = a.document_id + where a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + select count(*) into memory_card_count + from public.document_memory_cards a + join public.documents d on d.id = a.document_id + where a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + select count(*) into section_count + from public.document_sections a + join public.documents d on d.id = a.document_id + where a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + if not coalesce(p_dry_run, true) then + delete from public.document_chunks c + using public.documents d + where d.id = c.document_id + and c.document_id = any(target_document_ids) + and c.index_generation_id is not null + and c.index_generation_id::text is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + delete from public.document_images a + using public.documents d + where d.id = a.document_id + and a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + delete from public.document_table_facts a + using public.documents d + where d.id = a.document_id + and a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + delete from public.document_embedding_fields a + using public.documents d + where d.id = a.document_id + and a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + delete from public.document_index_units a + using public.documents d + where d.id = a.document_id + and a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + delete from public.document_memory_cards a + using public.documents d + where d.id = a.document_id + and a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + + delete from public.document_sections a + using public.documents d + where d.id = a.document_id + and a.document_id = any(target_document_ids) + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is not null + and nullif(coalesce(a.metadata, '{}'::jsonb)->>'index_generation_id', '') is distinct from nullif(coalesce(d.metadata, '{}'::jsonb)->>'index_generation_id', ''); + end if; + + return jsonb_build_object( + 'ok', true, + 'dry_run', coalesce(p_dry_run, true), + 'document_count', coalesce(array_length(target_document_ids, 1), 0), + 'document_ids', to_jsonb(target_document_ids), + 'counts', jsonb_build_object( + 'document_chunks', chunk_count, + 'document_images', image_count, + 'document_table_facts', table_fact_count, + 'document_embedding_fields', embedding_field_count, + 'document_index_units', index_unit_count, + 'document_memory_cards', memory_card_count, + 'document_sections', section_count + ) + ); +end; +$$; + create or replace function public.refresh_import_batch_status(p_batch_id uuid) returns jsonb language plpgsql @@ -3366,6 +3598,8 @@ grant select on table public.document_index_units to authenticated; grant execute on function public.match_document_index_units_hybrid(extensions.vector, text, integer, double precision, uuid[], uuid) to service_role; revoke execute on function public.commit_document_index_generation(uuid, uuid, text, integer, integer, integer, jsonb, jsonb, jsonb) from public, anon, authenticated; grant execute on function public.commit_document_index_generation(uuid, uuid, text, integer, integer, integer, jsonb, jsonb, jsonb) to service_role; +revoke execute on function public.cleanup_abandoned_document_index_generations(uuid, integer, boolean) from public, anon, authenticated; +grant execute on function public.cleanup_abandoned_document_index_generations(uuid, integer, boolean) to service_role; revoke execute on function public.is_committed_document_generation(uuid, jsonb) from public, anon, authenticated; grant execute on function public.is_committed_document_generation(uuid, jsonb) to service_role; revoke execute on function public.is_committed_artifact_generation(jsonb, jsonb) from public, anon, authenticated; diff --git a/tests/reindex-pipeline.test.ts b/tests/reindex-pipeline.test.ts index 2d63e22727..630ba559ea 100644 --- a/tests/reindex-pipeline.test.ts +++ b/tests/reindex-pipeline.test.ts @@ -1,6 +1,9 @@ import { describe, expect, it } from "vitest"; import { + abandonedReindexGenerationTotal, committedIndexGeneration, + hasAbandonedReindexGenerations, + isAbandonedStagedGeneration, hasIncompleteDocumentsWithoutOpenJobs, isAtomicReindexCandidate, isCommittedGenerationMetadata, @@ -74,4 +77,37 @@ describe("reindex pipeline queue state", () => { }), ).toBe(true); }); + + it("identifies abandoned staged generation rows without flagging legacy generationless rows", () => { + expect( + isAbandonedStagedGeneration({ + rowGenerationId: "generation-b", + committedGeneration: "generation-a", + }), + ).toBe(true); + expect( + isAbandonedStagedGeneration({ + rowMetadata: { index_generation_id: "generation-a" }, + committedGeneration: "generation-a", + }), + ).toBe(false); + expect( + isAbandonedStagedGeneration({ + rowMetadata: {}, + committedGeneration: "generation-a", + }), + ).toBe(false); + }); + + it("summarizes abandoned staged generation cleanup counts", () => { + expect( + abandonedReindexGenerationTotal({ + document_chunks: 2, + document_images: 1, + document_table_facts: 0, + }), + ).toBe(3); + expect(hasAbandonedReindexGenerations({ document_chunks: 0, document_images: 0 })).toBe(false); + expect(hasAbandonedReindexGenerations({ document_chunks: 0, document_images: 1 })).toBe(true); + }); }); diff --git a/tests/supabase-schema.test.ts b/tests/supabase-schema.test.ts index 1de2411fcc..c06b588b86 100644 --- a/tests/supabase-schema.test.ts +++ b/tests/supabase-schema.test.ts @@ -38,6 +38,10 @@ const atomicReindexMigration = readFileSync( new URL("../supabase/migrations/20260628000000_atomic_reindex_generation_commit.sql", import.meta.url), "utf8", ).replace(/\s+/g, " "); +const abandonedReindexRecoveryMigration = readFileSync( + new URL("../supabase/migrations/20260629000000_abandoned_reindex_generation_recovery.sql", import.meta.url), + "utf8", +).replace(/\s+/g, " "); function extractTextChunkFunction(sql: string) { const start = sql.indexOf("function public.match_document_chunks_text"); @@ -154,6 +158,25 @@ describe("Supabase schema Data API grants", () => { expect(atomicReindexMigration).toContain("atomic reindex patch did not match match_document_index_units_hybrid"); }); + it("can identify and clean abandoned staged reindex generations", () => { + for (const sql of [schema, abandonedReindexRecoveryMigration]) { + expect(sql).toContain("create or replace function public.cleanup_abandoned_document_index_generations"); + expect(sql).toContain("p_dry_run boolean default true"); + expect(sql).toContain("j.status in ('pending', 'processing')"); + expect(sql).toContain("c.index_generation_id is not null"); + expect(sql).toContain("metadata, '{}'::jsonb)->>'index_generation_id'"); + expect(sql).toContain("if not coalesce(p_dry_run, true) then"); + expect(sql).toContain("'document_chunks', chunk_count"); + expect(sql).toContain("'document_index_units', index_unit_count"); + expect(sql).toContain( + "revoke execute on function public.cleanup_abandoned_document_index_generations(uuid, integer, boolean) from public, anon, authenticated", + ); + expect(sql).toContain( + "grant execute on function public.cleanup_abandoned_document_index_generations(uuid, integer, boolean) to service_role", + ); + } + }); + it("keeps indexing-v3 enrichment claiming separate from raw ingestion jobs", () => { expect(schema).toContain("create table if not exists public.ingestion_job_stages"); expect(schema).toContain("drop constraint if exists ingestion_job_stages_job_id_fkey");