feat(pg-delta): retry catalog extractors when pg_get_*def returns NULL (#238)

Co-authored-by: Claude <noreply@anthropic.com>

Author: avallete

.changeset/feat-extract-retries.md

@@ -0,0 +1,15 @@
+---
+"@supabase/pg-delta": minor
+---
+
+feat(pg-delta): retry catalog extractors when `pg_get_*def()` returns NULL
+
+`pg_get_indexdef`, `pg_get_constraintdef`, `pg_get_viewdef`, `pg_get_triggerdef`, `pg_get_ruledef`, and `pg_get_functiondef` can transiently return NULL when the underlying catalog row is dropped concurrently or the catalog state is in flux. Previously such rows were dropped silently after one attempt; now extraction retries the affected query a configurable number of times before falling back to filtering. In practice the second attempt no longer sees the dropped object (or successfully resolves the definition), so a real CREATE/DROP racing with `createPlan` is reliably preserved or excluded rather than half-captured.
+
+Configuration (precedence: option > env > default):
+
+- `CreatePlanOptions.extractRetries?: number` — public API option on `createPlan`.
+- `PGDELTA_EXTRACT_RETRIES` env var — same value, useful for CLI usage.
+- Default `1` (i.e. the first attempt plus one retry, 2 attempts total).
+
+After retries are exhausted, rows whose `pg_get_*def()` is still NULL are filtered out and a warning is emitted via `debug('pg-delta:extract')` (visible with `DEBUG=pg-delta:extract` or `DEBUG=pg-delta:*`). Setting `extractRetries: 0` disables retrying entirely and reproduces the previous "filter-on-first-attempt" behavior.

.changeset/fix-extractor-null-definitions.md

@@ -0,0 +1,5 @@
+---
+"@supabase/pg-delta": patch
+---
+
+fix(pg-delta): skip rows when `pg_get_viewdef`, `pg_get_triggerdef`, `pg_get_ruledef`, or `pg_get_functiondef` returns NULL instead of crashing the relevant `extract*` with a ZodError. Same race conditions as the prior `pg_get_indexdef` (#223) and `pg_get_constraintdef` fixes — the underlying catalog row can vanish (concurrent DDL, transient catalog state, recovery edges). A single unreadable view, materialized view, trigger, rule, or function no longer aborts the whole catalog extraction and `createPlan` call.

.changeset/fix-table-constraint-null-definition.md

@@ -0,0 +1,5 @@
+---
+"@supabase/pg-delta": patch
+---
+
+fix(pg-delta): skip table constraints where `pg_get_constraintdef()` returns NULL instead of crashing `extractTables` with a ZodError. Like `pg_get_indexdef`, `pg_get_constraintdef` can return NULL under race conditions with concurrent DDL or transient catalog inconsistencies. Such constraints are now filtered out at extraction time so a single unreadable constraint no longer aborts the whole catalog extraction and `createPlan` call.

packages/pg-delta/src/core/catalog.model.ts

@@ -302,7 +302,19 @@ export async function createEmptyCatalog(
   });
 }
 
-export async function extractCatalog(pool: Pool) {
+interface ExtractCatalogOptions {
+  /**
+   * Number of retry attempts for catalog extractors when `pg_get_*def()`
+   * returns NULL for at least one row. See `ExtractRetryOptions.retries`.
+   */
+  extractRetries?: number;
+}
+
+export async function extractCatalog(
+  pool: Pool,
+  options: ExtractCatalogOptions = {},
+) {
+  const retryOptions = { retries: options.extractRetries };
   const [
     aggregates,
     collations,
@@ -339,21 +351,21 @@ export async function extractCatalog(pool: Pool) {
     extractDomains(pool).then(listToRecord),
     extractEnums(pool).then(listToRecord),
     extractExtensions(pool).then(listToRecord),
-    extractIndexes(pool).then(listToRecord),
-    extractMaterializedViews(pool).then(listToRecord),
+    extractIndexes(pool, retryOptions).then(listToRecord),
+    extractMaterializedViews(pool, retryOptions).then(listToRecord),
     extractSubscriptions(pool).then(listToRecord),
     extractPublications(pool).then(listToRecord),
-    extractProcedures(pool).then(listToRecord),
+    extractProcedures(pool, retryOptions).then(listToRecord),
     extractRlsPolicies(pool).then(listToRecord),
     extractRoles(pool).then(listToRecord),
     extractSchemas(pool).then(listToRecord),
     extractSequences(pool).then(listToRecord),
-    extractTables(pool).then(listToRecord),
-    extractTriggers(pool).then(listToRecord),
+    extractTables(pool, retryOptions).then(listToRecord),
+    extractTriggers(pool, retryOptions).then(listToRecord),
     extractEventTriggers(pool).then(listToRecord),
-    extractRules(pool).then(listToRecord),
+    extractRules(pool, retryOptions).then(listToRecord),
     extractRanges(pool).then(listToRecord),
-    extractViews(pool).then(listToRecord),
+    extractViews(pool, retryOptions).then(listToRecord),
     extractForeignDataWrappers(pool).then(listToRecord),
     extractServers(pool).then(listToRecord),
     extractUserMappings(pool).then(listToRecord),

packages/pg-delta/src/core/objects/extract-with-retry.test.ts

@@ -0,0 +1,143 @@
+import { afterEach, describe, expect, test } from "bun:test";
+import {
+  extractWithDefinitionRetry,
+  resolveExtractRetries,
+} from "./extract-with-retry.ts";
+
+type Row = { id: string; definition: string | null };
+
+const hasNullDefinition = (r: Row) => r.definition === null;
+
+describe("resolveExtractRetries", () => {
+  const originalEnv = process.env.PGDELTA_EXTRACT_RETRIES;
+  afterEach(() => {
+    if (originalEnv === undefined) {
+      process.env.PGDELTA_EXTRACT_RETRIES = undefined;
+      delete process.env.PGDELTA_EXTRACT_RETRIES;
+    } else {
+      process.env.PGDELTA_EXTRACT_RETRIES = originalEnv;
+    }
+  });
+
+  test("defaults to 1 when option and env are unset", () => {
+    delete process.env.PGDELTA_EXTRACT_RETRIES;
+    expect(resolveExtractRetries()).toBe(1);
+  });
+
+  test("uses option when provided", () => {
+    process.env.PGDELTA_EXTRACT_RETRIES = "5";
+    expect(resolveExtractRetries(0)).toBe(0);
+    expect(resolveExtractRetries(1)).toBe(1);
+    expect(resolveExtractRetries(7)).toBe(7);
+  });
+
+  test("falls back to env when option is undefined", () => {
+    process.env.PGDELTA_EXTRACT_RETRIES = "4";
+    expect(resolveExtractRetries()).toBe(4);
+  });
+
+  test("clamps negative values to 0", () => {
+    delete process.env.PGDELTA_EXTRACT_RETRIES;
+    expect(resolveExtractRetries(-3)).toBe(0);
+    process.env.PGDELTA_EXTRACT_RETRIES = "-9";
+    expect(resolveExtractRetries()).toBe(0);
+  });
+
+  test("ignores non-numeric env values", () => {
+    process.env.PGDELTA_EXTRACT_RETRIES = "not-a-number";
+    expect(resolveExtractRetries()).toBe(1);
+  });
+
+  test("ignores empty env string", () => {
+    process.env.PGDELTA_EXTRACT_RETRIES = "";
+    expect(resolveExtractRetries()).toBe(1);
+  });
+});
+
+describe("extractWithDefinitionRetry", () => {
+  test("returns first attempt when no row has null definition", async () => {
+    let attempts = 0;
+    const rows = await extractWithDefinitionRetry<Row>({
+      label: "test",
+      query: async () => {
+        attempts++;
+        return [{ id: "a", definition: "OK" }];
+      },
+      hasNullDefinition,
+      options: { retries: 2, backoffMs: 0 },
+    });
+    expect(attempts).toBe(1);
+    expect(rows).toEqual([{ id: "a", definition: "OK" }]);
+  });
+
+  test("retries when definition is null and succeeds on attempt 2", async () => {
+    let attempts = 0;
+    const rows = await extractWithDefinitionRetry<Row>({
+      label: "test",
+      query: async () => {
+        attempts++;
+        if (attempts === 1) {
+          return [
+            { id: "a", definition: "OK" },
+            { id: "b", definition: null },
+          ];
+        }
+        return [{ id: "a", definition: "OK" }];
+      },
+      hasNullDefinition,
+      options: { retries: 2, backoffMs: 0 },
+    });
+    expect(attempts).toBe(2);
+    expect(rows).toEqual([{ id: "a", definition: "OK" }]);
+  });
+
+  test("returns last-attempt rows (with offenders) once retries are exhausted", async () => {
+    let attempts = 0;
+    const rows = await extractWithDefinitionRetry<Row>({
+      label: "test",
+      query: async () => {
+        attempts++;
+        return [
+          { id: "a", definition: "OK" },
+          { id: "b", definition: null },
+        ];
+      },
+      hasNullDefinition,
+      options: { retries: 2, backoffMs: 0 },
+    });
+    expect(attempts).toBe(3);
+    expect(rows).toEqual([
+      { id: "a", definition: "OK" },
+      { id: "b", definition: null },
+    ]);
+  });
+
+  test("retries: 0 disables retrying entirely", async () => {
+    let attempts = 0;
+    const rows = await extractWithDefinitionRetry<Row>({
+      label: "test",
+      query: async () => {
+        attempts++;
+        return [{ id: "b", definition: null }];
+      },
+      hasNullDefinition,
+      options: { retries: 0, backoffMs: 0 },
+    });
+    expect(attempts).toBe(1);
+    expect(rows).toEqual([{ id: "b", definition: null }]);
+  });
+
+  test("retries: 5 attempts up to 6 times before giving up", async () => {
+    let attempts = 0;
+    await extractWithDefinitionRetry<Row>({
+      label: "test",
+      query: async () => {
+        attempts++;
+        return [{ id: "b", definition: null }];
+      },
+      hasNullDefinition,
+      options: { retries: 5, backoffMs: 0 },
+    });
+    expect(attempts).toBe(6);
+  });
+});

packages/pg-delta/src/core/objects/extract-with-retry.ts

@@ -0,0 +1,87 @@
+import debug from "debug";
+
+const log = debug("pg-delta:extract");
+
+const DEFAULT_RETRIES = 1;
+const DEFAULT_BACKOFF_MS = 50;
+
+export interface ExtractRetryOptions {
+  /**
+   * Number of retry attempts to make when a `pg_get_*def()` call returns NULL
+   * for at least one row. Total attempts is `retries + 1`. Negative values are
+   * clamped to 0. When this option is undefined the value is read from the
+   * `PGDELTA_EXTRACT_RETRIES` environment variable, falling back to a default
+   * of 1 (i.e. the first attempt plus one retry, 2 attempts total).
+   */
+  retries?: number;
+  /**
+   * Delay between retry attempts in milliseconds; the actual wait is
+   * `backoffMs * attemptNumber` (linear). Defaults to 50. Set to 0 in tests.
+   */
+  backoffMs?: number;
+}
+
+export function resolveExtractRetries(option?: number): number {
+  if (typeof option === "number" && Number.isFinite(option)) {
+    return Math.max(0, Math.floor(option));
+  }
+  const envVal = process.env.PGDELTA_EXTRACT_RETRIES;
+  if (envVal !== undefined && envVal !== "") {
+    const n = Number(envVal);
+    if (Number.isFinite(n)) return Math.max(0, Math.floor(n));
+  }
+  return DEFAULT_RETRIES;
+}
+
+const sleep = (ms: number) =>
+  ms > 0 ? new Promise<void>((r) => setTimeout(r, ms)) : Promise.resolve();
+
+/**
+ * Runs `query()` up to `retries + 1` times, retrying as long as at least one
+ * row in the result satisfies `hasNullDefinition`. The retry exists because
+ * `pg_get_<x>def()` can return NULL transiently when the underlying catalog
+ * row is dropped concurrently or the catalog state is in flux; in practice a
+ * second attempt either no longer sees the dropped row or succeeds in
+ * resolving the definition.
+ *
+ * Returns the rows from the first attempt with no offenders, or — once
+ * retries are exhausted — the rows from the final attempt (still containing
+ * offenders). The caller is responsible for the final filter so this helper
+ * works for both flat schemas (definition on the row) and nested schemas
+ * (definition on a child collection, e.g. table constraints).
+ */
+export async function extractWithDefinitionRetry<TRow>(params: {
+  label: string;
+  query: () => Promise<TRow[]>;
+  hasNullDefinition: (row: TRow) => boolean;
+  options?: ExtractRetryOptions;
+}): Promise<TRow[]> {
+  const retries = resolveExtractRetries(params.options?.retries);
+  const backoffMs = params.options?.backoffMs ?? DEFAULT_BACKOFF_MS;
+  const maxAttempts = retries + 1;
+
+  let rows: TRow[] = [];
+  for (let attempt = 1; attempt <= maxAttempts; attempt++) {
+    rows = await params.query();
+    const offenders = rows.filter(params.hasNullDefinition).length;
+    if (offenders === 0) return rows;
+    if (attempt < maxAttempts) {
+      log(
+        "%s: pg_get_*def() returned NULL for %d row(s) on attempt %d/%d; retrying",
+        params.label,
+        offenders,
+        attempt,
+        maxAttempts,
+      );
+      await sleep(backoffMs * attempt);
+    } else {
+      log(
+        "%s: pg_get_*def() returned NULL for %d row(s) after %d attempt(s); skipping",
+        params.label,
+        offenders,
+        maxAttempts,
+      );
+    }
+  }
+  return rows;
+}

packages/pg-delta/src/core/objects/index/index.model.test.ts

@@ -36,8 +36,19 @@ const baseRow = {
 const mockPool = (rows: unknown[]): Pool =>
   ({ query: async () => ({ rows }) }) as unknown as Pool;
 
+const mockPoolSequence = (...attempts: unknown[][]): Pool => {
+  let i = 0;
+  return {
+    query: async () => ({
+      rows: attempts[Math.min(i++, attempts.length - 1)],
+    }),
+  } as unknown as Pool;
+};
+
+const NO_BACKOFF = { backoffMs: 0 } as const;
+
 describe("extractIndexes", () => {
-  test("skips rows where pg_get_indexdef returned NULL", async () => {
+  test("skips rows where pg_get_indexdef returned NULL after exhausting retries", async () => {
     const indexes = await extractIndexes(
       mockPool([
         {
@@ -47,6 +58,7 @@ describe("extractIndexes", () => {
         },
         { ...baseRow, name: '"orphan_idx"', definition: null },
       ]),
+      NO_BACKOFF,
     );
 
     expect(indexes).toHaveLength(1);
@@ -59,6 +71,7 @@ describe("extractIndexes", () => {
     await expect(
       extractIndexes(
         mockPool([{ ...baseRow, name: '"orphan"', definition: null }]),
+        NO_BACKOFF,
       ),
     ).resolves.toEqual([]);
   });
@@ -77,7 +90,30 @@ describe("extractIndexes", () => {
           definition: "CREATE INDEX b ON users (id)",
         },
       ]),
+      NO_BACKOFF,
     );
     expect(indexes.map((i) => i.name)).toEqual(['"a"', '"b"']);
   });
+
+  test("recovers when pg_get_indexdef is NULL on first attempt but resolved on retry", async () => {
+    const indexes = await extractIndexes(
+      mockPoolSequence(
+        // attempt 1: definition is NULL (transient race)
+        [{ ...baseRow, name: '"racy_idx"', definition: null }],
+        // attempt 2: catalog scan no longer sees the dropped row, or
+        // pg_get_indexdef successfully resolves the definition
+        [
+          {
+            ...baseRow,
+            name: '"racy_idx"',
+            definition: "CREATE INDEX racy_idx ON users (id)",
+          },
+        ],
+      ),
+      { retries: 2, backoffMs: 0 },
+    );
+    expect(indexes).toHaveLength(1);
+    expect(indexes[0]?.name).toBe('"racy_idx"');
+    expect(indexes[0]?.definition).toBe("CREATE INDEX racy_idx ON users (id)");
+  });
 });