mtgodot-poc/script/playable_fault_proxy.mjs

#!/usr/bin/env node
/* STB-DISCONNECT-01 local transport-fault proxy for ONE test client connection.
 *
 *   playable_fault_proxy.mjs --route LISTEN_PORT=UPSTREAM_HOST:UPSTREAM_PORT [--route ...]
 *     --events events.jsonl --log fault-proxy.jsonl [--run-id RUN_ID] [--poll-ms 200]
 *
 * - Listens on 127.0.0.1 only and forwards bytes to the upstream test server. The
 *   scenario's server.* points the client at these loopback ports; nothing else on
 *   the machine is touched: no routes, firewall rules, other clients or production.
 * - Tails the client's events.jsonl and acts on each `fault_request` exactly once:
 *     close        destroy the proxied connections ("connection closed")
 *     unreachable  destroy them AND stop listening for payload.unreachable_seconds
 *                  (capped at 20 s) so reconnect attempts are refused, then listen again
 * - Every action is appended to --log as JSON (wall_ms, action, type, attempt, ...).
 *   The client, not this proxy, decides whether recovery happened.
 * - Exits 0 on SIGTERM/SIGINT after closing its sockets. It never signals any process.
 *
 * Known limit: a server-side warp that hands out a direct game address bypasses the proxy.
 * exit: 0 stopped, 2 usage error or a listen port could not be opened. */
import fs from 'node:fs';
import net from 'node:net';
import path from 'node:path';
import { fileURLToPath } from 'node:url';

const LISTEN_HOST = '127.0.0.1';
const MAX_UNREACHABLE_S = 20;

function usage(stream = process.stderr) {
  stream.write('usage: playable_fault_proxy.mjs --route LISTEN_PORT=HOST:PORT [--route ...] --events events.jsonl --log fault-proxy.jsonl [--run-id ID] [--poll-ms 200]\n'
    + '  listens on 127.0.0.1 only; acts on fault_request events (close | unreachable)\n');
}

const isPort = (value) => Number.isInteger(value) && value >= 1 && value <= 65535;

export function parseArgs(argv) {
  const options = { routes: [], poll_ms: 200 };
  for (let i = 0; i < argv.length; i += 1) {
    const arg = argv[i];
    if (arg === '--help') { options.help = true; continue; }
    const value = argv[i + 1];
    if (value === undefined) throw new Error(`missing value for ${arg}`);
    i += 1;
    if (arg === '--route') {
      const match = /^(\d+)=([A-Za-z0-9.:-]+):(\d+)$/.exec(value);
      if (!match) throw new Error(`bad --route ${value}`);
      const route = { listen: Number(match[1]), host: match[2], port: Number(match[3]) };
      if (!isPort(route.listen) || !isPort(route.port)) throw new Error(`bad port in --route ${value}`);
      if (['127.0.0.1', 'localhost', '::1'].includes(route.host) && route.port === route.listen) throw new Error(`route ${value} loops back to itself`);
      if (options.routes.some((r) => r.listen === route.listen)) throw new Error(`duplicate listen port ${route.listen}`);
      options.routes.push(route);
    } else if (arg === '--events') options.events = value;
    else if (arg === '--log') options.log = value;
    else if (arg === '--run-id') options.run_id = value;
    else if (arg === '--poll-ms') options.poll_ms = Number(value);
    else throw new Error(`unknown option ${arg}`);
  }
  if (!options.help && (options.routes.length === 0 || !options.events || !options.log
      || !Number.isInteger(options.poll_ms) || options.poll_ms < 20)) throw new Error('--route, --events and --log are required');
  return options;
}

/** Incremental JSONL reader: returns complete new lines since the last call. */
export function tailer(file) {
  let offset = 0;
  let partial = '';
  return () => {
    let size;
    try { size = fs.statSync(file).size; } catch { return []; }
    if (size < offset) { offset = 0; partial = ''; }
    if (size === offset) return [];
    const fd = fs.openSync(file, 'r');
    const buffer = Buffer.alloc(size - offset);
    fs.readSync(fd, buffer, 0, buffer.length, offset);
    fs.closeSync(fd);
    offset = size;
    const text = partial + buffer.toString('utf8');
    const lines = text.split('\n');
    partial = lines.pop();
    return lines.filter((line) => line.trim() !== '');
  };
}

export class FaultProxy {
  constructor(options) {
    this.options = options;
    this.servers = new Map();
    this.pairs = new Set();
    this.seen = new Set();
    this.unreachableTimer = null;
    this.stopped = false;
    this.read = tailer(options.events);
  }

  log(action, detail = {}) {
    fs.appendFileSync(this.options.log, `${JSON.stringify({ wall_ms: Date.now(), action, ...detail })}\n`);
  }

  listen(route) {
    return new Promise((resolve, reject) => {
      const server = net.createServer((client) => this.accept(route, client));
      server.once('error', reject);
      server.listen(route.listen, LISTEN_HOST, () => {
        server.off('error', reject);
        this.servers.set(route.listen, server);
        resolve();
      });
    });
  }

  accept(route, client) {
    const upstream = net.connect(route.port, route.host);
    const pair = { client, upstream };
    this.pairs.add(pair);
    const drop = () => {
      if (!this.pairs.delete(pair)) return;
      client.destroy();
      upstream.destroy();
    };
    client.on('error', drop).on('close', drop);
    upstream.on('error', drop).on('close', drop);
    client.pipe(upstream);
    upstream.pipe(client);
    this.log('accepted', { listen_port: route.listen, open: this.pairs.size });
  }

  dropAll() {
    const count = this.pairs.size;
    for (const pair of [...this.pairs]) {
      this.pairs.delete(pair);
      pair.client.destroy();
      pair.upstream.destroy();
    }
    return count;
  }

  /** Stops accepting, drops open connections (server.close waits for them), returns the drop count. */
  async closeListeners() {
    const closing = [...this.servers.values()].map((server) => new Promise((resolve) => server.close(resolve)));
    this.servers.clear();
    const dropped = this.dropAll();
    await Promise.all(closing);
    return dropped;
  }

  async start() {
    for (const route of this.options.routes) await this.listen(route);
    this.log('listening', { listen_ports: this.options.routes.map((r) => r.listen) });
    this.poller = setInterval(() => { this.poll().catch((error) => this.log('error', { reason: String(error.message).slice(0, 120) })); }, this.options.poll_ms);
  }

  async poll() {
    // setInterval does not wait for an async fault; never run two polls at once.
    if (this.polling) return;
    this.polling = true;
    try { await this.drain(); } finally { this.polling = false; }
  }

  async drain() {
    for (const line of this.read()) {
      let event;
      try { event = JSON.parse(line); } catch { continue; }
      if (event?.kind !== 'fault_request') continue;
      if (this.options.run_id && event.run_id !== this.options.run_id) { this.log('ignored_foreign_run', {}); continue; }
      const payload = event.payload ?? {};
      const key = `${event.monotonic_us}:${payload.type}:${payload.attempt}`;
      if (this.seen.has(key)) continue;
      this.seen.add(key);
      await this.fault(payload);
    }
  }

  async fault(payload) {
    const detail = { type: payload.type, attempt: payload.attempt, round: payload.round };
    if (payload.type === 'close') {
      this.log('close', { ...detail, closed: this.dropAll() });
    } else if (payload.type === 'unreachable') {
      if (this.unreachableTimer) { this.log('rejected', { ...detail, reason: 'unreachable window already active' }); return; }
      const seconds = Math.min(MAX_UNREACHABLE_S, Math.max(1, Number.parseInt(payload.unreachable_seconds, 10) || 1));
      const closed = await this.closeListeners();
      this.log('unreachable_start', { ...detail, closed, seconds });
      this.unreachableTimer = setTimeout(async () => {
        this.unreachableTimer = null;
        if (this.stopped) return;
        try {
          for (const route of this.options.routes) await this.listen(route);
          this.log('unreachable_end', detail);
        } catch (error) {
          this.log('error', { ...detail, reason: `relisten failed: ${error.code || error.message}` });
        }
      }, seconds * 1000);
    } else {
      this.log('rejected', { ...detail, reason: 'unknown fault type' });
    }
  }

  async stop() {
    if (this.stopped) return;
    this.stopped = true;
    clearInterval(this.poller);
    if (this.unreachableTimer) clearTimeout(this.unreachableTimer);
    this.dropAll();
    await this.closeListeners();
    this.log('stopped', {});
  }
}

async function main() {
  let options;
  try { options = parseArgs(process.argv.slice(2)); } catch (error) { usage(); console.error(error.message); return 2; }
  if (options.help) { usage(process.stdout); return 0; }
  const proxy = new FaultProxy(options);
  try { await proxy.start(); } catch (error) {
    console.error(`FAULT PROXY: cannot listen (${error.code || error.message})`);
    await proxy.stop();
    return 2;
  }
  await new Promise((resolve) => {
    process.once('SIGTERM', resolve);
    process.once('SIGINT', resolve);
  });
  await proxy.stop();
  return 0;
}

if (process.argv[1] && path.resolve(process.argv[1]) === fileURLToPath(import.meta.url)) process.exitCode = await main();