From bdb2026f570666b5b5868362ec6f4ae7d5e515cb Mon Sep 17 00:00:00 2001 From: Marcel Wege Date: Thu, 30 Jul 2026 11:12:42 +0200 Subject: [PATCH 1/4] feat(plugins): allow .sql in distributed plugin packages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A distributed plugin ships its own schema as `migrations/*.sql`, but the zip extractor rejected the extension, so the package failed ingest with `zip.forbidden_extension`. Adding `.sql` was previously refused because it would have escalated the manifest-identity path traversal (an id of `..` with version `migrations` resolved onto the real migrations directory) from a directory delete into arbitrary SQL executed at boot. That traversal was closed in 09ff9cd0, and the remaining routes from uploaded content to a migrator-scanned path were re-verified before this change: - all eight SQL migrators resolve their directory from their own `import.meta.url`, the fixed `middleware/migrations`, or an operator env override — never from package content; - the three extraction call sites land only under state dirs (uploaded-packages staging/final, builder previews), never in the shipped code tree; - the `node_modules` symlink at the packages root is charset-valid as an id and is blocked by the reserved-root check, not by the charset gate. Tests pin those conditions rather than just the happy path: `.sql` accepted by the default allowlist, still rejected under an explicit override that omits it, and no `.sql` surviving outside the packages root for a traversing identity, the reserved `node_modules` id, or a Zip-Slip entry name. Shared package-zip fixtures move to `test/_helpers/pluginPackageZip.ts` so both install-pipeline suites use one builder. --- middleware/src/plugins/zipExtractor.ts | 9 + middleware/test/_helpers/pluginPackageZip.ts | 85 ++++++ .../test/pluginInstallPathTraversal.test.ts | 75 +---- .../test/pluginPackageSqlAllowlist.test.ts | 265 ++++++++++++++++++ 4 files changed, 369 insertions(+), 65 deletions(-) create mode 100644 middleware/test/_helpers/pluginPackageZip.ts create mode 100644 middleware/test/pluginPackageSqlAllowlist.test.ts diff --git a/middleware/src/plugins/zipExtractor.ts b/middleware/src/plugins/zipExtractor.ts index 92f2aa8d8..30799a72a 100644 --- a/middleware/src/plugins/zipExtractor.ts +++ b/middleware/src/plugins/zipExtractor.ts @@ -37,6 +37,15 @@ const EXTENSION_ALLOWLIST: ReadonlySet = new Set([ // fills via the `admin-ui-body` slot; without `.html` here the upload // pipeline rejects the package as "disallowed extension". '.html', + // #548 follow-up — a distributed plugin ships its own schema as + // `migrations/*.sql` next to its manifest. Inert on extraction: every + // migrator resolves its directory from its own `import.meta.url` (or the + // fixed `middleware/migrations`), and an uploaded package can only land + // under the packages root, whose `/` segments are charset- + // gated in manifestLoader and containment-re-checked in + // packageUploadService. Adding this before 09ff9cd0 would have turned that + // traversal into arbitrary SQL executed at boot. + '.sql', ]); const DECL_EXTENSIONS: ReadonlySet = new Set(['.ts', '.mts', '.cts']); diff --git a/middleware/test/_helpers/pluginPackageZip.ts b/middleware/test/_helpers/pluginPackageZip.ts new file mode 100644 index 000000000..0440ce2db --- /dev/null +++ b/middleware/test/_helpers/pluginPackageZip.ts @@ -0,0 +1,85 @@ +/** + * Fixtures for the plugin-package ingest pipeline: a minimal schema-v1 + * package zip plus the two collaborators `PackageUploadService` needs. + * + * Shared by `pluginInstallPathTraversal.test.ts` (identity/containment gates) + * and `pluginPackageSqlAllowlist.test.ts` (the `.sql` extension entry, which + * is only admissible while those gates hold). + */ + +import yazl from 'yazl'; + +import type { PluginCatalog } from '../../src/plugins/manifestLoader.js'; +import type { + UploadedPackage, + UploadedPackageStore, +} from '../../src/plugins/uploadedPackageStore.js'; + +/** Builds an in-memory zip from a `path → utf-8 content` map. */ +export function buildZip(files: Record): Promise { + return new Promise((resolve, reject) => { + const zip = new yazl.ZipFile(); + const chunks: Buffer[] = []; + zip.outputStream.on('data', (c: Buffer) => chunks.push(c)); + zip.outputStream.on('end', () => resolve(Buffer.concat(chunks))); + zip.outputStream.on('error', reject); + for (const [name, content] of Object.entries(files)) { + zip.addBuffer(Buffer.from(content, 'utf-8'), name, { mtime: new Date(0) }); + } + zip.end(); + }); +} + +/** The smallest manifest `adaptManifestV1` accepts, with a caller-set identity. */ +export function manifestYaml(id: string, version: string): string { + return `schema_version: "1" + +identity: + id: ${JSON.stringify(id)} + name: "Install Fixture" + version: ${JSON.stringify(version)} + kind: "tool" + description: "Fixture plugin for install-pipeline tests." + +compat: + core: ">=1.0 <2.0" + +lifecycle: + entry: "dist/plugin.js" +`; +} + +/** + * A well-formed package zip. `extraFiles` are merged on top, which is how the + * `.sql` suite adds `migrations/*.sql` without a second builder. + */ +export function packageZip( + id: string, + version: string, + extraFiles: Record = {}, +): Promise { + return buildZip({ + 'manifest.yaml': manifestYaml(id, version), + 'dist/plugin.js': 'module.exports = { activate() {} };\n', + ...extraFiles, + }); +} + +export function fakeStore(): UploadedPackageStore { + const packages = new Map(); + return { + get: (id: string) => packages.get(id), + list: () => [...packages.values()], + register: async (pkg: UploadedPackage) => { + packages.set(pkg.id, pkg); + }, + } as unknown as UploadedPackageStore; +} + +export function fakeCatalog(): PluginCatalog { + return { + get: () => undefined, + load: async () => undefined, + list: () => [], + } as unknown as PluginCatalog; +} diff --git a/middleware/test/pluginInstallPathTraversal.test.ts b/middleware/test/pluginInstallPathTraversal.test.ts index d8ddbc540..1820829e3 100644 --- a/middleware/test/pluginInstallPathTraversal.test.ts +++ b/middleware/test/pluginInstallPathTraversal.test.ts @@ -13,6 +13,10 @@ * 1. `adaptManifestV1` rejects an id/version outside the documented charset; * 2. `resolveContainedPackageDir` re-proves containment inside the upload * service, immediately before the destructive `fs.rm`. + * + * `pluginPackageSqlAllowlist.test.ts` depends on both holding: `.sql` is only + * in the extension allowlist because a package can never land in a + * migrator-scanned directory. */ import { describe, it, beforeEach, afterEach } from 'node:test'; @@ -21,18 +25,17 @@ import { promises as fs } from 'node:fs'; import os from 'node:os'; import path from 'node:path'; -import yazl from 'yazl'; - import { adaptManifestV1 } from '../src/plugins/manifestLoader.js'; -import type { PluginCatalog } from '../src/plugins/manifestLoader.js'; import { PackageUploadService, resolveContainedPackageDir, } from '../src/plugins/packageUploadService.js'; -import type { - UploadedPackage, - UploadedPackageStore, -} from '../src/plugins/uploadedPackageStore.js'; +import type { UploadedPackage } from '../src/plugins/uploadedPackageStore.js'; +import { + fakeCatalog, + fakeStore, + packageZip, +} from './_helpers/pluginPackageZip.js'; // --------------------------------------------------------------------------- // Layer 1 — manifest loader charset gate @@ -169,64 +172,6 @@ describe('resolveContainedPackageDir', () => { // End-to-end — the destructive rm must not be reached for a rejected package // --------------------------------------------------------------------------- -function buildZip(files: Record): Promise { - return new Promise((resolve, reject) => { - const zip = new yazl.ZipFile(); - const chunks: Buffer[] = []; - zip.outputStream.on('data', (c: Buffer) => chunks.push(c)); - zip.outputStream.on('end', () => resolve(Buffer.concat(chunks))); - zip.outputStream.on('error', reject); - for (const [name, content] of Object.entries(files)) { - zip.addBuffer(Buffer.from(content, 'utf-8'), name, { mtime: new Date(0) }); - } - zip.end(); - }); -} - -function manifestYaml(id: string, version: string): string { - return `schema_version: "1" - -identity: - id: ${JSON.stringify(id)} - name: "Traversal Fixture" - version: ${JSON.stringify(version)} - kind: "tool" - description: "Fixture plugin for traversal tests." - -compat: - core: ">=1.0 <2.0" - -lifecycle: - entry: "dist/plugin.js" -`; -} - -function packageZip(id: string, version: string): Promise { - return buildZip({ - 'manifest.yaml': manifestYaml(id, version), - 'dist/plugin.js': 'module.exports = { activate() {} };\n', - }); -} - -function fakeStore(): UploadedPackageStore { - const packages = new Map(); - return { - get: (id: string) => packages.get(id), - list: () => [...packages.values()], - register: async (pkg: UploadedPackage) => { - packages.set(pkg.id, pkg); - }, - } as unknown as UploadedPackageStore; -} - -function fakeCatalog(): PluginCatalog { - return { - get: () => undefined, - load: async () => undefined, - list: () => [], - } as unknown as PluginCatalog; -} - describe('PackageUploadService ingest × path traversal', () => { let root: string; let packagesDir: string; diff --git a/middleware/test/pluginPackageSqlAllowlist.test.ts b/middleware/test/pluginPackageSqlAllowlist.test.ts new file mode 100644 index 000000000..6fbdc78b9 --- /dev/null +++ b/middleware/test/pluginPackageSqlAllowlist.test.ts @@ -0,0 +1,265 @@ +/** + * `.sql` in the plugin-package extension allowlist. + * + * A distributed plugin ships its own schema as `migrations/*.sql`, so the + * extractor has to accept the extension. That entry is only admissible + * because uploaded content can never reach a directory a migrator scans: + * + * - all eight SQL migrators resolve their directory from their own + * `import.meta.url` (or the fixed `middleware/migrations` / an operator + * env override), never from package content; + * - uploaded packages land only under the packages root, whose + * `/` segments are charset-gated in `manifestLoader` and + * containment-re-checked in `packageUploadService` (commit 09ff9cd0); + * - the `node_modules` symlink at the packages root — the one path that + * would lead into the host tree — is a reserved id. + * + * Adding `.sql` before that traversal was closed would have turned a + * directory-delete into arbitrary SQL executed at boot. These tests pin the + * conditions, not just the happy path: if any of them starts failing, `.sql` + * has to come back out of the allowlist. + * + * The identity/containment gates themselves are covered by + * `pluginInstallPathTraversal.test.ts`. + */ + +import { describe, it, beforeEach, afterEach } from 'node:test'; +import { strict as assert } from 'node:assert'; +import { promises as fs } from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; + +import { PackageUploadService } from '../src/plugins/packageUploadService.js'; +import { extractZipToDir, ZipExtractionError } from '../src/plugins/zipExtractor.js'; +import { + buildZip, + fakeCatalog, + fakeStore, + manifestYaml, + packageZip, +} from './_helpers/pluginPackageZip.js'; + +const MIGRATION_SQL = 'CREATE TABLE plugin_notes (id TEXT PRIMARY KEY);\n'; + +function sqlPackageZip(id: string, version: string): Promise { + return packageZip(id, version, { 'migrations/001_init.sql': MIGRATION_SQL }); +} + +/** + * A package zip carrying a Zip-Slip `.sql` entry. yazl refuses to *author* a + * traversing entry name, so the name is swapped in afterwards at the byte + * level: the placeholder and the payload are the same length, and the name is + * stored verbatim in both the local header and the central directory, so a + * global replace over the buffer keeps every offset valid. + */ +const SLIP_PLACEHOLDER = 'xx/xx/migrations/999_pwn.sql'; +const SLIP_REAL = '../../migrations/999_pwn.sql'; + +async function zipSlipSqlZip(): Promise { + assert.equal(SLIP_PLACEHOLDER.length, SLIP_REAL.length); + const raw = await buildZip({ + 'manifest.yaml': manifestYaml('@omadia/plugin-notes', '1.0.0'), + 'dist/plugin.js': 'module.exports = { activate() {} };\n', + [SLIP_PLACEHOLDER]: 'DROP TABLE users;\n', + }); + const patched = Buffer.from( + raw.toString('latin1').split(SLIP_PLACEHOLDER).join(SLIP_REAL), + 'latin1', + ); + assert.equal(patched.length, raw.length); + return patched; +} + +/** Every `.sql` file under `dir`, as paths relative to `dir`, sorted. */ +async function sqlFilesUnder(dir: string): Promise { + const found: string[] = []; + const walk = async (current: string): Promise => { + const entries = await fs.readdir(current, { withFileTypes: true }); + for (const e of entries) { + const abs = path.join(current, e.name); + if (e.isDirectory()) await walk(abs); + else if (e.name.endsWith('.sql')) found.push(path.relative(dir, abs)); + } + }; + await walk(dir); + return found.sort(); +} + +// --------------------------------------------------------------------------- +// Extractor-level — the allowlist entry itself +// --------------------------------------------------------------------------- + +describe('zipExtractor × .sql', () => { + let root: string; + + beforeEach(async () => { + root = await fs.mkdtemp(path.join(os.tmpdir(), 'zip-sql-')); + }); + + afterEach(async () => { + await fs.rm(root, { recursive: true, force: true }); + }); + + async function extract( + files: Record, + overrides: Partial[2]> = {}, + ): Promise { + const zipPath = path.join(root, 'in.zip'); + await fs.writeFile(zipPath, await buildZip(files)); + const result = await extractZipToDir(zipPath, path.join(root, 'out'), { + maxEntries: 50, + maxExtractedBytes: 1024 * 1024, + ...overrides, + }); + return result.files; + } + + it('accepts .sql under the default plugin-package allowlist', async () => { + const files = await extract({ 'migrations/001_init.sql': MIGRATION_SQL }); + assert.deepEqual(files, ['migrations/001_init.sql']); + assert.equal( + await fs.readFile(path.join(root, 'out', 'migrations', '001_init.sql'), 'utf8'), + MIGRATION_SQL, + ); + }); + + // The Profile-Bundle importer passes its own allowlist. A caller that opts + // out of the default set must not silently inherit `.sql` — the override is + // documented as replacing the default wholesale, and that has to stay true. + it('rejects .sql when an explicit extensionAllowlist omits it', async () => { + await assert.rejects( + () => + extract( + { 'migrations/001_init.sql': MIGRATION_SQL }, + { extensionAllowlist: new Set(['.json', '.yaml']) }, + ), + (err: unknown) => { + assert.ok(err instanceof ZipExtractionError); + assert.equal(err.code, 'zip.forbidden_extension'); + return true; + }, + ); + }); + + it('still rejects an extension outside both sets', async () => { + await assert.rejects( + () => extract({ 'run.sh': '#!/bin/sh\n' }), + (err: unknown) => + err instanceof ZipExtractionError && err.code === 'zip.forbidden_extension', + ); + }); +}); + +// --------------------------------------------------------------------------- +// Ingest-level — where the .sql is allowed to end up +// --------------------------------------------------------------------------- + +describe('PackageUploadService ingest × .sql payloads', () => { + let root: string; + let packagesDir: string; + let victimDir: string; + + beforeEach(async () => { + root = await fs.mkdtemp(path.join(os.tmpdir(), 'upload-sql-')); + packagesDir = path.join(root, '.uploaded-packages'); + await fs.mkdir(packagesDir, { recursive: true }); + // Stand-in for the real `migrations/` directory that sits next to the + // packages root in the production image and is readdir+executed at boot. + victimDir = path.join(root, 'migrations'); + await fs.mkdir(victimDir, { recursive: true }); + await fs.writeFile(path.join(victimDir, '001_init.sql'), 'SELECT 1;\n'); + }); + + afterEach(async () => { + await fs.rm(root, { recursive: true, force: true }); + }); + + /** The only `.sql` that may exist after a rejected ingest: the victim's own. */ + const untouched = (): string[] => [path.join('migrations', '001_init.sql')]; + + function service(): PackageUploadService { + return new PackageUploadService({ + store: fakeStore(), + catalog: fakeCatalog(), + packagesDir, + limits: { + maxBytes: 1024 * 1024, + maxExtractedBytes: 4 * 1024 * 1024, + maxEntries: 50, + }, + hostDependencies: {}, + log: () => undefined, + }); + } + + it('installs a package that ships its own .sql migrations', async () => { + const result = await service().ingest({ + fileBuffer: await sqlPackageZip('@omadia/plugin-notes', '1.0.0'), + originalFilename: 'plugin-notes.zip', + uploadedBy: 'operator@example.com', + }); + + assert.equal(result.ok, true); + const finalDir = path.join(packagesDir, '@omadia', 'plugin-notes', '1.0.0'); + assert.equal( + await fs.readFile(path.join(finalDir, 'migrations', '001_init.sql'), 'utf8'), + MIGRATION_SQL, + ); + }); + + // The escalation this allowlist entry was previously blocked on. Rejection + // happens after the staged extract but before the `fs.rm` + rename, so the + // assertion is that no `.sql` survives outside the packages root — not that + // the zip was never opened. + it('never lands .sql outside the packages root for a traversing identity', async () => { + const result = await service().ingest({ + fileBuffer: await sqlPackageZip('..', 'migrations'), + originalFilename: 'evil.zip', + uploadedBy: 'attacker@example.com', + }); + + assert.equal(result.ok, false); + assert.equal((result as { code: string }).code, 'package.manifest_invalid'); + assert.deepEqual(await sqlFilesUnder(root), untouched()); + }); + + // `node_modules` is a syntactically valid npm name, so the loader lets it + // through — the packages root's symlink to the host tree makes it the one + // id that would walk `.sql` straight into the shipped migrator directories. + it('never lands .sql through the reserved node_modules entry', async () => { + const result = await service().ingest({ + fileBuffer: await sqlPackageZip('node_modules', '1.0.0'), + originalFilename: 'evil.zip', + uploadedBy: 'attacker@example.com', + }); + + assert.equal(result.ok, false); + assert.equal((result as { code: string }).code, 'package.path_traversal'); + assert.deepEqual(await sqlFilesUnder(root), untouched()); + }); + + // Zip-Slip is the second way a `.sql` could reach a migrator dir, and it is + // independent of the manifest: the entry path itself does the traversing. + // + // Three guards stack here, and the OUTERMOST one wins: yauzl's own + // `validateFileName` rejects the entry while reading the central directory, + // so it never reaches the extractor's `zip.path_escape` branch (which stays + // as defence in depth) nor the manifest gate. yauzl surfaces that as a plain + // `Error` on the zipfile, which `extractZipToDir` re-throws unwrapped and + // `ingest` propagates — hence `rejects` rather than an `IngestFailure`. The + // security property is the same either way and is what is asserted. + it('rejects a .sql entry whose zip path escapes the staging root', async () => { + const fileBuffer = await zipSlipSqlZip(); + await assert.rejects( + () => + service().ingest({ + fileBuffer, + originalFilename: 'evil.zip', + uploadedBy: 'attacker@example.com', + }), + /invalid relative path/, + ); + + assert.deepEqual(await sqlFilesUnder(root), untouched()); + }); +}); From 3d18a4bb7dad9edd39b9b3595bd6f7ab65ce520f Mon Sep 17 00:00:00 2001 From: Marcel Wege Date: Thu, 30 Jul 2026 11:18:51 +0200 Subject: [PATCH 2/4] fix(migrations): make all 8 SQL migrators safe against concurrent multi-replica boot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every migrator was read-ledger -> filter -> apply with no mutual exclusion, so two replicas booting together both executed the same pending list. `CREATE ... IF NOT EXISTS` masks that; `ALTER TABLE ... ADD CONSTRAINT` does not — the loser gets 42710 and its boot fails. `conductorWebhookEndpointStore.pg.test.ts` already carried a test-level `migrateWithRetry` workaround for the same race. Design (identical in all eight, no shared helper — that is separate work): - Read the ledger BEFORE locking and return early when nothing is pending, so the steady-state boot takes no lock and can never queue behind a migrating replica. - Acquire with `pg_try_advisory_lock(4410, hashtext())` in a bounded 2s poll, never the blocking `pg_advisory_lock`: three of the eight run inside a plugin `activate()` that ToolPluginRuntime hard-caps at 10s, and the knowledge-graph plugin can already have spent 6s in `waitForPostgres` first. The try-variant also returns a boolean the code actually reads, so the session always knows whether it holds the lock. - Re-read the ledger UNDER the lock, so a replica that queued behind a winner never re-applies what the winner just applied. - A loser that never gets the lock re-reads the ledger: if the winner finished, it continues; otherwise it throws a retryable error naming the pending files. The message says "timed out" so `bootstrap.retryErroredPlugins` classifies it as transient instead of latching the plugin `errored`. - `pg_advisory_unlock` runs on the success path only, inside `try`, and its boolean is read. It is never awaited in `finally`, where it could hang on a half-open connection (these pools set no statement_timeout) or replace the original migration error. Any path that cannot prove the lock was released ends at `client.release(true)`, which destroys the connection and releases the session lock with it. `client.release()` always runs. - `ensureLedger` retries once on 42P07/23505: `CREATE TABLE IF NOT EXISTS` is not atomic against a concurrent `CREATE TABLE` of the same name. Tests cover, for all eight: no lock when there is no work, lock taken before the first migration, the under-lock re-read, the bounded loser error, the loser whose winner finished, an original error preserved through a failing migration, a throwing unlock, an unlock that reports not-held, a driver that models no advisory locks, the ledger DDL race, and that 6s + 2s fits the 10s activate cap. --- .../src/migrator.ts | 201 +++++++- .../harness-memory-postgres/src/migrator.ts | 195 ++++++- .../src/registry/migrator.ts | 203 +++++++- middleware/src/auth/migrator.ts | 200 +++++++- middleware/src/conductor/migrator.ts | 185 ++++++- middleware/src/plugins/routines/migrator.ts | 195 ++++++- middleware/src/profileSnapshots/migrator.ts | 197 ++++++- middleware/src/profileStorage/migrator.ts | 197 ++++++- .../test/migratorConcurrentBoot.test.ts | 483 ++++++++++++++++++ 9 files changed, 1898 insertions(+), 158 deletions(-) create mode 100644 middleware/test/migratorConcurrentBoot.test.ts diff --git a/middleware/packages/harness-knowledge-graph-neon/src/migrator.ts b/middleware/packages/harness-knowledge-graph-neon/src/migrator.ts index a585b6f83..8fb82e4a7 100644 --- a/middleware/packages/harness-knowledge-graph-neon/src/migrator.ts +++ b/middleware/packages/harness-knowledge-graph-neon/src/migrator.ts @@ -1,40 +1,106 @@ import { readdir, readFile } from 'node:fs/promises'; import { dirname, join } from 'node:path'; import { fileURLToPath } from 'node:url'; -import type { Pool } from 'pg'; +import type { Pool, PoolClient } from 'pg'; const MIGRATIONS_DIR = join( dirname(fileURLToPath(import.meta.url)), 'migrations', ); +const LEDGER_DDL = ` + CREATE TABLE IF NOT EXISTS _graph_migrations ( + id TEXT PRIMARY KEY, + applied_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + `; + +/** + * Advisory-lock coordinates. The namespace is shared by every SQL migrator; + * 4400 is `LOCK_NS_REGISTRY` (embedding registry) and 4401 the stale-vector + * clear, so 4410 keeps migrations clear of both. The second key is + * `hashtext()`, so each subsystem serialises against its own + * replicas only and never against a different subsystem's migrations. + */ +const LOCK_NS_MIGRATIONS = 4_410; +const LOCK_KEY = '_graph_migrations'; + +/** + * How long a replica waits for the migration lock before giving up. + * + * This is the tightest of the eight budgets and therefore sets the value for + * all of them: `activate()` in `plugin.ts` calls `waitForPostgres` (6s budget) + * and then this migrator, inside a `ToolPluginRuntime` timeout of 10s. 6s + 2s + * leaves 2s of headroom for the migrations themselves, and the blocking + * `pg_advisory_lock` is unusable here for exactly that reason. Exported so the + * budget can be asserted in a test instead of trusted in a comment. + */ +export const GRAPH_MIGRATION_LOCK_WAIT_MS = 2_000; +const LOCK_POLL_MS = 100; + +/** + * Apply pending knowledge-graph SQL migrations against the graph pool. + * + * Concurrency: read-ledger → filter → apply is not safe on its own. Two + * replicas booting together both see the same pending list and both execute + * it; `IF NOT EXISTS` hides that, `ADD CONSTRAINT` does not (42710 → the + * loser's boot fails, and the kernel treats the knowledge graph as required, + * so that is a crash-loop). The apply loop therefore runs under a + * session-scoped advisory lock, taken with `pg_try_advisory_lock` and only + * after the ledger says there is work to do — the steady-state boot takes no + * lock at all and pays nothing. + */ export async function runGraphMigrations( pool: Pool, log: (msg: string) => void = () => undefined, ): Promise { const client = await pool.connect(); + // Tracks whether THIS session provably holds the advisory lock. It is the + // only input to `client.release()` below: a connection that cannot prove it + // released the lock is destroyed rather than pooled, because ending the + // session is the only other way a session-scoped lock goes away. + let lockHeld = false; try { - await client.query(` - CREATE TABLE IF NOT EXISTS _graph_migrations ( - id TEXT PRIMARY KEY, - applied_at TIMESTAMPTZ NOT NULL DEFAULT now() + await ensureLedger(client); + + // Ledger first, lock second. The overwhelmingly common boot has nothing + // pending, and that boot must not pay for — or queue behind — a lock. + let pending = await pendingMigrations(client); + if (pending.length === 0) return; + + const deadline = Date.now() + GRAPH_MIGRATION_LOCK_WAIT_MS; + for (;;) { + lockHeld = await tryAcquireMigrationLock(client); + if (lockHeld) break; + const remaining = deadline - Date.now(); + if (remaining <= 0) break; + await sleep(Math.min(LOCK_POLL_MS, remaining)); + } + + if (!lockHeld) { + // Never a silent skip: re-read the ledger. If the holder finished while + // we waited, this replica's schema IS current and the boot continues. + pending = await pendingMigrations(client); + if (pending.length === 0) { + log('[graph] migrations applied by another replica while waiting'); + return; + } + // Otherwise the work is genuinely still owed. Failing loudly is the only + // honest option; the message says "timed out" deliberately, so + // `bootstrap.retryErroredPlugins` classifies it as transient and + // re-attempts activation instead of latching the plugin `errored`. + throw new Error( + `[graph] timed out after ${String(GRAPH_MIGRATION_LOCK_WAIT_MS)}ms waiting for the ${LOCK_KEY} advisory lock; ` + + `${String(pending.length)} migration(s) still pending (${pending.join(', ')}) — another replica is mid-migration, retry the boot`, ); - `); - - const applied = new Set( - ( - await client.query<{ id: string }>( - 'SELECT id FROM _graph_migrations', - ) - ).rows.map((r) => r.id), - ); + } - const files = (await readdir(MIGRATIONS_DIR)) - .filter((f) => f.endsWith('.sql')) - .sort(); + // Re-read UNDER the lock. The pre-lock read is a fast path, not a + // decision: the replica we queued behind may have applied part or all of + // that list before it released. + pending = await pendingMigrations(client); - for (const file of files) { - if (applied.has(file)) continue; + for (const file of pending) { const sql = await readFile(join(MIGRATIONS_DIR, file), 'utf8'); log(`[graph] applying migration ${file}`); await client.query('BEGIN'); @@ -50,7 +116,102 @@ export async function runGraphMigrations( throw err; } } + + // Unlock on the success path only, and inside `try` — never in `finally`. + // In `finally` it would run on a possibly half-open connection whose pool + // sets no `statement_timeout`, so it could hang the release indefinitely, + // and an unlock that throws there would replace the original migration + // error. On the failure path `lockHeld` stays true and the connection is + // destroyed instead, which releases the lock with the session. + if (await releaseMigrationLock(client)) lockHeld = false; } finally { - client.release(); + client.release(lockHeld); + } +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +/** + * `CREATE TABLE IF NOT EXISTS` is not atomic against a concurrent `CREATE + * TABLE` of the same name: the existence check and the catalog insert are + * separate steps, so two replicas booting together can both pass the check and + * the loser fails with 42P07 (duplicate_table) or 23505 (a unique violation on + * a system catalog index). The table exists either way, so one retry settles + * it — the second attempt takes the IF NOT EXISTS short-circuit. This runs + * outside any transaction, so the failed statement leaves nothing to roll back. + */ +async function ensureLedger(client: PoolClient): Promise { + try { + await client.query(LEDGER_DDL); + } catch (err) { + if (!isDuplicateObjectError(err)) throw err; + await client.query(LEDGER_DDL); + } +} + +function isDuplicateObjectError(err: unknown): boolean { + const code = (err as { code?: unknown } | null)?.code; + return code === '42P07' || code === '23505'; +} + +/** + * The ledger read, expressed as the list of files still owed. Called twice on + * the locking path — once before the lock and once after acquiring it. + */ +async function pendingMigrations(client: PoolClient): Promise { + const applied = new Set( + ( + await client.query<{ id: string }>( + 'SELECT id FROM _graph_migrations', + ) + ).rows.map((r) => r.id), + ); + + const files = (await readdir(MIGRATIONS_DIR)) + .filter((f) => f.endsWith('.sql')) + .sort(); + + return files.filter((f) => !applied.has(f)); +} + +/** + * Take the migration lock without ever blocking the backend, and REPORT + * whether it was taken. The boolean is the whole point: a caller that cannot + * distinguish "acquired" from "someone else holds it" cannot release anything + * either. + */ +async function tryAcquireMigrationLock(client: PoolClient): Promise { + const result = await client.query<{ locked: boolean }>( + 'SELECT pg_try_advisory_lock($1::int, hashtext($2)::int) AS locked', + [LOCK_NS_MIGRATIONS, LOCK_KEY], + ); + // A fake/limited driver that does not model advisory locks returns no row; + // treat that as acquired so unit tests still exercise the migrations. Mirrors + // `tryAcquireRegistryLock` in `vectorColumnMigration.ts`. + const row = result.rows[0]; + return row === undefined || row.locked !== false; +} + +/** + * Release the session lock, and REPORT whether it actually went. `false` is + * what makes the caller destroy the connection instead of pooling it — the + * only other way a session-scoped lock is released. Swallowing the answer + * hands a connection that may still hold the lock back to the pool, where it + * blocks every later replica's migration for the connection's lifetime. + */ +async function releaseMigrationLock(client: PoolClient): Promise { + try { + const result = await client.query<{ unlocked: boolean }>( + 'SELECT pg_advisory_unlock($1::int, hashtext($2)::int) AS unlocked', + [LOCK_NS_MIGRATIONS, LOCK_KEY], + ); + // "No row" mirrors the acquire side: a driver that does not model advisory + // locks never took one, so nothing is leaked by pooling the connection. + const row = result.rows[0]; + return row === undefined || row.unlocked !== false; + } catch { + return false; } } diff --git a/middleware/packages/harness-memory-postgres/src/migrator.ts b/middleware/packages/harness-memory-postgres/src/migrator.ts index 6aa1c63cc..449ad705e 100644 --- a/middleware/packages/harness-memory-postgres/src/migrator.ts +++ b/middleware/packages/harness-memory-postgres/src/migrator.ts @@ -1,13 +1,42 @@ import { readdir, readFile } from 'node:fs/promises'; import { dirname, join } from 'node:path'; import { fileURLToPath } from 'node:url'; -import type { Pool } from 'pg'; +import type { Pool, PoolClient } from 'pg'; const MIGRATIONS_DIR = join( dirname(fileURLToPath(import.meta.url)), 'migrations', ); +const LEDGER_DDL = ` + CREATE TABLE IF NOT EXISTS _memory_migrations ( + id TEXT PRIMARY KEY, + applied_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + `; + +/** + * Advisory-lock coordinates. The namespace is shared by every SQL migrator + * (4400/4401 belong to the knowledge-graph embedding registry and the + * stale-vector clear); the second key is `hashtext()`, so each + * subsystem serialises against its own replicas only and never against a + * different subsystem's migrations. + */ +const LOCK_NS_MIGRATIONS = 4_410; +const LOCK_KEY = '_memory_migrations'; + +/** + * How long a replica waits for the migration lock before giving up. This + * migrator runs inside `activate()`, which `ToolPluginRuntime` hard-caps at + * 10s, so the wait has to be bounded well inside that — hence + * `pg_try_advisory_lock` in a polling loop rather than the blocking + * `pg_advisory_lock`, which would make a concurrent boot a deterministic + * activation timeout. Exported so the budget can be asserted in a test + * instead of trusted in a comment. + */ +export const MEMORY_MIGRATION_LOCK_WAIT_MS = 2_000; +const LOCK_POLL_MS = 100; + /** * Applies the memory-store schema migrations against the shared graphPool. * @@ -16,34 +45,65 @@ const MIGRATIONS_DIR = join( * name) have already run; each unapplied file executes inside its own * transaction. Safe to call on every activate() — already-applied migrations * are skipped. + * + * Concurrency: read-ledger → filter → apply is not safe on its own. Two + * replicas booting together both see the same pending list and both execute + * it; `IF NOT EXISTS` hides that, `ADD CONSTRAINT` does not (42710 → the + * loser's boot fails). So the apply loop runs under a session-scoped advisory + * lock, taken only after the ledger says there is work to do — the + * steady-state boot takes no lock at all. */ export async function runMemoryMigrations( pool: Pool, log: (msg: string) => void = () => undefined, ): Promise { const client = await pool.connect(); + // Tracks whether THIS session provably holds the advisory lock. It is the + // only input to `client.release()` below: a connection that cannot prove it + // released the lock is destroyed rather than pooled, because ending the + // session is the only other way a session-scoped lock goes away. + let lockHeld = false; try { - await client.query(` - CREATE TABLE IF NOT EXISTS _memory_migrations ( - id TEXT PRIMARY KEY, - applied_at TIMESTAMPTZ NOT NULL DEFAULT now() + await ensureLedger(client); + + // Ledger first, lock second. The overwhelmingly common boot has nothing + // pending, and that boot must not pay for — or queue behind — a lock. + let pending = await pendingMigrations(client); + if (pending.length === 0) return; + + const deadline = Date.now() + MEMORY_MIGRATION_LOCK_WAIT_MS; + for (;;) { + lockHeld = await tryAcquireMigrationLock(client); + if (lockHeld) break; + const remaining = deadline - Date.now(); + if (remaining <= 0) break; + await sleep(Math.min(LOCK_POLL_MS, remaining)); + } + + if (!lockHeld) { + // Never a silent skip: re-read the ledger. If the holder finished while + // we waited, this replica's schema IS current and the boot continues. + pending = await pendingMigrations(client); + if (pending.length === 0) { + log('[memory-pg] migrations applied by another replica while waiting'); + return; + } + // Otherwise the work is genuinely still owed. Failing loudly is the only + // honest option; the message says "timed out" deliberately, so + // `bootstrap.retryErroredPlugins` classifies it as transient and + // re-attempts activation instead of latching the plugin `errored`. + throw new Error( + `[memory-pg] timed out after ${String(MEMORY_MIGRATION_LOCK_WAIT_MS)}ms waiting for the ${LOCK_KEY} advisory lock; ` + + `${String(pending.length)} migration(s) still pending (${pending.join(', ')}) — another replica is mid-migration, retry the boot`, ); - `); - - const applied = new Set( - ( - await client.query<{ id: string }>( - 'SELECT id FROM _memory_migrations', - ) - ).rows.map((r) => r.id), - ); + } - const files = (await readdir(MIGRATIONS_DIR)) - .filter((f) => f.endsWith('.sql')) - .sort(); + // Re-read UNDER the lock. The pre-lock read is a fast path, not a + // decision: the replica we queued behind may have applied part or all of + // that list before it released. + pending = await pendingMigrations(client); - for (const file of files) { - if (applied.has(file)) continue; + for (const file of pending) { const sql = await readFile(join(MIGRATIONS_DIR, file), 'utf8'); log(`[memory-pg] applying migration ${file}`); await client.query('BEGIN'); @@ -59,7 +119,102 @@ export async function runMemoryMigrations( throw err; } } + + // Unlock on the success path only, and inside `try` — never in `finally`. + // In `finally` it would run on a possibly half-open connection whose pool + // sets no `statement_timeout`, so it could hang the release indefinitely, + // and an unlock that throws there would replace the original migration + // error. On the failure path `lockHeld` stays true and the connection is + // destroyed instead, which releases the lock with the session. + if (await releaseMigrationLock(client)) lockHeld = false; } finally { - client.release(); + client.release(lockHeld); + } +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +/** + * `CREATE TABLE IF NOT EXISTS` is not atomic against a concurrent `CREATE + * TABLE` of the same name: the existence check and the catalog insert are + * separate steps, so two replicas booting together can both pass the check and + * the loser fails with 42P07 (duplicate_table) or 23505 (a unique violation on + * a system catalog index). The table exists either way, so one retry settles + * it — the second attempt takes the IF NOT EXISTS short-circuit. This runs + * outside any transaction, so the failed statement leaves nothing to roll back. + */ +async function ensureLedger(client: PoolClient): Promise { + try { + await client.query(LEDGER_DDL); + } catch (err) { + if (!isDuplicateObjectError(err)) throw err; + await client.query(LEDGER_DDL); + } +} + +function isDuplicateObjectError(err: unknown): boolean { + const code = (err as { code?: unknown } | null)?.code; + return code === '42P07' || code === '23505'; +} + +/** + * The ledger read, expressed as the list of files still owed. Called twice on + * the locking path — once before the lock and once after acquiring it. + */ +async function pendingMigrations(client: PoolClient): Promise { + const applied = new Set( + ( + await client.query<{ id: string }>( + 'SELECT id FROM _memory_migrations', + ) + ).rows.map((r) => r.id), + ); + + const files = (await readdir(MIGRATIONS_DIR)) + .filter((f) => f.endsWith('.sql')) + .sort(); + + return files.filter((f) => !applied.has(f)); +} + +/** + * Take the migration lock without ever blocking the backend, and REPORT + * whether it was taken. The boolean is the whole point: a caller that cannot + * distinguish "acquired" from "someone else holds it" cannot release anything + * either. + */ +async function tryAcquireMigrationLock(client: PoolClient): Promise { + const result = await client.query<{ locked: boolean }>( + 'SELECT pg_try_advisory_lock($1::int, hashtext($2)::int) AS locked', + [LOCK_NS_MIGRATIONS, LOCK_KEY], + ); + // A fake/limited driver that does not model advisory locks returns no row; + // treat that as acquired so unit tests still exercise the migrations. Mirrors + // `tryAcquireRegistryLock` in @omadia/knowledge-graph-neon. + const row = result.rows[0]; + return row === undefined || row.locked !== false; +} + +/** + * Release the session lock, and REPORT whether it actually went. `false` is + * what makes the caller destroy the connection instead of pooling it — the + * only other way a session-scoped lock is released. Swallowing the answer + * hands a connection that may still hold the lock back to the pool, where it + * blocks every later replica's migration for the connection's lifetime. + */ +async function releaseMigrationLock(client: PoolClient): Promise { + try { + const result = await client.query<{ unlocked: boolean }>( + 'SELECT pg_advisory_unlock($1::int, hashtext($2)::int) AS unlocked', + [LOCK_NS_MIGRATIONS, LOCK_KEY], + ); + // "No row" mirrors the acquire side: a driver that does not model advisory + // locks never took one, so nothing is leaked by pooling the connection. + const row = result.rows[0]; + return row === undefined || row.unlocked !== false; + } catch { + return false; } } diff --git a/middleware/packages/harness-orchestrator/src/registry/migrator.ts b/middleware/packages/harness-orchestrator/src/registry/migrator.ts index 21c8b873a..8c0c5b6cc 100644 --- a/middleware/packages/harness-orchestrator/src/registry/migrator.ts +++ b/middleware/packages/harness-orchestrator/src/registry/migrator.ts @@ -2,7 +2,7 @@ import { readdir, readFile } from 'node:fs/promises'; import { dirname, join, resolve } from 'node:path'; import { fileURLToPath } from 'node:url'; -import type { Pool } from 'pg'; +import type { Pool, PoolClient } from 'pg'; /** * Resolve the `migrations/` directory from this file's location. The path @@ -25,16 +25,52 @@ function defaultMigrationsDir(): string { return resolve(here, '..', '..', '..', '..', 'migrations'); } +const LEDGER_DDL = ` + CREATE TABLE IF NOT EXISTS _multi_orchestrator_migrations ( + id TEXT PRIMARY KEY, + applied_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + `; + +/** + * Advisory-lock coordinates. The namespace is shared by every SQL migrator + * (4400/4401 belong to the knowledge-graph embedding registry and the + * stale-vector clear); the second key is `hashtext()`, so each + * subsystem serialises against its own replicas only and never against a + * different subsystem's migrations. + */ +const LOCK_NS_MIGRATIONS = 4_410; +const LOCK_KEY = '_multi_orchestrator_migrations'; + +/** + * How long a replica waits for the migration lock before giving up. This + * migrator runs inside `activate()`, which `ToolPluginRuntime` hard-caps at + * 10s, so the wait has to be bounded well inside that — hence + * `pg_try_advisory_lock` in a polling loop rather than the blocking + * `pg_advisory_lock`, which would make a concurrent boot a deterministic + * activation timeout. Exported so the budget can be asserted in a test + * instead of trusted in a comment. + */ +export const MULTI_ORCH_MIGRATION_LOCK_WAIT_MS = 2_000; +const LOCK_POLL_MS = 100; + /** * Apply pending multi-orchestrator-runtime SQL migrations. * * Mirrors `runAuthMigrations` / `runGraphMigrations` / `runRoutineMigrations` - * line for line so the four migrators stay diff-comparable. Bookkeeping table + * line for line so the migrators stay diff-comparable. Bookkeeping table * `_multi_orchestrator_migrations` is independent of the other three because * the multi-orchestrator schema has its own evolution cadence. * * Each file is wrapped in a transaction and recorded only on commit, so a * partial failure leaves the tracking table consistent. + * + * Concurrency: read-ledger → filter → apply is not safe on its own. Two + * replicas booting together both see the same pending list and both execute + * it; `IF NOT EXISTS` hides that, `ADD CONSTRAINT` does not (42710 → the + * loser's boot fails). So the apply loop runs under a session-scoped advisory + * lock, taken only after the ledger says there is work to do — the + * steady-state boot takes no lock at all. */ export async function runMultiOrchestratorMigrations( pool: Pool, @@ -42,28 +78,54 @@ export async function runMultiOrchestratorMigrations( migrationsDir: string = defaultMigrationsDir(), ): Promise { const client = await pool.connect(); + // Tracks whether THIS session provably holds the advisory lock. It is the + // only input to `client.release()` below: a connection that cannot prove it + // released the lock is destroyed rather than pooled, because ending the + // session is the only other way a session-scoped lock goes away. + let lockHeld = false; try { - await client.query(` - CREATE TABLE IF NOT EXISTS _multi_orchestrator_migrations ( - id TEXT PRIMARY KEY, - applied_at TIMESTAMPTZ NOT NULL DEFAULT now() + await ensureLedger(client); + + // Ledger first, lock second. The overwhelmingly common boot has nothing + // pending, and that boot must not pay for — or queue behind — a lock. + let pending = await pendingMigrations(client, migrationsDir); + if (pending.length === 0) return; + + const deadline = Date.now() + MULTI_ORCH_MIGRATION_LOCK_WAIT_MS; + for (;;) { + lockHeld = await tryAcquireMigrationLock(client); + if (lockHeld) break; + const remaining = deadline - Date.now(); + if (remaining <= 0) break; + await sleep(Math.min(LOCK_POLL_MS, remaining)); + } + + if (!lockHeld) { + // Never a silent skip: re-read the ledger. If the holder finished while + // we waited, this replica's schema IS current and the boot continues. + pending = await pendingMigrations(client, migrationsDir); + if (pending.length === 0) { + log( + '[multi-orchestrator] migrations applied by another replica while waiting', + ); + return; + } + // Otherwise the work is genuinely still owed. Failing loudly is the only + // honest option; the message says "timed out" deliberately, so + // `bootstrap.retryErroredPlugins` classifies it as transient and + // re-attempts activation instead of latching the plugin `errored`. + throw new Error( + `[multi-orchestrator] timed out after ${String(MULTI_ORCH_MIGRATION_LOCK_WAIT_MS)}ms waiting for the ${LOCK_KEY} advisory lock; ` + + `${String(pending.length)} migration(s) still pending (${pending.join(', ')}) — another replica is mid-migration, retry the boot`, ); - `); - - const applied = new Set( - ( - await client.query<{ id: string }>( - 'SELECT id FROM _multi_orchestrator_migrations', - ) - ).rows.map((r) => r.id), - ); + } - const files = (await readdir(migrationsDir)) - .filter((f) => f.endsWith('.sql')) - .sort(); + // Re-read UNDER the lock. The pre-lock read is a fast path, not a + // decision: the replica we queued behind may have applied part or all of + // that list before it released. + pending = await pendingMigrations(client, migrationsDir); - for (const file of files) { - if (applied.has(file)) continue; + for (const file of pending) { const sql = await readFile(join(migrationsDir, file), 'utf8'); log(`[multi-orchestrator] applying migration ${file}`); await client.query('BEGIN'); @@ -79,7 +141,106 @@ export async function runMultiOrchestratorMigrations( throw err; } } + + // Unlock on the success path only, and inside `try` — never in `finally`. + // In `finally` it would run on a possibly half-open connection whose pool + // sets no `statement_timeout`, so it could hang the release indefinitely, + // and an unlock that throws there would replace the original migration + // error. On the failure path `lockHeld` stays true and the connection is + // destroyed instead, which releases the lock with the session. + if (await releaseMigrationLock(client)) lockHeld = false; } finally { - client.release(); + client.release(lockHeld); + } +} + +function sleep(ms: number): Promise { + // `done`, not `resolve` — `resolve` is already the path import above. + return new Promise((done) => setTimeout(done, ms)); +} + +/** + * `CREATE TABLE IF NOT EXISTS` is not atomic against a concurrent `CREATE + * TABLE` of the same name: the existence check and the catalog insert are + * separate steps, so two replicas booting together can both pass the check and + * the loser fails with 42P07 (duplicate_table) or 23505 (a unique violation on + * a system catalog index). The table exists either way, so one retry settles + * it — the second attempt takes the IF NOT EXISTS short-circuit. This runs + * outside any transaction, so the failed statement leaves nothing to roll back. + */ +async function ensureLedger(client: PoolClient): Promise { + try { + await client.query(LEDGER_DDL); + } catch (err) { + if (!isDuplicateObjectError(err)) throw err; + await client.query(LEDGER_DDL); + } +} + +function isDuplicateObjectError(err: unknown): boolean { + const code = (err as { code?: unknown } | null)?.code; + return code === '42P07' || code === '23505'; +} + +/** + * The ledger read, expressed as the list of files still owed. Called twice on + * the locking path — once before the lock and once after acquiring it. + */ +async function pendingMigrations( + client: PoolClient, + migrationsDir: string, +): Promise { + const applied = new Set( + ( + await client.query<{ id: string }>( + 'SELECT id FROM _multi_orchestrator_migrations', + ) + ).rows.map((r) => r.id), + ); + + const files = (await readdir(migrationsDir)) + .filter((f) => f.endsWith('.sql')) + .sort(); + + return files.filter((f) => !applied.has(f)); +} + +/** + * Take the migration lock without ever blocking the backend, and REPORT + * whether it was taken. The boolean is the whole point: a caller that cannot + * distinguish "acquired" from "someone else holds it" cannot release anything + * either. + */ +async function tryAcquireMigrationLock(client: PoolClient): Promise { + const result = await client.query<{ locked: boolean }>( + 'SELECT pg_try_advisory_lock($1::int, hashtext($2)::int) AS locked', + [LOCK_NS_MIGRATIONS, LOCK_KEY], + ); + // A fake/limited driver that does not model advisory locks returns no row; + // treat that as acquired so unit tests still exercise the migrations. Mirrors + // `tryAcquireRegistryLock` in @omadia/knowledge-graph-neon. + const row = result.rows[0]; + return row === undefined || row.locked !== false; +} + +/** + * Release the session lock, and REPORT whether it actually went. `false` is + * what makes the caller destroy the connection instead of pooling it — the + * only other way a session-scoped lock is released. Swallowing the answer + * hands a connection that may still hold the lock back to the pool, where it + * blocks every later replica's migration for the connection's lifetime. + */ +async function releaseMigrationLock(client: PoolClient): Promise { + try { + const result = await client.query<{ unlocked: boolean }>( + 'SELECT pg_advisory_unlock($1::int, hashtext($2)::int) AS unlocked', + [LOCK_NS_MIGRATIONS, LOCK_KEY], + ); + // "No row" mirrors the acquire side: a driver that does not model advisory + // locks never took one, so nothing is leaked by pooling the connection. + const row = result.rows[0]; + return row === undefined || row.unlocked !== false; + } catch { + return false; } } diff --git a/middleware/src/auth/migrator.ts b/middleware/src/auth/migrator.ts index 1115794ed..76b2e638e 100644 --- a/middleware/src/auth/migrator.ts +++ b/middleware/src/auth/migrator.ts @@ -1,13 +1,43 @@ import { readdir, readFile } from 'node:fs/promises'; import { dirname, join } from 'node:path'; import { fileURLToPath } from 'node:url'; -import type { Pool } from 'pg'; +import type { Pool, PoolClient } from 'pg'; const MIGRATIONS_DIR = join( dirname(fileURLToPath(import.meta.url)), 'migrations', ); +const LEDGER_DDL = ` + CREATE TABLE IF NOT EXISTS _auth_migrations ( + id TEXT PRIMARY KEY, + applied_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + `; + +/** + * Advisory-lock coordinates. The namespace is shared by every SQL migrator + * (4400/4401 belong to the knowledge-graph embedding registry and the + * stale-vector clear); the second key is `hashtext()`, so each + * subsystem serialises against its own replicas only and never against a + * different subsystem's migrations. + */ +const LOCK_NS_MIGRATIONS = 4_410; +const LOCK_KEY = '_auth_migrations'; + +/** + * How long a replica waits for the migration lock before giving up. + * + * Three of the eight migrators run inside a plugin `activate()`, which + * `ToolPluginRuntime` hard-caps at 10s, and the knowledge-graph plugin can + * already have spent 6s of that budget in `waitForPostgres` before its + * migrator is even called. 2s therefore has to be the ceiling for all eight — + * the tightest consumer sets the bound for everyone, and the value is exported + * so the budget can be asserted in a test instead of trusted in a comment. + */ +export const AUTH_MIGRATION_LOCK_WAIT_MS = 2_000; +const LOCK_POLL_MS = 100; + /** * Apply pending auth-subsystem SQL migrations against the shared Postgres * pool. Tracking happens in `_auth_migrations` so the lifecycle is fully @@ -18,36 +48,69 @@ const MIGRATIONS_DIR = join( * Idempotent: each file is wrapped in a transaction and recorded only on * commit, so a partial failure leaves the tracking table consistent. * + * Concurrency: read-ledger → filter → apply is not safe on its own. Two + * replicas booting together both see the same pending list and both execute + * it; `IF NOT EXISTS` hides that, `ADD CONSTRAINT` does not (42710 → the + * loser's boot fails). So the apply loop runs under a session-scoped advisory + * lock, taken with `pg_try_advisory_lock` (never the blocking variant — an + * unbounded server-side wait inside a 10s `activate()` would turn a rare race + * into a deterministic boot failure) and only after the ledger says there is + * work to do, so the steady-state boot takes no lock at all. + * * Mirrors `runGraphMigrations` and `runRoutineMigrations` line for line so - * the three migrators stay diff-comparable. + * the migrators stay diff-comparable. */ export async function runAuthMigrations( pool: Pool, log: (msg: string) => void = () => undefined, ): Promise { const client = await pool.connect(); + // Tracks whether THIS session provably holds the advisory lock. It is the + // only input to `client.release()` below: a connection that cannot prove it + // released the lock is destroyed rather than pooled, because ending the + // session is the only other way a session-scoped lock goes away. + let lockHeld = false; try { - await client.query(` - CREATE TABLE IF NOT EXISTS _auth_migrations ( - id TEXT PRIMARY KEY, - applied_at TIMESTAMPTZ NOT NULL DEFAULT now() + await ensureLedger(client); + + // Ledger first, lock second. The overwhelmingly common boot has nothing + // pending, and that boot must not pay for — or queue behind — a lock. + let pending = await pendingMigrations(client); + if (pending.length === 0) return; + + const deadline = Date.now() + AUTH_MIGRATION_LOCK_WAIT_MS; + for (;;) { + lockHeld = await tryAcquireMigrationLock(client); + if (lockHeld) break; + const remaining = deadline - Date.now(); + if (remaining <= 0) break; + await sleep(Math.min(LOCK_POLL_MS, remaining)); + } + + if (!lockHeld) { + // Never a silent skip: re-read the ledger. If the holder finished while + // we waited, this replica's schema IS current and the boot continues. + pending = await pendingMigrations(client); + if (pending.length === 0) { + log('[auth] migrations applied by another replica while waiting'); + return; + } + // Otherwise the work is genuinely still owed. Failing loudly is the only + // honest option; the message says "timed out" deliberately, so + // `bootstrap.retryErroredPlugins` classifies it as transient and + // re-attempts activation instead of latching the plugin `errored`. + throw new Error( + `[auth] timed out after ${String(AUTH_MIGRATION_LOCK_WAIT_MS)}ms waiting for the ${LOCK_KEY} advisory lock; ` + + `${String(pending.length)} migration(s) still pending (${pending.join(', ')}) — another replica is mid-migration, retry the boot`, ); - `); - - const applied = new Set( - ( - await client.query<{ id: string }>( - 'SELECT id FROM _auth_migrations', - ) - ).rows.map((r) => r.id), - ); + } - const files = (await readdir(MIGRATIONS_DIR)) - .filter((f) => f.endsWith('.sql')) - .sort(); + // Re-read UNDER the lock. The pre-lock read is a fast path, not a + // decision: the replica we queued behind may have applied part or all of + // that list before it released. + pending = await pendingMigrations(client); - for (const file of files) { - if (applied.has(file)) continue; + for (const file of pending) { const sql = await readFile(join(MIGRATIONS_DIR, file), 'utf8'); log(`[auth] applying migration ${file}`); await client.query('BEGIN'); @@ -63,7 +126,102 @@ export async function runAuthMigrations( throw err; } } + + // Unlock on the success path only, and inside `try` — never in `finally`. + // In `finally` it would run on a possibly half-open connection whose pool + // sets no `statement_timeout`, so it could hang the release indefinitely, + // and an unlock that throws there would replace the original migration + // error. On the failure path `lockHeld` stays true and the connection is + // destroyed instead, which releases the lock with the session. + if (await releaseMigrationLock(client)) lockHeld = false; } finally { - client.release(); + client.release(lockHeld); + } +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +/** + * `CREATE TABLE IF NOT EXISTS` is not atomic against a concurrent `CREATE + * TABLE` of the same name: the existence check and the catalog insert are + * separate steps, so two replicas booting together can both pass the check and + * the loser fails with 42P07 (duplicate_table) or 23505 (a unique violation on + * a system catalog index). The table exists either way, so one retry settles + * it — the second attempt takes the IF NOT EXISTS short-circuit. This runs + * outside any transaction, so the failed statement leaves nothing to roll back. + */ +async function ensureLedger(client: PoolClient): Promise { + try { + await client.query(LEDGER_DDL); + } catch (err) { + if (!isDuplicateObjectError(err)) throw err; + await client.query(LEDGER_DDL); + } +} + +function isDuplicateObjectError(err: unknown): boolean { + const code = (err as { code?: unknown } | null)?.code; + return code === '42P07' || code === '23505'; +} + +/** + * The ledger read, expressed as the list of files still owed. Called twice on + * the locking path — once before the lock and once after acquiring it. + */ +async function pendingMigrations(client: PoolClient): Promise { + const applied = new Set( + ( + await client.query<{ id: string }>( + 'SELECT id FROM _auth_migrations', + ) + ).rows.map((r) => r.id), + ); + + const files = (await readdir(MIGRATIONS_DIR)) + .filter((f) => f.endsWith('.sql')) + .sort(); + + return files.filter((f) => !applied.has(f)); +} + +/** + * Take the migration lock without ever blocking the backend, and REPORT + * whether it was taken. The boolean is the whole point: the previous attempt + * could not distinguish "acquired" from "someone else holds it", which is what + * made its release path inert. + */ +async function tryAcquireMigrationLock(client: PoolClient): Promise { + const result = await client.query<{ locked: boolean }>( + 'SELECT pg_try_advisory_lock($1::int, hashtext($2)::int) AS locked', + [LOCK_NS_MIGRATIONS, LOCK_KEY], + ); + // A fake/limited driver that does not model advisory locks returns no row; + // treat that as acquired so unit tests still exercise the migrations. Mirrors + // `tryAcquireRegistryLock` in @omadia/knowledge-graph-neon. + const row = result.rows[0]; + return row === undefined || row.locked !== false; +} + +/** + * Release the session lock, and REPORT whether it actually went. `false` is + * what makes the caller destroy the connection instead of pooling it — the + * only other way a session-scoped lock is released. Swallowing the answer + * hands a connection that may still hold the lock back to the pool, where it + * blocks every later replica's migration for the connection's lifetime. + */ +async function releaseMigrationLock(client: PoolClient): Promise { + try { + const result = await client.query<{ unlocked: boolean }>( + 'SELECT pg_advisory_unlock($1::int, hashtext($2)::int) AS unlocked', + [LOCK_NS_MIGRATIONS, LOCK_KEY], + ); + // "No row" mirrors the acquire side: a driver that does not model advisory + // locks never took one, so nothing is leaked by pooling the connection. + const row = result.rows[0]; + return row === undefined || row.unlocked !== false; + } catch { + return false; } } diff --git a/middleware/src/conductor/migrator.ts b/middleware/src/conductor/migrator.ts index 7199b36ff..f133baa03 100644 --- a/middleware/src/conductor/migrator.ts +++ b/middleware/src/conductor/migrator.ts @@ -1,10 +1,36 @@ import { readdir, readFile } from 'node:fs/promises'; import { dirname, join } from 'node:path'; import { fileURLToPath } from 'node:url'; -import type { Pool } from 'pg'; +import type { Pool, PoolClient } from 'pg'; const MIGRATIONS_DIR = join(dirname(fileURLToPath(import.meta.url)), 'migrations'); +const LEDGER_DDL = ` + CREATE TABLE IF NOT EXISTS _conductor_migrations ( + id TEXT PRIMARY KEY, + applied_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + `; + +/** + * Advisory-lock coordinates. The namespace is shared by every SQL migrator + * (4400/4401 belong to the knowledge-graph embedding registry and the stale-vector + * clear); the second key is `hashtext()`, so each subsystem serialises + * against its own replicas only and never against a different subsystem's migrations. + */ +const LOCK_NS_MIGRATIONS = 4_410; +const LOCK_KEY = '_conductor_migrations'; + +/** + * How long a replica waits for the migration lock before giving up. Held at the same + * 2s ceiling as the migrators that run inside a plugin `activate()` (which + * `ToolPluginRuntime` hard-caps at 10s) so the bound is one number across all eight + * rather than eight numbers to reason about. Exported so the budget can be asserted + * in a test instead of trusted in a comment. + */ +export const CONDUCTOR_MIGRATION_LOCK_WAIT_MS = 2_000; +const LOCK_POLL_MS = 100; + /** * Apply pending Conductor SQL migrations against the shared Postgres pool. * Tracking lives in `_conductor_migrations`, independent of the other @@ -12,30 +38,67 @@ const MIGRATIONS_DIR = join(dirname(fileURLToPath(import.meta.url)), 'migrations * migrators stay diff-comparable. * * Idempotent: each file runs in its own transaction, recorded only on commit. + * + * Concurrency: read-ledger → filter → apply is not safe on its own. Two replicas + * booting together both see the same pending list and both execute it; `IF NOT + * EXISTS` hides that, `ADD CONSTRAINT` does not (42710 → the loser's boot fails). + * So the apply loop runs under a session-scoped advisory lock, taken with + * `pg_try_advisory_lock` (never the blocking variant — an unbounded server-side wait + * inside a 10s `activate()` would turn a rare race into a deterministic boot failure) + * and only after the ledger says there is work to do, so the steady-state boot takes + * no lock at all. */ export async function runConductorMigrations( pool: Pool, log: (msg: string) => void = () => undefined, ): Promise { const client = await pool.connect(); + // Tracks whether THIS session provably holds the advisory lock. It is the only + // input to `client.release()` below: a connection that cannot prove it released + // the lock is destroyed rather than pooled, because ending the session is the + // only other way a session-scoped lock goes away. + let lockHeld = false; try { - await client.query(` - CREATE TABLE IF NOT EXISTS _conductor_migrations ( - id TEXT PRIMARY KEY, - applied_at TIMESTAMPTZ NOT NULL DEFAULT now() - ); - `); + await ensureLedger(client); - const applied = new Set( - (await client.query<{ id: string }>('SELECT id FROM _conductor_migrations')).rows.map( - (r) => r.id, - ), - ); + // Ledger first, lock second. The overwhelmingly common boot has nothing pending, + // and that boot must not pay for — or queue behind — a lock. + let pending = await pendingMigrations(client); + if (pending.length === 0) return; - const files = (await readdir(MIGRATIONS_DIR)).filter((f) => f.endsWith('.sql')).sort(); + const deadline = Date.now() + CONDUCTOR_MIGRATION_LOCK_WAIT_MS; + for (;;) { + lockHeld = await tryAcquireMigrationLock(client); + if (lockHeld) break; + const remaining = deadline - Date.now(); + if (remaining <= 0) break; + await sleep(Math.min(LOCK_POLL_MS, remaining)); + } - for (const file of files) { - if (applied.has(file)) continue; + if (!lockHeld) { + // Never a silent skip: re-read the ledger. If the holder finished while we + // waited, this replica's schema IS current and the boot continues. + pending = await pendingMigrations(client); + if (pending.length === 0) { + log('[conductor] migrations applied by another replica while waiting'); + return; + } + // Otherwise the work is genuinely still owed. Failing loudly is the only honest + // option; the message says "timed out" deliberately, so + // `bootstrap.retryErroredPlugins` classifies it as transient and re-attempts + // activation instead of latching the plugin `errored`. + throw new Error( + `[conductor] timed out after ${String(CONDUCTOR_MIGRATION_LOCK_WAIT_MS)}ms waiting for the ${LOCK_KEY} advisory lock; ` + + `${String(pending.length)} migration(s) still pending (${pending.join(', ')}) — another replica is mid-migration, retry the boot`, + ); + } + + // Re-read UNDER the lock. The pre-lock read is a fast path, not a decision: the + // replica we queued behind may have applied part or all of that list before it + // released. + pending = await pendingMigrations(client); + + for (const file of pending) { const sql = await readFile(join(MIGRATIONS_DIR, file), 'utf8'); log(`[conductor] applying migration ${file}`); await client.query('BEGIN'); @@ -48,7 +111,97 @@ export async function runConductorMigrations( throw err; } } + + // Unlock on the success path only, and inside `try` — never in `finally`. In + // `finally` it would run on a possibly half-open connection whose pool sets no + // `statement_timeout`, so it could hang the release indefinitely, and an unlock + // that throws there would replace the original migration error. On the failure + // path `lockHeld` stays true and the connection is destroyed instead, which + // releases the lock with the session. + if (await releaseMigrationLock(client)) lockHeld = false; } finally { - client.release(); + client.release(lockHeld); + } +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +/** + * `CREATE TABLE IF NOT EXISTS` is not atomic against a concurrent `CREATE TABLE` of + * the same name: the existence check and the catalog insert are separate steps, so + * two replicas booting together can both pass the check and the loser fails with + * 42P07 (duplicate_table) or 23505 (a unique violation on a system catalog index). + * The table exists either way, so one retry settles it — the second attempt takes + * the IF NOT EXISTS short-circuit. This runs outside any transaction, so the failed + * statement leaves nothing to roll back. + */ +async function ensureLedger(client: PoolClient): Promise { + try { + await client.query(LEDGER_DDL); + } catch (err) { + if (!isDuplicateObjectError(err)) throw err; + await client.query(LEDGER_DDL); + } +} + +function isDuplicateObjectError(err: unknown): boolean { + const code = (err as { code?: unknown } | null)?.code; + return code === '42P07' || code === '23505'; +} + +/** + * The ledger read, expressed as the list of files still owed. Called twice on the + * locking path — once before the lock and once after acquiring it. + */ +async function pendingMigrations(client: PoolClient): Promise { + const applied = new Set( + (await client.query<{ id: string }>('SELECT id FROM _conductor_migrations')).rows.map( + (r) => r.id, + ), + ); + + const files = (await readdir(MIGRATIONS_DIR)).filter((f) => f.endsWith('.sql')).sort(); + + return files.filter((f) => !applied.has(f)); +} + +/** + * Take the migration lock without ever blocking the backend, and REPORT whether it + * was taken. The boolean is the whole point: a caller that cannot distinguish + * "acquired" from "someone else holds it" cannot release anything either. + */ +async function tryAcquireMigrationLock(client: PoolClient): Promise { + const result = await client.query<{ locked: boolean }>( + 'SELECT pg_try_advisory_lock($1::int, hashtext($2)::int) AS locked', + [LOCK_NS_MIGRATIONS, LOCK_KEY], + ); + // A fake/limited driver that does not model advisory locks returns no row; treat + // that as acquired so unit tests still exercise the migrations. Mirrors + // `tryAcquireRegistryLock` in @omadia/knowledge-graph-neon. + const row = result.rows[0]; + return row === undefined || row.locked !== false; +} + +/** + * Release the session lock, and REPORT whether it actually went. `false` is what + * makes the caller destroy the connection instead of pooling it — the only other way + * a session-scoped lock is released. Swallowing the answer hands a connection that + * may still hold the lock back to the pool, where it blocks every later replica's + * migration for the connection's lifetime. + */ +async function releaseMigrationLock(client: PoolClient): Promise { + try { + const result = await client.query<{ unlocked: boolean }>( + 'SELECT pg_advisory_unlock($1::int, hashtext($2)::int) AS unlocked', + [LOCK_NS_MIGRATIONS, LOCK_KEY], + ); + // "No row" mirrors the acquire side: a driver that does not model advisory locks + // never took one, so nothing is leaked by pooling the connection. + const row = result.rows[0]; + return row === undefined || row.unlocked !== false; + } catch { + return false; } } diff --git a/middleware/src/plugins/routines/migrator.ts b/middleware/src/plugins/routines/migrator.ts index bcdcafdb8..2eda2b33c 100644 --- a/middleware/src/plugins/routines/migrator.ts +++ b/middleware/src/plugins/routines/migrator.ts @@ -1,13 +1,40 @@ import { readdir, readFile } from 'node:fs/promises'; import { dirname, join } from 'node:path'; import { fileURLToPath } from 'node:url'; -import type { Pool } from 'pg'; +import type { Pool, PoolClient } from 'pg'; const MIGRATIONS_DIR = join( dirname(fileURLToPath(import.meta.url)), 'migrations', ); +const LEDGER_DDL = ` + CREATE TABLE IF NOT EXISTS _routine_migrations ( + id TEXT PRIMARY KEY, + applied_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + `; + +/** + * Advisory-lock coordinates. The namespace is shared by every SQL migrator + * (4400/4401 belong to the knowledge-graph embedding registry and the + * stale-vector clear); the second key is `hashtext()`, so each + * subsystem serialises against its own replicas only and never against a + * different subsystem's migrations. + */ +const LOCK_NS_MIGRATIONS = 4_410; +const LOCK_KEY = '_routine_migrations'; + +/** + * How long a replica waits for the migration lock before giving up. Held at + * the same 2s ceiling as the migrators that run inside a plugin `activate()` + * (which `ToolPluginRuntime` hard-caps at 10s) so the bound is one number + * across all eight rather than eight numbers to reason about. Exported so the + * budget can be asserted in a test instead of trusted in a comment. + */ +export const ROUTINE_MIGRATION_LOCK_WAIT_MS = 2_000; +const LOCK_POLL_MS = 100; + /** * Apply pending routines-plugin SQL migrations against the shared Neon pool. * Tracks applied files in `_routine_migrations` so the lifecycle is @@ -15,34 +42,67 @@ const MIGRATIONS_DIR = join( * schema-evolution cadence). Idempotent: each file is wrapped in a * transaction and recorded only on commit, so a partial failure leaves the * tracking table consistent. + * + * Concurrency: read-ledger → filter → apply is not safe on its own. Two + * replicas booting together both see the same pending list and both execute + * it; `IF NOT EXISTS` hides that, `ADD CONSTRAINT` does not (42710 → the + * loser's boot fails). So the apply loop runs under a session-scoped advisory + * lock, taken with `pg_try_advisory_lock` (never the blocking variant — an + * unbounded server-side wait inside a 10s `activate()` would turn a rare race + * into a deterministic boot failure) and only after the ledger says there is + * work to do, so the steady-state boot takes no lock at all. */ export async function runRoutineMigrations( pool: Pool, log: (msg: string) => void = () => undefined, ): Promise { const client = await pool.connect(); + // Tracks whether THIS session provably holds the advisory lock. It is the + // only input to `client.release()` below: a connection that cannot prove it + // released the lock is destroyed rather than pooled, because ending the + // session is the only other way a session-scoped lock goes away. + let lockHeld = false; try { - await client.query(` - CREATE TABLE IF NOT EXISTS _routine_migrations ( - id TEXT PRIMARY KEY, - applied_at TIMESTAMPTZ NOT NULL DEFAULT now() + await ensureLedger(client); + + // Ledger first, lock second. The overwhelmingly common boot has nothing + // pending, and that boot must not pay for — or queue behind — a lock. + let pending = await pendingMigrations(client); + if (pending.length === 0) return; + + const deadline = Date.now() + ROUTINE_MIGRATION_LOCK_WAIT_MS; + for (;;) { + lockHeld = await tryAcquireMigrationLock(client); + if (lockHeld) break; + const remaining = deadline - Date.now(); + if (remaining <= 0) break; + await sleep(Math.min(LOCK_POLL_MS, remaining)); + } + + if (!lockHeld) { + // Never a silent skip: re-read the ledger. If the holder finished while + // we waited, this replica's schema IS current and the boot continues. + pending = await pendingMigrations(client); + if (pending.length === 0) { + log('[routines] migrations applied by another replica while waiting'); + return; + } + // Otherwise the work is genuinely still owed. Failing loudly is the only + // honest option; the message says "timed out" deliberately, so + // `bootstrap.retryErroredPlugins` classifies it as transient and + // re-attempts activation instead of latching the plugin `errored`. + throw new Error( + `[routines] timed out after ${String(ROUTINE_MIGRATION_LOCK_WAIT_MS)}ms waiting for the ${LOCK_KEY} advisory lock; ` + + `${String(pending.length)} migration(s) still pending (${pending.join(', ')}) — another replica is mid-migration, retry the boot`, ); - `); - - const applied = new Set( - ( - await client.query<{ id: string }>( - 'SELECT id FROM _routine_migrations', - ) - ).rows.map((r) => r.id), - ); + } - const files = (await readdir(MIGRATIONS_DIR)) - .filter((f) => f.endsWith('.sql')) - .sort(); + // Re-read UNDER the lock. The pre-lock read is a fast path, not a + // decision: the replica we queued behind may have applied part or all of + // that list before it released. + pending = await pendingMigrations(client); - for (const file of files) { - if (applied.has(file)) continue; + for (const file of pending) { const sql = await readFile(join(MIGRATIONS_DIR, file), 'utf8'); log(`[routines] applying migration ${file}`); await client.query('BEGIN'); @@ -58,7 +118,102 @@ export async function runRoutineMigrations( throw err; } } + + // Unlock on the success path only, and inside `try` — never in `finally`. + // In `finally` it would run on a possibly half-open connection whose pool + // sets no `statement_timeout`, so it could hang the release indefinitely, + // and an unlock that throws there would replace the original migration + // error. On the failure path `lockHeld` stays true and the connection is + // destroyed instead, which releases the lock with the session. + if (await releaseMigrationLock(client)) lockHeld = false; } finally { - client.release(); + client.release(lockHeld); + } +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +/** + * `CREATE TABLE IF NOT EXISTS` is not atomic against a concurrent `CREATE + * TABLE` of the same name: the existence check and the catalog insert are + * separate steps, so two replicas booting together can both pass the check and + * the loser fails with 42P07 (duplicate_table) or 23505 (a unique violation on + * a system catalog index). The table exists either way, so one retry settles + * it — the second attempt takes the IF NOT EXISTS short-circuit. This runs + * outside any transaction, so the failed statement leaves nothing to roll back. + */ +async function ensureLedger(client: PoolClient): Promise { + try { + await client.query(LEDGER_DDL); + } catch (err) { + if (!isDuplicateObjectError(err)) throw err; + await client.query(LEDGER_DDL); + } +} + +function isDuplicateObjectError(err: unknown): boolean { + const code = (err as { code?: unknown } | null)?.code; + return code === '42P07' || code === '23505'; +} + +/** + * The ledger read, expressed as the list of files still owed. Called twice on + * the locking path — once before the lock and once after acquiring it. + */ +async function pendingMigrations(client: PoolClient): Promise { + const applied = new Set( + ( + await client.query<{ id: string }>( + 'SELECT id FROM _routine_migrations', + ) + ).rows.map((r) => r.id), + ); + + const files = (await readdir(MIGRATIONS_DIR)) + .filter((f) => f.endsWith('.sql')) + .sort(); + + return files.filter((f) => !applied.has(f)); +} + +/** + * Take the migration lock without ever blocking the backend, and REPORT + * whether it was taken. The boolean is the whole point: a caller that cannot + * distinguish "acquired" from "someone else holds it" cannot release anything + * either. + */ +async function tryAcquireMigrationLock(client: PoolClient): Promise { + const result = await client.query<{ locked: boolean }>( + 'SELECT pg_try_advisory_lock($1::int, hashtext($2)::int) AS locked', + [LOCK_NS_MIGRATIONS, LOCK_KEY], + ); + // A fake/limited driver that does not model advisory locks returns no row; + // treat that as acquired so unit tests still exercise the migrations. Mirrors + // `tryAcquireRegistryLock` in @omadia/knowledge-graph-neon. + const row = result.rows[0]; + return row === undefined || row.locked !== false; +} + +/** + * Release the session lock, and REPORT whether it actually went. `false` is + * what makes the caller destroy the connection instead of pooling it — the + * only other way a session-scoped lock is released. Swallowing the answer + * hands a connection that may still hold the lock back to the pool, where it + * blocks every later replica's migration for the connection's lifetime. + */ +async function releaseMigrationLock(client: PoolClient): Promise { + try { + const result = await client.query<{ unlocked: boolean }>( + 'SELECT pg_advisory_unlock($1::int, hashtext($2)::int) AS unlocked', + [LOCK_NS_MIGRATIONS, LOCK_KEY], + ); + // "No row" mirrors the acquire side: a driver that does not model advisory + // locks never took one, so nothing is leaked by pooling the connection. + const row = result.rows[0]; + return row === undefined || row.unlocked !== false; + } catch { + return false; } } diff --git a/middleware/src/profileSnapshots/migrator.ts b/middleware/src/profileSnapshots/migrator.ts index 9946f6ddf..993b23bd9 100644 --- a/middleware/src/profileSnapshots/migrator.ts +++ b/middleware/src/profileSnapshots/migrator.ts @@ -1,13 +1,40 @@ import { readdir, readFile } from 'node:fs/promises'; import { dirname, join } from 'node:path'; import { fileURLToPath } from 'node:url'; -import type { Pool } from 'pg'; +import type { Pool, PoolClient } from 'pg'; const MIGRATIONS_DIR = join( dirname(fileURLToPath(import.meta.url)), 'migrations', ); +const LEDGER_DDL = ` + CREATE TABLE IF NOT EXISTS _profile_snapshot_migrations ( + id TEXT PRIMARY KEY, + applied_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + `; + +/** + * Advisory-lock coordinates. The namespace is shared by every SQL migrator + * (4400/4401 belong to the knowledge-graph embedding registry and the + * stale-vector clear); the second key is `hashtext()`, so each + * subsystem serialises against its own replicas only and never against a + * different subsystem's migrations. + */ +const LOCK_NS_MIGRATIONS = 4_410; +const LOCK_KEY = '_profile_snapshot_migrations'; + +/** + * How long a replica waits for the migration lock before giving up. Held at + * the same 2s ceiling as the migrators that run inside a plugin `activate()` + * (which `ToolPluginRuntime` hard-caps at 10s) so the bound is one number + * across all eight rather than eight numbers to reason about. Exported so the + * budget can be asserted in a test instead of trusted in a comment. + */ +export const PROFILE_SNAPSHOT_MIGRATION_LOCK_WAIT_MS = 2_000; +const LOCK_POLL_MS = 100; + /** * Apply pending profile-snapshot SQL migrations against the shared * Postgres pool. Tracking happens in `_profile_snapshot_migrations` so @@ -15,6 +42,15 @@ const MIGRATIONS_DIR = join( * `_profile_storage_migrations` / `_graph_migrations` — snapshots evolve * with the bundle/lifecycle stack, not with auth or graph. * + * Concurrency: read-ledger → filter → apply is not safe on its own. Two + * replicas booting together both see the same pending list and both execute + * it; `IF NOT EXISTS` hides that, `ADD CONSTRAINT` does not (42710 → the + * loser's boot fails). So the apply loop runs under a session-scoped advisory + * lock, taken with `pg_try_advisory_lock` (never the blocking variant — an + * unbounded server-side wait inside a 10s `activate()` would turn a rare race + * into a deterministic boot failure) and only after the ledger says there is + * work to do, so the steady-state boot takes no lock at all. + * * Mirrors `runAuthMigrations` and `runProfileStorageMigrations` line for * line so the migrators stay diff-comparable. */ @@ -23,28 +59,54 @@ export async function runProfileSnapshotMigrations( log: (msg: string) => void = () => undefined, ): Promise { const client = await pool.connect(); + // Tracks whether THIS session provably holds the advisory lock. It is the + // only input to `client.release()` below: a connection that cannot prove it + // released the lock is destroyed rather than pooled, because ending the + // session is the only other way a session-scoped lock goes away. + let lockHeld = false; try { - await client.query(` - CREATE TABLE IF NOT EXISTS _profile_snapshot_migrations ( - id TEXT PRIMARY KEY, - applied_at TIMESTAMPTZ NOT NULL DEFAULT now() + await ensureLedger(client); + + // Ledger first, lock second. The overwhelmingly common boot has nothing + // pending, and that boot must not pay for — or queue behind — a lock. + let pending = await pendingMigrations(client); + if (pending.length === 0) return; + + const deadline = Date.now() + PROFILE_SNAPSHOT_MIGRATION_LOCK_WAIT_MS; + for (;;) { + lockHeld = await tryAcquireMigrationLock(client); + if (lockHeld) break; + const remaining = deadline - Date.now(); + if (remaining <= 0) break; + await sleep(Math.min(LOCK_POLL_MS, remaining)); + } + + if (!lockHeld) { + // Never a silent skip: re-read the ledger. If the holder finished while + // we waited, this replica's schema IS current and the boot continues. + pending = await pendingMigrations(client); + if (pending.length === 0) { + log( + '[profile-snapshot] migrations applied by another replica while waiting', + ); + return; + } + // Otherwise the work is genuinely still owed. Failing loudly is the only + // honest option; the message says "timed out" deliberately, so + // `bootstrap.retryErroredPlugins` classifies it as transient and + // re-attempts activation instead of latching the plugin `errored`. + throw new Error( + `[profile-snapshot] timed out after ${String(PROFILE_SNAPSHOT_MIGRATION_LOCK_WAIT_MS)}ms waiting for the ${LOCK_KEY} advisory lock; ` + + `${String(pending.length)} migration(s) still pending (${pending.join(', ')}) — another replica is mid-migration, retry the boot`, ); - `); - - const applied = new Set( - ( - await client.query<{ id: string }>( - 'SELECT id FROM _profile_snapshot_migrations', - ) - ).rows.map((r) => r.id), - ); + } - const files = (await readdir(MIGRATIONS_DIR)) - .filter((f) => f.endsWith('.sql')) - .sort(); + // Re-read UNDER the lock. The pre-lock read is a fast path, not a + // decision: the replica we queued behind may have applied part or all of + // that list before it released. + pending = await pendingMigrations(client); - for (const file of files) { - if (applied.has(file)) continue; + for (const file of pending) { const sql = await readFile(join(MIGRATIONS_DIR, file), 'utf8'); log(`[profile-snapshot] applying migration ${file}`); await client.query('BEGIN'); @@ -60,7 +122,102 @@ export async function runProfileSnapshotMigrations( throw err; } } + + // Unlock on the success path only, and inside `try` — never in `finally`. + // In `finally` it would run on a possibly half-open connection whose pool + // sets no `statement_timeout`, so it could hang the release indefinitely, + // and an unlock that throws there would replace the original migration + // error. On the failure path `lockHeld` stays true and the connection is + // destroyed instead, which releases the lock with the session. + if (await releaseMigrationLock(client)) lockHeld = false; } finally { - client.release(); + client.release(lockHeld); + } +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +/** + * `CREATE TABLE IF NOT EXISTS` is not atomic against a concurrent `CREATE + * TABLE` of the same name: the existence check and the catalog insert are + * separate steps, so two replicas booting together can both pass the check and + * the loser fails with 42P07 (duplicate_table) or 23505 (a unique violation on + * a system catalog index). The table exists either way, so one retry settles + * it — the second attempt takes the IF NOT EXISTS short-circuit. This runs + * outside any transaction, so the failed statement leaves nothing to roll back. + */ +async function ensureLedger(client: PoolClient): Promise { + try { + await client.query(LEDGER_DDL); + } catch (err) { + if (!isDuplicateObjectError(err)) throw err; + await client.query(LEDGER_DDL); + } +} + +function isDuplicateObjectError(err: unknown): boolean { + const code = (err as { code?: unknown } | null)?.code; + return code === '42P07' || code === '23505'; +} + +/** + * The ledger read, expressed as the list of files still owed. Called twice on + * the locking path — once before the lock and once after acquiring it. + */ +async function pendingMigrations(client: PoolClient): Promise { + const applied = new Set( + ( + await client.query<{ id: string }>( + 'SELECT id FROM _profile_snapshot_migrations', + ) + ).rows.map((r) => r.id), + ); + + const files = (await readdir(MIGRATIONS_DIR)) + .filter((f) => f.endsWith('.sql')) + .sort(); + + return files.filter((f) => !applied.has(f)); +} + +/** + * Take the migration lock without ever blocking the backend, and REPORT + * whether it was taken. The boolean is the whole point: a caller that cannot + * distinguish "acquired" from "someone else holds it" cannot release anything + * either. + */ +async function tryAcquireMigrationLock(client: PoolClient): Promise { + const result = await client.query<{ locked: boolean }>( + 'SELECT pg_try_advisory_lock($1::int, hashtext($2)::int) AS locked', + [LOCK_NS_MIGRATIONS, LOCK_KEY], + ); + // A fake/limited driver that does not model advisory locks returns no row; + // treat that as acquired so unit tests still exercise the migrations. Mirrors + // `tryAcquireRegistryLock` in @omadia/knowledge-graph-neon. + const row = result.rows[0]; + return row === undefined || row.locked !== false; +} + +/** + * Release the session lock, and REPORT whether it actually went. `false` is + * what makes the caller destroy the connection instead of pooling it — the + * only other way a session-scoped lock is released. Swallowing the answer + * hands a connection that may still hold the lock back to the pool, where it + * blocks every later replica's migration for the connection's lifetime. + */ +async function releaseMigrationLock(client: PoolClient): Promise { + try { + const result = await client.query<{ unlocked: boolean }>( + 'SELECT pg_advisory_unlock($1::int, hashtext($2)::int) AS unlocked', + [LOCK_NS_MIGRATIONS, LOCK_KEY], + ); + // "No row" mirrors the acquire side: a driver that does not model advisory + // locks never took one, so nothing is leaked by pooling the connection. + const row = result.rows[0]; + return row === undefined || row.unlocked !== false; + } catch { + return false; } } diff --git a/middleware/src/profileStorage/migrator.ts b/middleware/src/profileStorage/migrator.ts index 69a218287..e169adea0 100644 --- a/middleware/src/profileStorage/migrator.ts +++ b/middleware/src/profileStorage/migrator.ts @@ -1,19 +1,55 @@ import { readdir, readFile } from 'node:fs/promises'; import { dirname, join } from 'node:path'; import { fileURLToPath } from 'node:url'; -import type { Pool } from 'pg'; +import type { Pool, PoolClient } from 'pg'; const MIGRATIONS_DIR = join( dirname(fileURLToPath(import.meta.url)), 'migrations', ); +const LEDGER_DDL = ` + CREATE TABLE IF NOT EXISTS _profile_storage_migrations ( + id TEXT PRIMARY KEY, + applied_at TIMESTAMPTZ NOT NULL DEFAULT now() + ); + `; + +/** + * Advisory-lock coordinates. The namespace is shared by every SQL migrator + * (4400/4401 belong to the knowledge-graph embedding registry and the + * stale-vector clear); the second key is `hashtext()`, so each + * subsystem serialises against its own replicas only and never against a + * different subsystem's migrations. + */ +const LOCK_NS_MIGRATIONS = 4_410; +const LOCK_KEY = '_profile_storage_migrations'; + +/** + * How long a replica waits for the migration lock before giving up. Held at + * the same 2s ceiling as the migrators that run inside a plugin `activate()` + * (which `ToolPluginRuntime` hard-caps at 10s) so the bound is one number + * across all eight rather than eight numbers to reason about. Exported so the + * budget can be asserted in a test instead of trusted in a comment. + */ +export const PROFILE_STORAGE_MIGRATION_LOCK_WAIT_MS = 2_000; +const LOCK_POLL_MS = 100; + /** * Apply pending profile-storage SQL migrations against the shared Postgres * pool. Tracking happens in `_profile_storage_migrations` so the lifecycle * is independent of `_auth_migrations` / `_graph_migrations` — profile- * storage evolves with the bundle/snapshot stack, not with auth or graph. * + * Concurrency: read-ledger → filter → apply is not safe on its own. Two + * replicas booting together both see the same pending list and both execute + * it; `IF NOT EXISTS` hides that, `ADD CONSTRAINT` does not (42710 → the + * loser's boot fails). So the apply loop runs under a session-scoped advisory + * lock, taken with `pg_try_advisory_lock` (never the blocking variant — an + * unbounded server-side wait inside a 10s `activate()` would turn a rare race + * into a deterministic boot failure) and only after the ledger says there is + * work to do, so the steady-state boot takes no lock at all. + * * Mirrors `runAuthMigrations` line for line so the migrators stay diff- * comparable. */ @@ -22,28 +58,54 @@ export async function runProfileStorageMigrations( log: (msg: string) => void = () => undefined, ): Promise { const client = await pool.connect(); + // Tracks whether THIS session provably holds the advisory lock. It is the + // only input to `client.release()` below: a connection that cannot prove it + // released the lock is destroyed rather than pooled, because ending the + // session is the only other way a session-scoped lock goes away. + let lockHeld = false; try { - await client.query(` - CREATE TABLE IF NOT EXISTS _profile_storage_migrations ( - id TEXT PRIMARY KEY, - applied_at TIMESTAMPTZ NOT NULL DEFAULT now() + await ensureLedger(client); + + // Ledger first, lock second. The overwhelmingly common boot has nothing + // pending, and that boot must not pay for — or queue behind — a lock. + let pending = await pendingMigrations(client); + if (pending.length === 0) return; + + const deadline = Date.now() + PROFILE_STORAGE_MIGRATION_LOCK_WAIT_MS; + for (;;) { + lockHeld = await tryAcquireMigrationLock(client); + if (lockHeld) break; + const remaining = deadline - Date.now(); + if (remaining <= 0) break; + await sleep(Math.min(LOCK_POLL_MS, remaining)); + } + + if (!lockHeld) { + // Never a silent skip: re-read the ledger. If the holder finished while + // we waited, this replica's schema IS current and the boot continues. + pending = await pendingMigrations(client); + if (pending.length === 0) { + log( + '[profile-storage] migrations applied by another replica while waiting', + ); + return; + } + // Otherwise the work is genuinely still owed. Failing loudly is the only + // honest option; the message says "timed out" deliberately, so + // `bootstrap.retryErroredPlugins` classifies it as transient and + // re-attempts activation instead of latching the plugin `errored`. + throw new Error( + `[profile-storage] timed out after ${String(PROFILE_STORAGE_MIGRATION_LOCK_WAIT_MS)}ms waiting for the ${LOCK_KEY} advisory lock; ` + + `${String(pending.length)} migration(s) still pending (${pending.join(', ')}) — another replica is mid-migration, retry the boot`, ); - `); - - const applied = new Set( - ( - await client.query<{ id: string }>( - 'SELECT id FROM _profile_storage_migrations', - ) - ).rows.map((r) => r.id), - ); + } - const files = (await readdir(MIGRATIONS_DIR)) - .filter((f) => f.endsWith('.sql')) - .sort(); + // Re-read UNDER the lock. The pre-lock read is a fast path, not a + // decision: the replica we queued behind may have applied part or all of + // that list before it released. + pending = await pendingMigrations(client); - for (const file of files) { - if (applied.has(file)) continue; + for (const file of pending) { const sql = await readFile(join(MIGRATIONS_DIR, file), 'utf8'); log(`[profile-storage] applying migration ${file}`); await client.query('BEGIN'); @@ -59,7 +121,102 @@ export async function runProfileStorageMigrations( throw err; } } + + // Unlock on the success path only, and inside `try` — never in `finally`. + // In `finally` it would run on a possibly half-open connection whose pool + // sets no `statement_timeout`, so it could hang the release indefinitely, + // and an unlock that throws there would replace the original migration + // error. On the failure path `lockHeld` stays true and the connection is + // destroyed instead, which releases the lock with the session. + if (await releaseMigrationLock(client)) lockHeld = false; } finally { - client.release(); + client.release(lockHeld); + } +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +/** + * `CREATE TABLE IF NOT EXISTS` is not atomic against a concurrent `CREATE + * TABLE` of the same name: the existence check and the catalog insert are + * separate steps, so two replicas booting together can both pass the check and + * the loser fails with 42P07 (duplicate_table) or 23505 (a unique violation on + * a system catalog index). The table exists either way, so one retry settles + * it — the second attempt takes the IF NOT EXISTS short-circuit. This runs + * outside any transaction, so the failed statement leaves nothing to roll back. + */ +async function ensureLedger(client: PoolClient): Promise { + try { + await client.query(LEDGER_DDL); + } catch (err) { + if (!isDuplicateObjectError(err)) throw err; + await client.query(LEDGER_DDL); + } +} + +function isDuplicateObjectError(err: unknown): boolean { + const code = (err as { code?: unknown } | null)?.code; + return code === '42P07' || code === '23505'; +} + +/** + * The ledger read, expressed as the list of files still owed. Called twice on + * the locking path — once before the lock and once after acquiring it. + */ +async function pendingMigrations(client: PoolClient): Promise { + const applied = new Set( + ( + await client.query<{ id: string }>( + 'SELECT id FROM _profile_storage_migrations', + ) + ).rows.map((r) => r.id), + ); + + const files = (await readdir(MIGRATIONS_DIR)) + .filter((f) => f.endsWith('.sql')) + .sort(); + + return files.filter((f) => !applied.has(f)); +} + +/** + * Take the migration lock without ever blocking the backend, and REPORT + * whether it was taken. The boolean is the whole point: a caller that cannot + * distinguish "acquired" from "someone else holds it" cannot release anything + * either. + */ +async function tryAcquireMigrationLock(client: PoolClient): Promise { + const result = await client.query<{ locked: boolean }>( + 'SELECT pg_try_advisory_lock($1::int, hashtext($2)::int) AS locked', + [LOCK_NS_MIGRATIONS, LOCK_KEY], + ); + // A fake/limited driver that does not model advisory locks returns no row; + // treat that as acquired so unit tests still exercise the migrations. Mirrors + // `tryAcquireRegistryLock` in @omadia/knowledge-graph-neon. + const row = result.rows[0]; + return row === undefined || row.locked !== false; +} + +/** + * Release the session lock, and REPORT whether it actually went. `false` is + * what makes the caller destroy the connection instead of pooling it — the + * only other way a session-scoped lock is released. Swallowing the answer + * hands a connection that may still hold the lock back to the pool, where it + * blocks every later replica's migration for the connection's lifetime. + */ +async function releaseMigrationLock(client: PoolClient): Promise { + try { + const result = await client.query<{ unlocked: boolean }>( + 'SELECT pg_advisory_unlock($1::int, hashtext($2)::int) AS unlocked', + [LOCK_NS_MIGRATIONS, LOCK_KEY], + ); + // "No row" mirrors the acquire side: a driver that does not model advisory + // locks never took one, so nothing is leaked by pooling the connection. + const row = result.rows[0]; + return row === undefined || row.unlocked !== false; + } catch { + return false; } } diff --git a/middleware/test/migratorConcurrentBoot.test.ts b/middleware/test/migratorConcurrentBoot.test.ts new file mode 100644 index 000000000..7960c02de --- /dev/null +++ b/middleware/test/migratorConcurrentBoot.test.ts @@ -0,0 +1,483 @@ +import { strict as assert } from 'node:assert'; +import { readdir } from 'node:fs/promises'; +import { dirname, join, resolve } from 'node:path'; +import { describe, it } from 'node:test'; +import { fileURLToPath } from 'node:url'; + +import type { Pool, PoolClient, QueryResult } from 'pg'; + +import { + GRAPH_MIGRATION_LOCK_WAIT_MS, + runGraphMigrations, +} from '../packages/harness-knowledge-graph-neon/src/migrator.js'; +import { + MEMORY_MIGRATION_LOCK_WAIT_MS, + runMemoryMigrations, +} from '../packages/harness-memory-postgres/src/migrator.js'; +import { + MULTI_ORCH_MIGRATION_LOCK_WAIT_MS, + runMultiOrchestratorMigrations, +} from '../packages/harness-orchestrator/src/registry/migrator.js'; +import { + AUTH_MIGRATION_LOCK_WAIT_MS, + runAuthMigrations, +} from '../src/auth/migrator.js'; +import { + CONDUCTOR_MIGRATION_LOCK_WAIT_MS, + runConductorMigrations, +} from '../src/conductor/migrator.js'; +import { + ROUTINE_MIGRATION_LOCK_WAIT_MS, + runRoutineMigrations, +} from '../src/plugins/routines/migrator.js'; +import { + PROFILE_SNAPSHOT_MIGRATION_LOCK_WAIT_MS, + runProfileSnapshotMigrations, +} from '../src/profileSnapshots/migrator.js'; +import { + PROFILE_STORAGE_MIGRATION_LOCK_WAIT_MS, + runProfileStorageMigrations, +} from '../src/profileStorage/migrator.js'; + +/** + * All eight SQL migrators are `read ledger → filter → apply` with no mutual + * exclusion. Two replicas booting together both read the same pending list and + * both execute it. `CREATE … IF NOT EXISTS` masks that; `ALTER TABLE … ADD + * CONSTRAINT` does not — the loser gets 42710 and its boot fails. Three of the + * eight run inside a plugin `activate()` that `ToolPluginRuntime` hard-caps at + * 10s, so the fix cannot be an unbounded `pg_advisory_lock`. + * + * These tests pin the mechanism, not Postgres: a real server proves a lock is + * free at the end, but it cannot force the cases that matter — an unlock that + * FAILS, a lock that is never granted, a migration that throws while the lock + * is held. A fake driver is the only way to reach those, and the observable + * behaviour they must produce is exactly three things: which SQL was issued, + * in which order, and whether the connection was POOLED (`release(false)`) or + * DESTROYED (`release(true)` — the only other way a session-scoped advisory + * lock is released). + */ + +const TEST_DIR = dirname(fileURLToPath(import.meta.url)); +const MIDDLEWARE_DIR = resolve(TEST_DIR, '..'); + +/** + * `waitForPostgres`' default budget (see `neonKnowledgeGraph.ts`), which the + * knowledge-graph plugin burns inside `activate()` BEFORE calling its + * migrator. The lock budget has to fit in what is left of the 10s cap. + */ +const WAIT_FOR_POSTGRES_BUDGET_MS = 6_000; +const ACTIVATE_TIMEOUT_MS = 10_000; + +interface MigratorCase { + readonly name: string; + readonly ledger: string; + readonly migrationsDir: string; + readonly budgetMs: number; + run(pool: Pool, log?: (msg: string) => void): Promise; +} + +const MIGRATORS: readonly MigratorCase[] = [ + { + name: 'auth', + ledger: '_auth_migrations', + migrationsDir: join(MIDDLEWARE_DIR, 'src', 'auth', 'migrations'), + budgetMs: AUTH_MIGRATION_LOCK_WAIT_MS, + run: (pool, log) => runAuthMigrations(pool, log), + }, + { + name: 'conductor', + ledger: '_conductor_migrations', + migrationsDir: join(MIDDLEWARE_DIR, 'src', 'conductor', 'migrations'), + budgetMs: CONDUCTOR_MIGRATION_LOCK_WAIT_MS, + run: (pool, log) => runConductorMigrations(pool, log), + }, + { + name: 'routines', + ledger: '_routine_migrations', + migrationsDir: join(MIDDLEWARE_DIR, 'src', 'plugins', 'routines', 'migrations'), + budgetMs: ROUTINE_MIGRATION_LOCK_WAIT_MS, + run: (pool, log) => runRoutineMigrations(pool, log), + }, + { + name: 'profile-snapshots', + ledger: '_profile_snapshot_migrations', + migrationsDir: join(MIDDLEWARE_DIR, 'src', 'profileSnapshots', 'migrations'), + budgetMs: PROFILE_SNAPSHOT_MIGRATION_LOCK_WAIT_MS, + run: (pool, log) => runProfileSnapshotMigrations(pool, log), + }, + { + name: 'profile-storage', + ledger: '_profile_storage_migrations', + migrationsDir: join(MIDDLEWARE_DIR, 'src', 'profileStorage', 'migrations'), + budgetMs: PROFILE_STORAGE_MIGRATION_LOCK_WAIT_MS, + run: (pool, log) => runProfileStorageMigrations(pool, log), + }, + { + name: 'knowledge-graph', + ledger: '_graph_migrations', + migrationsDir: join( + MIDDLEWARE_DIR, + 'packages', + 'harness-knowledge-graph-neon', + 'src', + 'migrations', + ), + budgetMs: GRAPH_MIGRATION_LOCK_WAIT_MS, + run: (pool, log) => runGraphMigrations(pool, log), + }, + { + name: 'memory-postgres', + ledger: '_memory_migrations', + migrationsDir: join( + MIDDLEWARE_DIR, + 'packages', + 'harness-memory-postgres', + 'src', + 'migrations', + ), + budgetMs: MEMORY_MIGRATION_LOCK_WAIT_MS, + run: (pool, log) => runMemoryMigrations(pool, log), + }, + { + name: 'multi-orchestrator', + ledger: '_multi_orchestrator_migrations', + migrationsDir: join(MIDDLEWARE_DIR, 'migrations'), + budgetMs: MULTI_ORCH_MIGRATION_LOCK_WAIT_MS, + run: (pool, log) => + runMultiOrchestratorMigrations(pool, log, join(MIDDLEWARE_DIR, 'migrations')), + }, +]; + +async function sqlFiles(dir: string): Promise { + return (await readdir(dir)).filter((f) => f.endsWith('.sql')).sort(); +} + +interface FakeScript { + /** Successive answers to the ledger SELECT; the last one repeats. */ + readonly ledgerReads?: readonly (readonly string[])[]; + /** Successive answers to `pg_try_advisory_lock`; the last one repeats. + * `null` models a driver that answers with no row at all. */ + readonly lockAnswers?: readonly (boolean | null)[]; + /** `pg_advisory_unlock` throws — what a connection stuck in an aborted + * transaction actually does. */ + readonly unlockThrows?: boolean; + /** `pg_advisory_unlock`'s answer; `null` models a driver with no row. */ + readonly unlockAnswer?: boolean | null; + /** Thrown by the FIRST migration body that executes. */ + readonly migrationThrows?: Error; + /** Errors thrown by successive `CREATE TABLE IF NOT EXISTS` attempts. */ + readonly ledgerDdlErrors?: readonly unknown[]; +} + +interface Fake { + readonly pool: Pool; + /** `true` = the connection was DESTROYED, `false` = returned to the pool. */ + readonly releases: boolean[]; + readonly sql: string[]; +} + +function makeFake(script: FakeScript = {}): Fake { + const releases: boolean[] = []; + const sql: string[] = []; + let ledgerReadCount = 0; + let lockCount = 0; + let ddlCount = 0; + let migrationCount = 0; + + const rows = (r: ReadonlyArray>): QueryResult => + ({ + command: '', + rowCount: r.length, + oid: 0, + rows: [...r], + fields: [], + }) as unknown as QueryResult; + + function pick(list: readonly T[] | undefined, index: number, fallback: T): T { + if (list === undefined || list.length === 0) return fallback; + return list[Math.min(index, list.length - 1)] as T; + } + + const query = async (text: string): Promise => { + sql.push(text); + + if (/CREATE TABLE IF NOT EXISTS _[a-z_]+_migrations/i.test(text)) { + const err = script.ledgerDdlErrors?.[ddlCount]; + ddlCount += 1; + if (err !== undefined) throw err; + return rows([]); + } + + if (/pg_try_advisory_lock/.test(text)) { + const answer = pick(script.lockAnswers, lockCount, true); + lockCount += 1; + return answer === null ? rows([]) : rows([{ locked: answer }]); + } + + if (/pg_advisory_unlock/.test(text)) { + if (script.unlockThrows === true) { + throw new Error('current transaction is aborted, commands ignored'); + } + const answer = script.unlockAnswer === undefined ? true : script.unlockAnswer; + return answer === null ? rows([]) : rows([{ unlocked: answer }]); + } + + if (/^\s*SELECT id FROM _[a-z_]+_migrations/i.test(text)) { + const applied = pick(script.ledgerReads, ledgerReadCount, []); + ledgerReadCount += 1; + return rows(applied.map((id) => ({ id }))); + } + + if (/^\s*(BEGIN|COMMIT|ROLLBACK)\s*$/i.test(text)) return rows([]); + if (/^\s*INSERT INTO _[a-z_]+_migrations/i.test(text)) return rows([]); + + // Everything else is a migration body. + migrationCount += 1; + if (script.migrationThrows !== undefined && migrationCount === 1) { + throw script.migrationThrows; + } + return rows([]); + }; + + const pool = { + async connect(): Promise { + return { + query, + release(destroy?: boolean): void { + releases.push(destroy === true); + }, + } as unknown as PoolClient; + }, + } as unknown as Pool; + + return { pool, releases, sql }; +} + +const count = (sql: readonly string[], re: RegExp): number => + sql.filter((s) => re.test(s)).length; + +const LOCK_RE = /pg_try_advisory_lock/; +const UNLOCK_RE = /pg_advisory_unlock/; +const INSERT_RE = /^\s*INSERT INTO _[a-z_]+_migrations/i; +const BEGIN_RE = /^\s*BEGIN\s*$/i; + +describe('SQL migrators — concurrent multi-replica boot', () => { + it('never touches the lock when the ledger says there is nothing to apply', async () => { + for (const m of MIGRATORS) { + const files = await sqlFiles(m.migrationsDir); + assert.ok(files.length > 0, `${m.name}: fixture check — expected .sql files`); + + const fake = makeFake({ ledgerReads: [files] }); + await m.run(fake.pool); + + // The steady-state boot is the common case by orders of magnitude. It + // must not pay a round-trip for the lock, and — more importantly — must + // not be able to queue behind a replica that IS migrating. + assert.equal(count(fake.sql, LOCK_RE), 0, `${m.name}: locked with no work to do`); + assert.equal(count(fake.sql, INSERT_RE), 0, `${m.name}: re-applied a migration`); + assert.deepEqual(fake.releases, [false], `${m.name}: connection must be pooled`); + } + }); + + it('takes the lock BEFORE the first migration and applies every pending file', async () => { + for (const m of MIGRATORS) { + const files = await sqlFiles(m.migrationsDir); + const fake = makeFake({ ledgerReads: [[]] }); + await m.run(fake.pool); + + const lockAt = fake.sql.findIndex((s) => LOCK_RE.test(s)); + const beginAt = fake.sql.findIndex((s) => BEGIN_RE.test(s)); + assert.ok(lockAt >= 0, `${m.name}: no lock was taken`); + assert.ok(beginAt > lockAt, `${m.name}: migration started before the lock`); + assert.equal(count(fake.sql, INSERT_RE), files.length, `${m.name}: ledger writes`); + assert.equal(count(fake.sql, UNLOCK_RE), 1, `${m.name}: released exactly once`); + assert.deepEqual(fake.releases, [false], `${m.name}: healthy connection is pooled`); + } + }); + + it('re-reads the ledger UNDER the lock, so the winner is never re-run', async () => { + for (const m of MIGRATORS) { + const files = await sqlFiles(m.migrationsDir); + // Pending before the lock, complete once we hold it: exactly what a + // replica sees when it queues behind a winner that then finishes. + const fake = makeFake({ ledgerReads: [[], files] }); + await m.run(fake.pool); + + assert.equal(count(fake.sql, LOCK_RE), 1, `${m.name}: expected one lock attempt`); + assert.equal( + count(fake.sql, INSERT_RE), + 0, + `${m.name}: applied migrations the winner had already applied`, + ); + assert.equal(count(fake.sql, UNLOCK_RE), 1, `${m.name}: released the lock`); + assert.deepEqual(fake.releases, [false], `${m.name}: connection pooled`); + } + }); + + it('gives a loser a CLEAR RETRYABLE ERROR — never a silent skip', async () => { + // Runs the eight in parallel: each burns its full budget by design, and + // 8 x 2s sequentially would dominate the suite. + await Promise.all( + MIGRATORS.map(async (m) => { + const fake = makeFake({ ledgerReads: [[]], lockAnswers: [false] }); + const started = Date.now(); + + await assert.rejects(m.run(fake.pool), (err: unknown) => { + assert.ok(err instanceof Error); + // "timed out" is load-bearing: `bootstrap.retryErroredPlugins` + // matches it as a transient activation error and re-attempts the + // plugin instead of latching it `errored` forever. + assert.match(err.message, /timed out/); + assert.match(err.message, /still pending/); + assert.match(err.message, new RegExp(m.ledger)); + return true; + }); + + const elapsed = Date.now() - started; + assert.ok( + elapsed >= m.budgetMs - 150, + `${m.name}: gave up after ${String(elapsed)}ms, before its ${String(m.budgetMs)}ms budget`, + ); + assert.ok( + elapsed < m.budgetMs + 2_000, + `${m.name}: overran its budget (${String(elapsed)}ms)`, + ); + assert.equal(count(fake.sql, INSERT_RE), 0, `${m.name}: applied without the lock`); + assert.equal( + count(fake.sql, UNLOCK_RE), + 0, + `${m.name}: unlocked a lock it never held`, + ); + // Never acquired ⇒ nothing to leak ⇒ a healthy connection must not be + // thrown away. + assert.deepEqual(fake.releases, [false], `${m.name}: connection pooled`); + }), + ); + }); + + it('lets a loser finish cleanly when the winner applied everything while it waited', async () => { + await Promise.all( + MIGRATORS.map(async (m) => { + const files = await sqlFiles(m.migrationsDir); + const logs: string[] = []; + const fake = makeFake({ ledgerReads: [[], files], lockAnswers: [false] }); + + await m.run(fake.pool, (msg) => logs.push(msg)); + + assert.ok( + logs.some((l) => /applied by another replica/.test(l)), + `${m.name}: the outcome must be stated, not silently assumed`, + ); + assert.equal(count(fake.sql, INSERT_RE), 0, `${m.name}: applied without the lock`); + assert.deepEqual(fake.releases, [false], `${m.name}: connection pooled`); + }), + ); + }); + + it('preserves the ORIGINAL error when a migration throws, and destroys the connection', async () => { + for (const m of MIGRATORS) { + const boom = Object.assign( + new Error('constraint "x_pkey" for relation "x" already exists'), + { code: '42710' }, + ); + const fake = makeFake({ ledgerReads: [[]], migrationThrows: boom }); + + await assert.rejects(m.run(fake.pool), (err: unknown) => { + // Identity, not shape: an unlock in `finally` would replace this. + assert.equal(err, boom, `${m.name}: the original error was replaced`); + return true; + }); + + assert.ok( + fake.sql.some((s) => /^\s*ROLLBACK\s*$/i.test(s)), + `${m.name}: the failed migration was not rolled back`, + ); + assert.equal( + count(fake.sql, UNLOCK_RE), + 0, + `${m.name}: no unlock may be issued on the failure path — the pools set no statement_timeout, so it can hang the release, and a throwing unlock would mask the migration error`, + ); + assert.deepEqual( + fake.releases, + [true], + `${m.name}: destroying the connection is what releases the session lock here`, + ); + } + }); + + it('destroys the connection when the unlock THROWS on the success path', async () => { + for (const m of MIGRATORS) { + const fake = makeFake({ ledgerReads: [[]], unlockThrows: true }); + // Everything committed; only the unlock failed. The migration itself + // succeeded, so this must NOT reject. + await m.run(fake.pool); + assert.deepEqual(fake.releases, [true], `${m.name}: unlock threw, connection pooled`); + } + }); + + it('destroys the connection when the driver says the lock was NOT released', async () => { + for (const m of MIGRATORS) { + const fake = makeFake({ ledgerReads: [[]], unlockAnswer: false }); + await m.run(fake.pool); + // Reading `pg_advisory_unlock`'s boolean is the whole point: `false` + // means this session never held it / did not release it, and pooling + // the connection would strand the lock. + assert.deepEqual(fake.releases, [true], `${m.name}: ignored the unlock verdict`); + } + }); + + it('treats a driver that models no advisory locks as clean (no row ⇒ acquired ⇒ released)', async () => { + for (const m of MIGRATORS) { + const files = await sqlFiles(m.migrationsDir); + const fake = makeFake({ + ledgerReads: [[]], + lockAnswers: [null], + unlockAnswer: null, + }); + await m.run(fake.pool); + assert.equal(count(fake.sql, INSERT_RE), files.length, `${m.name}: migrations skipped`); + assert.deepEqual(fake.releases, [false], `${m.name}: nothing was held, nothing to destroy`); + } + }); + + it('survives a concurrent CREATE TABLE of the ledger (42P07 / 23505) and rethrows anything else', async () => { + for (const m of MIGRATORS) { + const files = await sqlFiles(m.migrationsDir); + + for (const code of ['42P07', '23505']) { + const dup = Object.assign(new Error(`duplicate ${code}`), { code }); + const fake = makeFake({ ledgerReads: [files], ledgerDdlErrors: [dup] }); + await m.run(fake.pool); + assert.equal( + count(fake.sql, /CREATE TABLE IF NOT EXISTS/i), + 2, + `${m.name}: ${code} must be retried exactly once`, + ); + assert.deepEqual(fake.releases, [false], `${m.name}: connection pooled`); + } + + const denied = Object.assign(new Error('permission denied for schema public'), { + code: '42501', + }); + const fake = makeFake({ ledgerReads: [files], ledgerDdlErrors: [denied] }); + await assert.rejects(m.run(fake.pool), (err: unknown) => { + assert.equal(err, denied, `${m.name}: a real DDL failure must not be swallowed`); + return true; + }); + assert.deepEqual(fake.releases, [false], `${m.name}: connection pooled`); + } + }); + + it('fits the bounded wait inside the 10s activate() cap for every migrator', async () => { + for (const m of MIGRATORS) { + // The knowledge-graph plugin spends up to WAIT_FOR_POSTGRES_BUDGET_MS in + // `waitForPostgres` before its migrator is even called, and all eight + // share one budget so this is the binding constraint for all of them. + assert.ok(m.budgetMs > 0, `${m.name}: budget must be positive`); + assert.ok( + WAIT_FOR_POSTGRES_BUDGET_MS + m.budgetMs < ACTIVATE_TIMEOUT_MS, + `${m.name}: ${String(WAIT_FOR_POSTGRES_BUDGET_MS)}ms + ${String(m.budgetMs)}ms leaves no room inside the ${String(ACTIVATE_TIMEOUT_MS)}ms activate() cap`, + ); + } + }); +}); From 4dfe8aadd0252d78d904828abf9e787e29c37c0e Mon Sep 17 00:00:00 2001 From: Marcel Wege Date: Thu, 30 Jul 2026 11:34:29 +0200 Subject: [PATCH 3/4] feat(plugins): allow .sql in packages + make the 8 migrators concurrency-safe MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two of three wave-2 packets shipped after review; the third is held back again. .SQL IN THE ZIP ALLOWLIST — unblocked by #548. It was rejected in the first batch because it would have escalated the then-live path traversal from directory delete/replace into arbitrary SQL execution. #548 closed that, and the reviewer verified independently that no route remains from uploaded content to any migrator-scanned path: all 8 migrators derive their directory from import.meta.url or an operator env var, never from package content, and the 3 extraction destinations are staging/preview paths with generated names. The stronger argument, which the analysis had buried: .js/.mjs/.cjs are already allowlisted and dynamic-import()ed by the runtime, so anyone who can get a zip to ingest() already holds in-process code execution. .sql is strictly weaker than what the trust boundary already grants — which makes it security-inert independently of the migrator analysis. Also adds `migrations` to both boilerplate build-zip.mjs INCLUDE lists. Without it the directory is silently dropped at packaging time: the copy loop skips anything not on the list, so an author following the boilerplate gets a green install and no schema. The allowlist entry alone was necessary but not sufficient. MIGRATOR CONCURRENCY — shipped on the third attempt. Every migrator was read-ledger -> filter -> apply with no mutual exclusion, so two replicas booting together both execute; IF NOT EXISTS masks it, ADD CONSTRAINT does not (42710). Two earlier designs were rejected, and both rejections shaped this one: an unbounded pg_advisory_lock is unusable because three of the eight migrators run inside a plugin activate() that ToolPluginRuntime caps at 10s — turning a rare race into a deterministic boot failure; and a retry that called pg_advisory_unlock without reading its boolean return could not distinguish "released" from "never held", making the mechanism silently inert. STILL HELD BACK: the DynamicAgentRuntime rollback, now on its second rejection. It does not cover the timeout path — withTimeout is a bare Promise.race and does not cancel, so after the rollback runs and clears the in-flight marker, the orphaned activate() continues and re-registers. That is the same defect class already documented for ToolPluginRuntime, and it wants a cancellation token rather than another rollback tweak. 5,224 pass, typecheck clean, ratchet held at 3,303. --- .../agent-integration/scripts/build-zip.mjs | 4 +++ .../agent-pure-llm/scripts/build-zip.mjs | 4 +++ specs/470-dev-platform-plugin/README.md | 27 ++++++++++--------- 3 files changed, 22 insertions(+), 13 deletions(-) diff --git a/middleware/assets/boilerplate/agent-integration/scripts/build-zip.mjs b/middleware/assets/boilerplate/agent-integration/scripts/build-zip.mjs index e4c380d2f..b5eb16aac 100644 --- a/middleware/assets/boilerplate/agent-integration/scripts/build-zip.mjs +++ b/middleware/assets/boilerplate/agent-integration/scripts/build-zip.mjs @@ -68,6 +68,10 @@ const INCLUDE = [ 'manifest.yaml', 'package.json', 'dist', + // Plugin-owned SQL. The host allows .sql in a package (zipExtractor.ts); without + // this entry the directory is silently dropped here and the install succeeds with + // no schema at all — a green install and a missing table. + 'migrations', 'skills', 'assets', 'README.md', diff --git a/middleware/assets/boilerplate/agent-pure-llm/scripts/build-zip.mjs b/middleware/assets/boilerplate/agent-pure-llm/scripts/build-zip.mjs index e4c380d2f..b5eb16aac 100644 --- a/middleware/assets/boilerplate/agent-pure-llm/scripts/build-zip.mjs +++ b/middleware/assets/boilerplate/agent-pure-llm/scripts/build-zip.mjs @@ -68,6 +68,10 @@ const INCLUDE = [ 'manifest.yaml', 'package.json', 'dist', + // Plugin-owned SQL. The host allows .sql in a package (zipExtractor.ts); without + // this entry the directory is silently dropped here and the install succeeds with + // no schema at all — a green install and a missing table. + 'migrations', 'skills', 'assets', 'README.md', diff --git a/specs/470-dev-platform-plugin/README.md b/specs/470-dev-platform-plugin/README.md index 9331b3303..0711413bc 100644 --- a/specs/470-dev-platform-plugin/README.md +++ b/specs/470-dev-platform-plugin/README.md @@ -54,19 +54,20 @@ it at all. reached `path.join` and then a recursive `fs.rm`; reachable from remote registry ZIPs and imported profile bundles. Kept out of this PR so it does not wait on epic decisions. -### Held back after review - -Two fixes were implemented, reviewed, and **not shipped** — the review caught real harm: - -- **`.sql` in the ZIP allowlist** — would have escalated the #548 traversal from - delete/replace to arbitrary SQL execution. Blocked until #548 lands. -- **Advisory lock on the 8 migrators** — an unbounded wait, but three migrators run inside a - plugin `activate()` capped at 10s, so it would convert a rare race into a deterministic - boot failure. Redesign needed (bounded wait); a second attempt also failed review because - it never read `pg_advisory_unlock`'s return value. -- **DynamicAgentRuntime rollback** — the attempt left a zombie entry in `active`, and its - by-source rollback tore down the winner's registrations under two concurrent activations - of the same id. +### Wave 2 — shipped after #548 landed + +- **`.sql` in the ZIP allowlist** — unblocked once the traversal fix merged. Also adds + `migrations` to both boilerplate `build-zip.mjs` INCLUDE lists, without which the + directory is silently dropped and the install succeeds with no schema. +- **Migrator concurrency** — shipped on the third attempt. Two prior designs were rejected: + an unbounded `pg_advisory_lock` inside a 10s `activate()` budget, and a retry that never + read `pg_advisory_unlock`'s return value. + +### Still held back + +- **DynamicAgentRuntime rollback** — two attempts rejected. The current one does not cover + the timeout path: `withTimeout` does not cancel, so after the rollback the orphaned + `activate()` keeps running and re-registers. ### Next — decisions before code From add6fa7eccba5aa3f6d4b1d4989e6aff3e739abd Mon Sep 17 00:00:00 2001 From: Marcel Wege Date: Thu, 30 Jul 2026 14:33:35 +0200 Subject: [PATCH 4/4] =?UTF-8?q?chore(470):=20resync=20wave2=20with=20main?= =?UTF-8?q?=20(#549)=20=E2=80=94=20baseline=203,303=20=E2=86=92=203,306?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The channel-api work added 3 dev-platform references (test/packages). Same legitimate raise as on the C5 branch. --- specs/470-dev-platform-plugin/decoupling-baseline.json | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/specs/470-dev-platform-plugin/decoupling-baseline.json b/specs/470-dev-platform-plugin/decoupling-baseline.json index d81904525..f89dec4af 100644 --- a/specs/470-dev-platform-plugin/decoupling-baseline.json +++ b/specs/470-dev-platform-plugin/decoupling-baseline.json @@ -1,9 +1,9 @@ { - "total": 3303, + "total": 3306, "zones": { "middleware/src": 1636, - "middleware/test": 965, - "middleware/packages": 97, + "middleware/test": 966, + "middleware/packages": 99, "middleware/scripts": 8, "middleware/sidecars": 195, "middleware/migrations": 70,