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);