From 900878cbd9a3050ffc0190a3a2bf4155496c5816 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Sun, 13 Sep 2026 17:22:07 +0000 Subject: [PATCH] fix: keep migration sessions alive through long index builds --- .github/workflows/ci.yml | 13 +++++++++ packages/db/src/index.js | 22 +++++++++----- packages/db/src/migrate.js | 32 +++++++++++++------- test/migrate-postgres.test.js | 55 +++++++++++++++++++++++++++++++++++ 4 files changed, 105 insertions(+), 17 deletions(-) create mode 100644 test/migrate-postgres.test.js diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9d35d6f..77e28a1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -6,6 +6,18 @@ on: jobs: test: runs-on: ubuntu-latest + services: + postgres: + image: postgres:17-alpine + env: + POSTGRES_HOST_AUTH_METHOD: trust + ports: + - 5432:5432 + options: >- + --health-cmd pg_isready + --health-interval 5s + --health-timeout 5s + --health-retries 10 steps: - uses: actions/checkout@v4 - uses: oven-sh/setup-bun@v2 @@ -16,4 +28,5 @@ jobs: - run: bun test env: DATABASE_URL: postgres://unused:unused@localhost:5432/unused + MIGRATION_TEST_DATABASE_URL: postgres://postgres@localhost:5432/postgres SITE_URL: http://localhost:3000 diff --git a/packages/db/src/index.js b/packages/db/src/index.js index 783dc02..e4d188c 100644 --- a/packages/db/src/index.js +++ b/packages/db/src/index.js @@ -6,13 +6,21 @@ import { SQL } from 'bun'; * runtime and no native addons. `max` and the worker concurrency are chosen * together: every BullMQ slot can hold a connection. */ -export const sql = new SQL({ - url: config.databaseUrl, - max: Number(process.env.DB_POOL_MAX ?? 12), - idleTimeout: 30, - connectionTimeout: 15, - tls: config.databaseUrl.includes('sslmode=require') ? { rejectUnauthorized: false } : undefined, -}); +export function connect({ + url = config.databaseUrl, + max = Number(process.env.DB_POOL_MAX ?? 12), + idleTimeout = 30, +} = {}) { + return new SQL({ + url, + max, + idleTimeout, + connectionTimeout: 15, + tls: url.includes('sslmode=require') ? { rejectUnauthorized: false } : undefined, + }); +} + +export const sql = connect(); export async function healthcheck() { const [row] = await sql`select 1 as ok`; diff --git a/packages/db/src/migrate.js b/packages/db/src/migrate.js index c2de748..5489169 100644 --- a/packages/db/src/migrate.js +++ b/packages/db/src/migrate.js @@ -1,7 +1,7 @@ import { readdir, readFile } from 'node:fs/promises'; import { dirname, join } from 'node:path'; import { fileURLToPath } from 'node:url'; -import { sql } from './index.js'; +import { connect } from './index.js'; const MIGRATIONS_DIR = join(dirname(fileURLToPath(import.meta.url)), '..', 'migrations'); @@ -12,23 +12,35 @@ const MIGRATIONS_DIR = join(dirname(fileURLToPath(import.meta.url)), '..', 'migr * Called on boot by every process; the advisory lock makes that safe when web * and worker boot at the same instant. */ -export async function migrate({ log = console.log } = {}) { - await sql` - create table if not exists schema_migrations ( - filename text primary key, - applied_at timestamptz not null default now() - ) - `; +export async function migrate({ log = console.log, directory = MIGRATIONS_DIR, url } = {}) { + // A session advisory lock must stay on the connection doing the migration. + // Bun also applies idleTimeout while a long statement produces no messages: + // building an index over existing data can easily exceed the app pool's 30s. + const sql = connect({ url, max: 1, idleTimeout: 0 }); + try { + return await migrateOnConnection(sql, { log, directory }); + } finally { + await sql.end(); + } +} + +async function migrateOnConnection(sql, { log, directory }) { await sql`select pg_advisory_lock(8675310)`; try { - const files = (await readdir(MIGRATIONS_DIR)).filter((f) => f.endsWith('.sql')).sort(); + await sql` + create table if not exists schema_migrations ( + filename text primary key, + applied_at timestamptz not null default now() + ) + `; + const files = (await readdir(directory)).filter((f) => f.endsWith('.sql')).sort(); const applied = new Set( (await sql`select filename from schema_migrations`).map((r) => r.filename), ); let ran = 0; for (const file of files) { if (applied.has(file)) continue; - const body = await readFile(join(MIGRATIONS_DIR, file), 'utf8'); + const body = await readFile(join(directory, file), 'utf8'); log(`[migrate] applying ${file}`); await sql.begin(async (tx) => { await tx.unsafe(body); diff --git a/test/migrate-postgres.test.js b/test/migrate-postgres.test.js new file mode 100644 index 0000000..0fda4d0 --- /dev/null +++ b/test/migrate-postgres.test.js @@ -0,0 +1,55 @@ +import { expect, test } from 'bun:test'; +import { randomUUID } from 'node:crypto'; +import { mkdtemp, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { connect } from '../packages/db/src/index.js'; +import { migrate } from '../packages/db/src/migrate.js'; + +const postgresTest = process.env.MIGRATION_TEST_DATABASE_URL ? test : test.skip; + +postgresTest( + 'long migrations and concurrent boots retain one locked session', + async () => { + const url = new URL(process.env.MIGRATION_TEST_DATABASE_URL); + const admin = connect({ url: url.href, max: 1, idleTimeout: 0 }); + const name = `ndb_migrate_${randomUUID().replaceAll('-', '')}`; + const directory = await mkdtemp(join(tmpdir(), 'ndb-migrate-')); + let check; + try { + await admin.unsafe(`create database ${name}`); + url.pathname = `/${name}`; + await writeFile( + join(directory, '0001_slow.sql'), + ` + create table migration_probe (id integer primary key, backend integer); + insert into migration_probe values (1, pg_backend_pid()); + -- Production failed at 30s. A query without messages must survive longer. + select pg_sleep(35); + do $$ begin + if not exists ( + select 1 from pg_locks where pid=pg_backend_pid() + and locktype='advisory' and objid=8675310 and granted + ) then raise exception 'migration lost its session lock'; end if; + end $$; + insert into migration_probe values (2, pg_backend_pid()); + `, + ); + const options = { url: url.href, directory, log: () => {} }; + const results = await Promise.all([migrate(options), migrate(options)]); + expect(results.sort()).toEqual([0, 1]); + check = connect({ url: url.href, max: 1, idleTimeout: 0 }); + const rows = await check`select * from migration_probe order by id`; + expect(rows.map((row) => row.id)).toEqual([1, 2]); + expect(rows[0].backend).toBe(rows[1].backend); + expect(await check`select filename from schema_migrations`).toHaveLength(1); + expect(await migrate(options)).toBe(0); + } finally { + await check?.end(); + await admin.unsafe(`drop database if exists ${name} with (force)`); + await admin.end(); + await rm(directory, { recursive: true, force: true }); + } + }, + 90_000, +);