Close publication connections before frontend builds

.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

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]; };

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') {