path: root/packages/meshbay-hub/tests/harness/reconnect_race_harness.mjs

// Run the SHIPPED download over the SHIPPED reconnect, against a node that
// refuses what comes before the handshake, and see whether the download lives.
//
// transport.js is executed whole — connect(), _reconnectLoop, _sendAndWait,
// _dispatch and the framing are the shipped code — and the pipeline is lifted
// out of file-utils.js as text. What is modelled is the environment: a clock,
// a peer connection that opens its channel some time after the answer, a hub
// that answers the offer and the token, and a node that answers anything but
// the opening messages with "Handshake required" until the proof is in, as
// webrtc/dispatch.py does. Cryptography is not under test here: the checks
// that would refuse a peer are stubbed to accept.
//
// The scenario is the one that failed live: the node stops answering, the
// chunks in flight time out, and their retries come due while the reconnect's
// new channel is open and its handshake is not finished.
//
// Usage: node reconnect_race_harness.mjs <file-utils.js> <transport sources> <json>
import { readFileSync } from 'fs';
import { delimiter } from 'path';

const utils = readFileSync(process.argv[2], 'utf8');
const transportSrc = process.argv[3].split(delimiter)
  .map((p) => readFileSync(p, 'utf8')).join('\n');
const {
  chunks = 400,
  chunkMs = 100,
  hangAt = 5000,          // the node stops answering the first connection
  failAfter = 30200,      // ...and the peer connection reports `failed` this much later
  openMs = 100,           // answer → channel open
  rttMs = 300,            // one request → its reply, during the handshake
  lateCloseMs = 0,        // >0: the replaced channel's close event fires this late
  runUntil = 300000,
  stream = false,         // a film playing instead of a download (see below)
} = JSON.parse(process.argv[4] || '{}');

for (const level of ['log', 'warn', 'error', 'info', 'debug']) {
  console[level] = (...args) => process.stderr.write(args.join(' ') + '\n');
}
process.on('unhandledRejection', () => {});

// ── The clock ────────────────────────────────────────────────────────────────

let now = 0;
let nextId = 1;
let timers = [];
globalThis.setTimeout = (fn, ms) => {
  const t = { at: now + (ms || 0), fn, id: nextId++ };
  timers.push(t);
  return t.id;
};
globalThis.setInterval = (fn, ms) => {
  const t = { at: now + ms, fn, id: nextId++, every: ms };
  timers.push(t);
  return t.id;
};
globalThis.clearTimeout = (id) => { timers = timers.filter((t) => t.id !== id); };
globalThis.clearInterval = globalThis.clearTimeout;
Date.now = () => now;
const flush = async () => { for (let i = 0; i < 50; i++) await Promise.resolve(); };
async function run(until) {
  await flush();
  while (timers.length) {
    const due = timers.reduce((a, b) => (b.at < a.at ? b : a));
    if (due.at > until) break;
    if (due.every) due.at += due.every;
    else timers = timers.filter((t) => t !== due);
    now = Math.max(now, due.every ? due.at - due.every : due.at);
    due.fn();
    await flush();
  }
  now = until;
}

// ── The page around the transport ────────────────────────────────────────────

globalThis.window = globalThis;
globalThis.addEventListener = () => {};
globalThis.removeEventListener = () => {};
globalThis.location = { hash: '' };
globalThis.document = {
  addEventListener() {}, removeEventListener() {}, visibilityState: 'visible',
};
globalThis.localStorage = { getItem: () => null, setItem() {}, removeItem() {} };

const b64 = (u8) => Buffer.from(u8).toString('base64');
window.MeshBayCrypto = {
  b64encode: (u8) => b64(u8),
  b64decode: (s) => new Uint8Array(Buffer.from(String(s), 'base64')),
  verifyNodeSignature: async () => true,
  challengeTranscript: () => new Uint8Array(1),
  webrtcBinding: () => new Uint8Array(1),
  handshakeProof: async () => new Uint8Array([7]),
  handshakeTranscript: () => new Uint8Array(1),
  constantTimeEqual: () => true,
  openGroup: async () => window.__codec.msgpack_encode({ chat_epoch: 1 }),
  decryptChunkBin: async (_k, _f, index) => new Uint8Array([index & 255]),
};

const SDP = 'v=0\r\na=fingerprint:sha-256 AA:BB\r\n';
globalThis.fetch = async (url) => ({
  ok: true, status: 200, headers: { get: () => null },
  json: async () => (url.endsWith('/mnp-token') ? { mnp_token: 't' } : { sdp: SDP }),
});

// ── The node ─────────────────────────────────────────────────────────────────

const OPENING = new Set(['handshake', 'handshake_response', 'keypair_bundle_fetch',
  'gek_bundle_fetch', 'join_request', 'device_add_request', 'device_hello']);
let refusedBeforeHandshake = 0;
const refusedTypes = new Set();
let served = 0;
const pcs = [];

class NodeSession {
  constructor(channel) { this.channel = channel; this.authed = false; this.hung = false; }
  reply(obj, ms) {
    setTimeout(() => {
      if (this.hung || this.channel.readyState !== 'open') return;
      const body = window.__codec.msgpack_encode(obj);
      const frame = new Uint8Array(4 + body.byteLength);
      new DataView(frame.buffer).setUint32(0, body.byteLength, false);
      frame.set(body, 4);
      this.channel.onmessage({ data: frame.buffer });
    }, ms);
  }
  receive(frame) {
    if (this.hung) return;
    const msg = window.__codec.msgpack_decode(frame.slice(4));
    const req_id = msg.req_id;
    if (!this.authed && !OPENING.has(msg.type)) {
      refusedBeforeHandshake += 1;
      refusedTypes.add(msg.type);
      return this.reply({ type: 'error', detail: 'Handshake required', req_id }, 1);
    }
    switch (msg.type) {
      case 'handshake':
        return this.reply({ type: 'handshake_challenge', v: '99.0', v_min: '0.0',
          nonce: b64(new Uint8Array(32)), node_pk: 'pk', sig: 's', req_id }, rttMs);
      case 'handshake_response':
        setTimeout(() => { this.authed = true; }, rttMs);
        return this.reply({ type: 'handshake_ack', proof: 'AA==', node_pk: 'pk', sig: 's',
          nonce: 'n', ct: 'c', req_id }, rttMs);
      case 'device_hello':
        return this.reply({ type: 'device_hello_ack', req_id }, rttMs);
      case 'file_req':
        served += 1;
        return this.reply({ type: 'file_chunk', file_id: msg.file_id,
          chunk_index: msg.chunk_index, nonce: 1, ct: new Uint8Array([1]), req_id }, chunkMs);
      default:
        return undefined;
    }
  }
}

class FakeChannel {
  constructor() { this.readyState = 'connecting'; this.binaryType = 'arraybuffer'; }
  send(frame) {
    if (this.readyState !== 'open') throw new Error('InvalidStateError');
    this.session.receive(frame);
  }
  close() {
    if (this.readyState === 'closed') return;
    this.readyState = 'closed';
    if (this.onclose) setTimeout(() => this.onclose({}), lateCloseMs);
  }
}

class FakePC {
  constructor() {
    this.connectionState = 'new';
    this.iceConnectionState = 'new';
    this.iceGatheringState = 'complete';
    pcs.push(this);
  }
  createDataChannel() { this.channel = new FakeChannel(); return this.channel; }
  async createOffer() { return { type: 'offer', sdp: SDP }; }
  async setLocalDescription(d) { this.localDescription = d; }
  async setRemoteDescription() {
    setTimeout(() => {
      if (this.connectionState === 'closed') return;
      this.channel.session = new NodeSession(this.channel);
      this.channel.readyState = 'open';
      this.connectionState = 'connected';
      if (this.channel.onopen) this.channel.onopen();
    }, openMs);
  }
  close() {
    this.connectionState = 'closed';
    if (this.channel) this.channel.close();
  }
}
globalThis.RTCPeerConnection = FakePC;

// ── The shipped transport and pipeline ───────────────────────────────────────

new Function(`${transportSrc}\nwindow.__codec = { msgpack_encode, msgpack_decode };`)();
const Transport = window.MeshBayTransport;

const constant = (name) => utils.match(new RegExp(`^const ${name} = [^;]+;`, 'm'))[0];
const from = utils.indexOf('const CHUNK_RETRY_ATTEMPTS');
const to = utils.indexOf('async function downloadEntry');
const { pipelinedDownload } = new Function('window', 't', `
  ${constant('PIPELINE_WINDOW')}
  ${constant('LANE_MIN_CHUNKS')}
  ${utils.slice(from, utils.lastIndexOf('/**', to))}
  return { pipelinedDownload };`)(window, (key) => key);

const identity = { pkEdB64: 'ed', signAs: async () => 'sig', native: false };
const main = new Transport('https://hub.invalid', 'token');
let outcome = 'pending';
let reconnected = 0;
main.addReconnectListener(() => { reconnected += 1; });
const written = [];
const writable = { write: async (bytes) => { written.push(bytes[0]); } };

// A film playing (`stream`): the player goes on granting the node room for
// segments as it plays from its buffer, and on telling it what it sees, through
// the connection loss and the reconnect, as video-player.js does. A refusal of
// either is a node error the player shows in place of the film.
const opened = main.connect('node', 'token', 'group', new Uint8Array(32), identity, null,
                            'u', 'uid', undefined, undefined, '', 'pk');
if (stream) {
  opened.then(() => {
    main.requestStream('f'.repeat(64));
    setInterval(() => {
      main.grantStreamCredit(1);
      main.sendStreamDiag({ buffered: 1 });
    }, 50);
    outcome = 'done';
  }, (err) => { outcome = `failed: ${err.message}`; });
} else {
  opened
    .then(() => pipelinedDownload(main, 'gek', 'f'.repeat(64), chunks, null, writable, null, ''))
    .then(() => { outcome = 'done'; }, (err) => { outcome = `failed: ${err.message}`; });
}

await run(hangAt);
pcs[0].channel.session.hung = true;
await run(hangAt + failAfter);
pcs[0].connectionState = 'failed';
pcs[0].onconnectionstatechange();
await run(runUntil);

process.stdout.write(JSON.stringify({
  outcome, reconnected, refusedBeforeHandshake, refusedTypes: [...refusedTypes], served,
  written: written.length,
  inOrder: written.every((b, i) => b === (i & 255)),
  connections: pcs.length,
}));