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"); }), );