From 76cecacdddfd81b28d34b005a9a3953d9a4504db Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Tue, 21 Jul 2026 10:33:28 -0700 Subject: [PATCH] FN-8451: prevent embedded Postgres double starts Prevent TaskStore boot collisions when an existing PostgreSQL pid file cannot yet be parsed. - Read live postmaster pid files asynchronously with bounded retries. - Fail closed when a present pid file has no readable port. - Cover join, unreadable pid, and fresh-start lifecycle paths. Files changed: packages/core/src/__tests__/postgres/embedded-lifecycle.test.ts | 119 +++++++++++++++++++++ packages/core/src/postgres/embedded-lifecycle.ts | 49 ++++++--- 2 files changed, 155 insertions(+), 13 deletions(-) Fusion-Task-Id: FN-8451 Fusion-Task-Lineage: 823a874d-97ef-4380-97ce-e41981f660ff Co-authored-by: Fusion (runfusion.ai) --- .../postgres/embedded-lifecycle.test.ts | 119 ++++++++++++++++++ .../core/src/postgres/embedded-lifecycle.ts | 49 ++++++-- 2 files changed, 155 insertions(+), 13 deletions(-) diff --git a/packages/core/src/__tests__/postgres/embedded-lifecycle.test.ts b/packages/core/src/__tests__/postgres/embedded-lifecycle.test.ts index ec19d438e8..c2fcb1e37d 100644 --- a/packages/core/src/__tests__/postgres/embedded-lifecycle.test.ts +++ b/packages/core/src/__tests__/postgres/embedded-lifecycle.test.ts @@ -1109,6 +1109,125 @@ describe("embedded-lifecycle: readPortFromPostmasterPid (P1 code-review fix)", ( * both startup-factory boot and direct lifecycle callers. Mock starts reject * before database work so these tests stay deterministic and process-free. */ +/* + * FNXC:PostgresEmbedded 2026-07-21-10:00: + * A shared data directory's pid file is the cross-process join contract. These + * process-free tests make the original extension timeout reproducible: the real + * layout joins using index 3, while a persistent unreadable lock file must never + * instantiate a colliding embedded-postgres start. + */ +describe("embedded-lifecycle: postmaster.pid join safety", () => { + it("joins an issue-shaped live pid without instantiating a second postmaster", async () => { + const dataDir = makeDataDir(); + const ctor = vi.fn(); + class UnexpectedEmbeddedPostgres { + constructor() { + ctor(); + } + initialise = vi.fn(async () => {}); + start = vi.fn(async () => {}); + stop = vi.fn(async () => {}); + } + __setEmbeddedPostgresCtorForTests(UnexpectedEmbeddedPostgres as never); + const ensureJoinedDatabase = vi + .spyOn(EmbeddedPostgresLifecycle.prototype, "ensureJoinedDatabase") + .mockResolvedValue(undefined); + try { + writeFileSync( + join(dataDir, "postmaster.pid"), + ["1866", dataDir, "1784424901", "34643", "/tmp", "localhost", "79484 2", "ready"].join("\n") + "\n", + ); + const lifecycle = new EmbeddedPostgresLifecycle(baseOptions(dataDir)); + + await expect(lifecycle.start()).resolves.toMatchObject({ + runtimeUrl: expect.stringContaining(":34643/"), + }); + expect(ensureJoinedDatabase).toHaveBeenCalledWith(34643); + expect(ctor).not.toHaveBeenCalled(); + expect(lifecycle.getOwnsProcess()).toBe(false); + } finally { + rmSync(dataDir, { recursive: true, force: true }); + } + }); + + it("fails closed for an unparseable present pid without starting PostgreSQL", async () => { + const dataDir = makeDataDir(); + const ctor = vi.fn(); + class UnexpectedEmbeddedPostgres { + constructor() { + ctor(); + } + initialise = vi.fn(async () => {}); + start = vi.fn(async () => {}); + stop = vi.fn(async () => {}); + } + __setEmbeddedPostgresCtorForTests(UnexpectedEmbeddedPostgres as never); + try { + // `/tmp` at index 3 mirrors the prior off-by-one regression shape. + writeFileSync( + join(dataDir, "postmaster.pid"), + ["1866", dataDir, "1784424901", "/tmp", "localhost", "79484 2", "ready"].join("\n") + "\n", + ); + const lifecycle = new EmbeddedPostgresLifecycle(baseOptions(dataDir)); + + await expect(lifecycle.start()).rejects.toThrow( + /postmaster\.pid is present but its port could not be read.*second postmaster will not be started/i, + ); + expect(ctor).not.toHaveBeenCalled(); + } finally { + rmSync(dataDir, { recursive: true, force: true }); + } + }); + + it("fails closed for an empty present pid without starting PostgreSQL", async () => { + const dataDir = makeDataDir(); + const ctor = vi.fn(); + class UnexpectedEmbeddedPostgres { + constructor() { + ctor(); + } + initialise = vi.fn(async () => {}); + start = vi.fn(async () => {}); + stop = vi.fn(async () => {}); + } + __setEmbeddedPostgresCtorForTests(UnexpectedEmbeddedPostgres as never); + try { + writeFileSync(join(dataDir, "postmaster.pid"), ""); + const lifecycle = new EmbeddedPostgresLifecycle(baseOptions(dataDir)); + + await expect(lifecycle.start()).rejects.toThrow(/postmaster\.pid is present but its port could not be read/i); + expect(ctor).not.toHaveBeenCalled(); + } finally { + rmSync(dataDir, { recursive: true, force: true }); + } + }); + + it("allows a fresh start when postmaster.pid is absent", async () => { + const dataDir = makeDataDir(); + const ctor = vi.fn(); + const expected = new Error("mock start reached"); + class RecordingEmbeddedPostgres { + constructor() { + ctor(); + } + initialise = vi.fn(async () => {}); + start = vi.fn(async () => { + throw expected; + }); + stop = vi.fn(async () => {}); + } + __setEmbeddedPostgresCtorForTests(RecordingEmbeddedPostgres as never); + try { + const lifecycle = new EmbeddedPostgresLifecycle({ ...baseOptions(dataDir), startTimeoutMs: 0 }); + + await expect(lifecycle.start()).rejects.toBe(expected); + expect(ctor).toHaveBeenCalledOnce(); + } finally { + rmSync(dataDir, { recursive: true, force: true }); + } + }); +}); + describe("embedded-lifecycle: shared-memory-safe postgres flags", () => { const sentinel = new Error("mock postgres start complete"); diff --git a/packages/core/src/postgres/embedded-lifecycle.ts b/packages/core/src/postgres/embedded-lifecycle.ts index fa259599d2..ad2f0ab679 100644 --- a/packages/core/src/postgres/embedded-lifecycle.ts +++ b/packages/core/src/postgres/embedded-lifecycle.ts @@ -1000,22 +1000,45 @@ function isDuplicateDatabaseError(error: unknown): boolean { return code === "23505" && constraint === "pg_database_datname_index"; } -function isAlreadyRunning(dataDir: string): { port: number; database: string } | null { - // Check in-process registry first +const POSTMASTER_PID_READ_ATTEMPTS = 3; +const POSTMASTER_PID_READ_RETRY_MS = 10; + +function waitForPostmasterPidReread(): Promise { + return new Promise((resolve) => setTimeout(resolve, POSTMASTER_PID_READ_RETRY_MS)); +} + +/** + * Resolve an instance another lifecycle/process already owns. + * + * FNXC:PostgresEmbedded 2026-07-21-10:00: + * A present `postmaster.pid` is a live-lock signal, not permission to launch a + * second postmaster. PostgreSQL can write it while a joiner reads, so retry a + * small bounded number of times; if its port remains unreadable, fail closed + * instead of turning parser-null into the double-boot collision that exhausts + * extension TaskStore's caller budget. + */ +async function isAlreadyRunning(dataDir: string): Promise<{ port: number; database: string } | null> { + // Check in-process registry first. const cached = runningInstances.get(dataDir); if (cached) return cached; - // Check postmaster.pid — another process (or a prior call) may have started PG - if (!existsSync(join(dataDir, "postmaster.pid"))) return null; + const pidPath = join(dataDir, "postmaster.pid"); + if (!existsSync(pidPath)) return null; - // Read the port from postmaster.pid - const port = readPortFromPostmasterPid(dataDir); - if (!port) return null; + for (let attempt = 0; attempt < POSTMASTER_PID_READ_ATTEMPTS; attempt += 1) { + const port = readPortFromPostmasterPid(dataDir); + if (port) { + // Probe: can we connect to this port? We return it optimistically; the + // connection layer reports a stale-but-parseable pid without a second start. + return { port, database: "fusion" }; + } + if (!existsSync(pidPath)) return null; + if (attempt < POSTMASTER_PID_READ_ATTEMPTS - 1) await waitForPostmasterPidReread(); + } - // Probe: can we connect to this port? - // We return the port optimistically — the connection layer will fail fast - // if the port is stale (postmaster.pid left over from a crash). - return { port, database: "fusion" }; + throw new Error( + `embedded postgres: postmaster.pid is present but its port could not be read after ${POSTMASTER_PID_READ_ATTEMPTS} attempts; a second postmaster will not be started (data dir ${dataDir})`, + ); } /** @@ -1202,7 +1225,7 @@ export class EmbeddedPostgresLifecycle { // FNXC:PostgresCutover 2026-06-27-11:05: // Check if PG is already running for this data dir. If so, reuse it. - const existing = isAlreadyRunning(this.options.dataDir); + const existing = await isAlreadyRunning(this.options.dataDir); if (existing) { this.options.onLog( `embedded postgres: already running on port ${existing.port} (data dir ${this.options.dataDir}), connecting without starting a new instance`, @@ -1369,7 +1392,7 @@ export class EmbeddedPostgresLifecycle { and orphan a live postmaster nothing would ever stop. See isPostgresLockCollisionError. */ if (!isPostgresLockCollisionError(error)) throw error; - const existing = isAlreadyRunning(this.options.dataDir); + const existing = await isAlreadyRunning(this.options.dataDir); if (!existing) throw error; /*