From 5b0d59de083a6a1685ac5d87e9378d1c855605b0 Mon Sep 17 00:00:00 2001 From: Ben Jacobson Date: Thu, 12 Nov 2026 10:53:48 -0800 Subject: [PATCH] fix(postgres): create organization schemas before their pools connect A pooled connection that opened before another pool committed an organization's schema could keep a search path without it, so migrations tried to create pg_catalog.triplex_schema_migrations and failed. Create the schema on a separate unbound connection first, so every schema-bound connection starts after the schema exists. Closing DatabaseManager now also closes each cached database's pools, which previously stayed open until the process exited. Adds an ordering test that fails on the old setup and a concurrent setup stress test that checks for leaked connections. Co-Authored-By: Claude Opus 5.5 --- .changeset/postgres-organization-setup.md | 10 + PLAN.md | 13 ++ packages/postgres/src/PostgresqlBackend.ts | 47 ++-- .../organization-setup-stress.test.ts | 221 ++++++++++++++++++ packages/sql/src/DatabaseManagerLayer.ts | 2 + 5 files changed, 278 insertions(+), 15 deletions(-) create mode 100644 .changeset/postgres-organization-setup.md create mode 100644 packages/postgres/test/integration/organization-setup-stress.test.ts diff --git a/.changeset/postgres-organization-setup.md b/.changeset/postgres-organization-setup.md new file mode 100644 index 0000000..220169b --- /dev/null +++ b/.changeset/postgres-organization-setup.md @@ -0,0 +1,10 @@ +--- +"@triplex-build/triplex": patch +--- + +Fix two PostgreSQL problems when a host creates many organization databases at once. Each +database schema is now created on a separate connection before its schema-bound pools open. +Before, a pooled connection that opened before the schema existed could create migration tables +in `pg_catalog` and fail with "permission denied to create pg_catalog.triplex_schema_migrations". +Closing the database manager now also closes every cached database's connection pools; before, +they stayed open until the process exited. diff --git a/PLAN.md b/PLAN.md index 9a0d196..7b53f02 100644 --- a/PLAN.md +++ b/PLAN.md @@ -426,6 +426,19 @@ it. | W2.5 | S8: replace the incubation banner with the proof line and the family strip | C | WorldVM `rollout/family.json` | blocked | The canonical family file does not exist yet in WorldVM | | W3 | Runner implementation (D14) as the `/runner` subpath | B | W1 | pr-open | 2026-11-11: #86 (`ergonomics/runner`, on #84); corpus 34/34 KV+SQLite, 16/16 Postgres; fixed a lost-wakeup race | +## Ask 0007: release readiness for 0.3.0 + +From `worldvm-project/asks/0007-triplex-release-readiness.md` (meta orchestrator, at Ben's +request). Reply goes in that file's Status block. Do not publish, tag or change npm settings. + +| ID | Item | Lane | Depends | Status | Evidence | +| --- | ----------------------------------------------------------------------------------------------------------- | ---- | ------- | ----------- | ------------------------------------------------------------------------------------------------------------------- | +| R1 | Fix the Postgres multi-organization setup race, with a stress test that reproduces it first; own PR to main | A | | pr-open | 2026-11-12: `fix/postgres-org-setup-race`; ordering test fails on old code; manager pool leak fixed; Postgres 79/79 | +| R2 | Verify release PR #35 on its head: install, check, pack check, Postgres integration; tarball name and size | A | R1 | waiting | Rerun after R1 merges and the release PR regenerates | +| R3 | Confirm the 0.3.0 bump, the CHANGELOG and `docs/releasing.md` | A | | in-progress | | +| R4 | Publish path: trusted publisher, why Release runs failed, whether publish can fail | A | | in-progress | Earlier failures were the version job ("mixed changeset"), fixed by #87; 0.2.0 was published with a token, not OIDC | +| R5 | Exact `npm deprecate` commands for after the release | A | | in-progress | | + Proof line check (2026-11-11): WorldVM's current runtime keeps each world's subject records and bases in Triplex (`worldvm/packages/runtime/src/subject/triplex-records.ts`, `triplex-basis.ts`; one database per world), and its Cloudflare demo uses Triplex too. Proposed wording: "WorldVM keeps each diff --git a/packages/postgres/src/PostgresqlBackend.ts b/packages/postgres/src/PostgresqlBackend.ts index 458d4fe..0902cfc 100644 --- a/packages/postgres/src/PostgresqlBackend.ts +++ b/packages/postgres/src/PostgresqlBackend.ts @@ -6,7 +6,7 @@ * The registry uses the "public" schema. */ -import { Context, Effect, Layer, Redacted } from "effect"; +import { Effect, Layer, Redacted } from "effect"; import { SqlClient } from "effect/sql"; import { PgClient } from "@effect/sql-pg"; import { createHash } from "node:crypto"; @@ -102,25 +102,42 @@ const createScopedPoolLayer = (config: PostgresqlBackendConfig, schema: string) ...(config.pool?.idleTimeout !== undefined && { idleTimeout: config.pool.idleTimeout }), }); -const createSchemaLayer = (config: PostgresqlBackendConfig, schema: string) => - createScopedPoolLayer(config, schema).pipe( - Layer.tap((ctx) => +/** + * Create the schema on a short-lived connection that is not bound to it. + * + * A composed database layer may initialize an adapter pool and a query pool + * for the same new schema concurrently. PostgreSQL's `IF NOT EXISTS` check can + * still race at the namespace unique index, so lock this deterministic schema + * identity through creation. + * + * This must finish before any schema-bound connection opens. A backend that + * started before another session committed the schema can keep resolving its + * search path without it. It then sees the schema in catalog lookups but + * creates unqualified tables in `pg_catalog`, which fails migrations. + */ +const ensureSchema = (config: PostgresqlBackendConfig, schema: string) => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql.withTransaction( Effect.gen(function* () { - const sql = Context.get(ctx, SqlClient.SqlClient); - // A composed database layer may initialize an adapter pool and a query - // pool for the same new schema concurrently. PostgreSQL's `IF NOT - // EXISTS` check can still race at the namespace unique index, so lock - // this deterministic schema identity through creation. - yield* sql.withTransaction( - Effect.gen(function* () { - yield* sql`SELECT pg_advisory_xact_lock(hashtextextended(${schema}, 0))`; - yield* sql.unsafe(`CREATE SCHEMA IF NOT EXISTS ${quoteIdentifier(schema)}`); - }), - ); + yield* sql`SELECT pg_advisory_xact_lock(hashtextextended(${schema}, 0))`; + yield* sql.unsafe(`CREATE SCHEMA IF NOT EXISTS ${quoteIdentifier(schema)}`); + }), + ); + }).pipe( + Effect.provide( + PgClient.layer({ + url: Redacted.make(connectionUrl(config)), + minConnections: 0, + maxConnections: 1, }), ), + Effect.scoped, ); +const createSchemaLayer = (config: PostgresqlBackendConfig, schema: string) => + Layer.unwrap(ensureSchema(config, schema).pipe(Effect.as(createScopedPoolLayer(config, schema)))); + /** * A pool whose every physical connection is bound to one validated Triplex * database schema. This does not create the schema or run migrations. diff --git a/packages/postgres/test/integration/organization-setup-stress.test.ts b/packages/postgres/test/integration/organization-setup-stress.test.ts new file mode 100644 index 0000000..384eb52 --- /dev/null +++ b/packages/postgres/test/integration/organization-setup-stress.test.ts @@ -0,0 +1,221 @@ +/** + * Stress test for concurrent organization setup on PostgreSQL. + * + * Each organization gets its own schema, reached through two pools (adapter and + * query) whose connections bind `search_path` at startup. A connection that + * opened before another pool committed the schema could resolve unqualified + * names without it, so a migration tried to create + * `pg_catalog.triplex_schema_migrations` and failed. In the server log of a + * failure, the failing connection opened 36 ms before the other pool committed + * the schema. PostgreSQL 17 marks a recomputed search path valid after an + * invalidation that arrived during the computation, so a stale path can stick. + * + * The race shows up about once in several thousand creations, too rarely to + * reproduce on demand. The first test instead checks the ordering that + * prevents it: while schema creation is blocked, no schema-bound connection is + * open. Before the fix, the pool's three minimum connections were. + * + * The second test repeats concurrent organization creation in rounds and + * checks that closing the database manager releases every organization's + * pools. Before the fix they leaked (192 connections after the default run). + * Raise `PG_SETUP_STRESS_ROUNDS` and `PG_SETUP_STRESS_ORGS` to search harder, + * for example under CPU load. Keep `orgs * 2 pools * pool.max` below the + * server's `max_connections`. + */ + +import { describe, expect, it } from "vitest"; +import { Deferred, Effect, Exit, Fiber, Layer, Redacted } from "effect"; +import { SqlClient } from "effect/sql"; +import { PgClient } from "@effect/sql-pg"; +import { DatabaseId, DatabaseManager } from "@triplex-internal/core"; +import { + DatabaseRegistry, + IdGeneratorLive, + RuntimeClockLive, + TripleStoreRuntimeLayer, +} from "@triplex-internal/core/internal"; +import { + DatabaseManagerLive, + DatabaseRegistryLive, + StorageBackend, +} from "@triplex-build/triplex-sql"; +import { databaseToSchema, makePostgresqlBackend } from "@triplex-build/triplex-postgres"; +import { + checkDockerAvailable, + PgConnectionInfo, + PgContainerLayer, +} from "../fixtures/PgTestLayer.js"; + +const DOCKER_AVAILABLE = checkDockerAvailable(); +const ROUNDS = Number(process.env["PG_SETUP_STRESS_ROUNDS"] ?? 6); +const ORGS = Number(process.env["PG_SETUP_STRESS_ORGS"] ?? 8); + +const backendFor = (url: string) => { + const parsed = new URL(url); + return makePostgresqlBackend({ + host: parsed.hostname, + port: Number(parsed.port), + database: parsed.pathname.slice(1), + username: decodeURIComponent(parsed.username), + password: Redacted.make(decodeURIComponent(parsed.password)), + pool: { min: 2, max: 4 }, + }); +}; + +const openConnections = (url: string) => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const rows = yield* sql<{ count: number }>` + SELECT count(*)::int AS count + FROM pg_stat_activity + WHERE datname = current_database() AND pid <> pg_backend_pid() + `; + return rows[0]!.count; + }).pipe( + Effect.provide(PgClient.layer({ url: Redacted.make(url), maxConnections: 1 })), + Effect.scoped, + ); + +describe("PostgreSQL organization setup", () => { + it.skipIf(!DOCKER_AVAILABLE)( + "opens no schema-bound connection until the schema exists", + { timeout: 60_000 }, + async () => { + await Effect.runPromise( + Effect.gen(function* () { + const { url } = yield* PgConnectionInfo; + const parsed = new URL(url); + const backend = makePostgresqlBackend({ + host: parsed.hostname, + port: Number(parsed.port), + database: parsed.pathname.slice(1), + username: decodeURIComponent(parsed.username), + password: Redacted.make(decodeURIComponent(parsed.password)), + pool: { min: 3, max: 4 }, + }); + const database = DatabaseId.make("org-setup-order"); + const schema = databaseToSchema(database); + const client = (max: number) => + PgClient.layer({ url: Redacted.make(url), maxConnections: max }); + + yield* Effect.gen(function* () { + const storage = yield* StorageBackend; + yield* storage.deleteDatabaseStorage(database); + }).pipe(Effect.provide(backend)); + + // Hold the schema's creation lock so setup stops at schema creation. + const holding = yield* Deferred.make(); + const release = yield* Deferred.make(); + const holder = yield* Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql.withTransaction( + Effect.gen(function* () { + yield* sql`SELECT pg_advisory_xact_lock(hashtextextended(${schema}, 0))`; + const [row] = yield* sql<{ pid: number }>`SELECT pg_backend_pid() AS pid`; + yield* Deferred.succeed(holding, row!.pid); + yield* Deferred.await(release); + }), + ); + }).pipe(Effect.provide(client(1)), Effect.scoped, Effect.forkChild); + const holderPid = yield* Deferred.await(holding); + + const setup = yield* Effect.gen(function* () { + const storage = yield* StorageBackend; + yield* Layer.build(storage.createDatabaseClient(database)); + }).pipe(Effect.provide(backend), Effect.scoped, Effect.forkChild); + + const observe = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + // Wait until setup is blocked on the lock. + for (let attempt = 0; ; attempt++) { + const [waiting] = yield* sql<{ count: number }>` + SELECT count(*)::int AS count FROM pg_locks + WHERE locktype = 'advisory' AND NOT granted + `; + if (waiting!.count > 0) break; + if (attempt > 200) return yield* Effect.die("setup never waited for the lock"); + yield* Effect.sleep("25 millis"); + } + // Give a pool that opens eagerly time to connect. + yield* Effect.sleep("500 millis"); + const [others] = yield* sql<{ count: number }>` + SELECT count(*)::int AS count FROM pg_stat_activity + WHERE datname = current_database() + AND backend_type = 'client backend' + AND pid NOT IN (pg_backend_pid(), ${holderPid}) + `; + return others!.count; + }).pipe(Effect.provide(client(1)), Effect.scoped); + + // Only the unbound connection creating the schema may be open. + expect(yield* observe).toBe(1); + + yield* Deferred.succeed(release, undefined); + yield* Fiber.join(holder); + yield* Fiber.join(setup); + }).pipe(Effect.provide(PgContainerLayer)), + ); + }, + ); + + it.skipIf(!DOCKER_AVAILABLE)( + "creates many organizations at once and releases their pools", + { timeout: 60_000 + ROUNDS * ORGS * 1_000 }, + async () => { + await Effect.runPromise( + Effect.gen(function* () { + const { url } = yield* PgConnectionInfo; + const backend = backendFor(url); + const baseline = yield* openConnections(url); + const failures: Array = []; + + for (let round = 0; round < ROUNDS; round++) { + yield* Effect.gen(function* () { + const storage = yield* StorageBackend; + yield* storage.deleteAllStorage(); + }).pipe(Effect.provide(backend)); + + const registry = DatabaseRegistryLive.pipe( + Layer.provide(backend), + Layer.provide(RuntimeClockLive), + ); + const manager = DatabaseManagerLive.pipe( + Layer.provide(backend), + Layer.provide(registry), + Layer.provide(TripleStoreRuntimeLayer), + Layer.provide(IdGeneratorLive), + ); + const names = Array.from({ length: ORGS }, (_, i) => `org-stress-${i}`); + const exit = yield* Effect.gen(function* () { + const databases = yield* DatabaseManager; + yield* DatabaseRegistry; + yield* Effect.all( + names.map((name) => databases.create(name)), + { concurrency: "unbounded" }, + ); + const stores = yield* Effect.all( + names.map((name) => databases.getTriples(name)), + { concurrency: "unbounded" }, + ); + yield* Effect.all( + stores.map((store) => store.match({})), + { concurrency: "unbounded" }, + ); + }).pipe(Effect.provide(Layer.merge(manager, registry)), Effect.exit); + if (Exit.isFailure(exit)) failures.push(`round ${round}: ${String(exit.cause)}`); + } + + expect(failures).toEqual([]); + // Every round's organization pools were closed with its manager. + // Backends exit shortly after their clients disconnect, so allow a moment. + let open = yield* openConnections(url); + for (let attempt = 0; attempt < 50 && open > baseline; attempt++) { + yield* Effect.sleep("100 millis"); + open = yield* openConnections(url); + } + expect(open).toBeLessThanOrEqual(baseline); + }).pipe(Effect.provide(PgContainerLayer)), + ); + }, + ); +}); diff --git a/packages/sql/src/DatabaseManagerLayer.ts b/packages/sql/src/DatabaseManagerLayer.ts index 6c9c448..04dd02e 100644 --- a/packages/sql/src/DatabaseManagerLayer.ts +++ b/packages/sql/src/DatabaseManagerLayer.ts @@ -463,6 +463,8 @@ export const DatabaseManagerLive = Layer.effect( Effect.gen(function* () { // Stop the cleanup fiber yield* Fiber.interrupt(cleanupFiber); + // Each cached database owns its connection pools; release them with the manager. + yield* closeAllConnections(); yield* Effect.logInfo("Database manager finalized"); }), );