Postgres: create organization schemas before their pools connect

.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.

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

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.

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<number>();
+          const release = yield* Deferred.make<void>();
+          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<string> = [];
+
+          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)),
+      );
+    },
+  );
+});

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