test: clean up throwaway Postgres resources

reference-implementation/scripts/run-tests-failure.js

@@ -0,0 +1,15 @@
+// Copyright The PDP-Connect Contributors
+// SPDX-License-Identifier: Apache-2.0
+
+export function combineTestAndCleanupFailure(testError, cleanupError, context) {
+  if (!testError) return cleanupError;
+  if (!cleanupError) return testError;
+  return new AggregateError([testError, cleanupError], `${context} and database cleanup both failed`);
+}
+
+export function testProcessExitFailure(filePath, code, signal, output) {
+  const outcome = signal ? `exited via signal ${signal}` : `exited with code ${code}`;
+  const error = new Error(`Test process for ${filePath} ${outcome}\n${output}`);
+  error.output = output;
+  return error;
+}

reference-implementation/scripts/run-tests-failure.test.mjs

@@ -0,0 +1,27 @@
+// Copyright The PDP-Connect Contributors
+// SPDX-License-Identifier: Apache-2.0
+
+import assert from 'node:assert/strict';
+import test from 'node:test';
+
+import { combineTestAndCleanupFailure, testProcessExitFailure } from './run-tests-failure.js';
+
+test('runner preserves spawn and cleanup failures together', () => {
+  const spawnError = new Error('spawn failed');
+  const cleanupError = new Error('drop failed');
+  const failure = combineTestAndCleanupFailure(spawnError, cleanupError, 'could not start test process');
+
+  assert.ok(failure instanceof AggregateError);
+  assert.deepEqual(failure.errors, [spawnError, cleanupError]);
+});
+
+test('runner preserves failed child output and cleanup failure together', () => {
+  const output = '\n==> test/example.test.js\nnot ok 1 - expected pass\n';
+  const childError = testProcessExitFailure('test/example.test.js', 1, null, output);
+  const cleanupError = new Error('drop failed');
+  const failure = combineTestAndCleanupFailure(childError, cleanupError, 'test process failed');
+
+  assert.ok(failure instanceof AggregateError);
+  assert.deepEqual(failure.errors, [childError, cleanupError]);
+  assert.equal(failure.errors[0].output, output);
+});

reference-implementation/scripts/run-tests.js

@@ -6,8 +6,10 @@ import { availableParallelism } from 'node:os';
 import { dirname, join } from 'node:path';
 import { fileURLToPath } from 'node:url';
 import { spawn } from 'node:child_process';
+import { randomBytes } from 'node:crypto';
 import pg from 'pg';
 import { buildScrubbedTestEnv } from './test-env.js';
+import { combineTestAndCleanupFailure, testProcessExitFailure } from './run-tests-failure.js';
 
 const __dirname = dirname(fileURLToPath(import.meta.url));
 const repoRoot = dirname(__dirname);
@@ -26,6 +28,7 @@ const requestedConcurrency = Number.parseInt(process.env.PDPP_TEST_CONCURRENCY |
 // requiring any changes to individual test files.
 
 let fileCounter = 0;
+const runnerId = randomBytes(4).toString('hex');
 
 /**
  * Derive the admin connection URL from a per-test URL by replacing the
@@ -40,8 +43,8 @@ function adminUrlFromBase(baseUrl) {
 }
 
 /**
- * Derive a short, safe DB name from the test file path and a monotonic
- * counter so concurrent workers never collide.
+ * Derive a short, safe DB name from the test file path, a per-run random ID,
+ * and a monotonic counter so concurrent runners and workers never collide.
  */
 function deriveDbName(filePath) {
   // Strip directory and extension; keep only alphanumeric/underscore chars.
@@ -52,9 +55,11 @@ function deriveDbName(filePath) {
     .replace(/\.[^.]+$/, '')
     .replace(/[^a-z0-9_]/gi, '_')
     .toLowerCase()
-    .slice(0, 40);
+    // PostgreSQL identifiers are limited to 63 bytes. The fixed prefix,
+    // random ID, and a base-36 counter leave 38 bytes for the file stem.
+    .slice(0, 38);
   fileCounter += 1;
-  return `pdpp_test_${base}_${fileCounter}`;
+  return `pdpp_test_${base}_${runnerId}_${fileCounter.toString(36)}`;
 }
 
 // Per-file databases currently allocated. Tracked so that if the runner
@@ -93,6 +98,10 @@ async function allocateTestDb(filePath, baseUrl) {
   try {
     await client.connect();
     // Identifier is safe: deriveDbName produces only [a-z0-9_] chars.
+    // A prior hard-killed runner can leave an allocated name behind. Remove
+    // it before allocating a fresh DB; the per-run random ID avoids collisions
+    // with any concurrently running test runner.
+    await client.query(`DROP DATABASE IF EXISTS "${dbName}" WITH (FORCE)`);
     await client.query(`CREATE DATABASE "${dbName}"`);
     await client.end();
   } catch (err) {
@@ -106,18 +115,31 @@ async function allocateTestDb(filePath, baseUrl) {
   testUrl.pathname = `/${dbName}`;
 
   const allocation = { url: testUrl.toString() };
+  let releasePromise;
 
-  allocation.release = async function release() {
-    activeAllocations.delete(allocation);
-    const drop = new pg.Client({ connectionString: adminUrl });
-    try {
-      await drop.connect();
-      await drop.query(`DROP DATABASE IF EXISTS "${dbName}"`);
-      await drop.end();
-    } catch (err) {
-      try { await drop.end(); } catch (_) {}
-      process.stderr.write(`[run-tests] WARN: could not drop test DB ${dbName}: ${err.message}\n`);
-    }
+  allocation.release = function release() {
+    if (releasePromise) return releasePromise;
+
+    releasePromise = (async () => {
+      const drop = new pg.Client({ connectionString: adminUrl });
+      try {
+        await drop.connect();
+        // FORCE also closes a leaked test pool, so cleanup cannot silently leave
+        // a per-file database behind after a failed test.
+        await drop.query(`DROP DATABASE IF EXISTS "${dbName}" WITH (FORCE)`);
+        await drop.end();
+        activeAllocations.delete(allocation);
+      } catch (err) {
+        try { await drop.end(); } catch (_) {}
+        throw new Error(`[run-tests] could not drop test DB ${dbName}: ${err.message}`, { cause: err });
+      }
+    })().catch((error) => {
+      // Keep failed allocations eligible for signal cleanup to retry.
+      releasePromise = undefined;
+      throw error;
+    });
+
+    return releasePromise;
   };
 
   // Track the live allocation and arm the runner-level signal cleanup so a
@@ -149,6 +171,13 @@ async function runNodeTest(filePath, extraArgs) {
       env: childEnv,
     });
     let output = '';
+    let settled = false;
+
+    const settle = (outcome, value) => {
+      if (settled) return;
+      settled = true;
+      outcome(value);
+    };
 
     child.stdout.on('data', (chunk) => {
       output += chunk.toString();
@@ -157,24 +186,45 @@ async function runNodeTest(filePath, extraArgs) {
       output += chunk.toString();
     });
 
-    child.on('error', (err) => {
-      if (allocation) allocation.release().finally(() => reject(err));
-      else reject(err);
+    child.on('error', async (err) => {
+      let cleanupError;
+      try {
+        if (allocation) await allocation.release();
+      } catch (error) {
+        cleanupError = error;
+      }
+      settle(reject, combineTestAndCleanupFailure(err, cleanupError, `could not start test process for ${filePath}`));
     });
-    child.on('exit', (code, signal) => {
+    child.on('exit', async (code, signal) => {
+      const testError = signal || code !== 0
+        ? testProcessExitFailure(filePath, code, signal, `\n==> ${filePath}\n${output}`)
+        : undefined;
       const finish = () => {
         if (signal) {
-          reject(new Error(`Test process for ${filePath} exited via signal ${signal}`));
+          settle(reject, testError);
           return;
         }
-        resolve({
+        settle(resolve, {
           filePath,
           exitCode: code ?? 1,
           output: `\n==> ${filePath}\n${output}`,
         });
       };
       if (allocation) {
-        allocation.release().finally(finish);
+        let cleanupError;
+        try {
+          await allocation.release();
+        } catch (error) {
+          cleanupError = error;
+        }
+        if (cleanupError) {
+          settle(
+            reject,
+            combineTestAndCleanupFailure(testError, cleanupError, `test process failed for ${filePath}`),
+          );
+          return;
+        }
+        finish();
       } else {
         finish();
       }

reference-implementation/test/aggregation-rows-conformance-postgres.test.js

@@ -17,7 +17,7 @@
  * registered so the suite remains visible in CI output without failing.
  *
  * Target for local runs:
- *   docker run -d --name pg-pilot -p 55463:5432 \
+ *   docker run --rm -d --name pg-pilot -p 55463:5432 \
  *     -e POSTGRES_USER=pdpp -e POSTGRES_PASSWORD=pdpp \
  *     -e POSTGRES_DB=pdpp_pilot \
  *     pgvector/pgvector:pg16

reference-implementation/test/connector-summary-evidence-throughput-integration.test.js

@@ -35,6 +35,7 @@ import { reconcileConnectorSummaryEvidence } from '../server/connector-summary-e
 import { deleteAllRecordsForConnector, ingestRecord } from '../server/records.js';
 import { closePostgresStorage, initPostgresStorage, postgresQuery } from '../server/postgres-storage.js';
 import { dedicatedPostgresTestUrl } from './helpers/dedicated-postgres-test-url.js';
+import { withTemporaryPostgresDatabase } from './helpers/postgres-temp-database.js';
 
 const OWNER = 'owner_local';
 const NOW = '2026-07-17T00:00:00.000Z';
@@ -99,18 +100,6 @@ async function seedInstancePostgres(connectorInstanceId, connectorId) {
   );
 }
 
-function databaseUrl(url, name) {
-  const parsed = new URL(url);
-  parsed.pathname = `/${name}`;
-  return parsed.toString();
-}
-
-function adminUrl(url) {
-  const parsed = new URL(url);
-  parsed.pathname = '/postgres';
-  return parsed.toString();
-}
-
 let uniquePgDb = 0;
 
 /**
@@ -121,25 +110,15 @@ let uniquePgDb = 0;
  * loopback-only test listener, dropped again on the way out.
  */
 async function withTemporaryPostgres(fn) {
-  const pg = await import('pg');
-  const { Pool } = pg.default;
   uniquePgDb += 1;
-  const admin = new Pool({ connectionString: adminUrl(DEDICATED_POSTGRES_URL) });
   const database = `pdpp_summary_throughput_${process.pid}_${Date.now()}_${uniquePgDb}`;
-  await admin.query(`DROP DATABASE IF EXISTS "${database}"`);
-  await admin.query(`CREATE DATABASE "${database}"`);
-  try {
-    await initPostgresStorage({ backend: 'postgres', databaseUrl: databaseUrl(DEDICATED_POSTGRES_URL, database) });
-    await fn();
-  } finally {
-    await closePostgresStorage().catch(() => undefined);
-    await admin.query(
-      `SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = $1 AND pid <> pg_backend_pid()`,
-      [database],
-    ).catch(() => undefined);
-    await admin.query(`DROP DATABASE IF EXISTS "${database}"`).catch(() => undefined);
-    await admin.end();
-  }
+  return withTemporaryPostgresDatabase(
+    { connectionString: DEDICATED_POSTGRES_URL, databaseName: database, closeConnections: closePostgresStorage },
+    async (databaseUrl) => {
+      await initPostgresStorage({ backend: 'postgres', databaseUrl });
+      await fn();
+    },
+  );
 }
 
 function storageTargetFor(connectorId, connectorInstanceId) {

reference-implementation/test/device-batch-summary-evidence-convergence.test.js

@@ -64,6 +64,7 @@ import {
 import { __setPostgresRecordSortBackfillPhaseHookForTest } from '../server/postgres-records.js';
 import { closePostgresStorage, postgresQuery } from '../server/postgres-storage.js';
 import { dedicatedPostgresTestUrl } from './helpers/dedicated-postgres-test-url.js';
+import { withTemporaryPostgresDatabase } from './helpers/postgres-temp-database.js';
 
 const DEDICATED_POSTGRES_URL = dedicatedPostgresTestUrl(process.env.PDPP_TEST_POSTGRES_URL);
 const PROTOCOL_HEADERS = { 'X-PDPP-Collector-Protocol': COLLECTOR_PROTOCOL_VERSION };
@@ -189,36 +190,12 @@ function deviceIngestUrl(asUrl, device) {
   return `${asUrl}/_ref/device-exporters/${encodeURIComponent(device.device_id)}/ingest-batches`;
 }
 
-function databaseUrl(url, name) {
-  const parsed = new URL(url);
-  parsed.pathname = `/${name}`;
-  return parsed.toString();
-}
-
-function adminUrl(url) {
-  const parsed = new URL(url);
-  parsed.pathname = '/postgres';
-  return parsed.toString();
-}
-
 async function withTemporaryPostgres(fn) {
-  const pg = await import('pg');
-  const { Pool } = pg.default;
-  const admin = new Pool({ connectionString: adminUrl(DEDICATED_POSTGRES_URL) });
   const database = `pdpp_devbatch_summary_${process.pid}_${Date.now()}_${unique}`;
-  await admin.query(`DROP DATABASE IF EXISTS "${database}"`);
-  await admin.query(`CREATE DATABASE "${database}"`);
-  try {
-    await fn(databaseUrl(DEDICATED_POSTGRES_URL, database));
-  } finally {
-    await closePostgresStorage().catch(() => undefined);
-    await admin.query(
-      `SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = $1 AND pid <> pg_backend_pid()`,
-      [database],
-    ).catch(() => undefined);
-    await admin.query(`DROP DATABASE IF EXISTS "${database}"`).catch(() => undefined);
-    await admin.end();
-  }
+  return withTemporaryPostgresDatabase(
+    { connectionString: DEDICATED_POSTGRES_URL, databaseName: database, closeConnections: closePostgresStorage },
+    fn,
+  );
 }
 
 function createDriver(kind, server) {

reference-implementation/test/device-exporter-postgres-proof.test.js

@@ -18,6 +18,7 @@ import { closeDb, getDb, initDb } from '../server/db.js';
 import { ingestRecord, setClientEventEnqueueHook } from '../server/records.js';
 import { makeLocalTransformerBackend } from '../server/search-semantic.js';
 import { dedicatedPostgresTestUrl } from './helpers/dedicated-postgres-test-url.js';
+import { withTemporaryPostgresDatabase } from './helpers/postgres-temp-database.js';
 import {
   bootstrapPostgresSchema,
   closePostgresStorage,
@@ -43,37 +44,12 @@ function tempDbName() {
   return `pdpp_device_ingest_${process.pid}_${Date.now()}_${tempCounter}`;
 }
 
-function databaseUrl(url, name) {
-  const parsed = new URL(url);
-  parsed.pathname = `/${name}`;
-  return parsed.toString();
-}
-
-function adminUrl(url) {
-  const parsed = new URL(url);
-  parsed.pathname = '/postgres';
-  return parsed.toString();
-}
-
 async function withTempPostgres(fn) {
-  const admin = new Pool({ connectionString: adminUrl(POSTGRES_URL) });
   const name = tempDbName();
-  await admin.query(`DROP DATABASE IF EXISTS "${name}"`);
-  await admin.query(`CREATE DATABASE "${name}"`);
-  const url = databaseUrl(POSTGRES_URL, name);
-  try {
-    await fn(url);
-  } finally {
-    await closePostgresStorage().catch(() => undefined);
-    await admin.query(
-      `SELECT pg_terminate_backend(pid)
-         FROM pg_stat_activity
-        WHERE datname = $1 AND pid <> pg_backend_pid()`,
-      [name],
-    ).catch(() => undefined);
-    await admin.query(`DROP DATABASE IF EXISTS "${name}"`).catch(() => undefined);
-    await admin.end();
-  }
+  return withTemporaryPostgresDatabase(
+    { connectionString: POSTGRES_URL, databaseName: name, closeConnections: closePostgresStorage },
+    fn,
+  );
 }
 
 async function closeServer(server) {