ci(preview): bisect invalid seed rows instead of replaying batches row-by-row (stage 3)

.github/scripts/preview/prepare_preview_database.sh

@@ -15,6 +15,7 @@ published_modal_secret=false
 schema_upgraded=false
 schema_rebuilt=false
 schema_healed=false
+seed_stats_file=""
 
 read_output_value() {
   local file="$1"
@@ -32,6 +33,20 @@ load_env_file() {
   done < "$file"
 }
 
+summarize_seed_line() {
+  if [ -z "$seed_stats_file" ] || [ ! -s "$seed_stats_file" ]; then
+    echo "- Seed stats: unavailable (seed did not run)"
+    return
+  fi
+  python3 -c '
+import json, sys
+d = json.load(open(sys.argv[1]))
+print("- Seed: `{batches_attempted}` batches attempted, `{batches_split}` split,"
+      " `{rows_skipped}` rows skipped in `{seed_seconds}s`".format(
+          **{**d, "seed_seconds": d.get("seed_seconds", "?")}))
+' "$seed_stats_file" 2>/dev/null || echo "- Seed stats: unparseable"
+}
+
 summarize_database_phase() {
   {
     echo "## Preview database"
@@ -44,6 +59,7 @@ summarize_database_phase() {
     echo "- Schema upgraded to head: \`$schema_upgraded\`"
     echo "- Schema rebuilt from prod snapshot: \`$schema_rebuilt\`"
     echo "- Auto-healed after a failed incremental upgrade: \`$schema_healed\`"
+    summarize_seed_line
     echo "- Modal DB secret published: \`$published_modal_secret\`"
   } >> "$GITHUB_STEP_SUMMARY"
 }
@@ -66,6 +82,8 @@ branch_was_created="$(read_output_value "$supabase_output" branch_was_created)"
 
 schema_rebuilt_file="$(mktemp)"
 export SCHEMA_REBUILT_FILE="$schema_rebuilt_file"
+seed_stats_file="$(mktemp)"
+export PREVIEW_SEED_STATS_FILE="$seed_stats_file"
 
 # Migrate on any backend deploy, not just migration-file changes, so a reused
 # branch can't run new code against a stale schema.
@@ -90,3 +108,10 @@ if [ "$DEPLOY_BACKEND" = "true" ] || [ "$branch_was_created" = "true" ]; then
   "$script_dir/publish_modal_db_secret.sh"
   published_modal_secret=true
 fi
+
+# Hand the run-level seed report to the gate's timing artifact. Single-line
+# JSON written by seed_preview_db.py (both the rebuild-internal and the
+# converge seed paths); whichever ran last is the run's authoritative draw.
+if [ -s "$seed_stats_file" ] && [ -n "$github_output" ]; then
+  echo "seed_stats=$(cat "$seed_stats_file")" >> "$github_output"
+fi

.github/scripts/preview/record_preview_metrics.py

@@ -155,6 +155,21 @@ def parse_timings(raw):
     }
 
 
+def parse_seed_stats(raw):
+    """Run-level seed report emitted by the prepare job as a `seed_stats` output.
+
+    Malformed payloads degrade to no seed section; telemetry must never fail a
+    deployment step that produced it.
+    """
+    if not raw:
+        return {}
+    try:
+        data = json.loads(raw)
+    except json.JSONDecodeError:
+        return {}
+    return data if isinstance(data, dict) else {}
+
+
 def _parse_ts(value):
     if not value:
         return None
@@ -274,6 +289,7 @@ def build_report(run, jobs, env, now=None):
         "identifiers": collect_identifiers(env),
         "jobs": jobs_section,
         "phases": phases,
+        "seed": parse_seed_stats(env.get("SEED_STATS", "")),
     }
 
 

.github/scripts/preview/seed_preview_db.py

@@ -10,6 +10,7 @@
 variable unset (local runs) the seed is a no-op beyond legacy cleanup.
 """
 import asyncio
+import json
 import os
 import sys
 import time
@@ -72,13 +73,20 @@ async def _main() -> None:
     engine = _engine(branch_url)
     t0 = time.monotonic()
     try:
-        await seed(engine, sampled=sampled)
+        report = await seed(engine, sampled=sampled)
     finally:
         await engine.dispose()
+    elapsed = time.monotonic() - t0
     print(
-        f"seed_preview_db: seeded branch in {time.monotonic() - t0:.1f}s",
+        f"seed_preview_db: seeded branch in {elapsed:.1f}s"
+        f" ({report['batches_attempted']} batches,"
+        f" {report['batches_split']} split,"
+        f" {report['rows_skipped']} rows skipped)",
         file=sys.stderr,
     )
+    if stats_file := os.environ.get("PREVIEW_SEED_STATS_FILE"):
+        report["seed_seconds"] = round(elapsed, 1)
+        Path(stats_file).write_text(json.dumps(report, sort_keys=True))
 
 
 if __name__ == "__main__":

.github/workflows/pr-preview.yml

@@ -115,6 +115,7 @@ jobs:
       branch_ref: ${{ steps.prepare.outputs.branch_ref }}
       branch_id: ${{ steps.prepare.outputs.branch_id }}
       branch_was_created: ${{ steps.prepare.outputs.branch_was_created }}
+      seed_stats: ${{ steps.prepare.outputs.seed_stats }}
     container:
       image: ghcr.io/abundant-ai/oddish-ci-base:latest
       credentials:
@@ -608,6 +609,7 @@ jobs:
           DEPLOY_FRONTEND: ${{ needs.detect-changes.outputs.deploy_frontend }}
           SUPABASE_BRANCH_ID: ${{ needs.prepare-preview-database.outputs.branch_id }}
           SUPABASE_BRANCH_REF: ${{ needs.prepare-preview-database.outputs.branch_ref }}
+          SEED_STATS: ${{ needs.prepare-preview-database.outputs.seed_stats }}
           MODAL_APP_NAME: oddish-pr-${{ github.event.pull_request.number }}
           # The live preview backend URL: this run's deploy, else the prior
           # run's deterministic URL when one is still live. Never the

backend/preview_seed.py

@@ -408,8 +408,10 @@ def _prepare_row(table, row: dict) -> dict:
     return values
 
 
-async def seed(engine: AsyncEngine, *, sampled: dict | None = None) -> None:
+async def seed(engine: AsyncEngine, *, sampled: dict | None = None) -> dict:
     sample_rows = (sampled or {}).get("rows", {})
+    stats = {"batches_attempted": 0, "batches_split": 0, "rows_skipped": 0}
+    skips: dict[str, list[str]] = {}
     md = MetaData()
     async with engine.begin() as conn:
         await conn.run_sync(md.reflect)
@@ -424,14 +426,18 @@ async def seed(engine: AsyncEngine, *, sampled: dict | None = None) -> None:
         )
         await _cleanup_legacy_fixture_rows(md, conn, ordered)
         if sampled is None:
-            return
+            return _seed_report(stats, skips, {})
         await _reconcile_previous_draw(md, conn, sample_rows)
 
+    drawn = [t.name for t in ordered if sample_rows.get(t.name)]
+    before = await _table_counts(engine, drawn)
     for table in ordered:
         rows = sample_rows.get(table.name, [])
         if not rows:
             continue
-        await _load_table(engine, table, rows)
+        await _load_table(engine, table, rows, stats, skips)
+    after = await _table_counts(engine, drawn)
+    counts = {name: [before[name], after[name]] for name in drawn}
 
     async with engine.begin() as conn:
         for table_name, row_id, column, value in (sampled or {}).get("linkage", []):
@@ -458,9 +464,35 @@ async def seed(engine: AsyncEngine, *, sampled: dict | None = None) -> None:
                     ),
                     {"t": name, "rids": rids[start : start + 10000]},
                 )
+    return _seed_report(stats, skips, counts)
+
+
+def _seed_report(stats: dict, skips: dict[str, list[str]], counts: dict) -> dict:
+    """Deterministic run summary: counters, sorted skip causes, row counts."""
+    return {
+        **stats,
+        "skips": {cause: sorted(keys)[:100] for cause, keys in sorted(skips.items())},
+        "tables": counts,
+    }
+
+
+async def _table_counts(engine: AsyncEngine, names: list[str]) -> dict[str, int]:
+    """Row counts for the drawn tables in one round trip (NullPool makes each
+    fresh connection a full pooler handshake, so per-table COUNTs would add
+    seconds to every seed)."""
+    if not names:
+        return {}
+    union = " UNION ALL ".join(
+        f"SELECT '{name}', count(*) FROM \"{name}\"" for name in names  # noqa: S608
+    )
+    async with engine.connect() as conn:
+        rows = await conn.execute(text(union))
+        return {name: int(count) for name, count in rows}
 
 
-async def _load_table(engine: AsyncEngine, table, rows: list[dict]) -> None:
+async def _load_table(
+    engine: AsyncEngine, table, rows: list[dict], stats: dict, skips: dict
+) -> None:
     prepared = [_prepare_row(table, row) for row in rows]
     try:
         await _load_table_copy_merge(engine, table, prepared)
@@ -470,7 +502,12 @@ async def _load_table(engine: AsyncEngine, table, rows: list[dict]) -> None:
             f"copy fast-path failed for {table.name} "
             f"({_error_cause(exc)}); falling back to batched upserts"
         )
-    await _load_table_batches(engine, table, prepared)
+    table_skips: dict[str, list[str]] = {}
+    await _load_table_batches(engine, table, prepared, stats, table_skips)
+    for cause in sorted(table_skips):
+        skips.setdefault(cause, []).extend(
+            f"{table.name}.{key}" for key in sorted(table_skips[cause])
+        )
 
 
 async def _load_table_copy_merge(
@@ -518,7 +555,9 @@ def _rec_value(name: str, value):
             )
 
 
-async def _load_table_batches(engine: AsyncEngine, table, prepared: list[dict]) -> None:
+async def _load_table_batches(
+    engine: AsyncEngine, table, prepared: list[dict], stats: dict, skips: dict
+) -> None:
     pk_cols = [c.name for c in table.primary_key.columns]
     batch_size = max(1, _MAX_BIND_PARAMS // max(1, len(table.columns)))
     batches = [
@@ -529,8 +568,6 @@ async def _load_table_batches(engine: AsyncEngine, table, prepared: list[dict])
     for batch in batches:
         queue.put_nowait(batch)
 
-    skips: dict[str, list[str]] = {}
-
     async def worker() -> None:
         async with engine.connect() as conn:
             while True:
@@ -539,60 +576,74 @@ async def worker() -> None:
                 except asyncio.QueueEmpty:
                     return
                 async with conn.begin():
-                    await _upsert_batch(conn, table, pk_cols, chunk, skips)
+                    await _upsert_batch(conn, table, pk_cols, chunk, stats, skips)
 
     workers = min(_LOAD_STREAMS, len(batches))
     await asyncio.gather(*[worker() for _ in range(workers)])
 
-    for cause, row_keys in skips.items():
-        sample = ", ".join(row_keys[:3])
-        more = f" (+{len(row_keys) - 3} more)" if len(row_keys) > 3 else ""
+    # Deterministic report: sort causes and keys so concurrent streams can
+    # never reorder the output between runs.
+    for cause in sorted(skips):
+        keys = sorted(skips[cause])
+        sample = ", ".join(keys[:3])
+        more = f" (+{len(keys) - 3} more)" if len(keys) > 3 else ""
         _warn(
-            f"skipped {len(row_keys)} {table.name} row(s) on {cause}; "
+            f"skipped {len(keys)} {table.name} row(s) on {cause}; "
             f"existing rows kept -- e.g. {sample}{more}"
         )
 
 
+async def _isolate_bad_rows(execute, rows: list[dict], report_row, stats: dict) -> None:
+    """Upsert ``rows``, bisecting on constraint failures to isolate bad rows.
+
+    A failed batch is rolled back at its savepoint, split in half, and each
+    half retried, recursing only into failing halves until the invalid rows
+    stand alone: k bad rows in an n-row batch cost O(k log n) round trips
+    instead of O(n) row-by-row probes. Only IntegrityError (SQLSTATE class
+    23) is splittable -- network, timeout, and serialization errors are
+    transient infrastructure failures, not bad data, and abort the load.
+    """
+    stats["batches_attempted"] += 1
+    try:
+        await execute(rows)
+    except IntegrityError as exc:
+        if len(rows) == 1:
+            stats["rows_skipped"] += 1
+            report_row(rows[0], exc)
+            return
+        stats["batches_split"] += 1
+        mid = len(rows) // 2
+        await _isolate_bad_rows(execute, rows[:mid], report_row, stats)
+        await _isolate_bad_rows(execute, rows[mid:], report_row, stats)
+
+
 def _changed(table, stmt, keys: list[str]):
     return tuple_(*[table.c[k] for k in keys]).is_distinct_from(
         tuple_(*[stmt.excluded[k] for k in keys])
     )
 
 
 async def _upsert_batch(
-    conn, table, pk_cols: list[str], chunk: list[dict], skips: dict[str, list[str]]
+    conn, table, pk_cols: list[str], chunk: list[dict], stats: dict, skips: dict
 ) -> None:
     deferred = _LINKAGE_COLUMNS.get(table.name, set())
-    non_pk = [k for k in chunk[0] if k not in pk_cols and k not in deferred]
-    stmt = pg_insert(table).values(chunk)
-    set_ = {k: stmt.excluded[k] for k in non_pk}
-    try:
+
+    async def execute(rows: list[dict]) -> None:
+        non_pk = [k for k in rows[0] if k not in pk_cols and k not in deferred]
+        stmt = pg_insert(table).values(rows)
         async with conn.begin_nested():
             await conn.execute(
                 stmt.on_conflict_do_update(
                     index_elements=pk_cols,
-                    set_=set_,
+                    set_={k: stmt.excluded[k] for k in non_pk},
                     where=_changed(table, stmt, non_pk),
                 )
             )
-        return
-    except (IntegrityError, DBAPIError):
-        pass
-    for values in chunk:
-        non_pk = [k for k in values if k not in pk_cols and k not in deferred]
-        stmt = pg_insert(table).values(**values)
-        set_ = {k: stmt.excluded[k] for k in non_pk}
-        try:
-            async with conn.begin_nested():
-                await conn.execute(
-                    stmt.on_conflict_do_update(
-                        index_elements=pk_cols,
-                        set_=set_,
-                        where=_changed(table, stmt, non_pk),
-                    )
-                )
-        except (IntegrityError, DBAPIError) as exc:
-            skips.setdefault(_error_cause(exc), []).append(_row_key(table, values))
+
+    def report_row(row: dict, exc: IntegrityError) -> None:
+        skips.setdefault(_error_cause(exc), []).append(_row_key(table, row))
+
+    await _isolate_bad_rows(execute, chunk, report_row, stats)
 
 
 async def _reconcile_previous_draw(md: MetaData, conn, sample_rows: dict) -> None:

backend/tests/test_pr_preview_workflow.py

@@ -455,6 +455,7 @@ def test_gate_metrics_are_soft_and_retained():
         "DEPLOY_FRONTEND",
         "SUPABASE_BRANCH_ID",
         "SUPABASE_BRANCH_REF",
+        "SEED_STATS",
         "MODAL_APP_NAME",
         "MODAL_API_URL",
         "VERCEL_DEPLOYMENT_ID",
@@ -612,3 +613,14 @@ def test_reset_reuses_preview_scripts():
     )
     assert prepare_step["env"]["DEPLOY_BACKEND"] == "true"
     assert prepare_step["env"]["RUN_MIGRATIONS"] == "true"
+
+
+def test_seed_stats_flows_from_prepare_to_the_metrics_artifact():
+    jobs = _wf()["jobs"]
+    prepare_outputs = jobs["prepare-preview-database"].get("outputs", {})
+    assert prepare_outputs.get("seed_stats") == "${{ steps.prepare.outputs.seed_stats }}"
+    gate_steps = jobs["require-working-preview"]["steps"]
+    record = next(s for s in gate_steps if s.get("name") == "Record preview metrics")
+    assert record.get("env", {}).get("SEED_STATS") == (
+        "${{ needs.prepare-preview-database.outputs.seed_stats }}"
+    )