openvibers/openvibe.sdk Commits
Preserve active test databases and remove orphaned test volumes (#7)
Changes
@@@ -3,7 +3,10 @@# shape (ADR-035), for integration tests of openvibe-sdk/db, cache, queue, pubsub and limits.## scripts/test-services.sh up start (or reuse) and print the env to export--# scripts/test-services.sh down remove the containers++# scripts/test-services.sh down remove the containers (and their data volumes)++#++# The data is throwaway: createTestDb sweeps abandoned schemas, and `down` removes the containers with their anonymous++# volumes. `up` must not rebuild a reused database: a live test may have no connection after its pool becomes idle.## Tests read OV_TEST_PG_URL (through PgBouncer), OV_TEST_PG_DIRECT_URL and OV_TEST_VALKEY_URL; without them the# integration cases print "<label>: skipped (…)" and the suite still runs on PGlite and the memory stores.@@@ -17,24 +20,24 @@ up)docker network inspect $NET >/dev/null 2>&1 || docker network create $NET >/dev/nullif docker inspect ovsdk-pg >/dev/null 2>&1 && [ "$(docker inspect -f '{{.HostConfig.ShmSize}}' ovsdk-pg)" -lt 1073741824 ]; then# A reused ovsdk-pg from before --shm-size has the 64 MB default; remove it and PgBouncer (which points at it) so both start again.-- docker rm -f ovsdk-pg ovsdk-pgbouncer >/dev/null 2>&1 || true++ docker rm -f -v ovsdk-pg ovsdk-pgbouncer >/dev/null 2>&1 || truefiif ! docker ps --format '{{.Names}}' | grep -qx ovsdk-pg; then-- docker rm -f ovsdk-pg >/dev/null 2>&1 || true++ docker rm -f -v ovsdk-pg >/dev/null 2>&1 || truedocker run -d --name ovsdk-pg --network $NET -p 127.0.0.1:55432:5432 \--shm-size=1g \-e POSTGRES_USER=ov -e POSTGRES_PASSWORD=$PW -e POSTGRES_DB=ovtest -e POSTGRES_HOST_AUTH_METHOD=scram-sha-256 -e POSTGRES_INITDB_ARGS=--auth-host=scram-sha-256 \postgres:18-alpine -c max_connections=200 >/dev/nullfiif ! docker ps --format '{{.Names}}' | grep -qx ovsdk-pgbouncer; then-- docker rm -f ovsdk-pgbouncer >/dev/null 2>&1 || true++ docker rm -f -v ovsdk-pgbouncer >/dev/null 2>&1 || truedocker run -d --name ovsdk-pgbouncer --network $NET -p 127.0.0.1:56432:5432 \-e DATABASE_URL="postgres://ov:$PW@ovsdk-pg:5432/ovtest" -e POOL_MODE=transaction -e AUTH_TYPE=scram-sha-256 -e MAX_CLIENT_CONN=500 -e DEFAULT_POOL_SIZE=10 \-e MAX_DB_CONNECTIONS=120 -e SERVER_IDLE_TIMEOUT=5 \edoburu/pgbouncer:latest >/dev/nullfiif ! docker ps --format '{{.Names}}' | grep -qx ovsdk-valkey; then-- docker rm -f ovsdk-valkey >/dev/null 2>&1 || true++ docker rm -f -v ovsdk-valkey >/dev/null 2>&1 || truedocker run -d --name ovsdk-valkey --network $NET -p 127.0.0.1:56379:6379 valkey/valkey:9-alpine >/dev/nullfifor i in $(seq 1 60); do docker exec ovsdk-pg pg_isready -U ov -d ovtest >/dev/null 2>&1 && break; sleep 0.5; done@@@ -42,6 +45,6 @@ up)echo "export OV_TEST_PG_DIRECT_URL=postgres://ov:[email protected]:55432/ovtest"echo "export OV_TEST_VALKEY_URL=redis://127.0.0.1:56379/0";;--down) docker rm -f ovsdk-pg ovsdk-pgbouncer ovsdk-valkey >/dev/null 2>&1 || true; docker network rm $NET >/dev/null 2>&1 || true; echo removed ;;++down) docker rm -f -v ovsdk-pg ovsdk-pgbouncer ovsdk-valkey >/dev/null 2>&1 || true; docker network rm $NET >/dev/null 2>&1 || true; echo removed ;;*) echo "usage: $0 up|down" >&2; exit 2 ;;esac@@@ -21,6 +21,32 @@ const crypto = require('crypto');const quiet = { log() {}, warn() {}, error: (...a) => console.error(...a) };const SETUP_LOCK = 735_1_2026;const pgAvailable = () => !!(process.env.OV_TEST_PG_URL && process.env.OV_TEST_PG_DIRECT_URL);++// A leased run's schema and roles carry the time they were made (base 36 ms). The lease connection stays open even++// when the query pool goes idle; a killed process releases it so a later run can safely remove what it left behind.++const ORPHAN_MS = 6 * 3600e3;++const runName = (safe, now = Date.now()) => `${safe}_t${now.toString(36)}_${crypto.randomBytes(4).toString('hex')}_l`;++/** The time a run's name was made, or null (an older name without one). */++function runTime(name) { const m = /_t([0-9a-z]{8,9})_[0-9a-f]{8}_l$/.exec(String(name)); if (!m) return null; const t = parseInt(m[1], 36); return t > 1.6e12 && t < 4e12 ? t : null; }++const leaseKey = (name) => crypto.createHash('sha256').update(`openvibe-test-db:${name}`).digest().readBigInt64BE(0).toString();++/** Sweep only leased runs: older naming formats have no reliable end-of-run signal. */++async function sweepOrphans(su, { now = Date.now(), maxAge = ORPHAN_MS } = {}) {++ const schemas = (await su.query("SELECT nspname FROM pg_namespace WHERE nspname ~ '_t[0-9a-z]{8,9}_[0-9a-f]{8}_l$'")).rows.map((r) => r.nspname);++ const old = schemas.filter((n) => { const t = runTime(n); return t && now - t > maxAge; });++ let dropped = 0;++ for (const name of old.slice(0, 200)) {++ const removed = await su.tx(async (t) => {++ await t.query('SELECT pg_advisory_xact_lock($1)', [SETUP_LOCK]);++ if (!await t.value('SELECT pg_try_advisory_xact_lock($1::bigint)', [leaseKey(name)])) return false;++ const busy = await t.value('SELECT count(*) FROM pg_stat_activity WHERE usename = ANY($1)', [[name, `${name}_owner`]]);++ if (+busy) return false;++ await t.query(`DROP SCHEMA IF EXISTS ${name} CASCADE`);++ for (const r of [name, `${name}_owner`]) { if (await t.value('SELECT count(*) FROM pg_roles WHERE rolname = $1', [r]) > 0) { await t.query(`DROP OWNED BY ${r}`); await t.query(`DROP ROLE ${r}`); } }++ return true;++ });++ if (removed) dropped++;++ }++ return dropped;++}const valkeyAvailable = () => !!process.env.OV_TEST_VALKEY_URL;async function createTestDb({ migrations, store = process.env.OV_TEST_STORE || 'pglite', service = 'test', max = 4, log = quiet } = {}) {@@@ -32,28 +58,38 @@ async function createTestDb({ migrations, store = process.env.OV_TEST_STORE || '}if (!pgAvailable()) throw new Error('store pg needs OV_TEST_PG_URL and OV_TEST_PG_DIRECT_URL (openvibe-sdk scripts/test-services.sh up)');const safe = String(service).replace(/[^a-z0-9]/g, '').slice(0, 12) || 'svc';-- const name = `${safe}_t${process.pid}_${crypto.randomBytes(4).toString('hex')}`;++ const name = runName(safe);const owner = `${name}_owner`;const pw = crypto.randomBytes(16).toString('hex');const su = createDb({ url: process.env.OV_TEST_PG_DIRECT_URL, service: `${service}-test-admin`, max: 1, log });-- const database = await su.value('SELECT current_database()');-- await su.tx(async (t) => {-- await t.query('SELECT pg_advisory_xact_lock($1)', [SETUP_LOCK]);-- for (const stmt of [-- `CREATE ROLE ${owner} LOGIN PASSWORD '${pw}'`,-- `CREATE ROLE ${name} LOGIN PASSWORD '${pw}'`,-- `GRANT CONNECT ON DATABASE ${database} TO ${owner}, ${name}`,-- `CREATE SCHEMA ${name} AUTHORIZATION ${owner}`,-- `ALTER ROLE ${owner} SET search_path = ${name}`,-- `ALTER ROLE ${name} SET search_path = ${name}`,-- `ALTER ROLE ${name} SET statement_timeout = '15s'`,-- `ALTER ROLE ${name} SET lock_timeout = '5s'`,-- `GRANT USAGE ON SCHEMA ${name} TO ${name}`,-- `ALTER DEFAULT PRIVILEGES FOR ROLE ${owner} IN SCHEMA ${name} GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO ${name}`,-- `ALTER DEFAULT PRIVILEGES FOR ROLE ${owner} IN SCHEMA ${name} GRANT USAGE, SELECT, UPDATE ON SEQUENCES TO ${name}`,-- ]) await t.query(stmt);-- });++ const { Client } = require('pg');++ const lease = new Client({ connectionString: process.env.OV_TEST_PG_DIRECT_URL, application_name: `${service}-test-lease`, keepAlive: true });++ let releasingLease = false;++ try {++ await lease.connect();++ await lease.query('SELECT pg_advisory_lock($1::bigint)', [leaseKey(name)]);++ lease.on('end', () => { if (!releasingLease) throw new Error(`test database lease lost for ${name}`); });++ const database = await su.value('SELECT current_database()');++ await sweepOrphans(su).catch(() => {}); // what killed runs left behind; never in the way of this run++ await su.tx(async (t) => {++ await t.query('SELECT pg_advisory_xact_lock($1)', [SETUP_LOCK]);++ for (const stmt of [++ `CREATE ROLE ${owner} LOGIN PASSWORD '${pw}'`,++ `CREATE ROLE ${name} LOGIN PASSWORD '${pw}'`,++ `GRANT CONNECT ON DATABASE ${database} TO ${owner}, ${name}`,++ `CREATE SCHEMA ${name} AUTHORIZATION ${owner}`,++ `ALTER ROLE ${owner} SET search_path = ${name}`,++ `ALTER ROLE ${name} SET search_path = ${name}`,++ `ALTER ROLE ${name} SET statement_timeout = '15s'`,++ `ALTER ROLE ${name} SET lock_timeout = '5s'`,++ `GRANT USAGE ON SCHEMA ${name} TO ${name}`,++ `ALTER DEFAULT PRIVILEGES FOR ROLE ${owner} IN SCHEMA ${name} GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO ${name}`,++ `ALTER DEFAULT PRIVILEGES FOR ROLE ${owner} IN SCHEMA ${name} GRANT USAGE, SELECT, UPDATE ON SEQUENCES TO ${name}`,++ ]) await t.query(stmt);++ });++ } catch (e) { releasingLease = true; await lease.end().catch(() => {}); await su.close().catch(() => {}); throw e; }const as = (url, user) => { const u = new URL(url); u.username = user; u.password = pw; return u.toString(); };++ let db;async function drop() {try {await su.query('SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE usename = ANY($1)', [[name, owner]]);@@@ -62,9 +98,8 @@ async function createTestDb({ migrations, store = process.env.OV_TEST_STORE || 'await t.query(`DROP SCHEMA IF EXISTS ${name} CASCADE`);for (const r of [name, owner]) { await t.query(`DROP OWNED BY ${r}`); await t.query(`DROP ROLE ${r}`); }});-- } finally { await su.close(); }++ } finally { releasingLease = true; await lease.end().catch(() => {}); await su.close(); }}-- let db;try {if (migrations) {const ownerDb = createDb({ url: as(process.env.OV_TEST_PG_DIRECT_URL, owner), service: `${service}-test-migrate`, max: 1, log });@@@ -88,4 +123,4 @@ function createTestValkey({ prefix = 'test' } = {}) {return createValkey({ url: process.env.OV_TEST_VALKEY_URL, prefix: `ov:${prefix}-test:${crypto.randomBytes(4).toString('hex')}:`, log: quiet });}--module.exports = { createTestDb, createTestValkey, pgAvailable, valkeyAvailable };++module.exports = { createTestDb, createTestValkey, pgAvailable, valkeyAvailable, runName, runTime, leaseKey, sweepOrphans };nnew file mode 100644@@@ -0,0 +1,39 @@++'use strict';++/** Exercise the test service lifecycle with a Docker stub; no containers are started. */++const assert = require('node:assert/strict');++const fs = require('node:fs');++const os = require('node:os');++const path = require('node:path');++const { spawnSync } = require('node:child_process');++const { run } = require('./helpers');++++const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'sdk-test-services-'));++const trace = path.join(dir, 'docker.log');++fs.writeFileSync(path.join(dir, 'docker'), `#!/usr/bin/env node++const fs = require('node:fs');++const args = process.argv.slice(2);++fs.appendFileSync(process.env.DOCKER_TRACE, JSON.stringify(args) + '\\n');++if (args[0] === 'inspect' && args[1] === '-f') console.log(process.env.TEST_SHM || 1073741824);++if (args[0] === 'ps') console.log('ovsdk-pg\\novsdk-pgbouncer\\novsdk-valkey');++` , { mode: 0o755 });++++function up(extra = {}) {++ fs.writeFileSync(trace, '');++ const result = spawnSync('bash', [path.join(__dirname, '../scripts/test-services.sh'), 'up'], {++ encoding: 'utf8', env: { ...process.env, PATH: `${dir}:${process.env.PATH}`, DOCKER_TRACE: trace, ...extra },++ });++ assert.equal(result.status, 0, result.stderr);++ assert.match(result.stdout, /export OV_TEST_PG_URL=/);++ return fs.readFileSync(trace, 'utf8').trim().split('\n').map(JSON.parse);++}++++run([++ ['up preserves a reused database even when no test has an open query connection', () => {++ const calls = up();++ assert.equal(calls.some((a) => a.includes('psql') || a.includes('rm')), false);++ }],++ ['replacing a small-shm container removes its anonymous volume', () => {++ const calls = up({ TEST_SHM: '67108864' });++ assert.ok(calls.some((a) => a[0] === 'rm' && a.includes('-v') && a.includes('ovsdk-pg') && a.includes('ovsdk-pgbouncer')));++ }],++]);@@@ -6,6 +6,7 @@ const os = require('os');const path = require('path');const { run } = require('./helpers');const { createTestDb, pgAvailable } = require('../src/testing');++const { runName, runTime, leaseKey, sweepOrphans } = require('../src/testing/db');const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'sdk-testdb-'));fs.writeFileSync(path.join(dir, '0001_initial.sql'), '-- phase: expand\nCREATE TABLE notes (id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, body text NOT NULL);\n');@@@ -19,6 +20,12 @@ const tests = [assert.equal(await t.db.value('SELECT count(*)::int FROM notes'), 1);} finally { await t.close(); }}],++ ['a run\'s schema name carries when it was made (so a killed run\'s leftovers can be swept); an older name has none', async () => {++ const t0 = Date.now(), n = runName('sdk', t0);++ assert.match(n, /^sdk_t[0-9a-z]{8,9}_[0-9a-f]{8}_l$/); assert.ok(n.length < 50);++ assert.equal(runTime(n), t0); assert.equal(runTime('sdk_t12345_abcdef01'), null);++ assert.equal(runTime(n.slice(0, -2)), null, 'unleased names must never be swept');++ }],];if (pgAvailable()) {tests.push(['postgresql+pgbouncer: four setups at once each get their own schema, and close() drops them', async () => {@@@ -29,6 +36,31 @@ if (pgAvailable()) {for (const t of all) assert.equal(await t.db.value('SELECT count(*)::int FROM notes'), 1, 'isolated');} finally { await Promise.all(all.map((t) => t.close())); }}]);++ tests.push(['a live run keeps its schema after its query pool idles; a released lease permits sweeping', async () => {++ const { createDb } = require('../src/db');++ const { Client } = require('pg');++ const su = createDb({ url: process.env.OV_TEST_PG_DIRECT_URL, service: 'sdk-sweep-test', max: 1 });++ const stale = runName('sdk', Date.now() - 7 * 3600e3), fresh = runName('sdk', Date.now() - 60e3);++ const lease = new Client({ connectionString: process.env.OV_TEST_PG_DIRECT_URL });++ let leased = false;++ try {++ for (const n of [stale, fresh]) for (const q of [`CREATE ROLE ${n}_owner LOGIN`, `CREATE ROLE ${n} LOGIN`, `CREATE SCHEMA ${n} AUTHORIZATION ${n}_owner`, `CREATE TABLE ${n}.t (id int)`]) await su.query(q);++ await lease.connect(); leased = true;++ await lease.query('SELECT pg_advisory_lock($1::bigint)', [leaseKey(stale)]);++ const t = await createTestDb({ migrations: dir, store: 'pg', service: 'sdk' }); await t.close();++ let left = (await su.query('SELECT nspname FROM pg_namespace WHERE nspname = ANY($1)', [[stale, fresh]])).rows.map((r) => r.nspname);++ assert.deepEqual(left.sort(), [stale, fresh].sort(), 'the held lease preserves the stale run');++ await lease.end(); leased = false;++ await sweepOrphans(su);++ left = (await su.query('SELECT nspname FROM pg_namespace WHERE nspname = ANY($1)', [[stale, fresh]])).rows.map((r) => r.nspname);++ assert.deepEqual(left, [fresh], 'the released lease permits sweeping; the recent run stays');++ assert.equal(+(await su.value('SELECT count(*) FROM pg_roles WHERE rolname = ANY($1)', [[stale, `${stale}_owner`]])), 0, 'its roles too');++ } finally {++ if (leased) await lease.end().catch(() => {});++ for (const q of [`DROP SCHEMA IF EXISTS ${fresh} CASCADE`, `DROP ROLE IF EXISTS ${fresh}`, `DROP ROLE IF EXISTS ${fresh}_owner`]) await su.query(q).catch(() => {});++ await su.close();++ }++ }]);} else console.log('createTestDb on the containers: skipped (OV_TEST_PG_URL not set; scripts/test-services.sh up)');run(tests);
@@ -3,7 +3,10 @@
@@ -3,7 +3,10 @@
# shape (ADR-035), for integration tests of openvibe-sdk/db, cache, queue, pubsub and limits.
# shape (ADR-035), for integration tests of openvibe-sdk/db, cache, queue, pubsub and limits.
#
#
# scripts/test-services.sh up start (or reuse) and print the env to export
# scripts/test-services.sh up start (or reuse) and print the env to export
-# scripts/test-services.sh down remove the containers
+# scripts/test-services.sh down remove the containers (and their data volumes)
+#
+# The data is throwaway: createTestDb sweeps abandoned schemas, and `down` removes the containers with their anonymous
+# volumes. `up` must not rebuild a reused database: a live test may have no connection after its pool becomes idle.
#
#
# Tests read OV_TEST_PG_URL (through PgBouncer), OV_TEST_PG_DIRECT_URL and OV_TEST_VALKEY_URL; without them the
# Tests read OV_TEST_PG_URL (through PgBouncer), OV_TEST_PG_DIRECT_URL and OV_TEST_VALKEY_URL; without them the
# integration cases print "<label>: skipped (…)" and the suite still runs on PGlite and the memory stores.
# integration cases print "<label>: skipped (…)" and the suite still runs on PGlite and the memory stores.
@@ -17,24 +20,24 @@ up)
@@ -17,24 +20,24 @@ up)
docker network inspect $NET >/dev/null 2>&1 || docker network create $NET >/dev/null
docker network inspect $NET >/dev/null 2>&1 || docker network create $NET >/dev/null
if docker inspect ovsdk-pg >/dev/null 2>&1 && [ "$(docker inspect -f '{{.HostConfig.ShmSize}}' ovsdk-pg)" -lt 1073741824 ]; then
if docker inspect ovsdk-pg >/dev/null 2>&1 && [ "$(docker inspect -f '{{.HostConfig.ShmSize}}' ovsdk-pg)" -lt 1073741824 ]; then
# A reused ovsdk-pg from before --shm-size has the 64 MB default; remove it and PgBouncer (which points at it) so both start again.
# A reused ovsdk-pg from before --shm-size has the 64 MB default; remove it and PgBouncer (which points at it) so both start again.
- docker rm -f ovsdk-pg ovsdk-pgbouncer >/dev/null 2>&1 || true
+ docker rm -f -v ovsdk-pg ovsdk-pgbouncer >/dev/null 2>&1 || true
fi
fi
if ! docker ps --format '{{.Names}}' | grep -qx ovsdk-pg; then
if ! docker ps --format '{{.Names}}' | grep -qx ovsdk-pg; then
- docker rm -f ovsdk-pg >/dev/null 2>&1 || true
+ docker rm -f -v ovsdk-pg >/dev/null 2>&1 || true
docker run -d --name ovsdk-pg --network $NET -p 127.0.0.1:55432:5432 \
docker run -d --name ovsdk-pg --network $NET -p 127.0.0.1:55432:5432 \
--shm-size=1g \
--shm-size=1g \
-e POSTGRES_USER=ov -e POSTGRES_PASSWORD=$PW -e POSTGRES_DB=ovtest -e POSTGRES_HOST_AUTH_METHOD=scram-sha-256 -e POSTGRES_INITDB_ARGS=--auth-host=scram-sha-256 \
-e POSTGRES_USER=ov -e POSTGRES_PASSWORD=$PW -e POSTGRES_DB=ovtest -e POSTGRES_HOST_AUTH_METHOD=scram-sha-256 -e POSTGRES_INITDB_ARGS=--auth-host=scram-sha-256 \
postgres:18-alpine -c max_connections=200 >/dev/null
postgres:18-alpine -c max_connections=200 >/dev/null
fi
fi
if ! docker ps --format '{{.Names}}' | grep -qx ovsdk-pgbouncer; then
if ! docker ps --format '{{.Names}}' | grep -qx ovsdk-pgbouncer; then
- docker rm -f ovsdk-pgbouncer >/dev/null 2>&1 || true
+ docker rm -f -v ovsdk-pgbouncer >/dev/null 2>&1 || true
docker run -d --name ovsdk-pgbouncer --network $NET -p 127.0.0.1:56432:5432 \
docker run -d --name ovsdk-pgbouncer --network $NET -p 127.0.0.1:56432:5432 \
-e DATABASE_URL="postgres://ov:$PW@ovsdk-pg:5432/ovtest" -e POOL_MODE=transaction -e AUTH_TYPE=scram-sha-256 -e MAX_CLIENT_CONN=500 -e DEFAULT_POOL_SIZE=10 \
-e DATABASE_URL="postgres://ov:$PW@ovsdk-pg:5432/ovtest" -e POOL_MODE=transaction -e AUTH_TYPE=scram-sha-256 -e MAX_CLIENT_CONN=500 -e DEFAULT_POOL_SIZE=10 \
-e MAX_DB_CONNECTIONS=120 -e SERVER_IDLE_TIMEOUT=5 \
-e MAX_DB_CONNECTIONS=120 -e SERVER_IDLE_TIMEOUT=5 \
edoburu/pgbouncer:latest >/dev/null
edoburu/pgbouncer:latest >/dev/null
fi
fi
if ! docker ps --format '{{.Names}}' | grep -qx ovsdk-valkey; then
if ! docker ps --format '{{.Names}}' | grep -qx ovsdk-valkey; then
- docker rm -f ovsdk-valkey >/dev/null 2>&1 || true
+ docker rm -f -v ovsdk-valkey >/dev/null 2>&1 || true
docker run -d --name ovsdk-valkey --network $NET -p 127.0.0.1:56379:6379 valkey/valkey:9-alpine >/dev/null
docker run -d --name ovsdk-valkey --network $NET -p 127.0.0.1:56379:6379 valkey/valkey:9-alpine >/dev/null
fi
fi
for i in $(seq 1 60); do docker exec ovsdk-pg pg_isready -U ov -d ovtest >/dev/null 2>&1 && break; sleep 0.5; done
for i in $(seq 1 60); do docker exec ovsdk-pg pg_isready -U ov -d ovtest >/dev/null 2>&1 && break; sleep 0.5; done
@@ -42,6 +45,6 @@ up)
@@ -42,6 +45,6 @@ up)
echo "export OV_TEST_PG_DIRECT_URL=postgres://ov:[email protected]:55432/ovtest"
echo "export OV_TEST_PG_DIRECT_URL=postgres://ov:[email protected]:55432/ovtest"
echo "export OV_TEST_VALKEY_URL=redis://127.0.0.1:56379/0"
echo "export OV_TEST_VALKEY_URL=redis://127.0.0.1:56379/0"
;;
;;
-down) docker rm -f ovsdk-pg ovsdk-pgbouncer ovsdk-valkey >/dev/null 2>&1 || true; docker network rm $NET >/dev/null 2>&1 || true; echo removed ;;
+down) docker rm -f -v ovsdk-pg ovsdk-pgbouncer ovsdk-valkey >/dev/null 2>&1 || true; docker network rm $NET >/dev/null 2>&1 || true; echo removed ;;
*) echo "usage: $0 up|down" >&2; exit 2 ;;
*) echo "usage: $0 up|down" >&2; exit 2 ;;
esac
esac
@@ -21,6 +21,32 @@ const crypto = require('crypto');
@@ -21,6 +21,32 @@ const crypto = require('crypto');
const quiet = { log() {}, warn() {}, error: (...a) => console.error(...a) };
const quiet = { log() {}, warn() {}, error: (...a) => console.error(...a) };
const SETUP_LOCK = 735_1_2026;
const SETUP_LOCK = 735_1_2026;
const pgAvailable = () => !!(process.env.OV_TEST_PG_URL && process.env.OV_TEST_PG_DIRECT_URL);
const pgAvailable = () => !!(process.env.OV_TEST_PG_URL && process.env.OV_TEST_PG_DIRECT_URL);
+// A leased run's schema and roles carry the time they were made (base 36 ms). The lease connection stays open even
+// when the query pool goes idle; a killed process releases it so a later run can safely remove what it left behind.
+const ORPHAN_MS = 6 * 3600e3;
+const runName = (safe, now = Date.now()) => `${safe}_t${now.toString(36)}_${crypto.randomBytes(4).toString('hex')}_l`;
+/** The time a run's name was made, or null (an older name without one). */
+function runTime(name) { const m = /_t([0-9a-z]{8,9})_[0-9a-f]{8}_l$/.exec(String(name)); if (!m) return null; const t = parseInt(m[1], 36); return t > 1.6e12 && t < 4e12 ? t : null; }
+const leaseKey = (name) => crypto.createHash('sha256').update(`openvibe-test-db:${name}`).digest().readBigInt64BE(0).toString();
+/** Sweep only leased runs: older naming formats have no reliable end-of-run signal. */
+async function sweepOrphans(su, { now = Date.now(), maxAge = ORPHAN_MS } = {}) {
+ const schemas = (await su.query("SELECT nspname FROM pg_namespace WHERE nspname ~ '_t[0-9a-z]{8,9}_[0-9a-f]{8}_l$'")).rows.map((r) => r.nspname);
+ const old = schemas.filter((n) => { const t = runTime(n); return t && now - t > maxAge; });
+ let dropped = 0;
+ for (const name of old.slice(0, 200)) {
+ const removed = await su.tx(async (t) => {
+ await t.query('SELECT pg_advisory_xact_lock($1)', [SETUP_LOCK]);
+ if (!await t.value('SELECT pg_try_advisory_xact_lock($1::bigint)', [leaseKey(name)])) return false;
+ const busy = await t.value('SELECT count(*) FROM pg_stat_activity WHERE usename = ANY($1)', [[name, `${name}_owner`]]);
+ if (+busy) return false;
+ await t.query(`DROP SCHEMA IF EXISTS ${name} CASCADE`);
+ for (const r of [name, `${name}_owner`]) { if (await t.value('SELECT count(*) FROM pg_roles WHERE rolname = $1', [r]) > 0) { await t.query(`DROP OWNED BY ${r}`); await t.query(`DROP ROLE ${r}`); } }
+ return true;
+ });
+ if (removed) dropped++;
+ }
+ return dropped;
+}
const valkeyAvailable = () => !!process.env.OV_TEST_VALKEY_URL;
const valkeyAvailable = () => !!process.env.OV_TEST_VALKEY_URL;
async function createTestDb({ migrations, store = process.env.OV_TEST_STORE || 'pglite', service = 'test', max = 4, log = quiet } = {}) {
async function createTestDb({ migrations, store = process.env.OV_TEST_STORE || 'pglite', service = 'test', max = 4, log = quiet } = {}) {
@@ -32,28 +58,38 @@ async function createTestDb({ migrations, store = process.env.OV_TEST_STORE || '
@@ -32,28 +58,38 @@ async function createTestDb({ migrations, store = process.env.OV_TEST_STORE || '
}
}
if (!pgAvailable()) throw new Error('store pg needs OV_TEST_PG_URL and OV_TEST_PG_DIRECT_URL (openvibe-sdk scripts/test-services.sh up)');
if (!pgAvailable()) throw new Error('store pg needs OV_TEST_PG_URL and OV_TEST_PG_DIRECT_URL (openvibe-sdk scripts/test-services.sh up)');
const safe = String(service).replace(/[^a-z0-9]/g, '').slice(0, 12) || 'svc';
const safe = String(service).replace(/[^a-z0-9]/g, '').slice(0, 12) || 'svc';
- const name = `${safe}_t${process.pid}_${crypto.randomBytes(4).toString('hex')}`;
+ const name = runName(safe);
const owner = `${name}_owner`;
const owner = `${name}_owner`;
const pw = crypto.randomBytes(16).toString('hex');
const pw = crypto.randomBytes(16).toString('hex');
const su = createDb({ url: process.env.OV_TEST_PG_DIRECT_URL, service: `${service}-test-admin`, max: 1, log });
const su = createDb({ url: process.env.OV_TEST_PG_DIRECT_URL, service: `${service}-test-admin`, max: 1, log });
- const database = await su.value('SELECT current_database()');
- await su.tx(async (t) => {
- await t.query('SELECT pg_advisory_xact_lock($1)', [SETUP_LOCK]);
- for (const stmt of [
- `CREATE ROLE ${owner} LOGIN PASSWORD '${pw}'`,
- `CREATE ROLE ${name} LOGIN PASSWORD '${pw}'`,
- `GRANT CONNECT ON DATABASE ${database} TO ${owner}, ${name}`,
- `CREATE SCHEMA ${name} AUTHORIZATION ${owner}`,
- `ALTER ROLE ${owner} SET search_path = ${name}`,
- `ALTER ROLE ${name} SET search_path = ${name}`,
- `ALTER ROLE ${name} SET statement_timeout = '15s'`,
- `ALTER ROLE ${name} SET lock_timeout = '5s'`,
- `GRANT USAGE ON SCHEMA ${name} TO ${name}`,
- `ALTER DEFAULT PRIVILEGES FOR ROLE ${owner} IN SCHEMA ${name} GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO ${name}`,
- `ALTER DEFAULT PRIVILEGES FOR ROLE ${owner} IN SCHEMA ${name} GRANT USAGE, SELECT, UPDATE ON SEQUENCES TO ${name}`,
- ]) await t.query(stmt);
- });
+ const { Client } = require('pg');
+ const lease = new Client({ connectionString: process.env.OV_TEST_PG_DIRECT_URL, application_name: `${service}-test-lease`, keepAlive: true });
+ let releasingLease = false;
+ try {
+ await lease.connect();
+ await lease.query('SELECT pg_advisory_lock($1::bigint)', [leaseKey(name)]);
+ lease.on('end', () => { if (!releasingLease) throw new Error(`test database lease lost for ${name}`); });
+ const database = await su.value('SELECT current_database()');
+ await sweepOrphans(su).catch(() => {}); // what killed runs left behind; never in the way of this run
+ await su.tx(async (t) => {
+ await t.query('SELECT pg_advisory_xact_lock($1)', [SETUP_LOCK]);
+ for (const stmt of [
+ `CREATE ROLE ${owner} LOGIN PASSWORD '${pw}'`,
+ `CREATE ROLE ${name} LOGIN PASSWORD '${pw}'`,
+ `GRANT CONNECT ON DATABASE ${database} TO ${owner}, ${name}`,
+ `CREATE SCHEMA ${name} AUTHORIZATION ${owner}`,
+ `ALTER ROLE ${owner} SET search_path = ${name}`,
+ `ALTER ROLE ${name} SET search_path = ${name}`,
+ `ALTER ROLE ${name} SET statement_timeout = '15s'`,
+ `ALTER ROLE ${name} SET lock_timeout = '5s'`,
+ `GRANT USAGE ON SCHEMA ${name} TO ${name}`,
+ `ALTER DEFAULT PRIVILEGES FOR ROLE ${owner} IN SCHEMA ${name} GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO ${name}`,
+ `ALTER DEFAULT PRIVILEGES FOR ROLE ${owner} IN SCHEMA ${name} GRANT USAGE, SELECT, UPDATE ON SEQUENCES TO ${name}`,
+ ]) await t.query(stmt);
+ });
+ } catch (e) { releasingLease = true; await lease.end().catch(() => {}); await su.close().catch(() => {}); throw e; }
const as = (url, user) => { const u = new URL(url); u.username = user; u.password = pw; return u.toString(); };
const as = (url, user) => { const u = new URL(url); u.username = user; u.password = pw; return u.toString(); };
+ let db;
async function drop() {
async function drop() {
try {
try {
await su.query('SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE usename = ANY($1)', [[name, owner]]);
await su.query('SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE usename = ANY($1)', [[name, owner]]);
@@ -62,9 +98,8 @@ async function createTestDb({ migrations, store = process.env.OV_TEST_STORE || '
@@ -62,9 +98,8 @@ async function createTestDb({ migrations, store = process.env.OV_TEST_STORE || '
await t.query(`DROP SCHEMA IF EXISTS ${name} CASCADE`);
await t.query(`DROP SCHEMA IF EXISTS ${name} CASCADE`);
for (const r of [name, owner]) { await t.query(`DROP OWNED BY ${r}`); await t.query(`DROP ROLE ${r}`); }
for (const r of [name, owner]) { await t.query(`DROP OWNED BY ${r}`); await t.query(`DROP ROLE ${r}`); }
});
});
- } finally { await su.close(); }
+ } finally { releasingLease = true; await lease.end().catch(() => {}); await su.close(); }
}
}
- let db;
try {
try {
if (migrations) {
if (migrations) {
const ownerDb = createDb({ url: as(process.env.OV_TEST_PG_DIRECT_URL, owner), service: `${service}-test-migrate`, max: 1, log });
const ownerDb = createDb({ url: as(process.env.OV_TEST_PG_DIRECT_URL, owner), service: `${service}-test-migrate`, max: 1, log });
@@ -88,4 +123,4 @@ function createTestValkey({ prefix = 'test' } = {}) {
@@ -88,4 +123,4 @@ function createTestValkey({ prefix = 'test' } = {}) {
return createValkey({ url: process.env.OV_TEST_VALKEY_URL, prefix: `ov:${prefix}-test:${crypto.randomBytes(4).toString('hex')}:`, log: quiet });
return createValkey({ url: process.env.OV_TEST_VALKEY_URL, prefix: `ov:${prefix}-test:${crypto.randomBytes(4).toString('hex')}:`, log: quiet });
}
}
-module.exports = { createTestDb, createTestValkey, pgAvailable, valkeyAvailable };
+module.exports = { createTestDb, createTestValkey, pgAvailable, valkeyAvailable, runName, runTime, leaseKey, sweepOrphans };
new file mode 100644
new file mode 100644
@@ -0,0 +1,39 @@
@@ -0,0 +1,39 @@
+'use strict';
+/** Exercise the test service lifecycle with a Docker stub; no containers are started. */
+const assert = require('node:assert/strict');
+const fs = require('node:fs');
+const os = require('node:os');
+const path = require('node:path');
+const { spawnSync } = require('node:child_process');
+const { run } = require('./helpers');
+
+const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'sdk-test-services-'));
+const trace = path.join(dir, 'docker.log');
+fs.writeFileSync(path.join(dir, 'docker'), `#!/usr/bin/env node
+const fs = require('node:fs');
+const args = process.argv.slice(2);
+fs.appendFileSync(process.env.DOCKER_TRACE, JSON.stringify(args) + '\\n');
+if (args[0] === 'inspect' && args[1] === '-f') console.log(process.env.TEST_SHM || 1073741824);
+if (args[0] === 'ps') console.log('ovsdk-pg\\novsdk-pgbouncer\\novsdk-valkey');
+` , { mode: 0o755 });
+
+function up(extra = {}) {
+ fs.writeFileSync(trace, '');
+ const result = spawnSync('bash', [path.join(__dirname, '../scripts/test-services.sh'), 'up'], {
+ encoding: 'utf8', env: { ...process.env, PATH: `${dir}:${process.env.PATH}`, DOCKER_TRACE: trace, ...extra },
+ });
+ assert.equal(result.status, 0, result.stderr);
+ assert.match(result.stdout, /export OV_TEST_PG_URL=/);
+ return fs.readFileSync(trace, 'utf8').trim().split('\n').map(JSON.parse);
+}
+
+run([
+ ['up preserves a reused database even when no test has an open query connection', () => {
+ const calls = up();
+ assert.equal(calls.some((a) => a.includes('psql') || a.includes('rm')), false);
+ }],
+ ['replacing a small-shm container removes its anonymous volume', () => {
+ const calls = up({ TEST_SHM: '67108864' });
+ assert.ok(calls.some((a) => a[0] === 'rm' && a.includes('-v') && a.includes('ovsdk-pg') && a.includes('ovsdk-pgbouncer')));
+ }],
+]);
@@ -6,6 +6,7 @@ const os = require('os');
@@ -6,6 +6,7 @@ const os = require('os');
const path = require('path');
const path = require('path');
const { run } = require('./helpers');
const { run } = require('./helpers');
const { createTestDb, pgAvailable } = require('../src/testing');
const { createTestDb, pgAvailable } = require('../src/testing');
+const { runName, runTime, leaseKey, sweepOrphans } = require('../src/testing/db');
const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'sdk-testdb-'));
const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'sdk-testdb-'));
fs.writeFileSync(path.join(dir, '0001_initial.sql'), '-- phase: expand\nCREATE TABLE notes (id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, body text NOT NULL);\n');
fs.writeFileSync(path.join(dir, '0001_initial.sql'), '-- phase: expand\nCREATE TABLE notes (id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, body text NOT NULL);\n');
@@ -19,6 +20,12 @@ const tests = [
@@ -19,6 +20,12 @@ const tests = [
assert.equal(await t.db.value('SELECT count(*)::int FROM notes'), 1);
assert.equal(await t.db.value('SELECT count(*)::int FROM notes'), 1);
} finally { await t.close(); }
} finally { await t.close(); }
}],
}],
+ ['a run\'s schema name carries when it was made (so a killed run\'s leftovers can be swept); an older name has none', async () => {
+ const t0 = Date.now(), n = runName('sdk', t0);
+ assert.match(n, /^sdk_t[0-9a-z]{8,9}_[0-9a-f]{8}_l$/); assert.ok(n.length < 50);
+ assert.equal(runTime(n), t0); assert.equal(runTime('sdk_t12345_abcdef01'), null);
+ assert.equal(runTime(n.slice(0, -2)), null, 'unleased names must never be swept');
+ }],
];
];
if (pgAvailable()) {
if (pgAvailable()) {
tests.push(['postgresql+pgbouncer: four setups at once each get their own schema, and close() drops them', async () => {
tests.push(['postgresql+pgbouncer: four setups at once each get their own schema, and close() drops them', async () => {
@@ -29,6 +36,31 @@ if (pgAvailable()) {
@@ -29,6 +36,31 @@ if (pgAvailable()) {
for (const t of all) assert.equal(await t.db.value('SELECT count(*)::int FROM notes'), 1, 'isolated');
for (const t of all) assert.equal(await t.db.value('SELECT count(*)::int FROM notes'), 1, 'isolated');
} finally { await Promise.all(all.map((t) => t.close())); }
} finally { await Promise.all(all.map((t) => t.close())); }
}]);
}]);
+ tests.push(['a live run keeps its schema after its query pool idles; a released lease permits sweeping', async () => {
+ const { createDb } = require('../src/db');
+ const { Client } = require('pg');
+ const su = createDb({ url: process.env.OV_TEST_PG_DIRECT_URL, service: 'sdk-sweep-test', max: 1 });
+ const stale = runName('sdk', Date.now() - 7 * 3600e3), fresh = runName('sdk', Date.now() - 60e3);
+ const lease = new Client({ connectionString: process.env.OV_TEST_PG_DIRECT_URL });
+ let leased = false;
+ try {
+ for (const n of [stale, fresh]) for (const q of [`CREATE ROLE ${n}_owner LOGIN`, `CREATE ROLE ${n} LOGIN`, `CREATE SCHEMA ${n} AUTHORIZATION ${n}_owner`, `CREATE TABLE ${n}.t (id int)`]) await su.query(q);
+ await lease.connect(); leased = true;
+ await lease.query('SELECT pg_advisory_lock($1::bigint)', [leaseKey(stale)]);
+ const t = await createTestDb({ migrations: dir, store: 'pg', service: 'sdk' }); await t.close();
+ let left = (await su.query('SELECT nspname FROM pg_namespace WHERE nspname = ANY($1)', [[stale, fresh]])).rows.map((r) => r.nspname);
+ assert.deepEqual(left.sort(), [stale, fresh].sort(), 'the held lease preserves the stale run');
+ await lease.end(); leased = false;
+ await sweepOrphans(su);
+ left = (await su.query('SELECT nspname FROM pg_namespace WHERE nspname = ANY($1)', [[stale, fresh]])).rows.map((r) => r.nspname);
+ assert.deepEqual(left, [fresh], 'the released lease permits sweeping; the recent run stays');
+ assert.equal(+(await su.value('SELECT count(*) FROM pg_roles WHERE rolname = ANY($1)', [[stale, `${stale}_owner`]])), 0, 'its roles too');
+ } finally {
+ if (leased) await lease.end().catch(() => {});
+ for (const q of [`DROP SCHEMA IF EXISTS ${fresh} CASCADE`, `DROP ROLE IF EXISTS ${fresh}`, `DROP ROLE IF EXISTS ${fresh}_owner`]) await su.query(q).catch(() => {});
+ await su.close();
+ }
+ }]);
} else console.log('createTestDb on the containers: skipped (OV_TEST_PG_URL not set; scripts/test-services.sh up)');
} else console.log('createTestDb on the containers: skipped (OV_TEST_PG_URL not set; scripts/test-services.sh up)');
run(tests);
run(tests);