mtgodot-poc/script/playable_fault_proxy_test.mjs

#!/usr/bin/env node
/* STB-DISCONNECT-01 fault proxy regression. Uses only 127.0.0.1 ephemeral ports and a
 * local echo server as the "upstream"; never contacts the test server. Proves that the
 * proxy forwards bytes, closes the connection on `close`, refuses connections for the
 * `unreachable` window and listens again, acts on each request once, ignores foreign
 * run_ids, and stops cleanly on SIGTERM. */
import fs from 'node:fs';
import net from 'node:net';
import os from 'node:os';
import path from 'node:path';
import { spawn, spawnSync } from 'node:child_process';
import { parseArgs } from './playable_fault_proxy.mjs';

const script = path.join(path.dirname(new URL(import.meta.url).pathname), 'playable_fault_proxy.mjs');
const root = fs.mkdtempSync(path.join(os.tmpdir(), 'mt-fault-proxy.'));
let failures = 0;
const check = (ok, label) => { if (ok) console.log(`ok - ${label}`); else { failures += 1; console.error(`not ok - ${label}`); } };
const delay = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

async function until(predicate, timeoutMs, stepMs = 50) {
  const end = Date.now() + timeoutMs;
  while (Date.now() < end) {
    if (await predicate()) return true;
    await delay(stepMs);
  }
  return false;
}

function freePort() {
  return new Promise((resolve) => {
    const server = net.createServer();
    server.listen(0, '127.0.0.1', () => { const { port } = server.address(); server.close(() => resolve(port)); });
  });
}

function connect(port) {
  return new Promise((resolve) => {
    const socket = net.connect(port, '127.0.0.1');
    const state = { socket, closed: false, data: '', error: null };
    socket.on('data', (chunk) => { state.data += chunk.toString(); });
    socket.on('close', () => { state.closed = true; });
    socket.once('connect', () => resolve(state));
    socket.once('error', (error) => { state.error = error.code; resolve(state); });
  });
}

const logRows = (file) => (fs.existsSync(file) ? fs.readFileSync(file, 'utf8').trim().split('\n').filter(Boolean).map((l) => JSON.parse(l)) : []);
const actions = (file, action) => logRows(file).filter((row) => row.action === action);
let usClock = 1000;
const request = (file, type, attempt, extra = {}) => fs.appendFileSync(file, `${JSON.stringify({ monotonic_us: (usClock += 1000), run_id: 'run-a',
  case_id: 'STB-DISCONNECT-01', connection_epoch: 2, stage: 'FAULT', kind: 'fault_request', actor_vid: 0, target_vid: 0,
  payload: { type, attempt, round: 0, timeout_ms: 30000, unreachable_seconds: type === 'unreachable' ? 2 : 0, ...extra } })}\n`);

// Argument contract.
{
  const help = spawnSync(process.execPath, [script, '--help'], { encoding: 'utf8' });
  check(help.status === 0 && /127\.0\.0\.1/.test(help.stdout), '--help documents loopback-only listening');
  for (const [label, args] of [
    ['no routes', ['--events', 'e', '--log', 'l']],
    ['malformed route', ['--route', 'abc', '--events', 'e', '--log', 'l']],
    ['route to itself', ['--route', '4000=127.0.0.1:4000', '--events', 'e', '--log', 'l']],
    ['port out of range', ['--route', '70000=10.0.0.1:1', '--events', 'e', '--log', 'l']],
    ['unknown option', ['--route', '4000=10.0.0.1:1', '--events', 'e', '--log', 'l', '--listen-host', '0.0.0.0']],
  ]) {
    const run = spawnSync(process.execPath, [script, ...args], { encoding: 'utf8' });
    check(run.status === 2, `usage error exits 2: ${label}`);
  }
  check(parseArgs(['--route', '4000=192.0.2.1:11000', '--events', 'e', '--log', 'l']).routes[0].host === '192.0.2.1', 'route parses upstream host');
}

// Live behaviour against a local echo upstream.
const upstream = net.createServer((socket) => socket.pipe(socket));
await new Promise((resolve) => upstream.listen(0, '127.0.0.1', resolve));
const upstreamPort = upstream.address().port;
const listenPort = await freePort();
const events = path.join(root, 'events.jsonl');
const log = path.join(root, 'fault-proxy.jsonl');
const proxy = spawn(process.execPath, [script, '--route', `${listenPort}=127.0.0.1:${upstreamPort}`, '--events', events,
  '--log', log, '--run-id', 'run-a', '--poll-ms', '50'], { stdio: ['ignore', 'ignore', 'pipe'] });
let exited = null;
proxy.on('exit', (code) => { exited = code; });

check(await until(() => actions(log, 'listening').length === 1, 5000), 'proxy logs listening before any fault');
let a = await connect(listenPort);
a.socket.write('ping');
check(await until(() => a.data === 'ping', 2000), 'bytes are forwarded to the upstream and back');

request(events, 'close', 1);
check(await until(() => a.closed, 2000), 'close request drops the proxied connection');
check(actions(log, 'close')[0]?.closed === 1, 'close action records the dropped connection count');
let b = await connect(listenPort);
b.socket.write('again');
check(!b.error && await until(() => b.data === 'again', 2000), 'the listener stays open after close');

// The same line again (same monotonic_us/type/attempt) is not a second fault.
const lines = fs.readFileSync(events, 'utf8').trim().split('\n');
fs.appendFileSync(events, `${lines.at(-1)}\n`);
await delay(300);
check(!b.closed && actions(log, 'close').length === 1, 'a repeated request line is acted on once');

fs.appendFileSync(events, `${JSON.stringify({ monotonic_us: (usClock += 1000), run_id: 'other-run', kind: 'fault_request', payload: { type: 'close', attempt: 9 } })}\n`);
await delay(300);
check(!b.closed && actions(log, 'ignored_foreign_run').length === 1, 'fault requests from another run are ignored');

// Partial line: only acted on once the newline arrives.
const partial = JSON.stringify({ monotonic_us: (usClock += 1000), run_id: 'run-a', kind: 'fault_request',
  payload: { type: 'unreachable', attempt: 1, unreachable_seconds: 2 } });
fs.appendFileSync(events, partial.slice(0, 20));
await delay(300);
check(!b.closed, 'a half-written event line is not parsed early');
const started = Date.now();
fs.appendFileSync(events, `${partial.slice(20)}\n`);
check(await until(() => b.closed, 2000), 'unreachable request drops the connection');
await until(() => actions(log, 'unreachable_start').length === 1, 1000);
const refused = await connect(listenPort);
check(refused.error === 'ECONNREFUSED', `connections are refused during the unreachable window (${refused.error})`);
check(await until(() => actions(log, 'unreachable_end').length === 1, 5000), 'listener comes back after unreachable_seconds');
const elapsed = Date.now() - started;
check(elapsed >= 1900 && elapsed < 4500, `unreachable window lasts about 2 s (${elapsed} ms)`);
const c = await connect(listenPort);
c.socket.write('back');
check(!c.error && await until(() => c.data === 'back', 2000), 'forwarding works again after the window');
c.socket.destroy();

proxy.kill('SIGTERM');
check(await until(() => exited !== null, 5000) && exited === 0, 'SIGTERM stops the proxy with exit 0');
check(actions(log, 'stopped').length === 1, 'stop is logged');
const after = await connect(listenPort);
check(after.error === 'ECONNREFUSED', 'no listener remains after stop');
check(!fs.readFileSync(log, 'utf8').includes('ping'), 'the action log never contains payload bytes');

// A port already in use is a startup error, not a silent run.
{
  const busy = net.createServer();
  await new Promise((resolve) => busy.listen(0, '127.0.0.1', resolve));
  const run = spawnSync(process.execPath, [script, '--route', `${busy.address().port}=127.0.0.1:${upstreamPort}`, '--events', events,
    '--log', path.join(root, 'busy.jsonl')], { encoding: 'utf8', timeout: 10000 });
  check(run.status === 2 && /cannot listen/.test(run.stderr), 'busy listen port exits 2');
  busy.close();
}

upstream.close();
fs.rmSync(root, { recursive: true, force: true });
console.log(failures === 0 ? 'PASS: playable_fault_proxy_test' : `FAIL: playable_fault_proxy_test (${failures})`);
process.exitCode = failures === 0 ? 0 : 1;