diff --git a/cli/vitest.config.ts b/cli/vitest.config.ts index f624398e8de..13b72103824 100644 --- a/cli/vitest.config.ts +++ b/cli/vitest.config.ts @@ -3,5 +3,7 @@ import { defineConfig } from "vitest/config"; export default defineConfig({ test: { environment: "node", + hookTimeout: 30_000, + teardownTimeout: 30_000, }, }); diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index d2058a79615..cf2a16e3734 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -1058,8 +1058,55 @@ For normal issue work, start with the smallest targeted check that proves the ch #### Embedded Postgres template cache +To run the suite with one owned PostgreSQL server and a separate disposable +database for each fixture, use: + +```sh +node cli/node_modules/tsx/dist/cli.mjs packages/db/src/test-postgres-command.ts +node cli/node_modules/tsx/dist/cli.mjs packages/db/src/test-postgres-command.ts -- pnpm exec vitest run --project @paperclipai/db +``` + +The command starts an embedded server, prepares a migrated database template, +passes it to the child command, and waits for server shutdown before returning. +It preserves the child command's exit status. On cancellation it stops and waits +for the test process group before shutting down PostgreSQL. Database, server, +and CLI fixtures all use the common implementation. Tests of embedded startup itself continue to +start dedicated clusters. + +To reuse a disposable server that you already manage, set +`PAPERCLIP_TEST_POSTGRES_ADMIN_URL` to its administrative PostgreSQL URL. The role +must be able to create databases. Each fixture creates a random database and +closes only its clients before dropping it. It never stops the external server. +`DATABASE_URL` does not select this test path. The optional +`PAPERCLIP_TEST_POSTGRES_TEMPLATE` can name an existing migrated database with +the form `paperclip_test_template_`; `0` disables template reuse. + +On a supported platform, PostgreSQL startup failures fail test collection. +Resource exhaustion, unavailable configured servers, and broken native binaries +do not make integration tests skip. A genuinely unsupported platform or CPU can +still skip embedded PostgreSQL coverage. Startup retries apply only to port +collisions. Teardown waits for process exit, requests fast shutdown first, and +uses immediate PostgreSQL shutdown if needed. A failed stop fails the hook and +retains its data directory. Worker termination also requests owned server +shutdown. A forced `SIGKILL` can bypass these handlers; the separate host cleanup +job remains the backstop for stale macOS semaphore sets. +Database and CLI hooks use the same 30-second deadline as server hooks. This +allows the ten-second shutdown protocol and client/database cleanup to finish +before the test runner ends a worker. A shutdown timeout still fails the hook. +Template publication uses `pg_ctl` to request clean fast shutdown on every +supported platform and allows thirty seconds for it. A template is never +published after immediate shutdown or a crash. Cancelling the shared-server +command also waits for its owned cluster-directory cleanup after the test +processes and PostgreSQL exit. Direct worker termination stops PostgreSQL but +retains its disposable files for inspection; the synchronous exit guard cannot +await directory removal. +Stopped cluster directories are removed asynchronously, and cleanup awaits their +removal. Large suites can leave many relation files; deleting them must not block +the worker's event loop. The idle-sleep suite uses the existing 90-second fixture +budget for its final directory cleanup after hundreds of table truncations. + Suites that call `startEmbeddedPostgresTestDatabase` (from `@paperclipai/db`) -get their own embedded Postgres cluster. The first such suite on a host builds +without an administrative URL get their own embedded Postgres cluster. The first such suite on a host builds one fully migrated cluster and publishes its data directory under `$TMPDIR/paperclip-embedded-postgres-templates-/`. Every later suite copies that template and starts Postgres on the copy instead of running diff --git a/packages/db/src/__fixtures__/test-command-tree.mjs b/packages/db/src/__fixtures__/test-command-tree.mjs new file mode 100644 index 00000000000..ea63970c050 --- /dev/null +++ b/packages/db/src/__fixtures__/test-command-tree.mjs @@ -0,0 +1,31 @@ +import { spawn } from "node:child_process"; +import fs from "node:fs"; +import { setTimeout } from "node:timers/promises"; +import { fileURLToPath } from "node:url"; +import postgres from "postgres"; + +if (process.argv[2] === "root") { + spawn(process.execPath, [fileURLToPath(import.meta.url), "grandchild", String(process.pid)], { stdio: "inherit" }); + for (const signal of ["SIGTERM", "SIGINT"]) process.on(signal, () => process.exit(0)); +} else { + const sql = postgres(process.env.PAPERCLIP_TEST_POSTGRES_ADMIN_URL, { max: 1 }); + await sql`SELECT 1`; + fs.writeFileSync(process.env.TEST_COMMAND_TREE_READY, JSON.stringify({ root: Number(process.argv[3]), grandchild: process.pid })); + let stopping = false; + const stop = async () => { + if (stopping) return; + stopping = true; + try { + await setTimeout(200); + await sql`SELECT 1`; + fs.writeFileSync(process.env.TEST_COMMAND_TREE_DONE, "database available during child shutdown"); + await sql.end(); + process.exit(0); + } catch (error) { + console.error(error); + process.exit(1); + } + }; + for (const signal of ["SIGTERM", "SIGINT"]) process.on(signal, () => { void stop(); }); + setInterval(() => {}, 1000); +} diff --git a/packages/db/src/__fixtures__/test-postgres-exit.mjs b/packages/db/src/__fixtures__/test-postgres-exit.mjs new file mode 100644 index 00000000000..52002262f79 --- /dev/null +++ b/packages/db/src/__fixtures__/test-postgres-exit.mjs @@ -0,0 +1,9 @@ +import fs from "node:fs"; +import path from "node:path"; +import { __startEmbeddedPostgresWithRetryForTests as start } from "../test-embedded-postgres.ts"; + +const server = await start("paperclip-owner-process-"); +const pid = Number(fs.readFileSync(path.join(server.dataDir, "postmaster.pid"), "utf8").split("\n")[0]); +process.send({ pid, dataDir: server.dataDir }, () => { + if (process.argv[2] === "exit") process.exit(0); +}); diff --git a/packages/db/src/embedded-postgres-error.test.ts b/packages/db/src/embedded-postgres-error.test.ts index dba1ad46154..60b928ffcbe 100644 --- a/packages/db/src/embedded-postgres-error.test.ts +++ b/packages/db/src/embedded-postgres-error.test.ts @@ -2,6 +2,15 @@ import { describe, expect, it } from "vitest"; import { createEmbeddedPostgresLogBuffer, formatEmbeddedPostgresError } from "./embedded-postgres-error.js"; describe("formatEmbeddedPostgresError", () => { + it("identifies semaphore exhaustion without recommending SQL lock changes", () => { + const error = formatEmbeddedPostgresError(new Error(), { + fallbackMessage: "startup failed", + recentLogs: ["FATAL: could not create semaphores: No space left on device", "DETAIL: semget(123, 17, 03600)"], + }); + expect(error.message).toContain("operating-system semaphores"); + expect(error.message).toContain("integration tests must not be skipped"); + expect(error.message).toContain("not disk exhaustion or a SQL lock limit"); + }); it("adds a shared-memory hint when initdb logs expose the real cause", () => { const error = formatEmbeddedPostgresError("Postgres init script exited with code 1.", { fallbackMessage: "Failed to initialize embedded PostgreSQL cluster", diff --git a/packages/db/src/embedded-postgres-error.ts b/packages/db/src/embedded-postgres-error.ts index 9862a0f3be1..a297be143a3 100644 --- a/packages/db/src/embedded-postgres-error.ts +++ b/packages/db/src/embedded-postgres-error.ts @@ -24,6 +24,11 @@ function summarizeRecentLogs(recentLogs: string[]): string | null { function detectEmbeddedPostgresHint(recentLogs: string[]): string | null { const haystack = recentLogs.join("\n").toLowerCase(); + if (haystack.includes("could not create semaphores") || haystack.includes("semget(")) { + return "Embedded PostgreSQL could not allocate operating-system semaphores. " + + "Check for stale PostgreSQL IPC resources and unfinished test servers. " + + "This is not disk exhaustion or a SQL lock limit; integration tests must not be skipped."; + } if (!haystack.includes("could not create shared memory segment")) { return null; } @@ -75,7 +80,7 @@ export function formatEmbeddedPostgresError( const baseError = toError(error, input.fallbackMessage); const recentLogs = input.recentLogs ?? []; const parts = [baseError.message]; - const hint = detectEmbeddedPostgresHint(recentLogs); + const hint = detectEmbeddedPostgresHint([baseError.message, ...recentLogs]); const recentSummary = summarizeRecentLogs(recentLogs); if (hint) { diff --git a/packages/db/src/test-embedded-postgres.test.ts b/packages/db/src/test-embedded-postgres.test.ts index aeec953d73d..84856638701 100644 --- a/packages/db/src/test-embedded-postgres.test.ts +++ b/packages/db/src/test-embedded-postgres.test.ts @@ -2,8 +2,9 @@ import fs from "node:fs"; import os from "node:os"; import path from "node:path"; import postgres from "postgres"; -import { afterEach, describe, expect, it } from "vitest"; +import { afterAll, afterEach, describe, expect, it } from "vitest"; import { inspectMigrations } from "./client.js"; +import { drainTestPostgresOwners, TestPostgresShutdownError } from "./test-postgres-owner.js"; import { __embeddedPostgresStartMaxAttemptsForTests as MAX_ATTEMPTS, __embeddedPostgresTemplateRootDirForTests as templateRootDir, @@ -18,6 +19,12 @@ import { startEmbeddedPostgresTestDatabase, } from "./test-embedded-postgres.js"; +const originalAdminUrl = process.env.PAPERCLIP_TEST_POSTGRES_ADMIN_URL; +delete process.env.PAPERCLIP_TEST_POSTGRES_ADMIN_URL; +afterAll(() => { + if (originalAdminUrl !== undefined) process.env.PAPERCLIP_TEST_POSTGRES_ADMIN_URL = originalAdminUrl; +}); + // A fake embedded-postgres constructor. It records every constructed instance so // the test can assert the retry uses a fresh port and a fresh data directory each // attempt. `start()` emits the same output the real cluster writes for a port @@ -266,6 +273,29 @@ describeWithPostgres("startEmbeddedPostgresTestDatabase template path", () => { else process.env.PAPERCLIP_TEST_POSTGRES_TEMPLATE = originalTemplateSetting; }); + it("does not start a replacement when a copied cluster cannot stop", { timeout: EMBEDDED_POSTGRES_TEST_TIMEOUT_MS }, async () => { + delete process.env.PAPERCLIP_TEST_POSTGRES_TEMPLATE; + const ready = await startEmbeddedPostgresTestDatabase("paperclip-template-stop-ready-"); + await ready.cleanup(); + const { ctor, constructed } = makeFakeCtor(Number.POSITIVE_INFINITY); + let stops = 0; + class FailedStopPostgres extends ctor { + async stop(): Promise { + if (++stops === 1) throw new Error("owned stop failed"); + } + } + __setEmbeddedPostgresCtorProviderForTests(async () => FailedStopPostgres); + try { + await expect(startEmbeddedPostgresTestDatabase("paperclip-template-stop-fail-")).rejects.toBeInstanceOf(TestPostgresShutdownError); + expect(constructed).toHaveLength(1); + expect(fs.existsSync(constructed[0].databaseDir)).toBe(true); + } finally { + await drainTestPostgresOwners(); + __setEmbeddedPostgresCtorProviderForTests(null); + for (const options of constructed) fs.rmSync(path.dirname(options.databaseDir), { recursive: true, force: true }); + } + }); + it("publishes a complete template and starts suites from a copy of it", { timeout: EMBEDDED_POSTGRES_TEST_TIMEOUT_MS }, async () => { // This test covers the template path, so it enables the cache even when // the suite runs under PAPERCLIP_TEST_POSTGRES_TEMPLATE=0. The first @@ -452,6 +482,44 @@ describe("support probe when the template cannot be built", () => { resetTemplateState(); }); + it("fails the suite when PostgreSQL cannot allocate semaphores", async () => { + const original = process.env.PAPERCLIP_TEST_POSTGRES_TEMPLATE; + process.env.PAPERCLIP_TEST_POSTGRES_TEMPLATE = "0"; + const { ctor, constructed } = makeFakeCtor(Number.POSITIVE_INFINITY); + class ExhaustedPostgres extends ctor { + async start(): Promise { + throw new Error("could not create semaphores: No space left on device (semget)"); + } + } + __setEmbeddedPostgresCtorProviderForTests(async () => ExhaustedPostgres); + resetTemplateState(); + try { + await expect(getEmbeddedPostgresTestSupport()).rejects.toThrow("could not create semaphores"); + expect(constructed).toHaveLength(1); + } finally { + if (original === undefined) delete process.env.PAPERCLIP_TEST_POSTGRES_TEMPLATE; + else process.env.PAPERCLIP_TEST_POSTGRES_TEMPLATE = original; + } + }); + + it("still reports an explicitly unsupported platform as unsupported", async () => { + const original = process.env.PAPERCLIP_TEST_POSTGRES_TEMPLATE; + process.env.PAPERCLIP_TEST_POSTGRES_TEMPLATE = "0"; + __setEmbeddedPostgresCtorProviderForTests(async () => { + throw new Error('Unsupported arch "test-cpu" for platform "test-os"'); + }); + resetTemplateState(); + try { + expect(await getEmbeddedPostgresTestSupport()).toMatchObject({ + supported: false, + reason: expect.stringContaining('Unsupported arch "test-cpu"'), + }); + } finally { + if (original === undefined) delete process.env.PAPERCLIP_TEST_POSTGRES_TEMPLATE; + else process.env.PAPERCLIP_TEST_POSTGRES_TEMPLATE = original; + } + }); + it("falls through to the fresh probe instead of reporting the host as unsupported", { timeout: EMBEDDED_POSTGRES_TEST_TIMEOUT_MS }, async () => { const originalTemplateSetting = process.env.PAPERCLIP_TEST_POSTGRES_TEMPLATE; delete process.env.PAPERCLIP_TEST_POSTGRES_TEMPLATE; diff --git a/packages/db/src/test-embedded-postgres.ts b/packages/db/src/test-embedded-postgres.ts index 862829c8234..c2a4b4fa984 100644 --- a/packages/db/src/test-embedded-postgres.ts +++ b/packages/db/src/test-embedded-postgres.ts @@ -1,3 +1,4 @@ +import { execFile } from "node:child_process"; import { createHash } from "node:crypto"; import fs from "node:fs"; import { createRequire } from "node:module"; @@ -5,12 +6,16 @@ import net from "node:net"; import os from "node:os"; import path from "node:path"; import { fileURLToPath } from "node:url"; +import { promisify } from "node:util"; import { applyPendingMigrations, closeRegisteredClients, ensurePostgresDatabase } from "./client.js"; import { createEmbeddedPostgresLogBuffer, formatEmbeddedPostgresError, } from "./embedded-postgres-error.js"; import { prepareEmbeddedPostgresNativeRuntime } from "./embedded-postgres-native.js"; +import { loadWithoutEmbeddedPostgresExitHooks } from "./embedded-postgres-lifecycle.js"; +import { ownTestPostgres, TestPostgresShutdownError, type TestPostgresInstance } from "./test-postgres-owner.js"; +import { checkTestPostgresAdmin, startTestPostgresDatabase, testPostgresAdminUrl } from "./test-postgres-admin.js"; // Time budget (ms) for a vitest test in the embedded-Postgres cost class: a // test that starts an embedded Postgres cluster and runs migrations. Measured @@ -20,11 +25,7 @@ import { prepareEmbeddedPostgresNativeRuntime } from "./embedded-postgres-native // passes while a genuine hang still fails fast. export const EMBEDDED_POSTGRES_TEST_TIMEOUT_MS = 90_000; -type EmbeddedPostgresInstance = { - initialise(): Promise; - start(): Promise; - stop(): Promise; -}; +type EmbeddedPostgresInstance = ReturnType; type EmbeddedPostgresCtor = new (opts: { databaseDir: string; @@ -35,7 +36,7 @@ type EmbeddedPostgresCtor = new (opts: { initdbFlags?: string[]; onLog?: (message: unknown) => void; onError?: (message: unknown) => void; -}) => EmbeddedPostgresInstance; +}) => TestPostgresInstance; export type EmbeddedPostgresTestSupport = { supported: boolean; @@ -65,9 +66,9 @@ function getReservedTestPorts(): Set { type EmbeddedPostgresCtorProvider = () => Promise; async function loadEmbeddedPostgresCtor(): Promise { - const mod = await import("embedded-postgres"); + const mod = await loadWithoutEmbeddedPostgresExitHooks(() => import("embedded-postgres")); await prepareEmbeddedPostgresNativeRuntime(); - return mod.default as EmbeddedPostgresCtor; + return mod.default as unknown as EmbeddedPostgresCtor; } let embeddedPostgresCtorProvider: EmbeddedPostgresCtorProvider = loadEmbeddedPostgresCtor; @@ -122,7 +123,12 @@ const EMBEDDED_POSTGRES_INITDB_FLAGS = ["--encoding=UTF8", "--locale=C", "--lc-m async function createEmbeddedPostgresTestInstance(tempDirPrefix: string) { const dataDir = fs.mkdtempSync(path.join(os.tmpdir(), tempDirPrefix)); - return await createEmbeddedPostgresTestInstanceAt(dataDir); + try { + return await createEmbeddedPostgresTestInstanceAt(dataDir); + } catch (error) { + await cleanupEmbeddedPostgresTestDirs(dataDir); + throw error; + } } // Wraps an embedded-postgres instance around `dataDir`. The caller decides @@ -136,7 +142,7 @@ async function createEmbeddedPostgresTestInstanceAt(dataDir: string) { // The `start()` rejection carries an empty message, so we capture the output // in a bounded buffer and surface it in the thrown error. const logBuffer = createEmbeddedPostgresLogBuffer(); - const instance = new EmbeddedPostgres({ + const instance = ownTestPostgres(new EmbeddedPostgres({ databaseDir: dataDir, user: EMBEDDED_POSTGRES_USER, password: EMBEDDED_POSTGRES_PASSWORD, @@ -145,7 +151,7 @@ async function createEmbeddedPostgresTestInstanceAt(dataDir: string) { initdbFlags: EMBEDDED_POSTGRES_INITDB_FLAGS, onLog: (message) => logBuffer.append(message), onError: (message) => logBuffer.append(message), - }); + }), dataDir); return { dataDir, port, instance, getRecentLogs: () => logBuffer.getRecentLogs() }; } @@ -158,63 +164,24 @@ function databaseConnectionStringFor(port: number): string { return `postgres://${EMBEDDED_POSTGRES_USER}:${EMBEDDED_POSTGRES_PASSWORD}@127.0.0.1:${port}/paperclip`; } -function cleanupEmbeddedPostgresTestDirs(dataDir: string) { - fs.rmSync(dataDir, { recursive: true, force: true }); +async function cleanupEmbeddedPostgresTestDirs(dataDir: string): Promise { + if (process.platform !== "win32") { + try { + await promisify(execFile)("/bin/rm", ["-rf", "--", dataDir], { maxBuffer: 64 * 1024 }); + return; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; + } + } + await fs.promises.rm(dataDir, { recursive: true, force: true }); } -// Upper bound (ms) on how long we wait for the embedded Postgres cluster to -// stop gracefully before abandoning the wait and returning from the hook. -const EMBEDDED_POSTGRES_STOP_TIMEOUT_MS = 5000; - -// `embedded-postgres@18.1.0-beta.16` exposes only `stop(): Promise` — no -// shutdown-mode argument. Internally it SIGINTs the postgres process (already -// PostgreSQL "fast shutdown") and resolves *only* on the child's `exit` event, -// with no time bound of its own. Under the loaded serial server shard a slow -// shutdown checkpoint can push that past vitest's hookTimeout and hang the -// afterAll hook. So we bound the graceful stop: if it overruns, we stop waiting -// and return so the hook completes. The SIGINT has already been delivered, so -// the abandoned process still exits on its own (and again when the runner exits). -// Errors are swallowed, matching prior behavior. -// -// `cleanupFn` (data-dir reclaim) is chained on the raw `stop()` promise, not on -// the timeout race, so the disposable data dir is removed *only after* `stop()` -// actually settles — i.e. once the child Postgres process has exited. Removing -// it on the timeout path would pull the data files out from under a still-running -// cluster and provoke checkpoint / WAL I/O errors. In the fast path `cleanupFn` -// has run by the time this resolves; in the timeout path it runs asynchronously -// once the abandoned process finally exits. async function stopEmbeddedPostgresBounded( instance: EmbeddedPostgresInstance | null, - cleanupFn?: () => void, + cleanupFn?: () => Promise, ): Promise { - if (!instance) { - cleanupFn?.(); - return; - } - let timer: ReturnType | undefined; - const stopped = instance - .stop() - .catch(() => { - // Swallow shutdown errors — the data dir is reclaimed regardless. - }) - .finally(() => { - try { - cleanupFn?.(); - } catch { - // Best-effort reclaim; ignore removal errors. - } - }); - try { - await Promise.race([ - stopped, - new Promise((resolve) => { - timer = setTimeout(resolve, EMBEDDED_POSTGRES_STOP_TIMEOUT_MS); - timer.unref?.(); - }), - ]); - } finally { - if (timer) clearTimeout(timer); - } + await instance?.stop(); + await cleanupFn?.(); } // Upper bound on start attempts. `getAvailablePort` uses a check-then-use probe: @@ -251,6 +218,7 @@ async function startEmbeddedPostgresWithRetry(tempDirPrefix: string): Promise<{ await stopEmbeddedPostgresBounded(created.instance, () => cleanupEmbeddedPostgresTestDirs(created.dataDir), ); + if (!/address already in use/i.test(lastError.message)) throw lastError; } } @@ -306,10 +274,6 @@ const TEMPLATE_ROOT_DIR_NAME = "paperclip-embedded-postgres-templates"; // Templates whose key no longer matches are pruned once they have not been // used for this long. Every reuse refreshes the template's timestamp. const TEMPLATE_PRUNE_AGE_MS = 7 * 24 * 60 * 60 * 1000; -// Publishing a template requires the builder cluster to have exited for real, -// so its data files are quiescent before the rename. Give that stop more room -// than the per-suite bounded stop; a template build happens once per host. -const TEMPLATE_BUILD_STOP_TIMEOUT_MS = 30_000; export type EmbeddedPostgresStartMode = "template" | "fresh"; @@ -482,18 +446,10 @@ function pruneStaleTemplates(rootDir: string, keepDir: string): void { // Waits for the raw `stop()` so the builder cluster has really exited before // its data directory is published. Returns false if the stop overran. async function stopForTemplatePublish(instance: EmbeddedPostgresInstance): Promise { - let timer: ReturnType | undefined; try { - return await Promise.race([ - instance.stop().then(() => true, () => false), - new Promise((resolve) => { - timer = setTimeout(() => resolve(false), TEMPLATE_BUILD_STOP_TIMEOUT_MS); - timer.unref?.(); - }), - ]); - } finally { - if (timer) clearTimeout(timer); - } + await instance.stopGracefully(); + return true; + } catch { return false; } } // Builds a migrated cluster and publishes its data directory at `templateDir`. @@ -533,12 +489,6 @@ async function buildEmbeddedPostgresTemplate(templateDir: string): Promise | null = null; -// Resolves the template data directory for this host, building it on first -// use. Resolves null when the cache is disabled, the cache root is not private -// to this user, or the template could not be built or published; every caller -// then takes the fresh path, which reports the real error itself (a cluster -// that cannot start, a migration that cannot apply). Never rejects, so a -// broken cache can never turn into `supported: false` and skip a suite. async function getEmbeddedPostgresTemplateDir(): Promise { if (isTemplateCacheDisabled()) return null; if (!templateDirPromise) { @@ -554,6 +504,7 @@ async function getEmbeddedPostgresTemplateDir(): Promise { const built = await buildEmbeddedPostgresTemplate(templateDir); return built && isUsableTemplate(templateDir) ? templateDir : null; } catch (error) { + if (error instanceof TestPostgresShutdownError) throw error; warnTemplateOnce( "build", `template build failed, using fresh clusters: ${(error as Error).message}`, @@ -604,6 +555,7 @@ async function startEmbeddedPostgresFromTemplateWithRetry( await stopEmbeddedPostgresBounded(created?.instance ?? null, () => cleanupEmbeddedPostgresTestDirs(rootDir), ); + if (!/address already in use/i.test(lastError.message)) throw lastError; } } @@ -636,12 +588,6 @@ export function __resetEmbeddedPostgresTemplateStateForTests(): void { } async function probeEmbeddedPostgresSupport(): Promise { - // A published template proves this host can run the cluster: it was built - // here. Reusing it makes the probe free for every file after the first. - // Building it costs one cluster start, the same as the throwaway probe did. - // When the template cannot be built or used, the throwaway probe below - // decides, so only a cluster that cannot start reports `supported: false`; - // a migration that cannot apply fails the suite loudly on the fresh path. if ((await getEmbeddedPostgresTemplateDir()) !== null) return { supported: true }; let started: { dataDir: string; instance: EmbeddedPostgresInstance } | null = null; @@ -650,12 +596,13 @@ async function probeEmbeddedPostgresSupport(): Promise { + const adminUrl = testPostgresAdminUrl(); + if (adminUrl) { + await checkTestPostgresAdmin(adminUrl); + return { supported: true }; + } if (!embeddedPostgresSupportPromise) { embeddedPostgresSupportPromise = probeEmbeddedPostgresSupport(); } @@ -684,6 +636,7 @@ async function startEmbeddedPostgresClusterForSuite( const started = await startEmbeddedPostgresFromTemplateWithRetry(templateDir, tempDirPrefix); return { ...started, mode: "template" }; } catch (error) { + if (error instanceof TestPostgresShutdownError) throw error; // The next caller in this process re-resolves the template: if it was // pruned it is rebuilt, instead of retrying a path that is gone. resetTemplateDirCache(); @@ -705,6 +658,8 @@ async function startEmbeddedPostgresClusterForSuite( export async function startEmbeddedPostgresTestDatabase( tempDirPrefix: string, ): Promise { + const adminUrl = testPostgresAdminUrl(); + if (adminUrl) return startTestPostgresDatabase(adminUrl); const { port, dataDir, instance, mode } = await startEmbeddedPostgresClusterForSuite(tempDirPrefix); lastStartMode = mode; @@ -723,8 +678,10 @@ export async function startEmbeddedPostgresTestDatabase( // stops can crash the process: the stop kills the backend socket, // but a queued write on that connection still fires later and finds // a null socket. - await closeRegisteredClients(connectionString); - await stopEmbeddedPostgresBounded(instance, () => cleanupEmbeddedPostgresTestDirs(dataDir)); + try { await closeRegisteredClients(connectionString); } + finally { + await stopEmbeddedPostgresBounded(instance, () => cleanupEmbeddedPostgresTestDirs(dataDir)); + } }, }; } catch (error) { diff --git a/packages/db/src/test-postgres-admin.test.ts b/packages/db/src/test-postgres-admin.test.ts new file mode 100644 index 00000000000..7498a4a467c --- /dev/null +++ b/packages/db/src/test-postgres-admin.test.ts @@ -0,0 +1,86 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { createDb } from "./client.js"; +import { + __setEmbeddedPostgresCtorProviderForTests, + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, + type EmbeddedPostgresTestDatabase, +} from "./test-embedded-postgres.js"; + +const previousAdmin = process.env.PAPERCLIP_TEST_POSTGRES_ADMIN_URL; +const previousTemplate = process.env.PAPERCLIP_TEST_POSTGRES_TEMPLATE; +let server: EmbeddedPostgresTestDatabase; +const databases: EmbeddedPostgresTestDatabase[] = []; +const support = await getEmbeddedPostgresTestSupport(); + +describe.skipIf(!support.supported)("shared PostgreSQL fixtures", () => { + beforeAll(async () => { + server = await startEmbeddedPostgresTestDatabase("paperclip-admin-regression-"); + process.env.PAPERCLIP_TEST_POSTGRES_ADMIN_URL = server.connectionString; + }, 90_000); + + afterEach(async () => { + __setEmbeddedPostgresCtorProviderForTests(null); + process.env.PAPERCLIP_TEST_POSTGRES_ADMIN_URL = server.connectionString; + if (previousTemplate === undefined) delete process.env.PAPERCLIP_TEST_POSTGRES_TEMPLATE; + else process.env.PAPERCLIP_TEST_POSTGRES_TEMPLATE = previousTemplate; + while (databases.length) await databases.pop()!.cleanup(); + }); + + afterAll(async () => { + await server?.cleanup(); + if (previousAdmin === undefined) delete process.env.PAPERCLIP_TEST_POSTGRES_ADMIN_URL; + else process.env.PAPERCLIP_TEST_POSTGRES_ADMIN_URL = previousAdmin; + if (previousTemplate === undefined) delete process.env.PAPERCLIP_TEST_POSTGRES_TEMPLATE; + else process.env.PAPERCLIP_TEST_POSTGRES_TEMPLATE = previousTemplate; + }); + + it("uses one server for isolated databases and closes clients before dropping them", async () => { + __setEmbeddedPostgresCtorProviderForTests(async () => { throw new Error("Must not start another PostgreSQL server"); }); + expect(await getEmbeddedPostgresTestSupport()).toEqual({ supported: true }); + const first = await startEmbeddedPostgresTestDatabase("ignored-"); databases.push(first); + const second = await startEmbeddedPostgresTestDatabase("ignored-"); databases.push(second); + expect(first.connectionString).not.toBe(second.connectionString); + const a = createDb(first.connectionString), b = createDb(second.connectionString); + await a.$client`CREATE TABLE isolated_probe (id integer)`; + expect((await b.$client`SELECT to_regclass('isolated_probe') AS name`)[0]?.name).toBeNull(); + await first.cleanup(); + await first.cleanup(); + const control = createDb(server.connectionString); + try { + const name = new URL(first.connectionString).pathname.slice(1); + expect(await control.$client`SELECT datname FROM pg_database WHERE datname = ${name}`).toHaveLength(0); + expect((await b.$client`SELECT 1 AS value`)[0]?.value).toBe(1); + } finally { await control.$client.end(); } + }, 90_000); + + it("rejects unavailable configured servers rather than reporting support", async () => { + process.env.PAPERCLIP_TEST_POSTGRES_ADMIN_URL = "postgres://paperclip:paperclip@127.0.0.1:1/postgres"; + await expect(getEmbeddedPostgresTestSupport()).rejects.toThrow(); + }); + + it("rejects unsafe templates and treats zero as a disabled template", async () => { + process.env.PAPERCLIP_TEST_POSTGRES_TEMPLATE = 'postgres"; DROP DATABASE postgres; --'; + await expect(startEmbeddedPostgresTestDatabase("ignored-")).rejects.toThrow("Invalid disposable test template name"); + process.env.PAPERCLIP_TEST_POSTGRES_TEMPLATE = "0"; + databases.push(await startEmbeddedPostgresTestDatabase("ignored-")); + }, 90_000); + + it("applies pending migrations when a supplied database template is not current", async () => { + const template = `paperclip_test_template_${randomUUID().replaceAll("-", "")}`; + const control = createDb(server.connectionString); + await control.$client.unsafe(`CREATE DATABASE "${template}"`); + try { + process.env.PAPERCLIP_TEST_POSTGRES_TEMPLATE = template; + const database = await startEmbeddedPostgresTestDatabase("ignored-"); + databases.push(database); + const db = createDb(database.connectionString); + expect((await db.$client`SELECT to_regclass('companies') AS name`)[0]?.name).toBe("companies"); + } finally { + await control.$client.unsafe(`DROP DATABASE "${template}"`); + await control.$client.end(); + } + }, 90_000); + +}); diff --git a/packages/db/src/test-postgres-admin.ts b/packages/db/src/test-postgres-admin.ts new file mode 100644 index 00000000000..a76848fcc05 --- /dev/null +++ b/packages/db/src/test-postgres-admin.ts @@ -0,0 +1,50 @@ +import { randomUUID } from "node:crypto"; +import postgres from "postgres"; +import { applyPendingMigrations, closeRegisteredClients } from "./client.js"; + +export function testPostgresAdminUrl(): string | undefined { + const value = process.env.PAPERCLIP_TEST_POSTGRES_ADMIN_URL?.trim(); + if (!value) return undefined; + const url = new URL(value); + if (url.protocol !== "postgres:" && url.protocol !== "postgresql:") { + throw new Error("PAPERCLIP_TEST_POSTGRES_ADMIN_URL must be a PostgreSQL URL"); + } + return value; +} + +export async function checkTestPostgresAdmin(url: string): Promise { + const sql = postgres(url, { max: 1, connect_timeout: 5, onnotice() {} }); + try { await sql`SELECT 1`; } + finally { await sql.end({ timeout: 1 }); } +} + +export async function startTestPostgresDatabase(adminUrl: string): Promise<{ + connectionString: string; + cleanup(): Promise; +}> { + const name = `paperclip_test_${randomUUID().replaceAll("-", "")}`; + const setting = process.env.PAPERCLIP_TEST_POSTGRES_TEMPLATE; + const template = setting === "0" ? undefined : setting; + if (template && !/^paperclip_test_template_[a-f0-9]+$/.test(template)) { + throw new Error("Invalid disposable test template name"); + } + const admin = postgres(adminUrl, { max: 1, connect_timeout: 5, onnotice() {} }); + const url = new URL(adminUrl); + url.pathname = `/${name}`; + const connectionString = url.toString(); + let cleaning: Promise | null = null; + const cleanup = () => cleaning ??= (async () => { + try { + await closeRegisteredClients(connectionString, { scope: "database" }); + await admin.unsafe(`DROP DATABASE IF EXISTS "${name}" WITH (FORCE)`); + } finally { await admin.end({ timeout: 1 }); } + })(); + try { + await admin.unsafe(`CREATE DATABASE "${name}"${template ? ` TEMPLATE "${template}"` : ""}`); + await applyPendingMigrations(connectionString); + return { connectionString, cleanup }; + } catch (error) { + await cleanup(); + throw error; + } +} diff --git a/packages/db/src/test-postgres-command.test.ts b/packages/db/src/test-postgres-command.test.ts new file mode 100644 index 00000000000..6795fec273c --- /dev/null +++ b/packages/db/src/test-postgres-command.test.ts @@ -0,0 +1,57 @@ +import { execFile, spawn } from "node:child_process"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { setTimeout } from "node:timers/promises"; +import { fileURLToPath } from "node:url"; +import { promisify } from "node:util"; +import { expect, it } from "vitest"; +import { getEmbeddedPostgresTestSupport } from "./test-embedded-postgres.js"; + +const previousAdmin = process.env.PAPERCLIP_TEST_POSTGRES_ADMIN_URL; +delete process.env.PAPERCLIP_TEST_POSTGRES_ADMIN_URL; +let supported = false; +try { supported = process.platform !== "win32" && (await getEmbeddedPostgresTestSupport()).supported; } +finally { + if (previousAdmin !== undefined) process.env.PAPERCLIP_TEST_POSTGRES_ADMIN_URL = previousAdmin; +} +const execute = promisify(execFile); + +it.skipIf(!supported).each(["SIGTERM", "SIGINT"] as const)("drains the test command's descendants before PostgreSQL on %s", async (signal) => { + const directory = fs.mkdtempSync(path.join(os.tmpdir(), "paperclip-command-cancel-")); + const ready = path.join(directory, "ready"), done = path.join(directory, "done"); + const env: NodeJS.ProcessEnv = { + ...process.env, TEST_COMMAND_TREE_READY: ready, TEST_COMMAND_TREE_DONE: done, + TMPDIR: directory, TEMP: directory, TMP: directory, + }; + delete env.PAPERCLIP_TEST_POSTGRES_ADMIN_URL; + delete env.PAPERCLIP_TEST_POSTGRES_TEMPLATE; + const wrapper = spawn(process.execPath, [ + "--import", fileURLToPath(new URL("../node_modules/tsx/dist/loader.mjs", import.meta.url)), + fileURLToPath(new URL("./test-postgres-command.ts", import.meta.url)), "--", process.execPath, + fileURLToPath(new URL("./__fixtures__/test-command-tree.mjs", import.meta.url)), "root", + ], { env, stdio: ["ignore", "pipe", "pipe"] }); + let output = ""; + wrapper.stdout.on("data", (data) => { output = (output + String(data)).slice(-8000); }); + wrapper.stderr.on("data", (data) => { output = (output + String(data)).slice(-8000); }); + const exited = new Promise((resolve, reject) => { + wrapper.once("error", reject); + wrapper.once("exit", (code) => resolve(code)); + }); + try { + for (let i = 0; i < 1200 && !fs.existsSync(ready) && wrapper.exitCode === null; i++) await setTimeout(25); + expect(fs.existsSync(ready), output).toBe(true); + const descendants = JSON.parse(fs.readFileSync(ready, "utf8")) as { root: number; grandchild: number }; + wrapper.kill(signal); + expect(await exited, output).toBe(signal === "SIGINT" ? 130 : 143); + expect(fs.readFileSync(done, "utf8")).toBe("database available during child shutdown"); + expect(fs.readdirSync(directory).filter((name) => name.startsWith("paperclip-test-command-"))).toEqual([]); + for (const pid of Object.values(descendants)) { + const result = await execute("/bin/ps", ["-p", String(pid), "-o", "stat="]).catch(() => ({ stdout: "" })); + expect(result.stdout.trim() === "" || result.stdout.trim().startsWith("Z"), `descendant ${pid} still running`).toBe(true); + } + } finally { + if (wrapper.exitCode === null && wrapper.signalCode === null) { wrapper.kill("SIGTERM"); await exited; } + fs.rmSync(directory, { recursive: true, force: true }); + } +}, 90_000); diff --git a/packages/db/src/test-postgres-command.ts b/packages/db/src/test-postgres-command.ts new file mode 100644 index 00000000000..5edb86f2d5a --- /dev/null +++ b/packages/db/src/test-postgres-command.ts @@ -0,0 +1,46 @@ +import path from "node:path"; +import { createDb } from "./client.js"; +import { computeEmbeddedPostgresTemplateKey, startEmbeddedPostgresTestDatabase } from "./test-embedded-postgres.js"; +import { checkTestPostgresAdmin, testPostgresAdminUrl } from "./test-postgres-admin.js"; +import { beforeTestPostgresShutdown } from "./test-postgres-owner.js"; +import { startTestProcessTree } from "./test-process-tree.js"; + +const args = process.argv.slice(2); +if (args[0] === "--") args.shift(); +const command = args.shift() ?? process.execPath; +if (command === process.execPath && args.length === 0) args.push(path.resolve("scripts/run-vitest-stable.mjs")); + +let server: Awaited> | undefined; +try { + const env = { ...process.env }; + let adminUrl = testPostgresAdminUrl(); + if (adminUrl) await checkTestPostgresAdmin(adminUrl); + else { + server = await startEmbeddedPostgresTestDatabase("paperclip-test-command-"); + const url = new URL(server.connectionString); + url.pathname = "/postgres"; + adminUrl = url.toString(); + if (env.PAPERCLIP_TEST_POSTGRES_TEMPLATE !== "0") { + const template = `paperclip_test_template_${computeEmbeddedPostgresTemplateKey()}`; + const admin = createDb(adminUrl, { maxConnections: 1 }); + try { await admin.$client.unsafe(`CREATE DATABASE "${template}" TEMPLATE "paperclip"`); } + finally { await admin.$client.end({ timeout: 1 }); } + env.PAPERCLIP_TEST_POSTGRES_TEMPLATE = template; + } + } + env.PAPERCLIP_TEST_POSTGRES_ADMIN_URL = adminUrl; + console.log("[test:postgres] Each suite gets a disposable database on one test PostgreSQL server"); + const child = startTestProcessTree(command, args, env); + const removeShutdownTask = beforeTestPostgresShutdown({ + ...child, + afterStop: async () => { await server?.cleanup(); }, + }); + try { + process.exitCode = await child.wait(); + } finally { + try { await child.stop(); } + finally { removeShutdownTask(); } + } +} finally { + await server?.cleanup(); +} diff --git a/packages/db/src/test-postgres-owner-process.test.ts b/packages/db/src/test-postgres-owner-process.test.ts new file mode 100644 index 00000000000..8eb7d37cb68 --- /dev/null +++ b/packages/db/src/test-postgres-owner-process.test.ts @@ -0,0 +1,54 @@ +import { fork } from "node:child_process"; +import fs from "node:fs"; +import path from "node:path"; +import { setTimeout } from "node:timers/promises"; +import { fileURLToPath } from "node:url"; +import { expect, it } from "vitest"; +import { getEmbeddedPostgresTestSupport } from "./test-embedded-postgres.js"; + +const configuredAdmin = process.env.PAPERCLIP_TEST_POSTGRES_ADMIN_URL; +delete process.env.PAPERCLIP_TEST_POSTGRES_ADMIN_URL; +let embeddedSupported = false; +try { embeddedSupported = process.platform !== "win32" && (await getEmbeddedPostgresTestSupport()).supported; } +finally { + if (configuredAdmin !== undefined) process.env.PAPERCLIP_TEST_POSTGRES_ADMIN_URL = configuredAdmin; +} + +function exists(pid: number): boolean { + try { process.kill(pid, 0); return true; } + catch (error) { return (error as NodeJS.ErrnoException).code !== "ESRCH"; } +} + +it.skipIf(!embeddedSupported).each(["exit", "SIGTERM"])("stops the owned PostgreSQL process when its worker receives %s", async (mode) => { + const child = fork(fileURLToPath(new URL("./__fixtures__/test-postgres-exit.mjs", import.meta.url)), [mode], { + execArgv: ["--import", fileURLToPath(new URL("../node_modules/tsx/dist/loader.mjs", import.meta.url))], + stdio: ["ignore", "pipe", "pipe", "ipc"], + }); + let output = ""; + child.stdout?.on("data", (data) => { output = (output + String(data)).slice(-8000); }); + child.stderr?.on("data", (data) => { output = (output + String(data)).slice(-8000); }); + const exited = new Promise((resolve, reject) => { + child.once("error", reject); + child.once("exit", (code) => resolve(code)); + }); + let server: { pid: number; dataDir: string } | undefined; + try { + const ready = await new Promise<{ pid: number; dataDir: string }>((resolve, reject) => { + child.once("message", (value) => resolve(value as { pid: number; dataDir: string })); + child.once("exit", () => reject(new Error(`Worker exited before readiness: ${output}`))); + }); + server = ready; + if (mode === "SIGTERM") child.kill("SIGTERM"); + expect(await exited, output).toBe(mode === "exit" ? 0 : 143); + for (let i = 0; i < 200 && exists(ready.pid); i++) await setTimeout(25); + expect(exists(ready.pid), output).toBe(false); + } finally { + if (child.exitCode === null && child.signalCode === null) { child.kill("SIGTERM"); await exited; } + if (server && exists(server.pid)) { + const contents = fs.readFileSync(path.join(server.dataDir, "postmaster.pid"), "utf8"); + if (Number(contents.split("\n")[0]) === server.pid) process.kill(server.pid, "SIGINT"); + for (let i = 0; i < 200 && exists(server.pid); i++) await setTimeout(25); + } + if (server && !exists(server.pid)) fs.rmSync(server.dataDir, { recursive: true, force: true }); + } +}, 30_000); diff --git a/packages/db/src/test-postgres-owner.test.ts b/packages/db/src/test-postgres-owner.test.ts new file mode 100644 index 00000000000..ab984e09849 --- /dev/null +++ b/packages/db/src/test-postgres-owner.test.ts @@ -0,0 +1,105 @@ +import { EventEmitter } from "node:events"; +import type { ChildProcess } from "node:child_process"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { ownTestPostgres } from "./test-postgres-owner.js"; + +function fixture() { + const child = Object.assign(new EventEmitter(), { + exitCode: null as number | null, + signalCode: null as NodeJS.Signals | null, + kill: vi.fn(() => true), + }); + const raw = { + initialise: vi.fn(async () => {}), + start: vi.fn(async () => {}), + stop: vi.fn(async () => {}), + process: child as unknown as ChildProcess, + }; + const owner = ownTestPostgres(raw); + function exit() { child.exitCode = 0; child.emit("exit", 0, null); } + return { child, raw, owner, exit }; +} + +afterEach(() => { vi.useRealTimers(); }); + +describe.skipIf(process.platform === "win32")("test PostgreSQL ownership", () => { + it("waits for fast shutdown and makes concurrent cleanup idempotent", async () => { + const before = process.listenerCount("exit"); + const f = fixture(); + const stop = f.owner.stop(); + expect(f.owner.stop()).toBe(stop); + expect(f.child.kill).toHaveBeenCalledWith("SIGINT"); + f.exit(); + await stop; + await f.owner.stop(); + expect(f.child.kill).toHaveBeenCalledTimes(1); + expect(process.listenerCount("exit")).toBe(before); + }); + + it("does not call the dependency's stop after the process already exited", async () => { + const f = fixture(); + f.exit(); + await f.owner.stop(); + expect(f.raw.stop).not.toHaveBeenCalled(); + expect(f.child.kill).not.toHaveBeenCalled(); + }); + + it("uses immediate PostgreSQL shutdown when fast shutdown times out", async () => { + vi.useFakeTimers(); + const f = fixture(); + const stop = f.owner.stop(); + await vi.advanceTimersByTimeAsync(5_000); + expect(f.child.kill.mock.calls).toEqual([["SIGINT"], ["SIGQUIT"]]); + f.exit(); + await stop; + expect(f.child.kill).not.toHaveBeenCalledWith("SIGKILL"); + }); + + it("reports a shutdown failure instead of declaring the process stopped", async () => { + vi.useFakeTimers(); + const f = fixture(); + const failure = expect(f.owner.stop()).rejects.toThrow("data directory is retained"); + await vi.advanceTimersByTimeAsync(10_000); + await failure; + f.exit(); + await f.owner.stop(); + }); + + it("keeps one exit hook while two servers have different owners", async () => { + const before = process.listenerCount("exit"); + const a = fixture(), b = fixture(); + expect(process.listenerCount("exit")).toBe(before + 1); + a.exit(); await a.owner.stop(); + expect(process.listenerCount("exit")).toBe(before + 1); + b.exit(); await b.owner.stop(); + expect(process.listenerCount("exit")).toBe(before); + }); + + it("never uses immediate shutdown when preparing a clean template", async () => { + vi.useFakeTimers(); + const f = fixture(); + const clean = expect(f.owner.stopGracefully()).rejects.toThrow("template cannot be published"); + await vi.advanceTimersByTimeAsync(30_000); + await clean; + expect(f.child.kill.mock.calls).toEqual([["SIGINT"]]); + f.exit(); + await f.owner.stop(); + }); + + it("rejects a crashed process as a publishable template", async () => { + const f = fixture(); + f.child.exitCode = 1; + await expect(f.owner.stopGracefully()).rejects.toThrow("without clean shutdown"); + await f.owner.stop(); + }); + + it("does not reuse an earlier immediate stop as a clean template shutdown", async () => { + vi.useFakeTimers(); + const f = fixture(); + const stopped = f.owner.stop(); + await vi.advanceTimersByTimeAsync(5_000); + f.exit(); + await stopped; + await expect(f.owner.stopGracefully()).rejects.toThrow("without clean shutdown"); + }); +}); diff --git a/packages/db/src/test-postgres-owner.ts b/packages/db/src/test-postgres-owner.ts new file mode 100644 index 00000000000..6ccc849501a --- /dev/null +++ b/packages/db/src/test-postgres-owner.ts @@ -0,0 +1,193 @@ +import type { ChildProcess } from "node:child_process"; +import { execFile, spawnSync } from "node:child_process"; +import path from "node:path"; +import { promisify } from "node:util"; + +const execute = promisify(execFile); + +export type TestPostgresInstance = { + initialise(): Promise; + start(): Promise; + stop(): Promise; + process?: ChildProcess; +}; + +type Owner = { stop(clean?: boolean): Promise; interrupt(): void }; +export class TestPostgresShutdownError extends Error {} +const owners = new Set(); +type ShutdownTask = { stop(signal: NodeJS.Signals): Promise; interrupt(): void; afterStop?(): Promise }; +const shutdownTasks = new Set(); +let shuttingDown = false; +let hooksInstalled = false; + +function interruptOwnedServers(): void { + for (const task of shutdownTasks) task.interrupt(); + for (const owner of owners) owner.interrupt(); +} + +export function beforeTestPostgresShutdown(task: ShutdownTask): () => void { + shutdownTasks.add(task); + updateHooks(); + return () => { shutdownTasks.delete(task); updateHooks(); }; +} + +export async function drainTestPostgresOwners(): Promise { + const results = await Promise.allSettled([...owners].map((owner) => owner.stop())); + const failures = results.filter((result) => result.status === "rejected"); + if (failures.length) throw new AggregateError(failures.map((result) => result.reason), "Test PostgreSQL shutdown failed"); +} + +async function shutdown(signal: "SIGINT" | "SIGTERM" | "SIGHUP"): Promise { + if (shuttingDown) { interruptOwnedServers(); return; } + shuttingDown = true; + const registered = [...shutdownTasks]; + const tasks = await Promise.allSettled(registered.map((task) => task.stop(signal))); + for (const result of tasks) { + if (result.status === "rejected") console.error(result.reason); + } + try { await drainTestPostgresOwners(); } + catch (error) { console.error(error); } + const cleanups = await Promise.allSettled(registered.map((task) => task.afterStop?.())); + for (const result of cleanups) { + if (result.status === "rejected") console.error(result.reason); + } + process.exit(128 + { SIGINT: 2, SIGTERM: 15, SIGHUP: 1 }[signal]); +} + +const onInterrupt = () => { void shutdown("SIGINT"); }; +const onTerminate = () => { void shutdown("SIGTERM"); }; +const onHangup = () => { void shutdown("SIGHUP"); }; + +function updateHooks(): void { + if ((owners.size > 0 || shutdownTasks.size > 0) && !hooksInstalled) { + hooksInstalled = true; + process.on("exit", interruptOwnedServers); + process.on("SIGINT", onInterrupt); + process.on("SIGTERM", onTerminate); + process.on("SIGHUP", onHangup); + } else if (owners.size === 0 && shutdownTasks.size === 0) { + hooksInstalled = false; + process.removeListener("exit", interruptOwnedServers); + process.removeListener("SIGINT", onInterrupt); + process.removeListener("SIGTERM", onTerminate); + process.removeListener("SIGHUP", onHangup); + } +} + +async function settlesWithin(promise: Promise, timeoutMs: number): Promise { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + promise.then(() => true), + new Promise((resolve) => { timer = setTimeout(() => resolve(false), timeoutMs); }), + ]); + } finally { + if (timer) clearTimeout(timer); + } +} + +export function ownTestPostgres(instance: TestPostgresInstance, dataDir?: string): Pick & { + stopGracefully(): Promise; +} { + let pending: Promise | null = null; + let stopPromise: Promise | null = null; + let stopped = false; + let usedImmediateShutdown = false; + let stoppedCleanly = false; + + function controlPath(child: ChildProcess): string | undefined { + if (!dataDir || !child.spawnfile) return undefined; + return path.join(path.dirname(child.spawnfile), `pg_ctl${path.extname(child.spawnfile) === ".exe" ? ".exe" : ""}`); + } + + async function requestStop(child: ChildProcess, mode: "fast" | "immediate", clean: boolean): Promise { + const control = controlPath(child); + if (control && (clean || process.platform === "win32")) { + try { + await execute(control, ["-D", dataDir!, "stop", "-m", mode, "-W"], { + timeout: 2000, maxBuffer: 64 * 1024, windowsHide: true, + }); + } catch (error) { + if (child.exitCode === null && child.signalCode === null) throw error; + } + } else child.kill(mode === "fast" ? "SIGINT" : "SIGQUIT"); + } + + async function run(operation: () => Promise): Promise { + if (stopped || stopPromise) throw new Error("Test PostgreSQL is shutting down"); + const running = operation(); + pending = running; + try { await running; } + finally { if (pending === running) pending = null; } + } + + const owner: Owner = { + interrupt() { + const child = instance.process; + if (child && child.exitCode === null && child.signalCode === null) { + usedImmediateShutdown = true; + const control = controlPath(child); + if (process.platform === "win32" && control) { + spawnSync(control, ["-D", dataDir!, "stop", "-m", "immediate", "-W"], { timeout: 2000, windowsHide: true }); + } else { + try { child.kill("SIGQUIT"); } catch {} + } + } + }, + stop(clean = false) { + if (stopped) return Promise.resolve(); + if (stopPromise) return stopPromise; + stopPromise = (async () => { + if (pending && !await settlesWithin(pending.catch(() => {}), 30_000)) { + throw new TestPostgresShutdownError("Test PostgreSQL startup did not settle before shutdown; its data directory is retained"); + } + const child = instance.process; + if (!child) { + if (!await settlesWithin(instance.stop(), 10_000)) { + throw new TestPostgresShutdownError("Test PostgreSQL shutdown timed out; its data directory is retained"); + } + } else if (child.exitCode === null && child.signalCode === null) { + let onExit: () => void; + const exited = new Promise((resolve) => { onExit = resolve; child.once("exit", onExit); }); + try { + await requestStop(child, "fast", clean); + if (!await settlesWithin(exited, clean ? 30_000 : 5_000)) { + if (clean) throw new TestPostgresShutdownError("Test PostgreSQL did not complete clean shutdown; the template cannot be published"); + usedImmediateShutdown = true; + await requestStop(child, "immediate", false); + if (!await settlesWithin(exited, 5_000)) { + throw new TestPostgresShutdownError("Test PostgreSQL did not exit after fast and immediate shutdown; its data directory is retained"); + } + } + } finally { + child.removeListener("exit", onExit!); + } + } + stoppedCleanly = !usedImmediateShutdown && (!child || (child.exitCode === 0 && child.signalCode === null)); + if (clean && !stoppedCleanly) { + throw new TestPostgresShutdownError("Test PostgreSQL exited without clean shutdown; the template cannot be published"); + } + instance.process = undefined; + stopped = true; + owners.delete(owner); + updateHooks(); + })().catch((error) => { + stopPromise = null; + if (error instanceof TestPostgresShutdownError) throw error; + throw new TestPostgresShutdownError("Test PostgreSQL shutdown failed; its data directory is retained", { cause: error }); + }); + return stopPromise; + }, + }; + owners.add(owner); + if (owners.size === 1) updateHooks(); + return { + initialise: () => run(() => instance.initialise()), + start: () => run(() => instance.start()), + stop: () => owner.stop(), + stopGracefully: async () => { + await owner.stop(true); + if (!stoppedCleanly) throw new TestPostgresShutdownError("Test PostgreSQL exited without clean shutdown; the template cannot be published"); + }, + }; +} diff --git a/packages/db/src/test-process-tree.ts b/packages/db/src/test-process-tree.ts new file mode 100644 index 00000000000..e5f805f3931 --- /dev/null +++ b/packages/db/src/test-process-tree.ts @@ -0,0 +1,78 @@ +import { execFile, spawn } from "node:child_process"; +import { constants } from "node:os"; +import { setTimeout } from "node:timers/promises"; +import { promisify } from "node:util"; + +const execute = promisify(execFile); + +export function startTestProcessTree(command: string, args: string[], env: NodeJS.ProcessEnv) { + const posix = process.platform !== "win32"; + const child = spawn(command, args, { stdio: "inherit", env, detached: posix }); + let exited = false; + let stopping: Promise | null = null; + let retired = false; + const exit = new Promise((resolve, reject) => { + child.once("error", (error) => { exited = true; reject(error); }); + child.once("exit", (code, signal) => { + exited = true; + resolve(code ?? (signal ? 128 + constants.signals[signal] : 1)); + }); + }); + + async function alive(): Promise { + if (!child.pid) return false; + if (!posix) return !exited; + const { stdout } = await execute("/bin/ps", ["-axo", "pgid=,stat="], { timeout: 2000, maxBuffer: 1024 * 1024 }); + return stdout.split("\n").some((line) => { + const [group, state] = line.trim().split(/\s+/); + return Number(group) === child.pid && state && !state.startsWith("Z"); + }); + } + + function signal(signal: NodeJS.Signals): void { + if (!child.pid || retired) return; + try { + if (posix) process.kill(-child.pid, signal); + else child.kill(signal); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; + } + } + + async function waitUntilStopped(milliseconds: number): Promise { + const deadline = Date.now() + milliseconds; + do { + if (!await alive()) return true; + await setTimeout(50); + } while (Date.now() < deadline); + return false; + } + + function stop(requested: NodeJS.Signals = "SIGTERM"): Promise { + if (stopping) return stopping; + stopping = (async () => { + if (posix) { + signal(requested); + if (!await waitUntilStopped(15_000)) { + signal("SIGKILL"); + if (!await waitUntilStopped(5_000)) throw new Error("Test command process group did not stop"); + } + } else if (child.pid && !exited) { + await execute("taskkill", ["/PID", String(child.pid), "/T", "/F"], { timeout: 15_000 }); + } + await exit; + retired = true; + })(); + return stopping; + } + + return { + stop, + interrupt: () => signal("SIGTERM"), + async wait(): Promise { + const status = await exit; + await stop(); + return status; + }, + }; +} diff --git a/packages/db/vitest.config.ts b/packages/db/vitest.config.ts index c1433e6ef38..08a8d7b9b34 100644 --- a/packages/db/vitest.config.ts +++ b/packages/db/vitest.config.ts @@ -3,6 +3,8 @@ import { defineConfig } from "vitest/config"; export default defineConfig({ test: { environment: "node", + hookTimeout: 30_000, + teardownTimeout: 30_000, include: ["src/**/*.test.ts"], }, }); diff --git a/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts b/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts index b2bbe8b41a6..91eff8b4580 100644 --- a/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts +++ b/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts @@ -791,19 +791,7 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => { `)); await heartbeat.resumeQueuedRuns(); - await waitForCondition(async () => { - const [run, wakeup] = await Promise.all([ - db.select({ status: heartbeatRuns.status }) - .from(heartbeatRuns) - .where(eq(heartbeatRuns.id, runId)) - .then((rows) => rows[0] ?? null), - db.select({ status: agentWakeupRequests.status }) - .from(agentWakeupRequests) - .where(eq(agentWakeupRequests.id, wakeupRequestId)) - .then((rows) => rows[0] ?? null), - ]); - return run?.status === "cancelled" && wakeup?.status === "skipped"; - }); + await heartbeat.drainActiveRunExecutions(); const [run, wakeup, issue] = await Promise.all([ db.select({ status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode }) @@ -970,16 +958,10 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => { expect(ordering).toEqual(["validated", "parked"]); }; await heartbeat.resumeQueuedRuns(); - await waitForCondition(async () => { - const run = await db - .select({ status: heartbeatRuns.status }) - .from(heartbeatRuns) - .where(eq(heartbeatRuns.id, runId)) - .then((rows) => rows[0] ?? null); - return run?.status === "cancelled"; - }); + await heartbeat.drainActiveRunExecutions(); await parkPromise; + expect((await heartbeat.getRun(runId))?.status).toBe("cancelled"); const issue = await db .select({ status: issues.status }) .from(issues) @@ -1108,7 +1090,7 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => { }); expect(run).not.toBeNull(); - await waitForCondition(async () => countExecuteCallsForRun(run!.id) > 0); + await heartbeat.drainActiveRunExecutions(); expect(countExecuteCallsForRun(run!.id)).toBe(1); }); @@ -1126,7 +1108,7 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => { }); expect(run).not.toBeNull(); - await waitForCondition(async () => countExecuteCallsForRun(run!.id) > 0); + await heartbeat.drainActiveRunExecutions(); expect(countExecuteCallsForRun(run!.id)).toBe(1); }); @@ -1718,13 +1700,7 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => { .where(eq(agentWakeupRequests.id, wakeupRequestId)); await heartbeat.resumeQueuedRuns(); - await waitForCondition(async () => { - const [deferred] = await db - .select({ status: agentWakeupRequests.status, runId: agentWakeupRequests.runId }) - .from(agentWakeupRequests) - .where(eq(agentWakeupRequests.id, deferredWakeupId)); - return Boolean(deferred?.runId) && deferred?.status !== "deferred_issue_execution"; - }); + await heartbeat.drainActiveRunExecutions(); const [deferred] = await db .select({ status: agentWakeupRequests.status, runId: agentWakeupRequests.runId }) @@ -1788,15 +1764,7 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => { }); await heartbeat.resumeQueuedRuns(); - - await waitForCondition(async () => { - const run = await db - .select({ status: heartbeatRuns.status }) - .from(heartbeatRuns) - .where(eq(heartbeatRuns.id, runId)) - .then((rows) => rows[0] ?? null); - return run?.status === "cancelled"; - }); + await heartbeat.drainActiveRunExecutions(); const [run, wakeup, issue] = await Promise.all([ db diff --git a/server/src/__tests__/helpers/embedded-postgres.ts b/server/src/__tests__/helpers/embedded-postgres.ts index a5088faa573..a5b1cda116f 100644 --- a/server/src/__tests__/helpers/embedded-postgres.ts +++ b/server/src/__tests__/helpers/embedded-postgres.ts @@ -1,35 +1,7 @@ -import { randomUUID } from "node:crypto"; -import { - applyPendingMigrations, - createDb, - getEmbeddedPostgresTestSupport as embeddedSupport, - startEmbeddedPostgresTestDatabase as startEmbedded, +export { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, + EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, type EmbeddedPostgresTestDatabase, type EmbeddedPostgresTestSupport, } from "@paperclipai/db"; - -export type { EmbeddedPostgresTestDatabase, EmbeddedPostgresTestSupport }; - -export async function getEmbeddedPostgresTestSupport(): Promise { - return process.env.PAPERCLIP_TEST_POSTGRES_ADMIN_URL ? { supported: true } : embeddedSupport(); -} - -export async function startEmbeddedPostgresTestDatabase(prefix: string): Promise { - const adminUrl = process.env.PAPERCLIP_TEST_POSTGRES_ADMIN_URL; - if (!adminUrl) return startEmbedded(prefix); - const name = `paperclip_test_${randomUUID().replaceAll("-", "")}`; - const template = process.env.PAPERCLIP_TEST_POSTGRES_TEMPLATE; - if (template && !/^paperclip_test_template_[a-f0-9]+$/.test(template)) throw new Error("Invalid disposable test template name"); - const admin = createDb(adminUrl, { maxConnections: 1 }); - const url = new URL(adminUrl); - url.pathname = `/${name}`; - const cleanup = async () => { - try { await admin.$client.unsafe(`DROP DATABASE IF EXISTS "${name}" WITH (FORCE)`); } - finally { await admin.$client.end({ timeout: 1 }); } - }; - try { - await admin.$client.unsafe(`CREATE DATABASE "${name}"${template ? ` TEMPLATE "${template}"` : ""}`); - if (!template) await applyPendingMigrations(url.toString()); - return { connectionString: url.toString(), cleanup }; - } catch (error) { await cleanup(); throw error; } -} diff --git a/server/src/__tests__/idle-sleep-safety.test.ts b/server/src/__tests__/idle-sleep-safety.test.ts index 3d4bb27077c..187ff3d32e2 100644 --- a/server/src/__tests__/idle-sleep-safety.test.ts +++ b/server/src/__tests__/idle-sleep-safety.test.ts @@ -9,7 +9,7 @@ import { heartbeatRuns, issues, issueWatchdogs, issueThreadInteractions, issueQuestionResponseDeliveries, projects, projectWorkspaces, routines, routineTriggers, toolApplications, toolConnections, workspaceOperations, executionWorkspaceRuntimeLeases, type Db, } from "@paperclipai/db"; -import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { readIdleSleepSafety, type IdleSleepDrainStatus } from "../services/idle-sleep-safety.js"; const now = Date.UTC(2020, 0, 1); @@ -173,7 +173,7 @@ if (!support.supported) console.warn(`Skipping idle sleep Postgres tests: ${supp db = createDb(database.connectionString); }, 60_000); afterEach(async () => { await db.execute(sql`TRUNCATE companies, plugins, environments CASCADE`); }); - afterAll(async () => { await database?.cleanup(); }); + afterAll(async () => { await database?.cleanup(); }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); const read = () => readIdleSleepSafety(db, owned, () => now, ownerId, emptyLocal); const readScheduled = () => readIdleSleepSafety(db, owned, () => now, ownerId, emptyLocal, undefined, undefined, true);