f89e074dd1
Governed substrate for autonomous agents: scoped identity (passports), audited actions, MCP workspace. Infra IPs and secrets redacted for public release.
675 lines
24 KiB
JavaScript
675 lines
24 KiB
JavaScript
#!/usr/bin/env node
|
|
/**
|
|
* ADR-156 Phase 5A: BDD test runner worker
|
|
*
|
|
* Responsibilities:
|
|
* 1. LISTEN on Postgres channels `bdd.test_run.queued` and
|
|
* `bdd.criterion.claim_requested`. Any notification triggers a poll pass.
|
|
* 2. Poll `bdd_test_runs` logical rows where data.status='queued', claim
|
|
* one, execute its parent `bdd_tests.command` in a strict sandbox with a
|
|
* 30-second timeout.
|
|
* 3. POST the result back to /api/v3/bdd/tests/:id/runs with an internal
|
|
* JWT so the normal server-side state machine (claim/fail/disable)
|
|
* takes over.
|
|
*
|
|
* This is a SCAFFOLD (ADR-156 Phase 5A). It is intentionally conservative:
|
|
* - command must match an allowlist regex (npm|npx|curl|pytest|mcp|claude)
|
|
* - child runs with shell:false, scrubbed env, cwd=/tmp
|
|
* - if the process starts as root, drop to uid/gid 'nobody'
|
|
* - stdout/stderr are tailed to the last 4KB
|
|
* - timeout kills the process tree
|
|
*/
|
|
|
|
import { config } from '../config.js';
|
|
|
|
import { spawn } from 'node:child_process';
|
|
import { promisify } from 'node:util';
|
|
import { execFile } from 'node:child_process';
|
|
import jwt from 'jsonwebtoken';
|
|
import pg from 'pg';
|
|
|
|
import { dbGet, dbAll, dbRun, sqlNow, getAdapter } from '../database/connection.js';
|
|
import { generateBaseId } from '../utils/baseId.js';
|
|
import { logger } from '../utils/logger.js';
|
|
import { gateBash, gateSql, gateMcp } from './bdd-gates.js';
|
|
|
|
const log = logger.child({ worker: 'bdd-runner' });
|
|
|
|
const PORT = config.PORT || 5000;
|
|
const SELF_BASE_URL = process.env.BDD_WORKER_BASE_URL || `http://127.0.0.1:${PORT}`;
|
|
const JWT_SECRET = process.env.JWT_SECRET || config.JWT_SECRET;
|
|
const INTERNAL_USER_ID = parseInt(process.env.BDD_WORKER_USER_ID || '1', 10);
|
|
const POLL_INTERVAL_MS = parseInt(process.env.BDD_WORKER_POLL_MS || '15000', 10);
|
|
const EXEC_TIMEOUT_MS = parseInt(process.env.BDD_WORKER_EXEC_MS || '30000', 10);
|
|
const TAIL_BYTES = 4096;
|
|
const BDD_SPACE_ID = 11;
|
|
|
|
// ADR-156 Appendix C §2.1 — per-runner gates live in ./bdd-gates.js. The old
|
|
// single-binary prefix allowlist is superseded but preserved here as a last-
|
|
// ditch fallback when a test has no runner kind recorded (legacy rows from
|
|
// iteration 1). New rows MUST set data.runner to bash|sql|mcp|http.
|
|
const LEGACY_COMMAND_ALLOWLIST = /^(npm|npx|curl|pytest|mcp|claude)( |$)/;
|
|
|
|
const execFileP = promisify(execFile);
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Table-id resolution (same approach as backend/routes/v3/bdd.js)
|
|
// ---------------------------------------------------------------------------
|
|
const tableIdCache = new Map();
|
|
async function getBddTableId(name) {
|
|
if (tableIdCache.has(name)) return tableIdCache.get(name);
|
|
const row = await dbGet(`
|
|
SELECT ut.id
|
|
FROM universal_tables ut
|
|
JOIN projects p ON ut.project_id = p.id
|
|
WHERE p.space_id = ? AND ut.name = ?
|
|
ORDER BY ut.id ASC
|
|
LIMIT 1
|
|
`, [BDD_SPACE_ID, name]);
|
|
if (row?.id) {
|
|
tableIdCache.set(name, row.id);
|
|
return row.id;
|
|
}
|
|
return null;
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// JWT for self-POST
|
|
// ---------------------------------------------------------------------------
|
|
function issueInternalJwt() {
|
|
return jwt.sign(
|
|
{ id: INTERNAL_USER_ID, email: 'bdd-runner@internal', role: 'system' },
|
|
JWT_SECRET,
|
|
{ expiresIn: '5m' }
|
|
);
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Sandbox exec
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/**
|
|
* Parse a command string into [bin, ...args] without a shell.
|
|
* NOTE: This is a naive whitespace split; commands requiring quotes/pipes are
|
|
* rejected by the allowlist + the presence of shell metacharacters.
|
|
*/
|
|
function splitCommand(cmd) {
|
|
const SHELL_METACHARS = /[|&;`$(){}<>\\]/;
|
|
if (SHELL_METACHARS.test(cmd)) {
|
|
throw new Error('command contains shell metacharacters');
|
|
}
|
|
const parts = cmd.trim().split(/\s+/);
|
|
if (parts.length === 0) throw new Error('empty command');
|
|
return parts;
|
|
}
|
|
|
|
function tailBuffer(buf, n = TAIL_BYTES) {
|
|
if (!Buffer.isBuffer(buf)) buf = Buffer.from(String(buf || ''), 'utf8');
|
|
if (buf.length <= n) return buf.toString('utf8');
|
|
return buf.subarray(buf.length - n).toString('utf8');
|
|
}
|
|
|
|
/**
|
|
* Apply the per-runner security gate (ADR-156 Appendix C §2.1). Returns null
|
|
* if the command is allowed, or an error result if it is not.
|
|
*
|
|
* `test` here is the shape returned by getTestCommand().
|
|
*/
|
|
function applyRunnerGate(test, command) {
|
|
const runner = (test && test.runner) || 'bash';
|
|
const cfg = (test && test.runner_config) || {};
|
|
|
|
let gate;
|
|
switch (runner) {
|
|
case 'bash':
|
|
gate = gateBash(command, cfg.allow || []);
|
|
break;
|
|
case 'sql':
|
|
gate = gateSql(command);
|
|
break;
|
|
case 'mcp': {
|
|
// MCP command format: `tool_name {json_args}`. Extract leading token.
|
|
const toolName = String(command).trim().split(/\s+/, 1)[0] || '';
|
|
gate = gateMcp(toolName, cfg.tools || []);
|
|
break;
|
|
}
|
|
case 'http':
|
|
// No gate yet for http runner — built-in probe validates URL itself.
|
|
return null;
|
|
default:
|
|
// Legacy rows with no runner kind → fall back to the old prefix allowlist
|
|
// so we don't strand iteration-1 fixtures.
|
|
if (!LEGACY_COMMAND_ALLOWLIST.test(String(command).trim())) {
|
|
return {
|
|
status: 'error', exit_code: null, duration_ms: 0,
|
|
stdout_tail: '',
|
|
stderr_tail: `unknown runner "${runner}" and legacy allowlist miss: ${String(command).slice(0, 120)}`,
|
|
};
|
|
}
|
|
return null;
|
|
}
|
|
if (gate && gate.ok === false) {
|
|
return {
|
|
status: 'error', exit_code: null, duration_ms: 0,
|
|
stdout_tail: '',
|
|
stderr_tail: `gate ${runner}: ${gate.reason}`,
|
|
};
|
|
}
|
|
return null;
|
|
}
|
|
|
|
/**
|
|
* Execute a SELECT-only SQL statement against the dedicated read-only role
|
|
* (ADR-156 Appendix C §2.1 "sql runner"). Uses DATABASE_URL_RO if set, else
|
|
* falls back to the worker's POSTGRES_* env with user=bdd_readonly.
|
|
*
|
|
* The gate has already validated the statement is SELECT/WITH-SELECT shaped.
|
|
*/
|
|
async function executeSql(sql) {
|
|
const started = Date.now();
|
|
const { Client } = pg;
|
|
const connConfig = process.env.DATABASE_URL_RO
|
|
? { connectionString: process.env.DATABASE_URL_RO }
|
|
: {
|
|
host: process.env.POSTGRES_HOST || 'localhost',
|
|
port: parseInt(process.env.POSTGRES_PORT || '5432', 10),
|
|
database: process.env.POSTGRES_DB || 'godcrm_prod',
|
|
user: process.env.BDD_RO_USER || 'bdd_readonly',
|
|
password: process.env.BDD_RO_PASSWORD || process.env.POSTGRES_PASSWORD,
|
|
};
|
|
const client = new Client(connConfig);
|
|
try {
|
|
await client.connect();
|
|
// Enforce statement timeout even if the gate missed something long-running.
|
|
await client.query("SET statement_timeout = '5000ms'");
|
|
const res = await client.query(sql);
|
|
const rowsJson = JSON.stringify(res.rows || []);
|
|
return {
|
|
status: 'passed', exit_code: 0,
|
|
duration_ms: Date.now() - started,
|
|
stdout_tail: rowsJson.length > TAIL_BYTES ? rowsJson.slice(-TAIL_BYTES) : rowsJson,
|
|
stderr_tail: '',
|
|
};
|
|
} catch (e) {
|
|
return {
|
|
status: 'failed', exit_code: null,
|
|
duration_ms: Date.now() - started,
|
|
stdout_tail: '',
|
|
stderr_tail: `sql: ${e.message}`.slice(-TAIL_BYTES),
|
|
};
|
|
} finally {
|
|
try { await client.end(); } catch { /* ignore */ }
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Execute an HTTP probe (ADR-156 Phase 5A §2.1 "http runner").
|
|
*
|
|
* Contract:
|
|
* - Fetch test.target with test.method/headers/body and a timeout.
|
|
* - Read the response body as text (we trim both sides for equals checks;
|
|
* this is deliberate so upstream whitespace/trailing-newline drift between
|
|
* servers doesn't flake an otherwise-correct body match).
|
|
* - If test.expected_signal.equals is set → strict string equality check
|
|
* (after .trim() on BOTH sides). Records {expected, actual, matched} in
|
|
* assertion_result.
|
|
* - If test.expected_signal.status is set (number) → exact HTTP status match;
|
|
* otherwise the default "OK range" is 200-299.
|
|
* - status='passed' iff (equals check matched OR no equals provided) AND
|
|
* HTTP status is in the expected range.
|
|
* - On failure, assertion_result is still populated so consumers can debug.
|
|
*/
|
|
async function executeHttp(test) {
|
|
const started = Date.now();
|
|
const target = typeof test.target === 'string' ? test.target.trim() : '';
|
|
if (!target) {
|
|
return {
|
|
status: 'error', exit_code: null, duration_ms: 0,
|
|
stdout_tail: '', stderr_tail: 'http runner: missing target URL',
|
|
assertion_result: { expected: null, actual: null, matched: false },
|
|
};
|
|
}
|
|
|
|
const method = (test.method || 'GET').toString().toUpperCase();
|
|
const headers = (test.headers && typeof test.headers === 'object') ? test.headers : {};
|
|
const body = test.body ?? undefined;
|
|
const expectedSignal = test.expected_signal && typeof test.expected_signal === 'object'
|
|
? test.expected_signal : null;
|
|
|
|
const controller = new AbortController();
|
|
const timer = setTimeout(() => controller.abort(), EXEC_TIMEOUT_MS);
|
|
|
|
let resp;
|
|
try {
|
|
resp = await fetch(target, {
|
|
method,
|
|
headers,
|
|
body: (method === 'GET' || method === 'HEAD') ? undefined : body,
|
|
signal: controller.signal,
|
|
});
|
|
} catch (e) {
|
|
clearTimeout(timer);
|
|
const duration_ms = Date.now() - started;
|
|
const aborted = e.name === 'AbortError' || /abort/i.test(e.message || '');
|
|
return {
|
|
status: aborted ? 'timeout' : 'error',
|
|
exit_code: null,
|
|
duration_ms,
|
|
stdout_tail: '',
|
|
stderr_tail: `http: ${e.message || String(e)}`.slice(-TAIL_BYTES),
|
|
assertion_result: { expected: expectedSignal, actual: null, matched: false },
|
|
};
|
|
}
|
|
clearTimeout(timer);
|
|
|
|
let text = '';
|
|
try { text = await resp.text(); } catch { /* ignore */ }
|
|
|
|
const duration_ms = Date.now() - started;
|
|
const httpStatus = resp.status;
|
|
|
|
// Status range check (default 2xx; override with expected_signal.status).
|
|
let statusOk;
|
|
if (expectedSignal && typeof expectedSignal.status === 'number') {
|
|
statusOk = httpStatus === expectedSignal.status;
|
|
} else {
|
|
statusOk = httpStatus >= 200 && httpStatus <= 299;
|
|
}
|
|
|
|
// Equals check (trim both sides — see contract above).
|
|
let equalsMatched = true; // no equals provided ⇒ vacuously true
|
|
let expected = null;
|
|
let actual = text;
|
|
if (expectedSignal && typeof expectedSignal.equals === 'string') {
|
|
expected = expectedSignal.equals;
|
|
const a = (text || '').trim();
|
|
const e = expected.trim();
|
|
equalsMatched = a === e;
|
|
actual = text;
|
|
}
|
|
|
|
const matched = equalsMatched && statusOk;
|
|
|
|
const assertion_result = {
|
|
expected: expectedSignal ?? null,
|
|
actual: {
|
|
http_status: httpStatus,
|
|
body: text.length > TAIL_BYTES ? text.slice(0, TAIL_BYTES) : text,
|
|
},
|
|
matched,
|
|
};
|
|
|
|
const stdout_tail = text.length > TAIL_BYTES ? text.slice(-TAIL_BYTES) : text;
|
|
const stderr_tail = matched ? '' :
|
|
`http status=${httpStatus} statusOk=${statusOk} equalsMatched=${equalsMatched}`;
|
|
|
|
return {
|
|
status: matched ? 'passed' : 'failed',
|
|
exit_code: matched ? 0 : 1,
|
|
duration_ms,
|
|
stdout_tail,
|
|
stderr_tail,
|
|
assertion_result,
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Execute a sandboxed command. Returns { status, exit_code, duration_ms,
|
|
* stdout_tail, stderr_tail }.
|
|
*/
|
|
async function executeSandboxed(test) {
|
|
// HTTP runner — validate target & expected_signal, fetch, assert.
|
|
// Must be checked BEFORE the command emptiness guard since http tests
|
|
// have no `command` field — they have `target`/`method`/`expected_signal`.
|
|
if (test && test.runner === 'http') {
|
|
return await executeHttp(test);
|
|
}
|
|
|
|
const command = test && test.command;
|
|
if (typeof command !== 'string' || command.trim() === '') {
|
|
return {
|
|
status: 'error', exit_code: null, duration_ms: 0,
|
|
stdout_tail: '', stderr_tail: 'empty command',
|
|
};
|
|
}
|
|
|
|
const denied = applyRunnerGate(test, command);
|
|
if (denied) return denied;
|
|
|
|
// SQL runner — execute via the dedicated read-only Postgres role. Keeps the
|
|
// process isolation for bash intact but avoids forking psql inside a
|
|
// sandboxed shell context.
|
|
if (test.runner === 'sql') {
|
|
return await executeSql(command);
|
|
}
|
|
|
|
let argv;
|
|
try {
|
|
argv = splitCommand(command);
|
|
} catch (e) {
|
|
return {
|
|
status: 'error', exit_code: null, duration_ms: 0,
|
|
stdout_tail: '', stderr_tail: e.message,
|
|
};
|
|
}
|
|
|
|
const [bin, ...args] = argv;
|
|
|
|
const spawnOpts = {
|
|
shell: false,
|
|
cwd: '/tmp',
|
|
env: {
|
|
PATH: process.env.PATH || '/usr/local/bin:/usr/bin:/bin',
|
|
NODE_ENV: process.env.NODE_ENV || 'production',
|
|
},
|
|
detached: true, // put child in its own process group so we can kill the tree
|
|
};
|
|
|
|
// If running as root, try to drop privileges to 'nobody'.
|
|
if (process.getuid && process.getuid() === 0) {
|
|
try {
|
|
const { stdout } = await execFileP('id', ['-u', 'nobody'], { timeout: 2000 });
|
|
const uid = parseInt(stdout.trim(), 10);
|
|
const { stdout: gOut } = await execFileP('id', ['-g', 'nobody'], { timeout: 2000 });
|
|
const gid = parseInt(gOut.trim(), 10);
|
|
if (Number.isFinite(uid) && Number.isFinite(gid)) {
|
|
spawnOpts.uid = uid;
|
|
spawnOpts.gid = gid;
|
|
}
|
|
} catch (e) {
|
|
log.warn({ err: e.message }, 'bdd-runner: could not resolve nobody uid/gid, running as current user');
|
|
}
|
|
}
|
|
|
|
const started = Date.now();
|
|
return await new Promise((resolve) => {
|
|
let child;
|
|
try {
|
|
child = spawn(bin, args, spawnOpts);
|
|
} catch (e) {
|
|
return resolve({
|
|
status: 'error', exit_code: null, duration_ms: Date.now() - started,
|
|
stdout_tail: '', stderr_tail: `spawn failed: ${e.message}`,
|
|
});
|
|
}
|
|
|
|
const stdoutChunks = [];
|
|
const stderrChunks = [];
|
|
let stdoutLen = 0, stderrLen = 0;
|
|
const MAX_BUFFER = 1024 * 1024; // 1MB hard cap per stream
|
|
|
|
child.stdout.on('data', d => {
|
|
if (stdoutLen < MAX_BUFFER) { stdoutChunks.push(d); stdoutLen += d.length; }
|
|
});
|
|
child.stderr.on('data', d => {
|
|
if (stderrLen < MAX_BUFFER) { stderrChunks.push(d); stderrLen += d.length; }
|
|
});
|
|
|
|
let timedOut = false;
|
|
const timer = setTimeout(() => {
|
|
timedOut = true;
|
|
try {
|
|
// Kill the whole process group. `detached:true` above made child the leader.
|
|
if (child.pid) process.kill(-child.pid, 'SIGKILL');
|
|
} catch { /* ignore */ }
|
|
}, EXEC_TIMEOUT_MS);
|
|
|
|
child.on('error', (e) => {
|
|
clearTimeout(timer);
|
|
resolve({
|
|
status: 'error', exit_code: null, duration_ms: Date.now() - started,
|
|
stdout_tail: tailBuffer(Buffer.concat(stdoutChunks)),
|
|
stderr_tail: tailBuffer(Buffer.concat(stderrChunks)) || e.message,
|
|
});
|
|
});
|
|
|
|
child.on('close', (code, signal) => {
|
|
clearTimeout(timer);
|
|
const duration_ms = Date.now() - started;
|
|
const stdout_tail = tailBuffer(Buffer.concat(stdoutChunks));
|
|
const stderr_tail = tailBuffer(Buffer.concat(stderrChunks));
|
|
if (timedOut) {
|
|
return resolve({ status: 'timeout', exit_code: null, duration_ms, stdout_tail, stderr_tail });
|
|
}
|
|
if (code === 0) {
|
|
return resolve({ status: 'passed', exit_code: 0, duration_ms, stdout_tail, stderr_tail });
|
|
}
|
|
return resolve({
|
|
status: 'failed',
|
|
exit_code: typeof code === 'number' ? code : null,
|
|
duration_ms, stdout_tail, stderr_tail,
|
|
});
|
|
});
|
|
});
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Poll & dispatch
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/**
|
|
* Atomically claim one queued run: flip its data.status from 'queued' to
|
|
* 'running' and return the row. Returns null if nothing to do.
|
|
*/
|
|
async function claimOneQueuedRun() {
|
|
const runsTableId = await getBddTableId('bdd_test_runs');
|
|
if (!runsTableId) return null;
|
|
|
|
// UPDATE ... RETURNING the first queued row.
|
|
const adapter = await getAdapter();
|
|
const res = await adapter.query(`
|
|
UPDATE table_rows
|
|
SET data = data || jsonb_build_object('status','running','claimed_at', to_char(now(),'YYYY-MM-DD"T"HH24:MI:SS"Z"')),
|
|
updated_at = NOW()
|
|
WHERE id = (
|
|
SELECT id FROM table_rows
|
|
WHERE table_id = $1 AND data->>'status' = 'queued'
|
|
ORDER BY id ASC
|
|
LIMIT 1
|
|
FOR UPDATE SKIP LOCKED
|
|
)
|
|
RETURNING id, base_id, data
|
|
`, [runsTableId]);
|
|
if (!res.rows || res.rows.length === 0) return null;
|
|
const row = res.rows[0];
|
|
row.data = typeof row.data === 'string' ? JSON.parse(row.data) : (row.data || {});
|
|
return row;
|
|
}
|
|
|
|
async function getTestCommand(testRowId) {
|
|
const testsTableId = await getBddTableId('bdd_tests');
|
|
if (!testsTableId) return null;
|
|
const row = await dbGet(`
|
|
SELECT id, data FROM table_rows WHERE table_id = ? AND id = ?
|
|
`, [testsTableId, testRowId]);
|
|
if (!row) return null;
|
|
const data = typeof row.data === 'string' ? JSON.parse(row.data) : (row.data || {});
|
|
if (data.disabled === true || data.disabled === 'true') return null;
|
|
// ADR-156 Appendix C §1.4: runner kind drives which security gate we apply.
|
|
// Legacy rows (pre-iteration-2) may have a `kind` value like 'curl' instead
|
|
// of a runner kind — fall back to bash in that case.
|
|
const runner = (data.runner || data.kind || 'bash').toString().toLowerCase();
|
|
const runner_config = data.runner_config && typeof data.runner_config === 'object'
|
|
? data.runner_config : {};
|
|
return {
|
|
id: row.id,
|
|
command: data.command,
|
|
runner,
|
|
runner_config,
|
|
review_status: data.review_status || 'pending',
|
|
is_blocking: !!data.is_blocking,
|
|
// HTTP-runner fields (ADR-156 Phase 5A §2.1 http runner). Ignored by other runners.
|
|
target: data.target || null,
|
|
method: data.method || 'GET',
|
|
headers: data.headers && typeof data.headers === 'object' ? data.headers : null,
|
|
body: data.body ?? null,
|
|
expected_signal: data.expected_signal && typeof data.expected_signal === 'object'
|
|
? data.expected_signal : null,
|
|
};
|
|
}
|
|
|
|
async function postRunResult(testRowId, result, triggered_by_id) {
|
|
const token = issueInternalJwt();
|
|
const body = {
|
|
status: result.status,
|
|
exit_code: result.exit_code,
|
|
duration_ms: result.duration_ms,
|
|
stdout_tail: result.stdout_tail,
|
|
stderr_tail: result.stderr_tail,
|
|
assertion_result: result.assertion_result ?? null,
|
|
triggered_by: 'worker',
|
|
triggered_by_id: triggered_by_id || 'bdd-runner',
|
|
};
|
|
const resp = await fetch(`${SELF_BASE_URL}/api/v3/bdd/tests/${testRowId}/runs`, {
|
|
method: 'POST',
|
|
headers: {
|
|
'Content-Type': 'application/json',
|
|
'Authorization': `Bearer ${token}`,
|
|
},
|
|
body: JSON.stringify(body),
|
|
});
|
|
if (!resp.ok) {
|
|
const text = await resp.text().catch(() => '');
|
|
throw new Error(`self-POST ${resp.status}: ${text.slice(0, 200)}`);
|
|
}
|
|
return resp.json();
|
|
}
|
|
|
|
/**
|
|
* After the sandbox has produced a result, finalise the *queued* run row by
|
|
* merging the real status + captured tails into its data column. This is the
|
|
* row we claimed at the start — we update it in-place rather than creating a
|
|
* new one so consumers can pair queue + result by id. We ALSO self-POST so
|
|
* the criterion state-machine in routes/v3/bdd.js fires.
|
|
*/
|
|
async function finaliseQueuedRunRow(runRow, testRowId, result) {
|
|
const runsTableId = await getBddTableId('bdd_test_runs');
|
|
if (!runsTableId) return;
|
|
const patch = {
|
|
status: result.status,
|
|
exit_code: result.exit_code,
|
|
duration_ms: result.duration_ms,
|
|
stdout_tail: result.stdout_tail,
|
|
stderr_tail: result.stderr_tail,
|
|
finished_at: new Date().toISOString(),
|
|
};
|
|
if (result.assertion_result !== undefined) {
|
|
patch.assertion_result = result.assertion_result;
|
|
}
|
|
await dbRun(`
|
|
UPDATE table_rows
|
|
SET data = COALESCE(data,'{}'::jsonb) || ?::jsonb,
|
|
updated_at = ${sqlNow()}
|
|
WHERE id = ?
|
|
`, [JSON.stringify(patch), runRow.id]);
|
|
}
|
|
|
|
let pollInFlight = false;
|
|
async function pollOnce() {
|
|
if (pollInFlight) return;
|
|
pollInFlight = true;
|
|
try {
|
|
// Drain: process queued rows until none remain.
|
|
for (;;) {
|
|
const runRow = await claimOneQueuedRun();
|
|
if (!runRow) break;
|
|
|
|
const testRowId = parseInt(runRow.data?.test_id, 10);
|
|
if (!Number.isFinite(testRowId)) {
|
|
await finaliseQueuedRunRow(runRow, null, {
|
|
status: 'error', exit_code: null, duration_ms: 0,
|
|
stdout_tail: '', stderr_tail: 'invalid test_id on queued row',
|
|
});
|
|
continue;
|
|
}
|
|
|
|
const test = await getTestCommand(testRowId);
|
|
// http runner has no `command` — it uses `target`. Only reject when BOTH
|
|
// are absent, or when the test row itself is missing/disabled.
|
|
const hasWork = test && (test.command || (test.runner === 'http' && test.target));
|
|
if (!hasWork) {
|
|
await finaliseQueuedRunRow(runRow, testRowId, {
|
|
status: 'error', exit_code: null, duration_ms: 0,
|
|
stdout_tail: '', stderr_tail: 'test not found or disabled',
|
|
});
|
|
continue;
|
|
}
|
|
|
|
log.info({
|
|
testRowId, runRowId: runRow.id,
|
|
runner: test.runner,
|
|
cmd: String(test.command || test.target || '').slice(0, 120),
|
|
}, 'bdd-runner: executing');
|
|
const result = await executeSandboxed(test);
|
|
|
|
await finaliseQueuedRunRow(runRow, testRowId, result);
|
|
|
|
try {
|
|
await postRunResult(testRowId, result, runRow.data?.triggered_by_id);
|
|
} catch (e) {
|
|
log.warn({ err: e.message, testRowId, runRowId: runRow.id },
|
|
'bdd-runner: self-POST failed (row was finalised directly in DB)');
|
|
}
|
|
}
|
|
} catch (e) {
|
|
log.error({ err: e.message }, 'bdd-runner: poll pass failed');
|
|
} finally {
|
|
pollInFlight = false;
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// LISTEN/NOTIFY bootstrap
|
|
// ---------------------------------------------------------------------------
|
|
async function startListener() {
|
|
const { Client } = pg;
|
|
const connString = process.env.POSTGRES_URL;
|
|
const connectionConfig = connString
|
|
? { connectionString: connString }
|
|
: {
|
|
host: process.env.POSTGRES_HOST || 'localhost',
|
|
port: parseInt(process.env.POSTGRES_PORT || '5432', 10),
|
|
database: process.env.POSTGRES_DB || 'godcrm_prod',
|
|
user: process.env.POSTGRES_USER || 'godcrm',
|
|
password: process.env.POSTGRES_PASSWORD,
|
|
};
|
|
|
|
const client = new pg.Client(connectionConfig);
|
|
client.on('error', (err) => log.error({ err: err.message }, 'bdd-runner LISTEN client error'));
|
|
client.on('notification', (msg) => {
|
|
log.info({ channel: msg.channel, payload: msg.payload?.slice?.(0, 200) }, 'bdd-runner: NOTIFY');
|
|
// Any signal = kick the poll.
|
|
pollOnce().catch(() => {});
|
|
});
|
|
|
|
await client.connect();
|
|
await client.query('LISTEN "bdd.test_run.queued"');
|
|
await client.query('LISTEN "bdd.criterion.claim_requested"');
|
|
log.info('bdd-runner: LISTEN active on bdd.test_run.queued, bdd.criterion.claim_requested');
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Entrypoint
|
|
// ---------------------------------------------------------------------------
|
|
async function main() {
|
|
log.info({
|
|
pollInterval: POLL_INTERVAL_MS,
|
|
execTimeout: EXEC_TIMEOUT_MS,
|
|
baseUrl: SELF_BASE_URL,
|
|
}, 'bdd-runner: starting');
|
|
|
|
try {
|
|
await startListener();
|
|
} catch (e) {
|
|
log.error({ err: e.message }, 'bdd-runner: failed to start LISTEN (will continue with polling only)');
|
|
}
|
|
|
|
// Immediate catch-up pass, then periodic safety-net poll.
|
|
await pollOnce().catch(() => {});
|
|
setInterval(() => { pollOnce().catch(() => {}); }, POLL_INTERVAL_MS);
|
|
}
|
|
|
|
main().catch((e) => {
|
|
log.error({ err: e.message, stack: e.stack }, 'bdd-runner: fatal');
|
|
process.exit(1);
|
|
});
|