diff --git a/.github/workflows/stack127-publish-pages.yml b/.github/workflows/stack127-publish-pages.yml index 724a7ced..3622f49b 100644 --- a/.github/workflows/stack127-publish-pages.yml +++ b/.github/workflows/stack127-publish-pages.yml @@ -113,19 +113,16 @@ jobs: await publishingMain(plan.kind, ['apply', '--plan', `${process.env.RUNNER_TEMP}/page-plan.json`, '--environment', 'production', '--confirm', 'stack127-production', '--release-sha', publication.releaseSha], { afterCommit: async ({ pool }) => { if (options.hosting !== 'cloudflare') return; - const client = await pool.connect(); - try { - await refreshCommittedContent(client, { - ...options, - buildDeploy: async () => { - await buildFrontend({ output: options.directory }); - await checkCurrentRelease(options); - await deployFrontend({ directory: options.directory, preview: true }); - await checkCurrentRelease(options); - await deployFrontend({ directory: options.directory }); - }, - }); - } finally { client.release(); } + await refreshCommittedContent(pool, { + ...options, + buildDeploy: async () => { + await buildFrontend({ output: options.directory }); + await checkCurrentRelease(options); + await deployFrontend({ directory: options.directory, preview: true }); + await checkCurrentRelease(options); + await deployFrontend({ directory: options.directory }); + }, + }); }, }); NODE diff --git a/scripts/stack127/frontend/content-queue.mjs b/scripts/stack127/frontend/content-queue.mjs index 109ab7d9..1e885986 100644 --- a/scripts/stack127/frontend/content-queue.mjs +++ b/scripts/stack127/frontend/content-queue.mjs @@ -122,7 +122,7 @@ export async function snapshotQueue(client, { sha, force = false, now = () => ne await client.query("commit"); return snapshot; } catch (error) { - await client.query("rollback"); + await client.query("rollback").catch(() => {}); throw error; } } @@ -362,7 +362,7 @@ async function commitCapturedSnapshot(client, snapshot, verifiedCompletion) { await client.query("commit"); return { acknowledged: result.rowCount, captured: snapshot.events.length }; } catch (error) { - await client.query("rollback"); + await client.query("rollback").catch(() => {}); throw error; } } @@ -393,24 +393,66 @@ export async function completeRefresh(client, snapshot, directory, options = {}) return await commitCapturedSnapshot(client, snapshot, true); } -// Pass this callback to the publisher after its SQL commit and before its -// public checks. The coordinator supplies the build and deploy operations so -// everything runs in the existing production concurrency job. -export async function refreshCommittedContent(client, { sha, directory, buildDeploy, hosting = globalThis.process?.env.STACK127_FRONTEND_HOSTING, ...options } = {}) { - if (hosting !== "cloudflare") return { skipped: true, hosting: "legacy" }; - await checkCurrentRelease({ ...options, sha }); - const snapshot = await snapshotQueue(client, { sha, force: true }); - if (options.dryRun) return { dryRun: true, captured: snapshot.events.length, acknowledged: 0 }; - if (typeof buildDeploy !== "function" || !directory) throw new Error("A build/deploy callback and export directory are required."); +async function withRefreshConnection(pool, operation, poolError = () => undefined) { + let connectionError; + const client = await new Promise((resolve, reject) => { + pool.connect((error, client) => { + if (error) { reject(error); return; } + // pg-pool removes its idle listener before calling back. Attach ours + // synchronously, before another protocol message or promise continuation. + client.on("error", error => { connectionError ??= error; }); + resolve(client); + }); + }); + const checkConnection = () => { + const error = connectionError ?? poolError(); + if (error) throw error; + }; try { - await buildDeploy(snapshot); - return await completeRefresh(client, snapshot, directory, options); - } catch (error) { - await retrySnapshot(client, snapshot, error); - throw error; + checkConnection(); + const result = await operation({ async query(...args) { + checkConnection(); + const result = await client.query(...args); + checkConnection(); + return result; + } }); + checkConnection(); + return result; + } finally { + // Even healthy phase connections are closed, so none sit idle during a + // build or get reused for completion/retry. Keep the listener for late + // socket errors while pg finishes closing the discarded client. + client.release(true); } } +// Pass the publisher's pool after its SQL commit and before its public checks. +// Each database phase owns a connection; build/deploy holds none. +export async function refreshCommittedContent(pool, { sha, directory, buildDeploy, hosting = globalThis.process?.env.STACK127_FRONTEND_HOSTING, ...options } = {}) { + if (hosting !== "cloudflare") return { skipped: true, hosting: "legacy" }; + let poolError; + const onPoolError = error => { poolError ??= error; }; + pool.on("error", onPoolError); + try { + await checkCurrentRelease({ ...options, sha }); + const snapshot = await withRefreshConnection(pool, client => snapshotQueue(client, { sha, force: true }), () => poolError); + if (options.dryRun) return { dryRun: true, captured: snapshot.events.length, acknowledged: 0 }; + if (typeof buildDeploy !== "function" || !directory) throw new Error("A build/deploy callback and export directory are required."); + try { + await buildDeploy(snapshot); + if (poolError) throw poolError; + return await withRefreshConnection(pool, client => completeRefresh(client, snapshot, directory, options), () => poolError); + } catch (error) { + try { + await withRefreshConnection(pool, client => retrySnapshot(client, snapshot, error)); + } catch (retryError) { + console.error(`Could not record the frontend refresh retry: ${retryError.message}`); + } + throw error; + } + } finally { pool.removeListener("error", onPoolError); } +} + export async function contentQueueMain(args = process.argv.slice(2), environment = process.env) { const { readFile, writeFile, appendFile } = await import("node:fs/promises"); const value = flag => { const index = args.indexOf(flag); return index < 0 ? undefined : args[index + 1]; }; diff --git a/scripts/stack127/publishing/cli.mjs b/scripts/stack127/publishing/cli.mjs index 4957e1c6..f316545d 100644 --- a/scripts/stack127/publishing/cli.mjs +++ b/scripts/stack127/publishing/cli.mjs @@ -177,6 +177,9 @@ export async function publishingMain(kind, argv = process.argv.slice(2), { after const configuredOrigin = process.env.NEXT_PUBLIC_SITE_URL ?? localEnvironment.match(/^\s*NEXT_PUBLIC_SITE_URL\s*=\s*(.*?)\s*$/m)?.[1]?.replace(/^['"]|['"]$/g, '') ?? 'https://stack127.com'; const canonicalOrigin = args.environment === 'production' ? 'https://stack127.com' : new URL(configuredOrigin).origin; const pool = new pg.Pool({ connectionString: resolveDatabaseUrl({ environment: args.environment, ...(['apply', 'restore'].includes(args.command) ? { apply: true, confirm: args.confirm } : {}) }), max: 1, connectionTimeoutMillis: 15000, query_timeout: 60000 }); + // pg removes disconnected idle clients itself, but emits an error that must + // have a listener, including while discarded refresh connections close. + pool.on('error', error => { console.error(`Publication database pool error: ${error.message}`); }); let applied = false; try { if (args.command === 'prepare') { diff --git a/tests/stack127/frontend-content-queue.test.mjs b/tests/stack127/frontend-content-queue.test.mjs index 11461805..3e1aca0a 100644 --- a/tests/stack127/frontend-content-queue.test.mjs +++ b/tests/stack127/frontend-content-queue.test.mjs @@ -4,6 +4,10 @@ import { mkdtemp, readFile, writeFile, rm } from "node:fs/promises"; import path from "node:path"; import os from "node:os"; import { createHmac } from "node:crypto"; +import { EventEmitter } from "node:events"; +import { execFile } from "node:child_process"; +import { promisify } from "node:util"; +import pg from "pg"; import ts from "typescript"; import { targetsForEvents } from "../../scripts/stack127/cache/targets.mjs"; import { cacheBodyDigest } from "../../scripts/stack127/cache/refresh.mjs"; @@ -578,31 +582,249 @@ test("main advancing during API refresh prevents acknowledgement", async () => { } finally { await files.close(); } }); -test("the publishing callback is inactive by default and preserves failed captured edits", async () => { +// Use real pg.Client event emitters without a database. Query models above +// supply the rows; every checkout must be discarded before another phase. +function refreshPool(...models) { + const pool = new EventEmitter(); + pool.clients = []; + pool.attempts = 0; + pool.connect = callback => { + const model = models[pool.attempts++]; + if (model instanceof Error) { callback(model); return; } + assert(model, "Unexpected database checkout"); + const client = new pg.Client(); + client.discarded = false; + client.query = (...args) => { + assert.equal(client.discarded, false, "A discarded connection was reused"); + return model.query(...args); + }; + client.release = destroy => { + assert.equal(destroy, true, "Refresh connections must not return to the idle pool"); + assert.equal(client.discarded, false, "Connection released twice"); + client.discarded = true; + }; + pool.clients.push(client); + callback(null, client); + }; + return pool; +} + +function refreshOptions(files) { + return { hosting: "cloudflare", sha, directory: files?.directory ?? "/unused", token: "read", zone: "zone", cloudflareToken: "purge", + fetchImplementation: async (url, init) => { + if (url.endsWith("/api/cache/revalidate")) return json({ ok: true }); + if (url.includes("purge_cache")) return json({ success: true }); + if (files) return files.fetchImplementation(url, init); + return url.includes("api.github.com") ? json({ object: { sha } }) : json({ ok: true, build: { sha } }); + }, + }; +} + +function refreshCompletionClient() { + const client = ackClient(); + const query = client.query; + client.query = async (text, values) => { + if (text.includes("select signing_key")) { + client.calls.push({ text, values }); + return { rows: [{ signing_key: "cache-key" }] }; + } + return query(text, values); + }; + return client; +} + +test("real pg checkout handles a fatal response in the ready packet during snapshot and retry acquisition", async t => { + for (const phase of ["snapshot", "retry"]) await t.test(phase, async () => { + // A child process makes an uncaught client error an observable failure. + // The local peer speaks only enough PostgreSQL to exercise real checkout. + const source = ` + import assert from 'node:assert/strict'; + import { createServer } from 'node:net'; + import { EventEmitter } from 'node:events'; + import pg from 'pg'; + import { refreshCommittedContent } from ${JSON.stringify(new URL("../../scripts/stack127/frontend/content-queue.mjs", import.meta.url).href)}; + const sha = ${JSON.stringify(sha)}; + const revision = ${JSON.stringify(revision)}; + const event = ${JSON.stringify(event)}; + const pages = ${JSON.stringify(pages)}; + const captureClient = ${captureClient.toString()}; + const original = new Error('Original build failure'); + const phase = ${JSON.stringify(phase)}; + const packet = (type, body) => { + const header = Buffer.alloc(5); + header.write(type); header.writeInt32BE(body.length + 4, 1); + return Buffer.concat([header, body]); + }; + let connections = 0, builds = 0; + const server = createServer(socket => { + connections++; + socket.on('error', () => {}); + socket.once('data', () => socket.end(Buffer.concat([ + packet('R', Buffer.alloc(4)), packet('Z', Buffer.from('I')), + packet('E', Buffer.from('SFATAL\\0C57P01\\0MCheckout terminated\\0\\0')), + ]))); + }); + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)); + const realPool = new pg.Pool({ host:'127.0.0.1', port:server.address().port, + user:'test', database:'test', password:'unused', ssl:false, + connectionTimeoutMillis:2000, max:1 }); + const pool = new EventEmitter(); + realPool.on('error', error => pool.emit('error', error)); + let checkouts = 0; + pool.connect = callback => { + if (phase === 'retry' && checkouts++ === 0) { + const client = new pg.Client(); + client.query = captureClient().query; + client.release = destroy => assert.equal(destroy, true); + if (callback) { callback(null, client); return; } + return Promise.resolve(client); + } + return realPool.connect(callback); + }; + try { + await assert.rejects(refreshCommittedContent(pool, { hosting:'cloudflare', sha, + directory:'/unused', token:'read', + fetchImplementation: async url => Response.json(url.includes('api.github.com') + ? {object:{sha}} : {ok:true,build:{sha}}), + buildDeploy: async () => { builds++; throw original; }, + }), error => phase === 'retry' ? error === original + : error.code === '57P01' && error.message === 'Checkout terminated'); + assert.equal(builds, phase === 'retry' ? 1 : 0); + assert.equal(connections, 1); + } finally { + await realPool.end(); + await new Promise(resolve => server.close(resolve)); + } + `; + await promisify(execFile)(process.execPath, ["--input-type=module", "--eval", source], { timeout: 10000 }); + }); +}); + +test("the publishing callback is inactive by default and records failed captured edits on a fresh connection", async () => { assert.deepEqual(await refreshCommittedContent(null, { hosting: "legacy" }), { skipped: true, hosting: "legacy" }); const client = captureClient(); - const fetchImplementation = async url => url.includes("api.github.com") ? json({ object: { sha } }) : json({ ok: true, build: { sha } }); - await assert.rejects(refreshCommittedContent(client, { hosting: "cloudflare", sha, directory: "/unused", token: "read", fetchImplementation, + const failure = captureClient(); + const pool = refreshPool(client, failure); + await assert.rejects(refreshCommittedContent(pool, { ...refreshOptions(), buildDeploy: async captured => { assert.equal(client.calls.at(-1).text, "commit"); assert.equal(captured.events[0].created_at, event.created_at); + assert.equal(pool.clients[0].discarded, true); throw new Error("Build failed"); }, }), /Build failed/); - const retry = client.calls.at(-1); + const retry = failure.calls.at(-1); assert.match(retry.text, /update public.stack127_revalidation_events/); assert.equal(JSON.parse(retry.values[0])[0].created_at, event.created_at); + assert.equal(pool.clients.length, 2); + assert(pool.clients.every(connection => connection.discarded)); + assert(client.calls.every(call => !/update |delete |insert /i.test(call.text))); +}); + +test("build holds no connection, late async client errors are handled, and completion uses a fresh client", async () => { + const files = await fixture(); + const captured = captureClient(); + const completion = refreshCompletionClient(); + const pool = refreshPool(captured, completion); + try { + const result = await refreshCommittedContent(pool, { ...refreshOptions(files), buildDeploy: async () => { + assert.equal(pool.clients.length, 1); + assert(pool.clients.every(client => client.discarded)); + // An error outside a query promise must not escape as an uncaught event, + // even when the old socket finishes closing during the build pause. + await new Promise(resolve => setImmediate(() => { + pool.clients[0].emit("error", new Error("Late socket close")); + resolve(); + })); + assert.equal(completion.calls.length, 0); + } }); + assert.deepEqual(result, { acknowledged: 1, captured: 1 }); + assert.equal(completion.queued, null); + assert.equal(pool.clients.length, 2); + assert(pool.clients.every(client => client.discarded)); + assert.equal(pool.listenerCount("error"), 0); + } finally { await files.close(); } +}); + +test("async disconnect during completion keeps edits queued and records retry with a third client", async () => { + const files = await fixture(); + const completion = refreshCompletionClient(); + const retry = captureClient(); + const pool = refreshPool(captureClient(), completion, retry); + const disconnect = new Error("Connection terminated unexpectedly"); + const options = refreshOptions(files); + let disconnected = false; + try { + await assert.rejects(refreshCommittedContent(pool, { ...options, buildDeploy: async () => {}, + fetchImplementation: async (url, init) => { + if (url.endsWith("/api/cache/revalidate")) { + await new Promise(resolve => setImmediate(() => { + assert.equal(pool.clients[1].discarded, false); + pool.clients[1].emit("error", disconnect); + disconnected = true; + resolve(); + })); + } + return options.fetchImplementation(url, init); + }, + }), error => error === disconnect); + assert.equal(disconnected, true); + assert.equal(completion.queued, event); + assert(completion.calls.every(call => !/delete |insert |update |^begin$/i.test(call.text))); + assert.equal(pool.clients.length, 3); + assert(pool.clients.every(client => client.discarded)); + assert.match(retry.calls.at(-1).text, /update public.stack127_revalidation_events/); + assert.equal(retry.calls.at(-1).values[1], String(disconnect)); + } finally { await files.close(); } +}); + +test("async pool errors during a paused build stay in the failure path and preserve the queue", async () => { + const retry = captureClient(); + const pool = refreshPool(captureClient(), retry); + const disconnect = new Error("Idle pool connection lost"); + await assert.rejects(refreshCommittedContent(pool, { ...refreshOptions(), buildDeploy: async () => { + assert(pool.clients.every(client => client.discarded)); + await new Promise(resolve => setImmediate(() => { pool.emit("error", disconnect); resolve(); })); + } }), error => error === disconnect); + assert.equal(pool.clients.length, 2); + assert(pool.clients.every(client => client.discarded)); + assert.equal(retry.calls.length, 1); + assert.match(retry.calls[0].text, /update public.stack127_revalidation_events/); + assert.equal(retry.calls[0].values[1], String(disconnect)); +}); + +test("retry connection, query, and async errors preserve the original refresh failure", async t => { + const messages = []; + t.mock.method(console, "error", message => { messages.push(message); }); + for (const mode of ["connect", "query", "async"]) { + const original = new Error(`Build failed before ${mode} retry failure`); + const retryError = new Error(`Retry ${mode} failed`); + let pool; + const retry = mode === "connect" ? retryError : { async query() { + if (mode === "query") throw retryError; + await new Promise(resolve => setImmediate(() => { pool.clients[1].emit("error", retryError); resolve(); })); + return { rows: [] }; + } }; + pool = refreshPool(captureClient(), retry); + await assert.rejects(refreshCommittedContent(pool, { ...refreshOptions(), buildDeploy: async () => { throw original; } }), error => error === original); + assert.equal(pool.attempts, 2, "Failed retry writes must not be repeated blindly"); + assert(pool.clients.every(client => client.discarded)); + assert(messages.at(-1).includes(retryError.message)); + } }); test("the publishing callback dry run captures without building, deploying, or writing", async () => { const client = captureClient(); - const result = await refreshCommittedContent(client, { hosting: "cloudflare", sha, directory: "/unused", token: "read", dryRun: true, + const pool = refreshPool(client); + const result = await refreshCommittedContent(pool, { hosting: "cloudflare", sha, directory: "/unused", token: "read", dryRun: true, fetchImplementation: async url => url.includes("api.github.com") ? json({ object: { sha } }) : json({ ok: true, build: { sha } }), buildDeploy: () => assert.fail("Unexpected build or deploy"), }); assert.equal(result.acknowledged, 0); assert.equal(result.captured, 1); assert(client.calls.every(call => !/update |delete |insert /i.test(call.text))); + assert.equal(pool.clients.length, 1); + assert.equal(pool.clients[0].discarded, true); }); // Exercise the real Edge Function entry point without a Supabase connection.