admin/gbrain
1
0
Fork 0
mirror of https://github.com/garrytan/gbrain.git synced 2026-07-31 04:07:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into garrytan/gbrain-evals

Browse Source 
# Conflicts:
#	CHANGELOG.md
#	VERSION
#	docs/benchmarks/2026-04-18-minions-vs-openclaw-production.md
#	llms-full.txt
#	llms.txt
#	package.json
#	scripts/llms-config.ts
#	src/core/engine.ts
#	src/core/migrate.ts
#	src/core/postgres-engine.ts
#	src/core/types.ts
#	test/migrate.test.ts
This commit is contained in:
Garry Tan
2026-04-23 11:09:20 -07:00
parent 871227cda2 08b3698e90
commit f26bc0d082
12 changed files with 1119 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/commands/apply-migrations.ts
+28
View File
@@ -278,6 +278,34 @@ export async function runApplyMigrations(args: string[]): Promise<void> {
return;
}
// Pre-flight: warn if schema migrations (migrate.ts) are behind.
// apply-migrations runs orchestrator migrations only; schema migrations
// run via connectEngine() / initSchema(). Users often expect this CLI
// to handle everything (Issue 1 from v0.18.0 field report).
try {
const { LATEST_VERSION } = await import('../core/migrate.ts');
const { loadConfig: lc, toEngineConfig } = await import('../core/config.ts');
const { createEngine } = await import('../core/engine-factory.ts');
const cfg = lc();
if (cfg) {
const eng = await createEngine(toEngineConfig(cfg));
await eng.connect(toEngineConfig(cfg));
const verStr = await eng.getConfig('version');
const schemaVer = parseInt(verStr || '1', 10);
await eng.disconnect();
if (schemaVer < LATEST_VERSION) {
console.warn(
`\n⚠️ Schema version ${schemaVer} is behind latest ${LATEST_VERSION}.\n` +
` Schema migrations run automatically on next connectEngine() / initSchema().\n` +
` To run them now: gbrain init --migrate-only\n`,
);
}
}
} catch {
// Non-fatal: if DB is unreachable, orchestrator migrations can still
// run their filesystem-only phases.
}
const completed = loadCompletedMigrations();
const idx = indexCompleted(completed);
const plan = buildPlan(idx, installed, cli.specificMigration);
src/commands/doctor.ts
+69 -1
View File
@@ -1,6 +1,6 @@
import type { BrainEngine } from '../core/engine.ts';
import * as db from '../core/db.ts';
import { LATEST_VERSION } from '../core/migrate.ts';
import { LATEST_VERSION, getIdleBlockers } from '../core/migrate.ts';
import { checkResolvable } from '../core/check-resolvable.ts';
import { autoFixDryViolations, type AutoFixReport, type FixOutcome } from '../core/dry-fix.ts';
import { findRepoRoot } from '../core/repo-root.ts';
@@ -33,6 +33,19 @@ export async function runDoctor(engine: BrainEngine | null, args: string[], dbSo
const fastMode = args.includes('--fast');
const doFix = args.includes('--fix');
const dryRun = args.includes('--dry-run');
const locksMode = args.includes('--locks');
// --locks is a focused diagnostic: it runs the same pg_stat_activity
// query that `runMigrations` pre-flight uses, prints any idle-in-tx
// backends, and exits. Used by a user (or the migrate.ts error 57014
// message) who just hit a statement_timeout and needs to find the
// blocker. Referenced from migrate.ts's 57014 diagnostic — that
// message promised this flag exists.
if (locksMode) {
await runLocksCheck(engine, jsonOutput);
return;
}
const checks: Check[] = [];
let autoFixReport: AutoFixReport | null = null;
@@ -754,3 +767,58 @@ function outputResults(checks: Check[], json: boolean): boolean {
}
return hasFail;
}
/**
* `gbrain doctor --locks` — list idle-in-transaction backends older
* than 5 minutes that could block DDL. Exits 0 on clean, 1 on blockers.
*
* Agents hitting a statement_timeout (SQLSTATE 57014) during migration
* need a one-command path to find and kill the blocker. migrate.ts's
* 57014 diagnostic references this flag by name; keep the two in sync.
*
* Postgres-only. PGLite has no pool, no idle-in-tx concept, so the
* check prints a one-liner and exits 0.
*/
async function runLocksCheck(engine: BrainEngine | null, jsonOutput: boolean): Promise<void> {
if (!engine) {
if (jsonOutput) {
console.log(JSON.stringify({ status: 'unavailable', reason: 'no_engine' }));
} else {
console.log('gbrain doctor --locks requires a database connection. Configure a URL and retry.');
}
process.exit(1);
}
if (engine.kind !== 'postgres') {
if (jsonOutput) {
console.log(JSON.stringify({ status: 'not_applicable', engine: engine.kind }));
} else {
console.log(`gbrain doctor --locks is Postgres-only. Current engine: ${engine.kind}. No blockers possible (no connection pool).`);
}
return;
}
const blockers = await getIdleBlockers(engine);
if (jsonOutput) {
console.log(JSON.stringify({ status: blockers.length === 0 ? 'ok' : 'blockers_found', blockers }, null, 2));
if (blockers.length > 0) process.exit(1);
return;
}
if (blockers.length === 0) {
console.log('✓ No idle-in-transaction backends older than 5 minutes.');
return;
}
console.log(`Found ${blockers.length} idle-in-transaction backend(s) older than 5 minutes:\n`);
for (const b of blockers) {
console.log(` PID ${b.pid} (idle since ${b.query_start})`);
console.log(` Query: ${b.query}`);
console.log(` Kill: SELECT pg_terminate_backend(${b.pid});`);
console.log('');
}
console.log('These connections may block ALTER TABLE DDL during migration.');
console.log('After terminating, retry: gbrain apply-migrations --yes');
process.exit(1);
}
src/core/db.ts
+25
View File
@@ -71,6 +71,29 @@ export function resolvePoolSize(explicit?: number): number {
return DEFAULT_POOL_SIZE_FALLBACK;
}
/**
* Apply session-level defaults to a fresh connection. Called from both
* the module-level `connect()` singleton and the PostgresEngine
* instance-level pool so the idle-in-transaction session timeout is set
* uniformly.
*
* `idle_in_transaction_session_timeout = 5 min` was the v0.18.0 field
* report's headline production issue: a 24-hour idle connection was
* holding a lock on `pages` and blocking all DDL. 5 minutes is generous
* for any legitimate transaction but catches crashed writers. The GUC
* is session-scoped (safe for shared pools — no cross-statement leak).
*
* Wrapped in try/catch because some managed Postgres tenants restrict
* SET on the GUC; non-fatal if it fails.
*/
export async function setSessionDefaults(sql: ReturnType<typeof postgres>): Promise<void> {
try {
await sql`SET idle_in_transaction_session_timeout = '300000'`;
} catch {
// Non-fatal: some managed Postgres may restrict this GUC
}
}
export function getConnection(): ReturnType<typeof postgres> {
if (!sql) {
throw new GBrainError(
@@ -124,6 +147,8 @@ export async function connect(config: EngineConfig): Promise<void> {
// Test connection
await sql`SELECT 1`;
connectedUrl = url;
await setSessionDefaults(sql);
} catch (e: unknown) {
sql = null;
connectedUrl = null;
src/core/engine.ts
+31
View File
@@ -60,6 +60,31 @@ export interface TimelineBatchInput {
source_id?: string;
}
/**
* A single dedicated database connection, isolated from the engine's pool.
*
* Used by migration paths that need session-level GUCs (e.g.
* `SET statement_timeout = '600000'` before a `CREATE INDEX CONCURRENTLY`)
* without leaking into the shared pool, and by write-quiesce designs
* that need a session-lifetime Postgres advisory lock that survives
* across transaction boundaries.
*
* On Postgres: backed by postgres-js `sql.reserve()`; the same backend
* process serves every `executeRaw` call within the callback. Released
* automatically when the callback returns or throws.
*
* On PGLite: a thin pass-through. PGLite has no pool, so every call is
* already on the single backing connection. The interface is still
* exposed so cross-engine callers don't need to branch.
*
* Not safe to call from inside `transaction()`. The transaction holds a
* different backend; reserving a second one can deadlock on a row the
* transaction itself is waiting to write.
*/
export interface ReservedConnection {
executeRaw<T = Record<string, unknown>>(sql: string, params?: unknown[]): Promise<T[]>;
}
/** Maximum results returned by search operations. Internal bulk operations (listPages) are not clamped. */
export const MAX_SEARCH_LIMIT = 100;
@@ -79,6 +104,12 @@ export interface BrainEngine {
disconnect(): Promise<void>;
initSchema(): Promise<void>;
transaction<T>(fn: (engine: BrainEngine) => Promise<T>): Promise<T>;
/**
* Run `fn` with a dedicated connection (Postgres: reserved backend;
* PGLite: pass-through). See `ReservedConnection` for semantics and
* usage constraints. Release is automatic.
*/
withReservedConnection<T>(fn: (conn: ReservedConnection) => Promise<T>): Promise<T>;
// Pages CRUD
getPage(slug: string): Promise<Page | null>;
src/core/migrate.ts
+302 -117
View File
@@ -472,64 +472,107 @@ export const MIGRATIONS: Migration[] = [
sql: '',
handler: async (engine) => {
if (engine.kind === 'pglite') return;
await engine.runMigration(19, `
-- 1a. source_id with DEFAULT 'default' (idempotent)
ALTER TABLE files ADD COLUMN IF NOT EXISTS source_id TEXT
NOT NULL DEFAULT 'default' REFERENCES sources(id) ON DELETE CASCADE;
CREATE INDEX IF NOT EXISTS idx_files_source_id ON files(source_id);
-- 1b. page_id (nullable; pre-v0.17 files pointed at page_slug
-- which was ON DELETE SET NULL, so we keep the same nullable
-- semantic — orphaned files are legal).
ALTER TABLE files ADD COLUMN IF NOT EXISTS page_id INTEGER
REFERENCES pages(id) ON DELETE SET NULL;
CREATE INDEX IF NOT EXISTS idx_files_page_id ON files(page_id);
`);
// Atomic: FK drop + UNIQUE swap + files.page_id addition +
// backfill + ledger, all in one transaction. Closes the
// pre-v23 integrity window where files_page_slug_fkey was
// dropped in v21 but the replacement files.page_id didn't
// exist until v23 ran — process death in between left files
// unconstrained while file_upload kept writing (codex finding).
//
// Rollback scenarios:
// - Die mid-transaction → Postgres rolls back, files_page_slug_fkey
// still exists, config.version stays at 22. Retry restarts cleanly.
// - Die after commit but before setConfig(version=23) → all DDL
// committed, config.version still 22, retry re-runs everything
// with IF NOT EXISTS / NOT EXISTS guards idempotently.
await engine.transaction(async (tx) => {
// 0a. Drop files_page_slug_fkey (deferred from v21 to keep
// the FK intact across v21/v22 and remove it inside the
// same txn that adds the replacement page_id path).
// Guard against PGLite just in case (already returned above).
await tx.runMigration(23, `
DO $$ BEGIN
IF EXISTS (SELECT 1 FROM information_schema.tables WHERE table_name = 'files') THEN
ALTER TABLE files DROP CONSTRAINT IF EXISTS files_page_slug_fkey;
END IF;
END $$;
`);
await engine.runMigration(19, `
-- 1c. Backfill page_id from existing page_slug. Scoped to
-- source_id='default' because pre-v0.17 pages ALL lived in
-- the default source. Without this scope, after new sources
-- get added mid-migration, the JOIN could hit the wrong
-- page (different source, same slug).
UPDATE files f
SET page_id = p.id
FROM pages p
WHERE f.page_slug = p.slug
AND p.source_id = 'default'
AND f.page_id IS NULL;
`);
// 0b. Swap pages.UNIQUE(slug) → UNIQUE(source_id, slug).
// Deferred from v21 so PR #356 closes the integrity
// window. PGLite already did this swap in its v21 path.
await tx.runMigration(23, `
ALTER TABLE pages DROP CONSTRAINT IF EXISTS pages_slug_key;
DO $$ BEGIN
IF NOT EXISTS (
SELECT 1 FROM pg_constraint WHERE conname = 'pages_source_slug_key'
) THEN
ALTER TABLE pages ADD CONSTRAINT pages_source_slug_key
UNIQUE (source_id, slug);
END IF;
END $$;
`);
await engine.runMigration(19, `
-- 2. file_migration_ledger — drives the storage object rewrite
-- in the v0_18_0 orchestrator's phase B. Seeded from current
-- files rows; re-seed is idempotent via NOT EXISTS guard.
CREATE TABLE IF NOT EXISTS file_migration_ledger (
file_id INTEGER PRIMARY KEY REFERENCES files(id) ON DELETE CASCADE,
storage_path_old TEXT NOT NULL,
storage_path_new TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
error TEXT,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
CONSTRAINT chk_ledger_status CHECK (status IN ('pending','copy_done','db_updated','complete','failed'))
);
CREATE INDEX IF NOT EXISTS idx_file_migration_ledger_status
ON file_migration_ledger(status) WHERE status != 'complete';
// 1a. source_id with DEFAULT 'default' (idempotent)
await tx.runMigration(23, `
ALTER TABLE files ADD COLUMN IF NOT EXISTS source_id TEXT
NOT NULL DEFAULT 'default' REFERENCES sources(id) ON DELETE CASCADE;
CREATE INDEX IF NOT EXISTS idx_files_source_id ON files(source_id);
-- Seed the ledger with every existing file. New path prefixes
-- source_id so multi-source can land assets under their own
-- bucket path without collision.
INSERT INTO file_migration_ledger (file_id, storage_path_old, storage_path_new, status)
SELECT
f.id,
f.storage_path,
COALESCE(f.source_id, 'default') || '/' || f.storage_path,
'pending'
FROM files f
WHERE NOT EXISTS (
SELECT 1 FROM file_migration_ledger l WHERE l.file_id = f.id
);
`);
-- 1b. page_id (nullable; pre-v0.17 files pointed at page_slug
-- which was ON DELETE SET NULL, so we keep the same nullable
-- semantic — orphaned files are legal).
ALTER TABLE files ADD COLUMN IF NOT EXISTS page_id INTEGER
REFERENCES pages(id) ON DELETE SET NULL;
CREATE INDEX IF NOT EXISTS idx_files_page_id ON files(page_id);
`);
// 1c. Backfill page_id from existing page_slug. Scoped to
// source_id='default' because pre-v0.17 pages ALL lived in
// the default source. Without this scope, after new sources
// get added mid-migration, the JOIN could hit the wrong
// page (different source, same slug).
await tx.runMigration(23, `
UPDATE files f
SET page_id = p.id
FROM pages p
WHERE f.page_slug = p.slug
AND p.source_id = 'default'
AND f.page_id IS NULL;
`);
// 2. file_migration_ledger — drives the storage object rewrite
// in the v0_18_0 orchestrator's phase B. Seeded from current
// files rows; re-seed is idempotent via NOT EXISTS guard.
await tx.runMigration(23, `
CREATE TABLE IF NOT EXISTS file_migration_ledger (
file_id INTEGER PRIMARY KEY REFERENCES files(id) ON DELETE CASCADE,
storage_path_old TEXT NOT NULL,
storage_path_new TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
error TEXT,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
CONSTRAINT chk_ledger_status CHECK (status IN ('pending','copy_done','db_updated','complete','failed'))
);
CREATE INDEX IF NOT EXISTS idx_file_migration_ledger_status
ON file_migration_ledger(status) WHERE status != 'complete';
-- Seed the ledger with every existing file. New path prefixes
-- source_id so multi-source can land assets under their own
-- bucket path without collision.
INSERT INTO file_migration_ledger (file_id, storage_path_old, storage_path_new, status)
SELECT
f.id,
f.storage_path,
COALESCE(f.source_id, 'default') || '/' || f.storage_path,
'pending'
FROM files f
WHERE NOT EXISTS (
SELECT 1 FROM file_migration_ledger l WHERE l.file_id = f.id
);
`);
});
},
},
{
@@ -560,39 +603,53 @@ export const MIGRATIONS: Migration[] = [
{
version: 21,
name: 'pages_source_id_composite_unique',
// v0.18.0 Step 2 (Lane B) — adds pages.source_id with DEFAULT 'default'
// and swaps the global UNIQUE(slug) for the composite UNIQUE(source_id,
// slug). Lands alongside the engine SQL rewrite that makes every
// ON CONFLICT (slug) → ON CONFLICT (source_id, slug) so the constraint
// swap is atomic with the code that writes under it.
// v0.18.0 Step 2 (Lane B) — adds pages.source_id. Engine-split after
// codex caught the pre-v23 integrity window:
//
// DEFAULT 'default' is load-bearing: closes the Codex-flagged race
// Original v21 dropped files_page_slug_fkey and swapped
// UNIQUE(slug) → UNIQUE(source_id, slug) in one go. Between v21
// committing and v23 (which adds the replacement files.page_id
// path), a process-death left files WITHOUT any FK to pages
// while file_upload / `gbrain files` kept accepting writes.
//
// On Postgres: additive-only here. The FK drop + UNIQUE swap move
// into v23's handler (wrapped in engine.transaction) so they commit
// atomically with the files.page_id addition + backfill. See v23.
//
// On PGLite: no concurrent writers, no pool, no partial-state risk.
// Do the full add + swap here so PGLite brains reach the composite
// unique immediately (PGLite has no files table, so no FK drop
// needed).
//
// DEFAULT 'default' on source_id is load-bearing: closes the race
// where an INSERT between ADD COLUMN and SET NOT NULL could leave
// source_id NULL. Because the default already references a valid
// sources row (seeded in v16), new INSERTs immediately get a valid FK.
//
// Idempotent: IF NOT EXISTS on ADD COLUMN, DROP IF EXISTS on the old
// constraint, DO block guard on the new constraint creation.
sql: `
ALTER TABLE pages ADD COLUMN IF NOT EXISTS source_id TEXT
NOT NULL DEFAULT 'default' REFERENCES sources(id) ON DELETE CASCADE;
// source_id NULL. The default already references a valid sources
// row (seeded in v16), so new INSERTs immediately get a valid FK.
sql: '',
sqlFor: {
postgres: `
ALTER TABLE pages ADD COLUMN IF NOT EXISTS source_id TEXT
NOT NULL DEFAULT 'default' REFERENCES sources(id) ON DELETE CASCADE;
CREATE INDEX IF NOT EXISTS idx_pages_source_id ON pages(source_id);
CREATE INDEX IF NOT EXISTS idx_pages_source_id ON pages(source_id);
`,
pglite: `
ALTER TABLE pages ADD COLUMN IF NOT EXISTS source_id TEXT
NOT NULL DEFAULT 'default' REFERENCES sources(id) ON DELETE CASCADE;
-- Swap global UNIQUE(slug) → composite UNIQUE(source_id, slug). The
-- original constraint is named pages_slug_key by Postgres convention
-- when the column was declared UNIQUE inline. Both drops are
-- idempotent.
ALTER TABLE pages DROP CONSTRAINT IF EXISTS pages_slug_key;
DO $$ BEGIN
IF NOT EXISTS (
SELECT 1 FROM pg_constraint WHERE conname = 'pages_source_slug_key'
) THEN
ALTER TABLE pages ADD CONSTRAINT pages_source_slug_key
UNIQUE (source_id, slug);
END IF;
END $$;
`,
CREATE INDEX IF NOT EXISTS idx_pages_source_id ON pages(source_id);
ALTER TABLE pages DROP CONSTRAINT IF EXISTS pages_slug_key;
DO $$ BEGIN
IF NOT EXISTS (
SELECT 1 FROM pg_constraint WHERE conname = 'pages_source_slug_key'
) THEN
ALTER TABLE pages ADD CONSTRAINT pages_source_slug_key
UNIQUE (source_id, slug);
END IF;
END $$;
`,
},
},
{
version: 20,
@@ -758,9 +815,117 @@ export const MIGRATIONS: Migration[] = [
];
export const LATEST_VERSION = MIGRATIONS.length > 0
? MIGRATIONS[MIGRATIONS.length - 1].version
? Math.max(...MIGRATIONS.map(m => m.version))
: 1;
/**
* Row returned by `getIdleBlockers`. The shape is the public contract
* for both `gbrain doctor --locks` output and the internal DDL pre-flight.
*/
export interface IdleBlocker {
pid: number;
state: string;
query_start: string;
query: string;
}
/**
* Find idle-in-transaction connections older than 5 minutes that might
* block DDL. Postgres-only. Returns `[]` on PGLite, query failure, or
* no blockers. The query-failure path is intentionally silent because
* some managed Postgres configs restrict `pg_stat_activity` — a partial
* view of the server is still useful for doctor/pre-flight.
*
* Single source of truth shared by:
* - `checkForBlockingConnections` (DDL pre-flight warning)
* - `gbrain doctor --locks` (CLI diagnostic)
* - any future `--exclusive` drain-wait logic
*/
export async function getIdleBlockers(engine: BrainEngine): Promise<IdleBlocker[]> {
if (engine.kind !== 'postgres') return [];
try {
return await engine.executeRaw<IdleBlocker>(
`SELECT pid, state, query_start::text, substring(query, 1, 120) as query
FROM pg_stat_activity
WHERE state = 'idle in transaction'
AND query_start < NOW() - INTERVAL '5 minutes'
AND pid != pg_backend_pid()`
);
} catch {
return [];
}
}
/**
* Check for idle-in-transaction connections that might block DDL.
* Returns true if blockers were found (logged as warnings).
*/
async function checkForBlockingConnections(engine: BrainEngine): Promise<boolean> {
const rows = await getIdleBlockers(engine);
if (rows.length > 0) {
console.warn(`\n⚠️ Found ${rows.length} idle-in-transaction connection(s) older than 5 minutes:`);
for (const r of rows) {
console.warn(` PID ${r.pid} — idle since ${r.query_start}`);
console.warn(` Query: ${r.query}`);
}
console.warn(` These may block ALTER TABLE DDL. To kill: SELECT pg_terminate_backend(<pid>);\n`);
return true;
}
return false;
}
/**
* Wrap migration SQL execution with Supabase-compatible timeout.
* Uses SET LOCAL statement_timeout inside a transaction to override
* server-enforced timeouts (required for Supabase Postgres).
*/
async function runMigrationSQL(
engine: BrainEngine,
m: Migration,
sql: string,
): Promise<void> {
const useTransaction = m.transaction !== false;
if (useTransaction || engine.kind === 'pglite') {
// Wrap in transaction with extended timeout for Supabase compatibility.
// SET LOCAL scopes the timeout to this transaction only.
await engine.transaction(async (tx) => {
if (engine.kind === 'postgres') {
try {
await tx.runMigration(m.version, "SET LOCAL statement_timeout = '600000'");
} catch {
// Non-fatal: PGLite or older Postgres versions may not support this
}
}
await tx.runMigration(m.version, sql);
});
} else {
// Postgres + transaction:false → can't use SET LOCAL (needs a txn),
// can't use plain SET on the pooled connection (leaks to other
// queries). Instead: reserve a dedicated backend, set session-level
// statement_timeout on just that connection, run the DDL there.
//
// On Supabase (both PgBouncer 6543 and direct 5432) a server-level
// statement_timeout of ~2 min is enforced. Without this override a
// CREATE INDEX CONCURRENTLY on a large table (e.g. 500K pages) hits
// the timeout and aborts. SET on the reserved connection cleanly
// overrides because the GUC scope is connection-local (session-scope
// is fine when nobody else uses the connection).
//
// The reserved-connection primitive is new in PR #356. See
// BrainEngine.withReservedConnection.
await engine.withReservedConnection(async (conn) => {
try {
await conn.executeRaw("SET statement_timeout = '600000'");
} catch {
// Non-fatal: some managed Postgres may restrict this GUC.
// Falling through means the DDL runs with the server default.
}
await conn.executeRaw(sql);
});
}
}
export async function runMigrations(engine: BrainEngine): Promise<{ applied: number; current: number }> {
const currentStr = await engine.getConfig('version');
const current = parseInt(currentStr || '1', 10);
@@ -771,39 +936,59 @@ export async function runMigrations(engine: BrainEngine): Promise<{ applied: num
// be skipped on the next iteration.
const sorted = [...MIGRATIONS].sort((a, b) => a.version - b.version);
let applied = 0;
for (const m of sorted) {
if (m.version > current) {
// Pick SQL: engine-specific `sqlFor` wins over engine-agnostic `sql`.
const sql = m.sqlFor?.[engine.kind] ?? m.sql;
if (sql) {
const useTransaction = m.transaction !== false;
// Non-transactional path is Postgres-only: `CREATE INDEX CONCURRENTLY`
// refuses to run inside a transaction. PGLite has no concurrent
// writers, so even if a migration sets transaction:false we wrap it
// anyway (harmless; keeps behavior consistent).
if (useTransaction || engine.kind === 'pglite') {
await engine.transaction(async (tx) => {
await tx.runMigration(m.version, sql);
});
} else {
// Postgres + transaction:false → direct execution, no BEGIN/COMMIT.
await engine.runMigration(m.version, sql);
}
}
// Application-level handler (runs outside transaction for flexibility)
if (m.handler) {
await m.handler(engine);
}
// Update version after both SQL and handler succeed
await engine.setConfig('version', String(m.version));
console.log(` Migration ${m.version} applied: ${m.name}`);
applied++;
}
const pending = sorted.filter(m => m.version > current);
if (pending.length === 0) {
return { applied: 0, current };
}
return { applied, current: applied > 0 ? MIGRATIONS[MIGRATIONS.length - 1].version : current };
console.log(` Schema version ${current} → ${LATEST_VERSION} (${pending.length} migration(s) pending)`);
// Pre-flight: warn about connections that might block DDL
await checkForBlockingConnections(engine);
let applied = 0;
for (const m of pending) {
console.log(` [${m.version}] ${m.name}...`);
// Pick SQL: engine-specific `sqlFor` wins over engine-agnostic `sql`.
const sql = m.sqlFor?.[engine.kind] ?? m.sql;
if (sql) {
try {
await runMigrationSQL(engine, m, sql);
} catch (err: unknown) {
// Actionable diagnostics for statement timeout (Postgres error 57014).
// Shape matches the 4-part error standard (what / why / fix / verify).
const code = (err as { code?: string })?.code;
if (code === '57014') {
console.error(`\n❌ Migration ${m.version} (${m.name}) hit statement_timeout (SQLSTATE 57014).`);
console.error('');
console.error(' Cause: another connection holds a lock on the target table, or the');
console.error(' server statement_timeout (~2 min on Supabase) is too short for this DDL.');
console.error('');
console.error(' Fix:');
console.error(' 1. gbrain doctor --locks # find idle-in-transaction blockers');
console.error(' 2. Terminate blocker(s) shown by step 1 via pg_terminate_backend(<pid>)');
console.error(' 3. gbrain apply-migrations --yes # re-run from the version that failed');
console.error('');
console.error(' Verify:');
console.error(' gbrain doctor # schema_version should match latest');
console.error('');
}
throw err;
}
}
// Application-level handler (runs outside transaction for flexibility)
if (m.handler) {
await m.handler(engine);
}
// Update version after both SQL and handler succeed
await engine.setConfig('version', String(m.version));
console.log(` [${m.version}] ✓ ${m.name}`);
applied++;
}
return { applied, current: LATEST_VERSION };
}
src/core/pglite-engine.ts
+14 -1
View File
@@ -2,7 +2,7 @@ import { PGlite } from '@electric-sql/pglite';
import { vector } from '@electric-sql/pglite/vector';
import { pg_trgm } from '@electric-sql/pglite/contrib/pg_trgm';
import type { Transaction } from '@electric-sql/pglite';
import type { BrainEngine, LinkBatchInput, TimelineBatchInput } from './engine.ts';
import type { BrainEngine, LinkBatchInput, TimelineBatchInput, ReservedConnection } from './engine.ts';
import { MAX_SEARCH_LIMIT, clampSearchLimit } from './engine.ts';
import { runMigrations } from './migrate.ts';
import { PGLITE_SCHEMA_SQL } from './pglite-schema.ts';
@@ -92,6 +92,19 @@ export class PGLiteEngine implements BrainEngine {
}
}
async withReservedConnection<T>(fn: (conn: ReservedConnection) => Promise<T>): Promise<T> {
// PGLite has no connection pool. The single backing connection is
// always effectively reserved — pass it through.
const db = this.db;
const conn: ReservedConnection = {
async executeRaw<R = Record<string, unknown>>(sql: string, params?: unknown[]): Promise<R[]> {
const { rows } = await db.query(sql, params);
return rows as R[];
},
};
return fn(conn);
}
async transaction<T>(fn: (engine: BrainEngine) => Promise<T>): Promise<T> {
return this.db.transaction(async (tx) => {
const txEngine = Object.create(this) as PGLiteEngine;
src/core/postgres-engine.ts
+20 -1
View File
@@ -1,5 +1,5 @@
import postgres from 'postgres';
import type { BrainEngine, LinkBatchInput, TimelineBatchInput } from './engine.ts';
import type { BrainEngine, LinkBatchInput, TimelineBatchInput, ReservedConnection } from './engine.ts';
import { MAX_SEARCH_LIMIT, clampSearchLimit } from './engine.ts';
import { runMigrations } from './migrate.ts';
import { SCHEMA_SQL } from './schema-embedded.ts';
@@ -54,6 +54,7 @@ export class PostgresEngine implements BrainEngine {
}
this._sql = postgres(url, opts);
await this._sql`SELECT 1`;
await db.setSessionDefaults(this._sql);
} else {
// Module-level singleton (backward compat for CLI main engine)
await db.connect(config);
@@ -98,6 +99,24 @@ export class PostgresEngine implements BrainEngine {
}) as Promise<T>;
}
async withReservedConnection<T>(fn: (conn: ReservedConnection) => Promise<T>): Promise<T> {
const pool = this._sql || db.getConnection();
const reserved = await pool.reserve();
try {
const conn: ReservedConnection = {
async executeRaw<R = Record<string, unknown>>(query: string, params?: unknown[]): Promise<R[]> {
const rows = params === undefined
? await reserved.unsafe(query)
: await reserved.unsafe(query, params as Parameters<typeof reserved.unsafe>[1]);
return rows as unknown as R[];
},
};
return await fn(conn);
} finally {
reserved.release();
}
}
// Pages CRUD
async getPage(slug: string): Promise<Page | null> {
const sql = this.sql;