From a3dda8eaffc8cc33f21b6c1927fbbd3089978975 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Sat, 18 Jul 2026 17:12:27 -0700 Subject: [PATCH] fix(core): prevent concurrent startup database failures (#2330) ## Summary Concurrent PostgreSQL project initialization no longer causes transient dashboard failures, including repeated `GET /api/remote/status` 500 responses. The failure was a database deadlock between project-row identity promotion and schema/plugin DDL, which previously acquired overlapping locks in inconsistent orders. This establishes one advisory-lock order across SQLite cutover, project identity promotion, and schema mutations. Focused regression coverage proves schema DDL waits behind an active migration transaction and that identity stamping acquires the migration lock before reading project-owned tables. ## Validation - 25 focused unit tests passed. - 3 focused real-PostgreSQL regression tests passed. - `@fusion/core` typecheck passed. - Strict changeset validation passed. - Fast workspace verification passed, including the CLI build and boot health check. --- [![Compound Engineering](https://img.shields.io/badge/Built_with-Compound_Engineering-6366f1)](https://github.com/EveryInc/compound-engineering-plugin) ## Summary by CodeRabbit * **Bug Fixes** * Prevented transient dashboard failures caused by PostgreSQL startup and migration deadlocks. * Improved serialization when multiple projects initialize or update database schemas concurrently. * Ensured migration state updates and schema changes occur in a consistent order. * **Tests** * Added coverage for migration lock ordering, concurrent schema operations, and recovery after lock contention. --- ...fix-postgres-startup-migration-deadlock.md | 7 +++ .../migration-stamping-lock-order.test.ts | 49 +++++++++++++++++++ .../postgres/plugin-schema-hook.test.ts | 33 ++++++++++--- .../__tests__/postgres/schema-applier.test.ts | 44 +++++++++++++++++ packages/core/src/postgres/advisory-locks.ts | 21 ++++++++ .../core/src/postgres/migration-stamping.ts | 7 ++- .../core/src/postgres/plugin-schema-hook.ts | 3 +- packages/core/src/postgres/schema-applier.ts | 3 +- packages/core/src/postgres/sqlite-migrator.ts | 6 ++- 9 files changed, 161 insertions(+), 12 deletions(-) create mode 100644 .changeset/fix-postgres-startup-migration-deadlock.md create mode 100644 packages/core/src/__tests__/postgres/migration-stamping-lock-order.test.ts create mode 100644 packages/core/src/postgres/advisory-locks.ts diff --git a/.changeset/fix-postgres-startup-migration-deadlock.md b/.changeset/fix-postgres-startup-migration-deadlock.md new file mode 100644 index 0000000000..1ea84c6e59 --- /dev/null +++ b/.changeset/fix-postgres-startup-migration-deadlock.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Prevent transient dashboard failures when multiple projects initialize PostgreSQL concurrently. +category: fix +dev: Uses one advisory-lock order for schema DDL, SQLite cutover, and project identity promotion. diff --git a/packages/core/src/__tests__/postgres/migration-stamping-lock-order.test.ts b/packages/core/src/__tests__/postgres/migration-stamping-lock-order.test.ts new file mode 100644 index 0000000000..5ccb22e0d2 --- /dev/null +++ b/packages/core/src/__tests__/postgres/migration-stamping-lock-order.test.ts @@ -0,0 +1,49 @@ +import { describe, expect, it, vi } from "vitest"; +import { + rekeyFallbackProjectPartition, + stampMigratedProjectRows, +} from "../../postgres/migration-stamping.js"; + +function queryText(query: unknown): string { + return (query as { queryChunks?: Array<{ value: string[] }> }).queryChunks + ?.flatMap((chunk) => chunk.value).join("") ?? ""; +} + +function recordingDb(statements: string[]) { + const execute = vi.fn(async (query: unknown) => { + statements.push(queryText(query)); + return []; + }); + return { + transaction: vi.fn(async (callback: (tx: { execute: typeof execute }) => Promise) => ( + callback({ execute }) + )), + }; +} + +describe("migration stamping advisory-lock order", () => { + it("locks out schema DDL before scanning every project-owned table", async () => { + const statements: string[] = []; + + await rekeyFallbackProjectPartition( + recordingDb(statements) as never, + "local-fallback", + "project-registered", + ); + + expect(statements[0]).toContain("fusion:sqlite-migration-state"); + expect(statements[1]).toContain("information_schema.columns"); + }); + + it("locks out schema DDL before stamping migrated rows across tables", async () => { + const statements: string[] = []; + + await stampMigratedProjectRows(recordingDb(statements) as never, { + projectId: "project-registered", + rootDir: "/project", + }); + + expect(statements[0]).toContain("fusion:sqlite-migration-state"); + expect(statements[1]).toContain("fusion_sqlite_migrations"); + }); +}); diff --git a/packages/core/src/__tests__/postgres/plugin-schema-hook.test.ts b/packages/core/src/__tests__/postgres/plugin-schema-hook.test.ts index d69a561766..580a7e662a 100644 --- a/packages/core/src/__tests__/postgres/plugin-schema-hook.test.ts +++ b/packages/core/src/__tests__/postgres/plugin-schema-hook.test.ts @@ -29,6 +29,20 @@ function executedSql(execute: ReturnType): string { } describe("PostgreSQL plugin schema registry", () => { + it("acquires the SQLite migration-state lock before the schema DDL lock", async () => { + const execute = vi.fn().mockResolvedValue([]); + + await runLoadedPluginSchemaInitHooks(transactionalDb(execute) as never, [{ + pluginId: "fusion-plugin-roadmap", + hook: vi.fn(), + }]); + + const statements = executedSql(execute); + expect(statements.indexOf("fusion:sqlite-migration-state")).toBeGreaterThanOrEqual(0); + expect(statements.indexOf("fusion:sqlite-migration-state")) + .toBeLessThan(statements.indexOf("fusion:schema-applier")); + }); + /* FNXC:PluginPostgresSchema 2026-07-14-18:45: Every bundled legacy onSchemaInit declaration requires either a named default @@ -94,7 +108,7 @@ describe("PostgreSQL plugin schema registry", () => { postgresSchema: definition, }]); - expect(execute).toHaveBeenCalledTimes(4); + expect(execute).toHaveBeenCalledTimes(5); }); it("rejects unscoped or privileged third-party DDL", () => { @@ -146,8 +160,8 @@ describe("PostgreSQL plugin schema registry", () => { }, }]); - expect(execute).toHaveBeenCalledTimes(3); - const envelope = (execute.mock.calls[2]?.[0] as { queryChunks: Array<{ value: string[] }> }) + expect(execute).toHaveBeenCalledTimes(4); + const envelope = (execute.mock.calls[3]?.[0] as { queryChunks: Array<{ value: string[] }> }) .queryChunks.flatMap((chunk) => chunk.value).join(""); expect(envelope).toContain('FORCE ROW LEVEL SECURITY'); expect(envelope).toContain('project."external_fixture_rows"'); @@ -187,7 +201,7 @@ describe("PostgreSQL plugin schema registry", () => { expect(committed).toEqual([]); }); - it("serializes concurrent contracts with the schema-applier advisory lock", async () => { + it("serializes concurrent contracts with the migration-state then schema advisory locks", async () => { const events: string[] = []; let release: (() => void) | undefined; let held = false; @@ -208,10 +222,12 @@ describe("PostgreSQL plugin schema registry", () => { execute: async (query: unknown) => { const text = (query as { queryChunks: Array<{ value: string[] }> }).queryChunks .flatMap((chunk) => chunk.value).join(""); - if (text.includes("pg_advisory_xact_lock")) { + if (text.includes("fusion:sqlite-migration-state")) { await acquire(); ownsLock = true; - events.push("lock"); + events.push("migration-lock"); + } else if (text.includes("fusion:schema-applier")) { + events.push("schema-lock"); } else { const table = text.match(/external_fixture_(one|two)/)?.[1] ?? "unknown"; events.push(table); @@ -235,7 +251,10 @@ describe("PostgreSQL plugin schema registry", () => { }]); await Promise.all([contract("one"), contract("two")]); - expect(events).toEqual(["lock", "one", "one", "lock", "two", "two"]); + expect(events).toEqual([ + "migration-lock", "schema-lock", "one", "one", + "migration-lock", "schema-lock", "two", "two", + ]); }); it("repairs Roadmap ownership outside legacy foreign keys and restores composite relationships", async () => { diff --git a/packages/core/src/__tests__/postgres/schema-applier.test.ts b/packages/core/src/__tests__/postgres/schema-applier.test.ts index f30007e103..ac12ac5d51 100644 --- a/packages/core/src/__tests__/postgres/schema-applier.test.ts +++ b/packages/core/src/__tests__/postgres/schema-applier.test.ts @@ -1221,6 +1221,50 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => { ]); }); + it("queues schema DDL behind an active SQLite migration transaction", async () => { + ctx = await setupFreshDb(); + const migrationSql = postgres(ctx.testUrl, { max: 1, prepare: false, onnotice: () => {} }); + const schemaSql = postgres(ctx.testUrl, { + max: 1, + prepare: false, + onnotice: () => {}, + connection: { lock_timeout: 100 }, + }); + const schemaDb = drizzle(schemaSql); + let releaseMigration!: () => void; + let migrationLockAcquired!: () => void; + const release = new Promise((resolve) => { releaseMigration = resolve; }); + const acquired = new Promise((resolve) => { migrationLockAcquired = resolve; }); + const holder = migrationSql.begin(async (tx) => { + await tx`SELECT pg_advisory_xact_lock(hashtext('fusion:sqlite-migration-state'))`; + migrationLockAcquired(); + await release; + }); + + try { + await acquired; + try { + let lockError: unknown; + try { + await applySchemaBaseline(schemaDb, { pluginHooks: [] }); + } catch (error) { + lockError = error; + } + expect(lockError).toBeInstanceOf(Error); + expect((lockError as Error & { cause?: { code?: string } }).cause?.code).toBe("55P03"); + } finally { + releaseMigration(); + await holder; + } + expect((await applySchemaBaseline(schemaDb, { pluginHooks: [] })).applied).toBe(true); + } finally { + releaseMigration(); + await holder.catch(() => undefined); + await migrationSql.end({ timeout: 5 }); + await schemaSql.end({ timeout: 5 }); + } + }); + /* FNXC:GitHubImportTranslate 2026-07-16-23:30: An upgrade has already recorded 0010, so editing its SQL only fixes fresh diff --git a/packages/core/src/postgres/advisory-locks.ts b/packages/core/src/postgres/advisory-locks.ts new file mode 100644 index 0000000000..4b6735df03 --- /dev/null +++ b/packages/core/src/postgres/advisory-locks.ts @@ -0,0 +1,21 @@ +import { sql } from "drizzle-orm"; +import type { PostgresJsDatabase } from "drizzle-orm/postgres-js"; + +type AdvisoryLockTransaction = Pick>, "execute">; + +/** + * Serialize schema DDL behind any active SQLite cutover transaction. + * + * SQLite migration takes `fusion:sqlite-migration-state` before it reads the + * target schema. Schema application and runtime plugin DDL must take the same + * lock first, then their narrower schema lock, so PostgreSQL never sees the + * inverse DDL/read lock order that can deadlock concurrent project startup. + */ +export async function acquireSqliteMigrationStateLock(tx: AdvisoryLockTransaction): Promise { + await tx.execute(sql`SELECT pg_advisory_xact_lock(hashtext('fusion:sqlite-migration-state'))`); +} + +export async function acquireSchemaMutationLocks(tx: AdvisoryLockTransaction): Promise { + await acquireSqliteMigrationStateLock(tx); + await tx.execute(sql`SELECT pg_advisory_xact_lock(hashtext('fusion:schema-applier'))`); +} diff --git a/packages/core/src/postgres/migration-stamping.ts b/packages/core/src/postgres/migration-stamping.ts index 167a9edef7..1b5f1ac511 100644 --- a/packages/core/src/postgres/migration-stamping.ts +++ b/packages/core/src/postgres/migration-stamping.ts @@ -19,6 +19,7 @@ */ import { sql } from "drizzle-orm"; +import { acquireSqliteMigrationStateLock } from "./advisory-locks.js"; import type { PostgresJsDatabase } from "drizzle-orm/postgres-js"; /** The Drizzle instance type startup-factory uses for its `connections.migration`. */ @@ -63,6 +64,7 @@ export async function rekeyFallbackProjectPartition( if (!fallbackProjectId || fallbackProjectId === registeredProjectId) return false; return db.transaction(async (tx) => { + await acquireSqliteMigrationStateLock(tx); const tables = (await tx.execute(sql` SELECT table_name FROM information_schema.columns @@ -197,7 +199,10 @@ export async function stampMigratedProjectRows( FNXC:ProjectMigrationStamping 2026-07-14-21:55: Partition stamping is one atomic promotion. If any table cannot be re-keyed, roll back every earlier update so startup never exposes a partially migrated project identity. */ - return db.transaction((tx) => stampMigratedProjectRowsWithinTransaction(tx, projectId, rootDir)); + return db.transaction(async (tx) => { + await acquireSqliteMigrationStateLock(tx); + return stampMigratedProjectRowsWithinTransaction(tx, projectId, rootDir); + }); } async function stampMigratedProjectRowsWithinTransaction( diff --git a/packages/core/src/postgres/plugin-schema-hook.ts b/packages/core/src/postgres/plugin-schema-hook.ts index eb1f8cdea7..03056bbd4d 100644 --- a/packages/core/src/postgres/plugin-schema-hook.ts +++ b/packages/core/src/postgres/plugin-schema-hook.ts @@ -17,6 +17,7 @@ import type { PostgresJsDatabase } from "drizzle-orm/postgres-js"; import { sql } from "drizzle-orm"; +import { acquireSchemaMutationLocks } from "./advisory-locks.js"; import type { PluginPostgresSchemaDefinition } from "../plugin-types.js"; export interface LoadedPluginSchemaContract { @@ -989,7 +990,7 @@ export async function runLoadedPluginSchemaInitHooks( Runtime load and hot reload share the schema-applier advisory lock. Each contract and its complete isolation envelope commit atomically, so concurrent Fusion processes serialize DDL and a rejected reload cannot leave partially-created or temporarily unprotected tables behind. */ await db.transaction(async (tx) => { - await tx.execute(sql`SELECT pg_advisory_xact_lock(hashtext('fusion:schema-applier'))`); + await acquireSchemaMutationLocks(tx); if (loaded.postgresSchema) { const tables = new Set(); for (const statement of loaded.postgresSchema.statements) { diff --git a/packages/core/src/postgres/schema-applier.ts b/packages/core/src/postgres/schema-applier.ts index 98be8d6f38..4b6a5b5857 100644 --- a/packages/core/src/postgres/schema-applier.ts +++ b/packages/core/src/postgres/schema-applier.ts @@ -26,6 +26,7 @@ import { fileURLToPath } from "node:url"; import type { PostgresJsDatabase } from "drizzle-orm/postgres-js"; import { sql } from "drizzle-orm"; import { runPluginSchemaInitHooks, DEFAULT_PLUGIN_SCHEMA_INIT_HOOKS, type PluginSchemaInitHook } from "./plugin-schema-hook.js"; +import { acquireSchemaMutationLocks } from "./advisory-locks.js"; /** The latest PostgreSQL schema version known to this applier. */ /* @@ -274,7 +275,7 @@ export async function applySchemaBaseline( * cannot both apply a version or race its primary-key marker. */ return db.transaction(async (tx) => { - await tx.execute(sql`SELECT pg_advisory_xact_lock(hashtext('fusion:schema-applier'))`); + await acquireSchemaMutationLocks(tx); await ensureBookkeepingTable(tx); /* FNXC:PostgresSchema 2026-07-16-00:55: diff --git a/packages/core/src/postgres/sqlite-migrator.ts b/packages/core/src/postgres/sqlite-migrator.ts index c0b26ead35..bd80550944 100644 --- a/packages/core/src/postgres/sqlite-migrator.ts +++ b/packages/core/src/postgres/sqlite-migrator.ts @@ -59,6 +59,7 @@ import { createHash } from "node:crypto"; import { basename, dirname, resolve } from "node:path"; import { existsSync } from "node:fs"; import { applySchemaBaseline } from "./schema-applier.js"; +import { acquireSqliteMigrationStateLock } from "./advisory-locks.js"; import { PROJECT_SCHEMA, CENTRAL_SCHEMA, @@ -423,7 +424,7 @@ export async function migrateSqliteToPostgres( }); if (options.dryRun !== true) { await migrationDb.transaction(async (tx) => { - await tx.execute(sql`SELECT pg_advisory_xact_lock(hashtext('fusion:sqlite-migration-state'))`); + await acquireSqliteMigrationStateLock(tx); await ensureMigrationStateTable(tx); await tx.execute(sql` INSERT INTO public.${sql.identifier(SQLITE_MIGRATION_STATE_TABLE)} @@ -462,7 +463,7 @@ async function migrateSqliteToPostgresOnSession( * copied task as proof that the whole migration finished. */ if (!dryRun) { - await migrationDb.execute(sql`SELECT pg_advisory_xact_lock(hashtext('fusion:sqlite-migration-state'))`); + await acquireSqliteMigrationStateLock(migrationDb); await ensureMigrationStateTable(migrationDb); /* * FNXC:PostgresMigrationSession 2026-07-14-00:14: @@ -892,6 +893,7 @@ async function migrateLegacyProjectPluginRowsOnSession( projectPath: string, ): Promise { if (!sqliteTableExists(sqlitePath, "plugins")) return; + await acquireSqliteMigrationStateLock(db); const canonicalProjectPath = resolve(projectPath); const migrationKey = `project-plugins:${canonicalProjectPath}`; await ensureMigrationStateTable(db);