diff --git a/apps/cloud/scripts/backfill-workos-mirror.ts b/apps/cloud/scripts/backfill-workos-mirror.ts index 945de792b2..060e040d56 100644 --- a/apps/cloud/scripts/backfill-workos-mirror.ts +++ b/apps/cloud/scripts/backfill-workos-mirror.ts @@ -42,6 +42,7 @@ import { WorkOS } from "@workos-inc/node"; import { backfillWorkOsMirror } from "../src/auth/workos-mirror-backfill"; import { makeWorkOsMirrorStore } from "../src/auth/workos-mirror-store"; import { organizations } from "../src/db/schema"; +import { describeRefusedAttempt, waitForConnectionSlot } from "../src/db/too-many-connections"; const dryRun = process.argv.includes("--dry-run"); @@ -71,32 +72,46 @@ const workos = new WorkOS(apiKey); const fromPromise = (fn: () => Promise) => Effect.tryPromise({ try: fn, catch: (cause) => cause }); +// A full server refuses the connection with SQLSTATE 53300 when it is opened. +// Open it first, waiting for a slot, rather than fail the deploy gate that +// spawned this run (src/db/too-many-connections.ts); the mirror store's own +// failures do not carry the driver code, so the wait cannot sit around the +// backfill itself. +const connected = waitForConnectionSlot(sql, { + onRefused: (_failure, attempt) => console.log(describeRefusedAttempt(attempt)), +}); + +const backfill = backfillWorkOsMirror( + { + listOrganizationIds: () => + fromPromise(async () => { + // Never a deleted organization: its row is a tombstone (its + // memberships are purged, WorkOS no longer has it) and the mirror + // refuses a scan of it anyway. + const rows = await db + .select({ id: organizations.id }) + .from(organizations) + .where(isNull(organizations.deletedAt)) + .orderBy(asc(organizations.createdAt)); + return rows.map((row) => row.id); + }), + listOrgMembers: (organizationId) => + fromPromise(async () => { + const page = await workos.userManagement.listOrganizationMemberships({ + organizationId, + statuses: ["active", "pending", "inactive"], + }); + return page.listMetadata.after ? page.autoPagination() : page.data; + }), + getUser: (userId) => fromPromise(() => workos.userManagement.getUser(userId)), + }, + makeWorkOsMirrorStore(db), + { dryRun, log: (line) => console.log(line) }, +); + await Effect.runPromise( - backfillWorkOsMirror( - { - listOrganizationIds: () => - fromPromise(async () => { - // Never a deleted organization: its row is a tombstone (its - // memberships are purged, WorkOS no longer has it) and the mirror - // refuses a scan of it anyway. - const rows = await db - .select({ id: organizations.id }) - .from(organizations) - .where(isNull(organizations.deletedAt)) - .orderBy(asc(organizations.createdAt)); - return rows.map((row) => row.id); - }), - listOrgMembers: (organizationId) => - fromPromise(async () => { - const page = await workos.userManagement.listOrganizationMemberships({ - organizationId, - statuses: ["active", "pending", "inactive"], - }); - return page.listMetadata.after ? page.autoPagination() : page.data; - }), - getUser: (userId) => fromPromise(() => workos.userManagement.getUser(userId)), - }, - makeWorkOsMirrorStore(db), - { dryRun, log: (line) => console.log(line) }, - ).pipe(Effect.ensuring(Effect.promise(() => sql.end({ timeout: 5 })))), + connected.pipe( + Effect.andThen(backfill), + Effect.ensuring(Effect.promise(() => sql.end({ timeout: 5 }))), + ), ); diff --git a/apps/cloud/scripts/drain-workos-events.ts b/apps/cloud/scripts/drain-workos-events.ts index b5cfc6f3db..843a7b7491 100644 --- a/apps/cloud/scripts/drain-workos-events.ts +++ b/apps/cloud/scripts/drain-workos-events.ts @@ -26,6 +26,7 @@ import { WorkOS } from "@workos-inc/node"; import { makeUserStore } from "../src/auth/user-store"; import { replayWorkOsEvents, type WorkOsEventsSyncReport } from "../src/auth/workos-events-replay"; import { makeWorkOsMirrorStore } from "../src/auth/workos-mirror-store"; +import { describeRefusedAttempt, waitForConnectionSlot } from "../src/db/too-many-connections"; const connectionString = process.env.DATABASE_URL; if (!connectionString) { @@ -111,8 +112,21 @@ const drain = Effect.gen(function* () { return last; }); +// A full server refuses the connection with SQLSTATE 53300 when it is opened. +// Open it first, waiting for a slot, rather than fail the deploy gate that +// spawned this run (src/db/too-many-connections.ts); the mirror store's own +// failures do not carry the driver code, so the wait cannot sit around the +// drain itself. +const connected = waitForConnectionSlot(sql, { + onRefused: (_failure, attempt) => + console.log(`[drain-events] ${describeRefusedAttempt(attempt)}`), +}); + const report = await Effect.runPromise( - drain.pipe(Effect.ensuring(Effect.promise(() => sql.end({ timeout: 5 })))), + connected.pipe( + Effect.andThen(drain), + Effect.ensuring(Effect.promise(() => sql.end({ timeout: 5 }))), + ), ); if (report === null || report.stopped !== "drained") { diff --git a/apps/cloud/scripts/ensure-workos-mirror-ready.ts b/apps/cloud/scripts/ensure-workos-mirror-ready.ts index d10b826d3e..1045ffb1fe 100644 --- a/apps/cloud/scripts/ensure-workos-mirror-ready.ts +++ b/apps/cloud/scripts/ensure-workos-mirror-ready.ts @@ -35,6 +35,7 @@ import { dirname, resolve } from "node:path"; import { fileURLToPath } from "node:url"; import { drizzle } from "drizzle-orm/postgres-js"; +import { Result } from "effect"; import postgres from "postgres"; import { @@ -42,6 +43,10 @@ import { describeMirrorReadiness, readMirrorReadiness, } from "../src/auth/mirror-readiness-store"; +import { + describeRefusedAttempt, + retryWhileTooManyConnections, +} from "../src/db/too-many-connections"; const __dirname = dirname(fileURLToPath(import.meta.url)); const BACKFILL_SCRIPT = resolve(__dirname, "backfill-workos-mirror.ts"); @@ -65,7 +70,19 @@ const db = drizzle(sql); const log = (line: string) => console.log(`[mirror-ready] ${line}`); -const readiness = () => readMirrorReadiness(db, new Date()); +// A full server refuses the connection with SQLSTATE 53300 before a statement +// runs — on the first read, or on a later one after postgres.js has reopened a +// dropped connection — so every read waits for a slot instead of failing the +// deploy, as the migration step before this one does +// (src/db/too-many-connections.ts). The backfill and drain scripts below each +// open their own connection and wait for their own slot. +const readiness = async () => { + const outcome = await retryWhileTooManyConnections(() => readMirrorReadiness(db, new Date()), { + onRefused: (_failure, attempt) => log(describeRefusedAttempt(attempt)), + }); + if (Result.isFailure(outcome)) throw outcome.failure; + return outcome.success; +}; // The backfill and drain scripts own their own WorkOS + database wiring; // running them as subprocesses (with this process's env) keeps that wiring diff --git a/apps/cloud/scripts/migrate.ts b/apps/cloud/scripts/migrate.ts index 9466a049b5..9c27efc8af 100644 --- a/apps/cloud/scripts/migrate.ts +++ b/apps/cloud/scripts/migrate.ts @@ -5,8 +5,13 @@ import { fileURLToPath } from "node:url"; import { drizzle } from "drizzle-orm/postgres-js"; import { migrate as migrateDrizzle } from "drizzle-orm/postgres-js/migrator"; +import { Result } from "effect"; import postgres from "postgres"; +import { + describeRefusedAttempt, + retryWhileTooManyConnections, +} from "../src/db/too-many-connections"; import { cloudCodeMigrations, runCodeMigrations } from "./code-migrations/index"; const __dirname = dirname(fileURLToPath(import.meta.url)); @@ -47,13 +52,29 @@ const sql = postgres(connectionString, { ...(usesLocalDatabase ? {} : { ssl: "require" as const }), }); +// A full server refuses the connection with SQLSTATE 53300 before a statement +// runs — on the first statement, or on a later one after postgres.js has +// reopened a dropped connection. So each step over the direct connection +// waits for a slot instead of failing the deploy (src/db/too-many-connections.ts). +// A step is safe to repeat: Drizzle and the code-migration ledger each skip +// what has already been applied. +const withConnectionSlot = async (run: () => Promise): Promise => { + const outcome = await retryWhileTooManyConnections(run, { + onRefused: (_failure, attempt) => console.warn(`[migrate] ${describeRefusedAttempt(attempt)}`), + }); + if (Result.isFailure(outcome)) throw outcome.failure; + return outcome.success; +}; + try { if (!codeOnly) { if (dryRun) { console.log("[schema-migrate] dry run: Drizzle SQL migrations are not applied"); } else { console.log(`[schema-migrate] running Drizzle migrations from ${MIGRATIONS_FOLDER}`); - await migrateDrizzle(drizzle(sql), { migrationsFolder: MIGRATIONS_FOLDER }); + await withConnectionSlot(() => + migrateDrizzle(drizzle(sql), { migrationsFolder: MIGRATIONS_FOLDER }), + ); console.log("[schema-migrate] complete"); } } @@ -63,7 +84,9 @@ try { if (migrations.length === 0) { console.log("[code-migrate] no code migrations configured"); } else { - const applied = await runCodeMigrations(sql, migrations, { dryRun }); + const applied = await withConnectionSlot(() => + runCodeMigrations(sql, migrations, { dryRun }), + ); console.log( dryRun ? `[code-migrate] dry run planned ${applied.length} migration(s)` diff --git a/apps/cloud/src/db/too-many-connections.test.ts b/apps/cloud/src/db/too-many-connections.test.ts new file mode 100644 index 0000000000..0dd25149d9 --- /dev/null +++ b/apps/cloud/src/db/too-many-connections.test.ts @@ -0,0 +1,137 @@ +/* oxlint-disable executor/no-try-catch-or-throw, executor/no-error-constructor -- test doubles: simulate the driver's rejected promise and its Error-shaped cause chain */ +import { describe, expect, it } from "@effect/vitest"; +import { Effect, Fiber, Result, Schedule } from "effect"; +import { TestClock } from "effect/testing"; + +import { + TOO_MANY_CONNECTIONS_RETRIES, + TOO_MANY_CONNECTIONS_SQLSTATE, + describeRefusedAttempt, + isTooManyConnectionsError, + retryTooManyConnections, + retryWhileTooManyConnections, +} from "./too-many-connections"; + +// The shape postgres.js + Drizzle produce in the deploy log: Drizzle's +// "Failed query" error with the driver's PostgresError (SQLSTATE `code`) as +// its cause. +const refused = () => + Object.assign(new Error('Failed query: CREATE SCHEMA IF NOT EXISTS "drizzle"'), { + cause: Object.assign( + new Error("remaining connection slots are reserved for roles with the SUPERUSER attribute"), + { code: TOO_MANY_CONNECTIONS_SQLSTATE }, + ), + }); + +const noDelay = Schedule.recurs(2); + +describe("isTooManyConnectionsError", () => { + it("matches SQLSTATE 53300 anywhere in the cause chain", () => { + expect(isTooManyConnectionsError(refused())).toBe(true); + expect(isTooManyConnectionsError({ code: "53300" })).toBe(true); + }); + + it("rejects other driver codes and non-errors", () => { + expect(isTooManyConnectionsError({ code: "CONNECT_TIMEOUT" })).toBe(false); + expect(isTooManyConnectionsError(new Error("boom"))).toBe(false); + expect(isTooManyConnectionsError(undefined)).toBe(false); + expect(isTooManyConnectionsError("53300")).toBe(false); + }); +}); + +describe("retryWhileTooManyConnections", () => { + it("retries a refused connection and resolves with the first success", async () => { + let calls = 0; + const refusals: number[] = []; + const result = await retryWhileTooManyConnections( + async () => { + calls += 1; + if (calls < 3) throw refused(); + return "applied"; + }, + { schedule: noDelay, onRefused: (_, attempt) => refusals.push(attempt) }, + ); + expect(result).toEqual(Result.succeed("applied")); + expect(calls).toBe(3); + expect(refusals).toEqual([1, 2]); + }); + + it("rethrows any other failure without retrying", async () => { + let calls = 0; + const failure = Object.assign(new Error("Failed query: alter table"), { + cause: { code: "42P01" }, + }); + const result = await retryWhileTooManyConnections( + async () => { + calls += 1; + throw failure; + }, + { schedule: noDelay }, + ); + expect(Result.isFailure(result) && result.failure).toBe(failure); + expect(calls).toBe(1); + }); + + it("rethrows the last refusal once the schedule is spent", async () => { + let calls = 0; + const failures: unknown[] = []; + const result = await retryWhileTooManyConnections( + async () => { + calls += 1; + const failure = refused(); + failures.push(failure); + throw failure; + }, + { schedule: noDelay }, + ); + expect(calls).toBe(3); + expect(Result.isFailure(result) && result.failure).toBe(failures[2]); + }); +}); + +describe("the production schedule", () => { + // Virtual time: the production cadence is thirty seconds between attempts, + // and a test that waited it out for real would take a quarter of an hour. + // `it.effect` runs under the TestClock, which the schedule's sleeps use. + it.effect("retries every thirty seconds and gives up after about fifteen minutes", () => + Effect.gen(function* () { + let calls = 0; + const refusals: number[] = []; + const fiber = yield* Effect.forkChild( + Effect.result( + retryTooManyConnections( + Effect.suspend(() => { + calls += 1; + return Effect.fail(refused()); + }), + { onRefused: (_, attempt) => refusals.push(attempt) }, + ), + ), + ); + + yield* TestClock.adjust("29 seconds"); + expect(calls).toBe(1); + yield* TestClock.adjust("1 second"); + expect(calls).toBe(2); + + // Attempt n runs at (n - 1) × 30 s: the thirtieth at 14:30, the last at 15:00. + yield* TestClock.adjust("14 minutes"); + expect(calls).toBe(TOO_MANY_CONNECTIONS_RETRIES); + yield* TestClock.adjust("30 seconds"); + const result = yield* Fiber.join(fiber); + + expect(calls).toBe(TOO_MANY_CONNECTIONS_RETRIES + 1); + expect(refusals.at(-1)).toBe(TOO_MANY_CONNECTIONS_RETRIES + 1); + expect(Result.isFailure(result) && isTooManyConnectionsError(result.failure)).toBe(true); + }), + ); +}); + +describe("describeRefusedAttempt", () => { + it("says when it will retry and when it is giving up", () => { + expect(describeRefusedAttempt(1)).toBe( + `Postgres refused the connection: no free connection slots (attempt 1 of ${TOO_MANY_CONNECTIONS_RETRIES + 1}); retrying in 30 seconds`, + ); + expect(describeRefusedAttempt(TOO_MANY_CONNECTIONS_RETRIES + 1)).toMatch(/giving up$/); + }); +}); diff --git a/apps/cloud/src/db/too-many-connections.ts b/apps/cloud/src/db/too-many-connections.ts new file mode 100644 index 0000000000..c641118888 --- /dev/null +++ b/apps/cloud/src/db/too-many-connections.ts @@ -0,0 +1,135 @@ +// --------------------------------------------------------------------------- +// Retry a direct Postgres call while the server refuses new connections. +// +// The deploy's out-of-band scripts (`scripts/migrate.ts`, +// `scripts/ensure-workos-mirror-ready.ts`, and the backfill and drain scripts +// it spawns) each open ONE direct connection to production Postgres. Those +// connections compete for the server's non-superuser slots with Hyperdrive's +// pools (through PSBouncer), operator sessions, and other CI jobs. When no +// slot is free the server answers the connect with SQLSTATE 53300 +// (`too_many_connections`, "remaining connection slots are reserved for roles +// with the SUPERUSER attribute") before a single statement runs. On +// 2026-09-18 two consecutive migration attempts nine minutes apart were +// refused that way after Worker redeploys; a rerun about an hour later +// passed. So that specific refusal is retried on a fixed cadence for about +// fifteen minutes, which lets the deploy wait for a slot instead of failing; +// every other failure (a bad migration, a lost network) surfaces unchanged on +// the first attempt. This is a mitigation for the transient refusal, not the +// capacity fix — that is the server's connection ceiling and the pool sizes +// in front of it. +// --------------------------------------------------------------------------- + +import { Effect, Result, Schedule } from "effect"; +import type postgres from "postgres"; + +/** SQLSTATE `too_many_connections`: the server's connection ceiling is reached. */ +export const TOO_MANY_CONNECTIONS_SQLSTATE = "53300"; + +// postgres.js sets the SQLSTATE as a string `code`; Drizzle wraps that in its +// own "Failed query" error with the driver error in `.cause`. Walk the chain +// (bounded) rather than inspect one level. +const MAX_CAUSE_DEPTH = 8; + +export const isTooManyConnectionsError = (failure: unknown): boolean => { + let current: unknown = failure; + for ( + let depth = 0; + depth < MAX_CAUSE_DEPTH && typeof current === "object" && current !== null; + depth++ + ) { + if ((current as { readonly code?: unknown }).code === TOO_MANY_CONNECTIONS_SQLSTATE) { + return true; + } + current = (current as { readonly cause?: unknown }).cause; + } + return false; +}; + +/** + * 30 retries, 30 seconds apart: about fifteen minutes of waiting for a slot, + * longer than the nine minutes the two refusals of 2026-09-18 spanned. + */ +export const TOO_MANY_CONNECTIONS_RETRIES = 30; +export const TOO_MANY_CONNECTIONS_RETRY_INTERVAL = "30 seconds"; + +export const TOO_MANY_CONNECTIONS_RETRY_SCHEDULE = Schedule.both( + Schedule.spaced(TOO_MANY_CONNECTIONS_RETRY_INTERVAL), + Schedule.recurs(TOO_MANY_CONNECTIONS_RETRIES), +); + +export type RetryTooManyConnectionsOptions = { + /** Overrides the production cadence; tests pass a delay-free schedule. */ + readonly schedule?: Schedule.Schedule; + /** Called on every refused attempt (including the last), before the wait. */ + readonly onRefused?: (failure: unknown, attempt: number) => void; +}; + +/** One log line per refused attempt, for the script's own logger. */ +export const describeRefusedAttempt = (attempt: number): string => + `Postgres refused the connection: no free connection slots (attempt ${attempt} of ${TOO_MANY_CONNECTIONS_RETRIES + 1}); ` + + (attempt > TOO_MANY_CONNECTIONS_RETRIES + ? "giving up" + : `retrying in ${TOO_MANY_CONNECTIONS_RETRY_INTERVAL}`); + +/** + * Run `attempt`, retrying only while it fails with SQLSTATE 53300; any other + * failure ends the loop on the spot. Fails with the ORIGINAL error — the + * non-53300 failure from the attempt that raised it, or the last refusal once + * the schedule is spent. + */ +export const retryTooManyConnections = ( + attempt: Effect.Effect, + options: RetryTooManyConnectionsOptions = {}, +): Effect.Effect => + Effect.suspend(() => { + let refusals = 0; + const observed = Effect.tapError(attempt, (failure) => + Effect.sync(() => { + if (!isTooManyConnectionsError(failure)) return; + refusals += 1; + options.onRefused?.(failure, refusals); + }), + ); + return Effect.retry(observed, { + schedule: options.schedule ?? TOO_MANY_CONNECTIONS_RETRY_SCHEDULE, + while: isTooManyConnectionsError, + }); + }); + +/** + * {@link retryTooManyConnections} over a Promise, for the scripts that work in + * raw driver promises. Never rejects: resolves with the first success, or with + * the original failure for the calling script to rethrow at its boundary. + */ +export const retryWhileTooManyConnections = ( + run: () => Promise, + options: RetryTooManyConnectionsOptions = {}, +): Promise> => + Effect.runPromise( + retryTooManyConnections(Effect.tryPromise({ try: run, catch: (cause) => cause }), options).pipe( + Effect.result, + ), + ); + +/** + * Open `sql`'s one connection, waiting for a slot: a `SELECT 1` retried while + * 53300. postgres.js keeps that connection (no idle timeout by default) for + * every statement that follows, so a script whose database failures are + * classified at a service boundary — `WorkOsMirrorError` drops the driver + * cause, see `auth/errors.ts` — gets the same wait as one that sees the raw + * driver error. This guards the connect, which is where 53300 is raised; a + * connection dropped and reopened mid-run is not guarded. + */ +export const waitForConnectionSlot = ( + sql: postgres.Sql, + options: RetryTooManyConnectionsOptions = {}, +): Effect.Effect => + retryTooManyConnections( + Effect.tryPromise({ + try: async () => { + await sql`select 1`; + }, + catch: (cause) => cause, + }), + options, + );