Add M2 atomic reindex staged generation recovery

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",

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;
+});

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;
+}

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;