root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

FN-8451: prevent embedded Postgres double starts

Browse Source 
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) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
2026-07-21 10:33:28 -07:00
parent 5f0502e166
commit 76cecacddd
2 changed files with 155 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

119
packages/core/src/__tests__/postgres/embedded-lifecycle.test.ts
View File

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

49
packages/core/src/postgres/embedded-lifecycle.ts
View File

@@ -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<void> {
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;
/*