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.
//

import { readFileSync } from 'node:fs';
import { dirname, join } from 'node:path';
import { fileURLToPath } from 'node:url';
import { describe, expect, it } from 'vitest';
import type { Id, NetworkRuntimeEngine } from '../src/index.js';
import { GAP_TOLERANCE_MILLISECONDS, gaps, javaFrames, watchedEngine, 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 nodes whose socket stands for Java's "client" and "server". */
interface Story {
  readonly client: Id;
  readonly play: (engine: NetworkRuntimeEngine, advanceTo: (milliseconds: number) => void) => void;
}

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);
    },
  },
};

/** 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.
 */
const JAVA_TIMEOUT_MILLISECONDS = 3 * 2_750;
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],
        ),
      },
    };
  },
};

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));
    }
  });

  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);

      const expected = normalized(javaFrames(java.frames));
      const actual = 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,
        );
      }

      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 shift = (states: [number, string][], syn: number): [number, string][] => 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) {
        expectSameStates('server', shift(engineStates(engine, SERVER, true), engineSyn), shift(java.states.server, javaSyn));
      }
    });
  }
});