Keep migration sessions alive through long index builds

.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

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`;

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);

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,
+);