openvibers/openvibe.sdk Commits
sweepOrphans drops only the exact generated run shape, quoted (#9)
Changes
@@@ -28,9 +28,14 @@ const runName = (safe, now = Date.now()) => `${safe}_t${now.toString(36)}_${cryp/** 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. */++/** Exactly what runName() makes (the service name sanitized to [a-z0-9]{1,12}): only this run shape is ever dropped. */++const RUN_SCHEMA = /^[a-z0-9]{1,12}_t[0-9a-z]{8,9}_[0-9a-f]{8}_l$/;++const isRunSchema = (name) => RUN_SCHEMA.test(String(name));++const quoteIdent = (id) => `"${String(id).replace(/"/g, '""')}"`;++/** Sweep only leased runs: older naming formats have no reliable end-of-run signal. A catalog name reaches SQL only++ * after matching the whole generated shape, and quoted. */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 schemas = (await su.query("SELECT nspname FROM pg_namespace WHERE nspname ~ '^[a-z0-9]{1,12}_t[0-9a-z]{8,9}_[0-9a-f]{8}_l$'")).rows.map((r) => r.nspname).filter(isRunSchema);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)) {@@@ -39,8 +44,8 @@ async function sweepOrphans(su, { now = Date.now(), maxAge = ORPHAN_MS } = {}) {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}`); } }++ await t.query(`DROP SCHEMA IF EXISTS ${quoteIdent(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 ${quoteIdent(r)}`); await t.query(`DROP ROLE ${quoteIdent(r)}`); } }return true;});if (removed) dropped++;@@@ -123,4 +128,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, runName, runTime, leaseKey, sweepOrphans };++module.exports = { createTestDb, createTestValkey, pgAvailable, valkeyAvailable, runName, runTime, isRunSchema, leaseKey, sweepOrphans };@@@ -6,7 +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 { runName, runTime, isRunSchema, 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');@@@ -26,6 +26,11 @@ const tests = [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');}],++ ['the sweep takes only the exact generated shape (a catalog name reaches SQL quoted, and only if it matches whole)', async () => {++ assert.equal(isRunSchema(runName('sdk')), true); assert.equal(isRunSchema(runName('abcdefghijkl')), true);++ const tail = runName('sdk').slice(3);++ for (const bad of [`x"; DROP ROLE ov; --${tail}`, `SDK${tail}`, `a-b${tail}`, `abcdefghijklm${tail}`, tail.slice(1), `sdk${tail}\n`]) assert.equal(isRunSchema(bad), false, bad);++ }],];if (pgAvailable()) {tests.push(['postgresql+pgbouncer: four setups at once each get their own schema, and close() drops them', async () => {
@@ -28,9 +28,14 @@ const runName = (safe, now = Date.now()) => `${safe}_t${now.toString(36)}_${cryp
@@ -28,9 +28,14 @@ const runName = (safe, now = Date.now()) => `${safe}_t${now.toString(36)}_${cryp
/** The time a run's name was made, or null (an older name without one). */
/** 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; }
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();
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. */
+/** Exactly what runName() makes (the service name sanitized to [a-z0-9]{1,12}): only this run shape is ever dropped. */
+const RUN_SCHEMA = /^[a-z0-9]{1,12}_t[0-9a-z]{8,9}_[0-9a-f]{8}_l$/;
+const isRunSchema = (name) => RUN_SCHEMA.test(String(name));
+const quoteIdent = (id) => `"${String(id).replace(/"/g, '""')}"`;
+/** Sweep only leased runs: older naming formats have no reliable end-of-run signal. A catalog name reaches SQL only
+ * after matching the whole generated shape, and quoted. */
async function sweepOrphans(su, { now = Date.now(), maxAge = ORPHAN_MS } = {}) {
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 schemas = (await su.query("SELECT nspname FROM pg_namespace WHERE nspname ~ '^[a-z0-9]{1,12}_t[0-9a-z]{8,9}_[0-9a-f]{8}_l$'")).rows.map((r) => r.nspname).filter(isRunSchema);
const old = schemas.filter((n) => { const t = runTime(n); return t && now - t > maxAge; });
const old = schemas.filter((n) => { const t = runTime(n); return t && now - t > maxAge; });
let dropped = 0;
let dropped = 0;
for (const name of old.slice(0, 200)) {
for (const name of old.slice(0, 200)) {
@@ -39,8 +44,8 @@ async function sweepOrphans(su, { now = Date.now(), maxAge = ORPHAN_MS } = {}) {
@@ -39,8 +44,8 @@ async function sweepOrphans(su, { now = Date.now(), maxAge = ORPHAN_MS } = {}) {
if (!await t.value('SELECT pg_try_advisory_xact_lock($1::bigint)', [leaseKey(name)])) return false;
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`]]);
const busy = await t.value('SELECT count(*) FROM pg_stat_activity WHERE usename = ANY($1)', [[name, `${name}_owner`]]);
if (+busy) return false;
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}`); } }
+ await t.query(`DROP SCHEMA IF EXISTS ${quoteIdent(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 ${quoteIdent(r)}`); await t.query(`DROP ROLE ${quoteIdent(r)}`); } }
return true;
return true;
});
});
if (removed) dropped++;
if (removed) dropped++;
@@ -123,4 +128,4 @@ function createTestValkey({ prefix = 'test' } = {}) {
@@ -123,4 +128,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, runName, runTime, leaseKey, sweepOrphans };
+module.exports = { createTestDb, createTestValkey, pgAvailable, valkeyAvailable, runName, runTime, isRunSchema, leaseKey, sweepOrphans };
@@ -6,7 +6,7 @@ const os = require('os');
@@ -6,7 +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 { runName, runTime, isRunSchema, 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');
@@ -26,6 +26,11 @@ const tests = [
@@ -26,6 +26,11 @@ const tests = [
assert.equal(runTime(n), t0); assert.equal(runTime('sdk_t12345_abcdef01'), null);
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');
assert.equal(runTime(n.slice(0, -2)), null, 'unleased names must never be swept');
}],
}],
+ ['the sweep takes only the exact generated shape (a catalog name reaches SQL quoted, and only if it matches whole)', async () => {
+ assert.equal(isRunSchema(runName('sdk')), true); assert.equal(isRunSchema(runName('abcdefghijkl')), true);
+ const tail = runName('sdk').slice(3);
+ for (const bad of [`x"; DROP ROLE ov; --${tail}`, `SDK${tail}`, `a-b${tail}`, `abcdefghijklm${tail}`, tail.slice(1), `sdk${tail}\n`]) assert.equal(isRunSchema(bad), false, bad);
+ }],
];
];
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 () => {