fix(migrations): v0_29_1 backfill misses connect() and uses BEGIN/COMMIT outside transaction

src/commands/migrations/v0_29_1.ts

@@ -48,26 +48,32 @@ async function phaseBBackfill(opts: OrchestratorOpts): Promise<OrchestratorPhase
     const { backfillEffectiveDate } = await import('../../core/backfill-effective-date.ts');
     const cfg = loadConfig();
     if (!cfg) throw new Error('No gbrain config; run `gbrain init` first.');
-    const engine = await createEngine(toEngineConfig(cfg));
+    const engineCfg = toEngineConfig(cfg);
+    const engine = await createEngine(engineCfg);
+    await engine.connect(engineCfg);
 
     let totalExamined = 0;
     let totalUpdated = 0;
 
-    const result = await backfillEffectiveDate(engine, {
-      onBatch: ({ batch, lastId, rowsTouched, cumulative }) => {
-        totalExamined = cumulative;
-        totalUpdated += rowsTouched;
-        if (batch % 10 === 0) {
-          process.stderr.write(`  [backfill] batch ${batch} | last_id=${lastId} | examined=${cumulative} | updated_so_far=${totalUpdated}\n`);
-        }
-      },
-    });
+    try {
+      const result = await backfillEffectiveDate(engine, {
+        onBatch: ({ batch, lastId, rowsTouched, cumulative }) => {
+          totalExamined = cumulative;
+          totalUpdated += rowsTouched;
+          if (batch % 10 === 0) {
+            process.stderr.write(`  [backfill] batch ${batch} | last_id=${lastId} | examined=${cumulative} | updated_so_far=${totalUpdated}\n`);
+          }
+        },
+      });
 
-    return {
-      name: 'backfill_effective_date',
-      status: 'complete',
-      detail: `examined=${result.examined} updated=${result.updated} fallback=${result.fallback} dur=${result.durationSec.toFixed(1)}s`,
-    };
+      return {
+        name: 'backfill_effective_date',
+        status: 'complete',
+        detail: `examined=${result.examined} updated=${result.updated} fallback=${result.fallback} dur=${result.durationSec.toFixed(1)}s`,
+      };
+    } finally {
+      try { await engine.disconnect(); } catch { /* best-effort */ }
+    }
   } catch (e) {
     return { name: 'backfill_effective_date', status: 'failed', detail: e instanceof Error ? e.message : String(e) };
   }
@@ -82,23 +88,29 @@ async function phaseCVerify(opts: OrchestratorOpts): Promise<OrchestratorPhaseRe
     const { loadConfig, toEngineConfig } = await import('../../core/config.ts');
     const cfg = loadConfig();
     if (!cfg) throw new Error('No gbrain config; run `gbrain init` first.');
-    const engine = await createEngine(toEngineConfig(cfg));
-    // Count rows where effective_date is still NULL but frontmatter HAS a
-    // parseable date — those are the rows the backfill should have touched
-    // but didn't. (Rows that fall through to 'fallback' have non-null
-    // effective_date already; this catches genuine misses.)
-    const rows = await engine.executeRaw<{ count: string }>(
-      `SELECT COUNT(*)::text AS count FROM pages WHERE effective_date IS NULL`,
-    );
-    const remaining = Number(rows[0]?.count ?? 0);
-    if (remaining > 0) {
-      return {
-        name: 'verify',
-        status: 'failed',
-        detail: `${remaining} pages still have NULL effective_date (backfill incomplete)`,
-      };
+    const engineCfg = toEngineConfig(cfg);
+    const engine = await createEngine(engineCfg);
+    await engine.connect(engineCfg);
+    try {
+      // Count rows where effective_date is still NULL but frontmatter HAS a
+      // parseable date — those are the rows the backfill should have touched
+      // but didn't. (Rows that fall through to 'fallback' have non-null
+      // effective_date already; this catches genuine misses.)
+      const rows = await engine.executeRaw<{ count: string }>(
+        `SELECT COUNT(*)::text AS count FROM pages WHERE effective_date IS NULL`,
+      );
+      const remaining = Number(rows[0]?.count ?? 0);
+      if (remaining > 0) {
+        return {
+          name: 'verify',
+          status: 'failed',
+          detail: `${remaining} pages still have NULL effective_date (backfill incomplete)`,
+        };
+      }
+      return { name: 'verify', status: 'complete', detail: '0 pages with NULL effective_date' };
+    } finally {
+      try { await engine.disconnect(); } catch { /* best-effort */ }
     }
-    return { name: 'verify', status: 'complete', detail: '0 pages with NULL effective_date' };
   } catch (e) {
     return { name: 'verify', status: 'failed', detail: e instanceof Error ? e.message : String(e) };
   }

src/core/backfill-effective-date.ts

@@ -175,14 +175,14 @@ export async function backfillEffectiveDate(
     if (!opts.dryRun) {
       // Compute effective_date for each row, then UPDATE in a batch wrapped
       // in its own transaction (so SET LOCAL statement_timeout scopes to it).
-      // postgres.js's `transaction` would be cleaner but we're using executeRaw
-      // for engine portability; explicit BEGIN/COMMIT does the same on both.
-      if (isPostgres) {
-        await engine.executeRaw(`BEGIN`);
-        await engine.executeRaw(`SET LOCAL statement_timeout = '600s'`);
-      }
-
-      try {
+      // postgres.js v3 refuses ad-hoc BEGIN/COMMIT spread across executeRaw
+      // calls on pooled connections (UNSAFE_TRANSACTION) — must go through
+      // engine.transaction() so all statements share one backend. PGLite path
+      // skips the wrapper (single-writer; no SET LOCAL scoping needed).
+      const applyBatch = async (exec: BrainEngine) => {
+        if (isPostgres) {
+          await exec.executeRaw(`SET LOCAL statement_timeout = '600s'`);
+        }
         for (const r of rows) {
           const fm = parseFrontmatter(r.frontmatter);
           const filename = r.import_filename
@@ -204,20 +204,19 @@ export async function backfillEffectiveDate(
 
           if (!opts.force && datesMatch && sourcesMatch) continue;
 
-          await engine.executeRaw(
+          await exec.executeRaw(
             `UPDATE pages SET effective_date = $1::timestamptz, effective_date_source = $2 WHERE id = $3`,
             [computed.date ? computed.date.toISOString() : null, computed.source, r.id],
           );
           touched++;
           if (computed.source === 'fallback') fallback++;
         }
+      };
 
-        if (isPostgres) await engine.executeRaw(`COMMIT`);
-      } catch (e) {
-        if (isPostgres) {
-          try { await engine.executeRaw(`ROLLBACK`); } catch { /* ignore */ }
-        }
-        throw e;
+      if (isPostgres) {
+        await engine.transaction(applyBatch);
+      } else {
+        await applyBatch(engine);
       }
     } else {
       // Dry run: still count what WOULD change.