mdg-filius/packages/core/test/tcpJava.test.ts

// TCP as Java FILIUS runs it. tools/java-oracle's TcpRun played stories of
// socket calls on tcp-lan's network in Java FILIUS and wrote every frame each
// card saw, and every state each socket went through (tcp-java-<story>.java.json).
// The engine plays the same stories. The frames must come in the same order,
// with the same content, at about the same distance from each other
// (javaRun.ts); each state Java saw must come at about the same time.
//
// Java polled the states every 2 ms from another thread; a state the engine
// enters and leaves within STATE_UNSEEN_MILLISECONDS (CLOSE_WAIT, when the
// passive side closes by itself) Java may not have seen. The free port a
// client draws differs (Java's random, the engine's seed); it is compared as
// "client". Times count from the client's SYN.
//
// The lost* stories lose one frame on its sender's card, as TcpRun's
// loseFirst does in Java: the card shows it, no one receives it.

import { readFileSync } from 'node:fs';
import { dirname, join } from 'node:path';
import { fileURLToPath } from 'node:url';
import { describe, expect, it } from 'vitest';
import { TCP_FLAG, type Id, type NetworkRuntimeEngine, type TcpSegment } from '../src/index.js';
import { loseFirst } from './fixtures.js';
import { expectSameCaptures, GAP_TOLERANCE_MILLISECONDS, gaps, javaFrames, watchedEngine, withClientPortText, type JavaFrame, type SeenFrame } from './javaRun.js';
import type { Scenario } from './scenario.js';

const scenarioDirectory = join(dirname(fileURLToPath(import.meta.url)), 'scenarios');

const STATE_UNSEEN_MILLISECONDS = 20;
// The engine's calls block one after the other: in exchange, the server's
// answer starts when the client's send returns, and the client closes when
// the server's send returns — two cables' delay each, later than in Java.
const STATE_TOLERANCE_MILLISECONDS = 250;
const SERVER = '00000000-0000-0000-0000-000000000101';
const CLIENT = '00000000-0000-0000-0000-000000000102';
const OTHER_CLIENT = '00000000-0000-0000-0000-000000000103';

interface JavaRun {
  frames: Record<string, JavaFrame[]>;
  events: [number, string][];
  states: Record<string, [number, string][]>;
}

const text = (value: string) => new TextEncoder().encode(value);

/**
 * The story as the engine plays it; the node whose socket stands for Java's
 * "client". Java watched the server's socket from ESTABLISHED on, or, with
 * `serverListening`, from the start (LISTEN).
 */
interface Story {
  readonly client: Id;
  readonly serverListening?: boolean;
  readonly play: (engine: NetworkRuntimeEngine, advanceTo: (milliseconds: number) => void) => void;
}

const isSynAck = (segment: TcpSegment) => (segment.flags & TCP_FLAG.synchronize) !== 0 && (segment.flags & TCP_FLAG.acknowledgement) !== 0;
const isFin = (segment: TcpSegment) => (segment.flags & TCP_FLAG.finish) !== 0;

/** Connected, then the client closes half a second later; the server closes by itself. */
function connectAndClose(engine: NetworkRuntimeEngine, advanceTo: (milliseconds: number) => void): void {
  engine.openTcpServerSocket(SERVER, 80);
  advanceTo(500);
  const client = engine.openTcpClientSocket(CLIENT, '192.168.0.10', 80)!;
  engine.connectTcpWithResult(client);
  advanceTo(engine.state.currentTimeMilliseconds + 500);
  engine.closeTcpSocket(client);
}

const STORIES: Record<string, Story> = {
  exchange: {
    client: CLIENT,
    play: (engine, advanceTo) => {
      const listener = engine.openTcpServerSocket(SERVER, 80)!;
      advanceTo(500);
      const client = engine.openTcpClientSocket(CLIENT, '192.168.0.10', 80)!;
      engine.connectTcpWithResult(client);
      const accepted = engine.state.tcpAcceptedSocketIdsByListenerId.get(listener)![0]!;
      engine.sendTcp(client, text('GET /'));
      engine.receiveTcp(accepted);
      engine.sendTcp(accepted, text('x'.repeat(3000)));
      engine.receiveTcp(client);
      engine.closeTcpSocket(client);
      // The server closes a second after it saw CLOSE_WAIT.
      while (engine.state.socketsById.get(accepted)?.tcpState === 'ESTABLISHED') {
        advanceTo(engine.state.currentTimeMilliseconds + 1);
      }
      advanceTo(engine.state.currentTimeMilliseconds + 1_000);
      engine.closeTcpSocket(accepted);
    },
  },
  closedPort: {
    client: OTHER_CLIENT,
    play: (engine, advanceTo) => {
      advanceTo(500);
      const client = engine.openTcpClientSocket(OTHER_CLIENT, '192.168.0.10', 81)!;
      engine.connectTcpWithResult(client);
    },
  },
  lostData: {
    client: CLIENT,
    play: (engine, advanceTo) => {
      engine.openTcpServerSocket(SERVER, 80);
      advanceTo(500);
      const client = engine.openTcpClientSocket(CLIENT, '192.168.0.10', 80)!;
      engine.connectTcpWithResult(client);
      advanceTo(engine.state.currentTimeMilliseconds + 500);
      engine.setGlobalPacketLossEnabled(true);
      engine.sendTcp(client, text('verloren'));
      engine.setGlobalPacketLossEnabled(false);
    },
  },
  lostSynAck: {
    client: CLIENT,
    serverListening: true,
    play: (engine, advanceTo) => {
      loseFirst(engine, SERVER, isSynAck);
      engine.openTcpServerSocket(SERVER, 80);
      advanceTo(500);
      engine.connectTcpWithResult(engine.openTcpClientSocket(CLIENT, '192.168.0.10', 80)!);
    },
  },
  lostFin: {
    client: CLIENT,
    play: (engine, advanceTo) => {
      loseFirst(engine, CLIENT, isFin);
      connectAndClose(engine, advanceTo);
    },
  },
  lostLastFin: {
    client: CLIENT,
    play: (engine, advanceTo) => {
      loseFirst(engine, SERVER, isFin);
      connectAndClose(engine, advanceTo);
    },
  },
};

/** The states of the node's connection socket, from the engine's traces: when each was entered. */
function engineStates(engine: NetworkRuntimeEngine, nodeId: Id, fromEstablished: boolean): [number, string][] {
  const transitions = engine.state.packetTraces
    .filter((trace) => trace.nodeId === nodeId && trace.detail?.startsWith('TCP ') === true && trace.detail.includes(' -> '))
    .map((trace) => [trace.timeMilliseconds, trace.detail!.slice(4).split(' -> ')] as const)
    // The listener's own (CLOSED -> LISTEN and back) are not the connection's.
    .filter(([, [from, to]]) => !(from === 'CLOSED' && to === 'LISTEN') && !(from === 'LISTEN' && to === 'CLOSED'));
  const states: [number, string][] = transitions.map(([time, [, to]]) => [time, to!]);
  if (!fromEstablished) return states;
  const established = states.findIndex(([, state]) => state === 'ESTABLISHED');
  return established < 0 ? [] : states.slice(established);
}

/** Each state Java saw comes in the engine at about the same time; the engine's others are brief. */
function expectSameStates(label: string, actual: [number, string][], expected: [number, string][]) {
  let next = 0;
  const matched = new Set<number>();
  for (const [time, state] of expected) {
    const index = actual.findIndex(([, each], candidate) => candidate >= next && each === state);
    expect(index, `${label}: ${state} (Java at ${time} ms)`).toBeGreaterThanOrEqual(0);
    expect(Math.abs(actual[index]![0] - time), `${label}: time of ${state}`).toBeLessThanOrEqual(STATE_TOLERANCE_MILLISECONDS);
    matched.add(index);
    next = index + 1;
  }
  for (const [index, [time, state]] of actual.entries()) {
    if (matched.has(index)) continue;
    const until = actual[index + 1]?.[0] ?? Number.POSITIVE_INFINITY;
    expect(until - time, `${label}: ${state} at ${time} ms, which Java did not see`).toBeLessThanOrEqual(STATE_UNSEEN_MILLISECONDS);
  }
}

/** Times from the first SYN, the client's free port as "client". */
function normalized(frames: readonly SeenFrame[]): SeenFrame[] {
  const syn = frames.find((frame) => / TCP .* S(A)?(F)?(P)? seq=/.test(frame.text) && !frame.text.includes(' SA'))!;
  return frames.map((frame) => ({
    time: frame.time - syn.time,
    text: frame.text.replace(/:(\d{5})\b/g, (whole, port: string) => (Number(port) >= 49_152 ? ':client' : whole)),
  }));
}

/**
 * Where Java errs, per story: its run corrected, as Java means it.
 *
 * closedPort: connectClientMode sends the SYN in a loop over
 * MAX_SENDEVERSUCHE attempts (with sendeSegment's repeat flag for the
 * second and third), but its timeout branch sets closeSocket, and the loop
 * ends after the first: one SYN, and verbinden fails after one wait. The
 * engine sends the SYN three times, a wait of three round trips each.
 *
 * lostFin, lostLastFin: close waits at most five times (TCP_CLOSING_WAITS),
 * each ended by a segment or after the timeout, and then the socket is gone —
 * but its state stays FIN_WAIT_1, FIN_WAIT_2 or LAST_ACK. The engine shows
 * CLOSED after the last wait.
 */
const JAVA_TIMEOUT_MILLISECONDS = 3 * 2_750;
const JAVA_CLOSING_WAITS = 5;

/** The socket's states, with CLOSED after Java's close loop: from `state`, `waits` more waits of the timeout. */
function closedAfterWaits(states: [number, string][], state: string, waits: number): [number, string][] {
  const from = states.find(([, each]) => each === state)![0];
  return [...states, [from + waits * JAVA_TIMEOUT_MILLISECONDS, 'CLOSED']];
}

const JAVA_BUGS: Record<string, (java: JavaRun) => JavaRun> = {
  closedPort: (java) => {
    const client = OTHER_CLIENT;
    const syn = java.frames[client]!.find((frame) => frame.tcpFlags === 'S')!;
    const waitFrom = java.states.client!.find(([, state]) => state === 'SYN_SENT')![0];
    const retries = [1, 2].map((attempt) => ({ ...syn, time: waitFrom + attempt * JAVA_TIMEOUT_MILLISECONDS }));
    return {
      ...java,
      frames: { ...java.frames, [client]: [...java.frames[client]!, ...retries] },
      states: {
        ...java.states,
        client: java.states.client!.map(([time, state]) =>
          state === 'CLOSED' && time > waitFrom ? [waitFrom + 3 * JAVA_TIMEOUT_MILLISECONDS, state] : [time, state],
        ),
      },
    };
  },
  // The client's FIN is lost: its five waits pass without a segment.
  lostFin: (java) => ({ ...java, states: { ...java.states, client: closedAfterWaits(java.states.client!, 'FIN_WAIT_1', JAVA_CLOSING_WAITS) } }),
  // The server's FIN is lost: the client's first wait ended with the ACK, the server's five pass.
  lostLastFin: (java) => ({
    ...java,
    states: {
      ...java.states,
      client: closedAfterWaits(java.states.client!, 'FIN_WAIT_2', JAVA_CLOSING_WAITS - 1),
      server: closedAfterWaits(java.states.server!, 'LAST_ACK', JAVA_CLOSING_WAITS),
    },
  }),
};

/**
 * Where a story is compared only up to a time (Java's, from its start):
 * after it, Java follows its own error and the engine does not.
 *
 * lostSynAck: Java's client sends one SYN (closedPort above) and gives up a
 * timeout later; the engine's client sends it again then, the server answers
 * it with its SYN+ACK once more, and they connect (engine.test.ts). Compared
 * up to just before the second SYN.
 */
const JAVA_UNTIL: Record<string, (java: JavaRun) => number> = {
  lostSynAck: (java) => java.states.client!.find(([, state]) => state === 'SYN_SENT')![0] + JAVA_TIMEOUT_MILLISECONDS - GAP_TOLERANCE_MILLISECONDS,
};

/**
 * Where each card's capture is compared only up to a time (Java's), beyond
 * JAVA_UNTIL. closedPort: JAVA_BUGS adds the client's two later SYNs to its
 * own card, not where else they show — the switch has forgotten the server
 * by then and floods them to every card. Compared up to the second SYN.
 */
const CAPTURE_UNTIL: Record<string, (java: JavaRun) => number> = {
  ...JAVA_UNTIL,
  closedPort: (java) => java.states.client!.find(([, state]) => state === 'SYN_SENT')![0] + JAVA_TIMEOUT_MILLISECONDS - GAP_TOLERANCE_MILLISECONDS,
};

describe('TCP as Java FILIUS runs it', () => {
  it('corrects Java only where it errs', () => {
    for (const [name, correct] of Object.entries(JAVA_BUGS)) {
      const java = JSON.parse(readFileSync(join(scenarioDirectory, `tcp-java-${name}.java.json`), 'utf8')) as JavaRun;
      expect(JSON.stringify(correct(java)), name).not.toBe(JSON.stringify(java));
    }
  });

  it('cuts only where Java goes on with its error', () => {
    for (const [name, until] of Object.entries(CAPTURE_UNTIL)) {
      const java = JSON.parse(readFileSync(join(scenarioDirectory, `tcp-java-${name}.java.json`), 'utf8')) as JavaRun;
      const cut = until(java);
      const later = [...Object.values(java.frames).flat().map((frame) => frame.time), ...Object.values(java.states).flat().map(([time]) => time)];
      expect(later.some((time) => time >= cut), name).toBe(true);
    }
  });

  const scenario = JSON.parse(readFileSync(join(scenarioDirectory, 'tcp-lan.scenario.json'), 'utf8')) as Scenario;

  for (const [name, story] of Object.entries(STORIES)) {
    it(name, () => {
      const measured = JSON.parse(readFileSync(join(scenarioDirectory, `tcp-java-${name}.java.json`), 'utf8')) as JavaRun;
      const java = JAVA_BUGS[name]?.(measured) ?? measured;
      const { engine, frames } = watchedEngine(scenario);
      const advanceTo = (milliseconds: number) => {
        engine.handle({ type: 'advance', toMilliseconds: milliseconds });
      };
      story.play(engine, advanceTo);
      advanceTo(Math.max(...java.events.map(([time]) => time), ...Object.values(java.states).flatMap((each) => each.map(([time]) => time))) + 5_000);

      // Times from the SYN; with JAVA_UNTIL, only what comes before the cut.
      const javaSyn = java.frames[story.client]!.find((frame) => frame.tcpFlags === 'S')!.time;
      const engineSyn = frames().find((frame) => / TCP .* S seq=/.test(frame.text))!.time;
      const until = (JAVA_UNTIL[name]?.(measured) ?? Number.POSITIVE_INFINITY) - javaSyn;
      const before = <T extends { time: number } | [number, string]>(items: T[]): T[] =>
        items.filter((item) => (Array.isArray(item) ? item[0] : item.time) < until);
      const expected = before(normalized(javaFrames(java.frames)));
      const actual = before(normalized(frames()));
      const expectedSteps = gaps(expected);
      const actualSteps = gaps(actual);
      for (let index = 0; index < Math.max(expectedSteps.length, actualSteps.length); index += 1) {
        expect(actualSteps[index]?.text, `frame #${index + 1}`).toBe(expectedSteps[index]?.text);
        expect(Math.abs(actualSteps[index]!.gap - expectedSteps[index]!.gap), `gap before frame #${index + 1}`).toBeLessThanOrEqual(
          GAP_TOLERANCE_MILLISECONDS,
        );
      }

      expectSameCaptures(engine, scenario, java.frames, { text: withClientPortText, until: CAPTURE_UNTIL[name]?.(measured) });

      const shift = (states: [number, string][], syn: number): [number, string][] => before(states.map(([time, state]) => [time - syn, state]));
      // From SYN_SENT: Java's first CLOSED is when the socket was made.
      const javaClient = java.states.client!.filter(([, state], index) => !(index === 0 && state === 'CLOSED'));
      expectSameStates('client', shift(engineStates(engine, story.client, false), engineSyn), shift(javaClient, javaSyn));
      if (java.states.server !== undefined) {
        // A listening server from its first change: Java's first LISTEN is when it began to listen (its
        // poll may have seen the socket CLOSED before).
        const listening = java.states.server.findIndex(([, state]) => state === 'LISTEN');
        const javaServer = story.serverListening === true ? java.states.server.slice(listening + 1) : java.states.server;
        expectSameStates('server', shift(engineStates(engine, SERVER, story.serverListening !== true), engineSyn), shift(javaServer, javaSyn));
      }
    });
  }
});