feat(core): TCP in the simulation core
engine.ts ports Swift's "Java-compatible TCP sockets" in one piece, since half a state machine could not be checked: server and client sockets (a free port from Java's random numbers on their own seed), the three-way handshake with its result (connected, invalid socket, unreachable, timed out), segments of 1460 bytes, messages up to the push, a timeout of 3 × 2.5 s with at most three attempts, closing through every Java state including TIME_WAIT (2.5 s), sequence numbers modulo 2^32. Three scenarios match the Swift engine trace for trace: a LAN with data over three segments, closing from both ends, a closed port and an unreachable host; two networks through a router with a firewall rule, data lost until the connection closes, and an accepted socket that does not hold the port of its closed listener; CLOSING, reached by a lost FIN. The engine delivers a FIN within the same close, so both ends close at once only with packet loss; an empty message does not arrive in Swift (it does not advance the sequence number), nor here. Swift's four TCP engine tests are ported, with tests for stopping, a foreign SYN+ACK at a listener and the flag names. 17 mutation probes, 16 caught; the other (delivered, not waited, not connected gives unreachable) cannot be reached through Swift's engine, which never sends a reset and always acknowledges correctly. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -16,9 +16,10 @@
|
||||
//
|
||||
// "UDP sockets and Router RIP" (lines ~4472–5187, the
|
||||
// UDP sockets; RIP not yet)
|
||||
// "Java-compatible TCP sockets" (lines ~5187–5913)
|
||||
//
|
||||
// Not ported yet — each place throws NotPortedError instead of guessing:
|
||||
// gateway NAT, TCP delivery, RIP routing, remote links. DHCP is not in the
|
||||
// gateway NAT, RIP routing, remote links. DHCP is not in the
|
||||
// snapshot yet, so Swift's DHCP handling of a datagram has nothing to do.
|
||||
|
||||
import { compareIds, type Id } from '../ids.js';
|
||||
@@ -51,6 +52,8 @@ import {
|
||||
type IcmpMessage,
|
||||
type IcmpMessageKind,
|
||||
type IPv4Packet,
|
||||
type TcpSegment,
|
||||
type TcpSocketState,
|
||||
type UdpDatagram,
|
||||
} from './packets.js';
|
||||
import { JavaCompatibleRandom } from './random.js';
|
||||
@@ -232,7 +235,7 @@ export type RuntimePhase = 'stopped' | 'running';
|
||||
|
||||
export type SocketProtocol = 'udp' | 'tcp';
|
||||
|
||||
/** Swift's TopologyRuntimeSocketRecord (UDP so far). */
|
||||
/** Swift's TopologyRuntimeSocketRecord. */
|
||||
export interface SocketRecord {
|
||||
readonly id: Id;
|
||||
readonly nodeId: Id;
|
||||
@@ -242,6 +245,39 @@ export interface SocketRecord {
|
||||
readonly localPort: number;
|
||||
remoteIPAddress: string | undefined;
|
||||
remotePort: number | undefined;
|
||||
/** TCP only. */
|
||||
tcpState: TcpSocketState | undefined;
|
||||
/** A socket a TCP listener accepted: that listener. */
|
||||
readonly parentListenerSocketId: Id | undefined;
|
||||
}
|
||||
|
||||
/** Swift's TopologyRuntimeTCPSessionRecord; sequence numbers are unsigned 32-bit. */
|
||||
export interface TcpSession {
|
||||
readonly id: Id;
|
||||
readonly socketId: Id;
|
||||
state: TcpSocketState;
|
||||
sendAttempts: number;
|
||||
nextSendSequenceNumber: number;
|
||||
lastSentSequenceNumber: number;
|
||||
remoteSequenceNumber: number;
|
||||
lastAcknowledgedSequenceNumber: number | undefined;
|
||||
lastSentSegment: TcpSegment | undefined;
|
||||
receiveAssembly: Uint8Array;
|
||||
receivedMessages: Uint8Array[];
|
||||
}
|
||||
|
||||
export type TcpConnectionResult = 'connected' | 'invalidSocket' | 'unreachable' | 'timedOut';
|
||||
|
||||
export const TCP_MAXIMUM_SEGMENT_SIZE = 1_460;
|
||||
export const TCP_MAXIMUM_SEND_ATTEMPTS = 3;
|
||||
export const TCP_ROUND_TRIP_TIME_MILLISECONDS = 2_500;
|
||||
export const TCP_DEFAULT_TIMEOUT_MILLISECONDS = TCP_MAXIMUM_SEND_ATTEMPTS * TCP_ROUND_TRIP_TIME_MILLISECONDS;
|
||||
export const TCP_EPHEMERAL_PORT_LOWER_BOUND = 49_152;
|
||||
export const TCP_EPHEMERAL_PORT_UPPER_BOUND_EXCLUSIVE = 65_535;
|
||||
|
||||
/** Addition modulo 2^32, as Swift's `&+` on UInt32. */
|
||||
function uint32Add(value: number, addend: number): number {
|
||||
return (value + addend) >>> 0;
|
||||
}
|
||||
|
||||
/** A datagram waiting in a UDP socket's queue. */
|
||||
@@ -281,7 +317,10 @@ export interface RuntimeState {
|
||||
firewallDecisions: FirewallDecision[];
|
||||
socketsById: Map<Id, SocketRecord>;
|
||||
udpReceiveQueuesBySocketId: Map<Id, UdpReceivedDatagram[]>;
|
||||
tcpSessionsById: Map<Id, TcpSession>;
|
||||
tcpAcceptedSocketIdsByListenerId: Map<Id, Id[]>;
|
||||
nextSocketSequenceNumber: number;
|
||||
nextTcpInitialSequenceValue: number;
|
||||
nextEventSequenceNumber: number;
|
||||
nextFrameIdentity: number;
|
||||
nextPacketIdentity: number;
|
||||
@@ -308,7 +347,10 @@ function initialState(seed: bigint): RuntimeState {
|
||||
firewallDecisions: [],
|
||||
socketsById: new Map(),
|
||||
udpReceiveQueuesBySocketId: new Map(),
|
||||
tcpSessionsById: new Map(),
|
||||
tcpAcceptedSocketIdsByListenerId: new Map(),
|
||||
nextSocketSequenceNumber: 1,
|
||||
nextTcpInitialSequenceValue: 1,
|
||||
nextEventSequenceNumber: 0,
|
||||
nextFrameIdentity: 1,
|
||||
nextPacketIdentity: 1,
|
||||
@@ -443,6 +485,8 @@ export class NetworkRuntimeEngine {
|
||||
state.icmpObservationsByNodeId.clear();
|
||||
state.socketsById.clear();
|
||||
state.udpReceiveQueuesBySocketId.clear();
|
||||
state.tcpSessionsById.clear();
|
||||
state.tcpAcceptedSocketIdsByListenerId.clear();
|
||||
state.firewallDecisions = [];
|
||||
return [this.output({ type: 'stopped' })];
|
||||
}
|
||||
@@ -617,7 +661,8 @@ export class NetworkRuntimeEngine {
|
||||
case 'dhcpTimeout':
|
||||
throw new NotPortedError('DHCP client');
|
||||
case 'tcpTimeout':
|
||||
throw new NotPortedError('TCP timeouts');
|
||||
this.handleTcpTimeout(kind.sessionId);
|
||||
return;
|
||||
case 'remoteLinkDelivery':
|
||||
case 'lanRemoteLinkTransmission':
|
||||
throw new NotPortedError('remote links');
|
||||
@@ -676,8 +721,7 @@ export class NetworkRuntimeEngine {
|
||||
) {
|
||||
return undefined;
|
||||
}
|
||||
const id = deterministicSocketId(state.nextSocketSequenceNumber);
|
||||
state.nextSocketSequenceNumber += 1;
|
||||
const id = this.allocateSocketId();
|
||||
state.socketsById.set(id, {
|
||||
id,
|
||||
nodeId,
|
||||
@@ -686,6 +730,8 @@ export class NetworkRuntimeEngine {
|
||||
localPort: reservedPort,
|
||||
remoteIPAddress: options.remoteIPAddress,
|
||||
remotePort: options.remotePort,
|
||||
tcpState: undefined,
|
||||
parentListenerSocketId: undefined,
|
||||
});
|
||||
state.udpReceiveQueuesBySocketId.set(id, []);
|
||||
return id;
|
||||
@@ -709,15 +755,38 @@ export class NetworkRuntimeEngine {
|
||||
}
|
||||
|
||||
closeSocket(socketId: Id): void {
|
||||
if (this.stateValue.socketsById.get(socketId)?.protocol === 'tcp') throw new NotPortedError('TCP sockets');
|
||||
if (this.stateValue.socketsById.get(socketId)?.protocol === 'tcp') {
|
||||
this.closeTcpSocket(socketId);
|
||||
return;
|
||||
}
|
||||
this.stateValue.socketsById.delete(socketId);
|
||||
this.stateValue.udpReceiveQueuesBySocketId.delete(socketId);
|
||||
}
|
||||
|
||||
/** Forgets a socket without closing it (Swift's discardSocket; TCP children once TCP is ported). */
|
||||
/** Forgets a socket without closing it, with the sockets a TCP listener accepted. */
|
||||
discardSocket(socketId: Id): void {
|
||||
this.stateValue.socketsById.delete(socketId);
|
||||
this.stateValue.udpReceiveQueuesBySocketId.delete(socketId);
|
||||
const state = this.stateValue;
|
||||
const children = state.tcpAcceptedSocketIdsByListenerId.get(socketId) ?? [];
|
||||
state.tcpAcceptedSocketIdsByListenerId.delete(socketId);
|
||||
for (const child of children) {
|
||||
state.socketsById.delete(child);
|
||||
state.tcpSessionsById.delete(child);
|
||||
}
|
||||
state.socketsById.delete(socketId);
|
||||
state.tcpSessionsById.delete(socketId);
|
||||
state.udpReceiveQueuesBySocketId.delete(socketId);
|
||||
for (const [listener, accepted] of state.tcpAcceptedSocketIdsByListenerId) {
|
||||
state.tcpAcceptedSocketIdsByListenerId.set(
|
||||
listener,
|
||||
accepted.filter((id) => id !== socketId),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
private allocateSocketId(): Id {
|
||||
const id = deterministicSocketId(this.stateValue.nextSocketSequenceNumber);
|
||||
this.stateValue.nextSocketSequenceNumber += 1;
|
||||
return id;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -851,6 +920,621 @@ export class NetworkRuntimeEngine {
|
||||
});
|
||||
}
|
||||
|
||||
// ── TCP sockets ──────────────────────────────────────────────────────────
|
||||
|
||||
/** A listening TCP socket; undefined while stopped, when the port is taken or the address is not the node's. */
|
||||
openTcpServerSocket(nodeId: Id, localPort: number, localIPAddress?: string): Id | undefined {
|
||||
const state = this.stateValue;
|
||||
if (
|
||||
state.phase !== 'running' ||
|
||||
this.isTcpPortReserved(nodeId, localPort) ||
|
||||
(localIPAddress !== undefined && !this.networkInterfaces(nodeId).some((each) => each.ipAddress === localIPAddress))
|
||||
) {
|
||||
return undefined;
|
||||
}
|
||||
const id = this.allocateSocketId();
|
||||
state.socketsById.set(id, {
|
||||
id,
|
||||
nodeId,
|
||||
protocol: 'tcp',
|
||||
localIPAddress,
|
||||
localPort,
|
||||
remoteIPAddress: undefined,
|
||||
remotePort: undefined,
|
||||
tcpState: 'LISTEN',
|
||||
parentListenerSocketId: undefined,
|
||||
});
|
||||
state.tcpSessionsById.set(id, this.makeTcpSession(id, 'LISTEN'));
|
||||
state.tcpAcceptedSocketIdsByListenerId.set(id, []);
|
||||
this.recordTcpStateTransition(id, 'CLOSED', 'LISTEN');
|
||||
return id;
|
||||
}
|
||||
|
||||
/** A closed TCP client socket towards the remote end, on the given or a free port. */
|
||||
openTcpClientSocket(
|
||||
nodeId: Id,
|
||||
remoteIPAddress: string,
|
||||
remotePort: number,
|
||||
options: { localIPAddress?: string; localPort?: number } = {},
|
||||
): Id | undefined {
|
||||
const state = this.stateValue;
|
||||
if (state.phase !== 'running') return undefined;
|
||||
let localIPAddress: string | undefined;
|
||||
if (options.localIPAddress !== undefined) {
|
||||
if (!this.networkInterfaces(nodeId).some((each) => each.ipAddress === options.localIPAddress)) return undefined;
|
||||
localIPAddress = options.localIPAddress;
|
||||
} else {
|
||||
const route = this.routeForPacket(nodeId, remoteIPAddress);
|
||||
localIPAddress = route.type === 'egress' ? route.interface.ipAddress : this.networkInterfaces(nodeId)[0]?.ipAddress;
|
||||
}
|
||||
if (localIPAddress === undefined) return undefined;
|
||||
let localPort: number;
|
||||
if (options.localPort !== undefined) {
|
||||
if (this.isTcpPortReserved(nodeId, options.localPort)) return undefined;
|
||||
localPort = options.localPort;
|
||||
} else {
|
||||
const port = this.reserveFreeTcpPort(nodeId);
|
||||
if (port === undefined) return undefined;
|
||||
localPort = port;
|
||||
}
|
||||
const id = this.allocateSocketId();
|
||||
state.socketsById.set(id, {
|
||||
id,
|
||||
nodeId,
|
||||
protocol: 'tcp',
|
||||
localIPAddress,
|
||||
localPort,
|
||||
remoteIPAddress,
|
||||
remotePort,
|
||||
tcpState: 'CLOSED',
|
||||
parentListenerSocketId: undefined,
|
||||
});
|
||||
state.tcpSessionsById.set(id, this.makeTcpSession(id, 'CLOSED'));
|
||||
return id;
|
||||
}
|
||||
|
||||
/** A port held by an open (not closed) TCP socket that is no listener's child. */
|
||||
isTcpPortReserved(nodeId: Id, port: number): boolean {
|
||||
for (const socket of this.stateValue.socketsById.values()) {
|
||||
if (
|
||||
socket.nodeId === nodeId &&
|
||||
socket.protocol === 'tcp' &&
|
||||
socket.localPort === port &&
|
||||
socket.parentListenerSocketId === undefined &&
|
||||
socket.tcpState !== 'CLOSED'
|
||||
) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
connectTcp(socketId: Id): boolean {
|
||||
return this.connectTcpWithResult(socketId) === 'connected';
|
||||
}
|
||||
|
||||
/** The three-way handshake from a closed client socket, waiting on the virtual clock as Java blocks. */
|
||||
connectTcpWithResult(socketId: Id): TcpConnectionResult {
|
||||
const state = this.stateValue;
|
||||
const socket = state.socketsById.get(socketId);
|
||||
if (
|
||||
socket === undefined ||
|
||||
socket.protocol !== 'tcp' ||
|
||||
socket.parentListenerSocketId !== undefined ||
|
||||
socket.remoteIPAddress === undefined ||
|
||||
socket.remotePort === undefined ||
|
||||
socket.tcpState !== 'CLOSED' ||
|
||||
[...state.socketsById.values()].some(
|
||||
(other) =>
|
||||
other.id !== socketId &&
|
||||
other.nodeId === socket.nodeId &&
|
||||
other.protocol === 'tcp' &&
|
||||
other.localPort === socket.localPort &&
|
||||
other.parentListenerSocketId === undefined &&
|
||||
other.tcpState !== 'CLOSED',
|
||||
)
|
||||
) {
|
||||
return 'invalidSocket';
|
||||
}
|
||||
this.updateTcpState(socketId, 'SYN_SENT');
|
||||
this.resetTcpSendAttempts(socketId);
|
||||
const initialDelivery = this.transmitTcpSegment(socketId, TCP_FLAG.synchronize, new Uint8Array(), 0, 'TCP SYN');
|
||||
const waitedForTimeout = this.advanceTcpUntilSettled(socketId, { waitingStates: ['SYN_SENT'] });
|
||||
if (state.socketsById.get(socketId)?.tcpState === 'ESTABLISHED') return 'connected';
|
||||
if (initialDelivery === undefined || initialDelivery.type !== 'delivered') return 'unreachable';
|
||||
return waitedForTimeout ? 'timedOut' : 'unreachable';
|
||||
}
|
||||
|
||||
/** Sends in segments of at most TCP_MAXIMUM_SEGMENT_SIZE bytes, each acknowledged before the next. */
|
||||
sendTcp(socketId: Id, payload: Uint8Array): boolean {
|
||||
const state = this.stateValue;
|
||||
if (state.socketsById.get(socketId)?.tcpState !== 'ESTABLISHED') return false;
|
||||
const chunks: Uint8Array[] = [];
|
||||
if (payload.length === 0) {
|
||||
chunks.push(new Uint8Array());
|
||||
} else {
|
||||
for (let offset = 0; offset < payload.length; offset += TCP_MAXIMUM_SEGMENT_SIZE) {
|
||||
chunks.push(payload.subarray(offset, Math.min(offset + TCP_MAXIMUM_SEGMENT_SIZE, payload.length)));
|
||||
}
|
||||
}
|
||||
for (const [index, chunk] of chunks.entries()) {
|
||||
if (state.socketsById.get(socketId)?.tcpState !== 'ESTABLISHED') return false;
|
||||
this.resetTcpSendAttempts(socketId);
|
||||
const flags = index === chunks.length - 1 ? TCP_FLAG.push : 0;
|
||||
this.transmitTcpSegment(socketId, flags, chunk, 0, `TCP data segment ${index + 1}/${chunks.length}`);
|
||||
this.advanceTcpUntilSettled(socketId, { waitingForAcknowledgement: true });
|
||||
const session = state.tcpSessionsById.get(socketId);
|
||||
if (session === undefined || session.state !== 'ESTABLISHED' || session.lastSentSegment !== undefined) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
/** The oldest whole message (up to a PSH) received on the socket. */
|
||||
receiveTcp(socketId: Id): Uint8Array | undefined {
|
||||
const session = this.stateValue.tcpSessionsById.get(socketId);
|
||||
if (session === undefined || session.receivedMessages.length === 0) return undefined;
|
||||
return session.receivedMessages.shift();
|
||||
}
|
||||
|
||||
closeTcpSocket(socketId: Id): void {
|
||||
const socket = this.stateValue.socketsById.get(socketId);
|
||||
if (socket === undefined || socket.protocol !== 'tcp' || socket.tcpState === undefined) return;
|
||||
switch (socket.tcpState) {
|
||||
case 'LISTEN':
|
||||
case 'SYN_SENT':
|
||||
this.updateTcpState(socketId, 'CLOSED');
|
||||
this.clearTcpPendingTransmission(socketId);
|
||||
return;
|
||||
case 'ESTABLISHED':
|
||||
case 'SYN_RCVD':
|
||||
this.updateTcpState(socketId, 'FIN_WAIT_1');
|
||||
this.resetTcpSendAttempts(socketId);
|
||||
this.transmitTcpSegment(socketId, TCP_FLAG.finish, new Uint8Array(), 0, 'TCP FIN');
|
||||
return;
|
||||
case 'CLOSE_WAIT':
|
||||
this.updateTcpState(socketId, 'LAST_ACK');
|
||||
this.resetTcpSendAttempts(socketId);
|
||||
this.transmitTcpSegment(socketId, TCP_FLAG.finish, new Uint8Array(), 0, 'TCP FIN after CLOSE_WAIT');
|
||||
return;
|
||||
default:
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
/** The socket at the other end of a connection, the first by id. */
|
||||
peerTcpSocketId(socketId: Id): Id | undefined {
|
||||
const socket = this.stateValue.socketsById.get(socketId);
|
||||
if (
|
||||
socket === undefined ||
|
||||
socket.protocol !== 'tcp' ||
|
||||
socket.localIPAddress === undefined ||
|
||||
socket.remoteIPAddress === undefined ||
|
||||
socket.remotePort === undefined
|
||||
) {
|
||||
return undefined;
|
||||
}
|
||||
return [...this.stateValue.socketsById.values()]
|
||||
.filter(
|
||||
(candidate) =>
|
||||
candidate.id !== socketId &&
|
||||
candidate.protocol === 'tcp' &&
|
||||
candidate.tcpState !== 'LISTEN' &&
|
||||
candidate.localIPAddress === socket.remoteIPAddress &&
|
||||
candidate.localPort === socket.remotePort &&
|
||||
candidate.remoteIPAddress === socket.localIPAddress &&
|
||||
candidate.remotePort === socket.localPort,
|
||||
)
|
||||
.sort((lhs, rhs) => compareIds(lhs.id, rhs.id))[0]?.id;
|
||||
}
|
||||
|
||||
/** Removes a closed TCP socket; false when it is not one. */
|
||||
cleanClosedTcpSocket(socketId: Id): boolean {
|
||||
const state = this.stateValue;
|
||||
const socket = state.socketsById.get(socketId);
|
||||
if (socket?.protocol !== 'tcp' || socket.tcpState !== 'CLOSED') return false;
|
||||
this.cancelTcpTimeouts(socketId);
|
||||
state.socketsById.delete(socketId);
|
||||
state.tcpSessionsById.delete(socketId);
|
||||
state.tcpAcceptedSocketIdsByListenerId.delete(socketId);
|
||||
for (const [listener, accepted] of state.tcpAcceptedSocketIdsByListenerId) {
|
||||
state.tcpAcceptedSocketIdsByListenerId.set(
|
||||
listener,
|
||||
accepted.filter((id) => id !== socketId),
|
||||
);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private deliverTcp(packet: IPv4Packet, segment: TcpSegment, nodeId: Id, networkInterface: NetworkInterface): void {
|
||||
const state = this.stateValue;
|
||||
this.recordTrace({
|
||||
packetIdentity: packet.identity,
|
||||
nodeId,
|
||||
interfaceId: networkInterface.portId,
|
||||
direction: 'local',
|
||||
layer: 'transport',
|
||||
operation: 'accepted',
|
||||
beforeHeaders: tcpHeaders(segment),
|
||||
detail: 'TCP segment received',
|
||||
});
|
||||
const byId = (lhs: SocketRecord, rhs: SocketRecord) => compareIds(lhs.id, rhs.id);
|
||||
const sockets = [...state.socketsById.values()];
|
||||
const exact = sockets
|
||||
.filter(
|
||||
(socket) =>
|
||||
socket.nodeId === nodeId &&
|
||||
socket.protocol === 'tcp' &&
|
||||
socket.tcpState !== 'LISTEN' &&
|
||||
socket.tcpState !== 'CLOSED' &&
|
||||
socket.localPort === segment.destinationPort &&
|
||||
(socket.localIPAddress === undefined || socket.localIPAddress === packet.receiverIPAddress) &&
|
||||
socket.remoteIPAddress === packet.senderIPAddress &&
|
||||
socket.remotePort === segment.sourcePort,
|
||||
)
|
||||
.sort(byId)[0];
|
||||
if (exact !== undefined) {
|
||||
this.processIncomingTcpSegment(exact.id, segment);
|
||||
return;
|
||||
}
|
||||
const listener = sockets
|
||||
.filter(
|
||||
(socket) =>
|
||||
socket.nodeId === nodeId &&
|
||||
socket.protocol === 'tcp' &&
|
||||
socket.tcpState === 'LISTEN' &&
|
||||
socket.localPort === segment.destinationPort &&
|
||||
(socket.localIPAddress === undefined || socket.localIPAddress === packet.receiverIPAddress),
|
||||
)
|
||||
.sort(byId)[0];
|
||||
if (listener === undefined || (segment.flags & TCP_FLAG.synchronize) === 0 || (segment.flags & TCP_FLAG.acknowledgement) !== 0) {
|
||||
this.recordTrace({
|
||||
packetIdentity: packet.identity,
|
||||
nodeId,
|
||||
interfaceId: networkInterface.portId,
|
||||
direction: 'local',
|
||||
layer: 'transport',
|
||||
operation: 'dropped',
|
||||
beforeHeaders: tcpHeaders(segment),
|
||||
detail: 'TCP destination port closed',
|
||||
});
|
||||
return;
|
||||
}
|
||||
const acceptedId = this.allocateSocketId();
|
||||
state.socketsById.set(acceptedId, {
|
||||
id: acceptedId,
|
||||
nodeId,
|
||||
protocol: 'tcp',
|
||||
localIPAddress: packet.receiverIPAddress,
|
||||
localPort: segment.destinationPort,
|
||||
remoteIPAddress: packet.senderIPAddress,
|
||||
remotePort: segment.sourcePort,
|
||||
tcpState: 'SYN_RCVD',
|
||||
parentListenerSocketId: listener.id,
|
||||
});
|
||||
const session = this.makeTcpSession(acceptedId, 'SYN_RCVD');
|
||||
session.remoteSequenceNumber = nextTcpSequenceNumber(segment);
|
||||
state.tcpSessionsById.set(acceptedId, session);
|
||||
getOrCreate(state.tcpAcceptedSocketIdsByListenerId, listener.id, () => []).push(acceptedId);
|
||||
this.recordTcpStateTransition(acceptedId, 'LISTEN', 'SYN_RCVD');
|
||||
this.resetTcpSendAttempts(acceptedId);
|
||||
this.transmitTcpSegment(
|
||||
acceptedId,
|
||||
TCP_FLAG.synchronize | TCP_FLAG.acknowledgement,
|
||||
new Uint8Array(),
|
||||
session.remoteSequenceNumber,
|
||||
'TCP SYN+ACK',
|
||||
);
|
||||
}
|
||||
|
||||
private processIncomingTcpSegment(socketId: Id, segment: TcpSegment): void {
|
||||
const socket = this.stateValue.socketsById.get(socketId);
|
||||
if (socket === undefined || socket.tcpState === undefined) return;
|
||||
const has = (flag: number) => (segment.flags & flag) !== 0;
|
||||
if (has(TCP_FLAG.reset)) {
|
||||
this.clearTcpPendingTransmission(socketId);
|
||||
this.updateTcpState(socketId, 'CLOSED');
|
||||
return;
|
||||
}
|
||||
switch (socket.tcpState) {
|
||||
case 'SYN_SENT': {
|
||||
if (
|
||||
!has(TCP_FLAG.synchronize) ||
|
||||
!has(TCP_FLAG.acknowledgement) ||
|
||||
!this.acknowledgeTcpIfExpected(socketId, segment.acknowledgementNumber)
|
||||
) {
|
||||
this.updateTcpState(socketId, 'CLOSED');
|
||||
return;
|
||||
}
|
||||
const session = this.stateValue.tcpSessionsById.get(socketId);
|
||||
if (session !== undefined) session.remoteSequenceNumber = nextTcpSequenceNumber(segment);
|
||||
this.updateTcpState(socketId, 'ESTABLISHED');
|
||||
this.sendTcpAcknowledgement(socketId, segment);
|
||||
return;
|
||||
}
|
||||
case 'SYN_RCVD':
|
||||
if (!has(TCP_FLAG.acknowledgement) || !this.acknowledgeTcpIfExpected(socketId, segment.acknowledgementNumber)) return;
|
||||
this.updateTcpState(socketId, 'ESTABLISHED');
|
||||
return;
|
||||
case 'ESTABLISHED':
|
||||
if (has(TCP_FLAG.finish)) {
|
||||
this.sendTcpAcknowledgement(socketId, segment);
|
||||
this.updateTcpState(socketId, 'CLOSE_WAIT');
|
||||
} else if (segment.payload.length > 0 || has(TCP_FLAG.push)) {
|
||||
this.receiveTcpPayload(socketId, segment);
|
||||
} else if (has(TCP_FLAG.acknowledgement)) {
|
||||
this.acknowledgeTcpIfExpected(socketId, segment.acknowledgementNumber);
|
||||
}
|
||||
return;
|
||||
case 'FIN_WAIT_1': {
|
||||
const hasAck = has(TCP_FLAG.acknowledgement);
|
||||
const hasFin = has(TCP_FLAG.finish);
|
||||
if (hasAck) this.acknowledgeTcpIfExpected(socketId, segment.acknowledgementNumber);
|
||||
if (hasAck && hasFin) {
|
||||
this.sendTcpAcknowledgement(socketId, segment);
|
||||
this.enterTcpTimeWait(socketId);
|
||||
} else if (hasAck) {
|
||||
this.updateTcpState(socketId, 'FIN_WAIT_2');
|
||||
} else if (hasFin) {
|
||||
this.sendTcpAcknowledgement(socketId, segment);
|
||||
this.updateTcpState(socketId, 'CLOSING');
|
||||
}
|
||||
return;
|
||||
}
|
||||
case 'FIN_WAIT_2':
|
||||
if (has(TCP_FLAG.finish)) {
|
||||
this.sendTcpAcknowledgement(socketId, segment);
|
||||
this.enterTcpTimeWait(socketId);
|
||||
}
|
||||
return;
|
||||
case 'CLOSING':
|
||||
if (has(TCP_FLAG.acknowledgement) && this.acknowledgeTcpIfExpected(socketId, segment.acknowledgementNumber)) {
|
||||
this.enterTcpTimeWait(socketId);
|
||||
}
|
||||
return;
|
||||
case 'LAST_ACK':
|
||||
if (has(TCP_FLAG.acknowledgement) && this.acknowledgeTcpIfExpected(socketId, segment.acknowledgementNumber)) {
|
||||
this.updateTcpState(socketId, 'CLOSED');
|
||||
}
|
||||
return;
|
||||
default:
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
private receiveTcpPayload(socketId: Id, segment: TcpSegment): void {
|
||||
const session = this.stateValue.tcpSessionsById.get(socketId);
|
||||
if (session === undefined) return;
|
||||
const nextRemoteSequence = nextTcpSequenceNumber(segment);
|
||||
if (nextRemoteSequence > session.remoteSequenceNumber) {
|
||||
session.remoteSequenceNumber = nextRemoteSequence;
|
||||
session.receiveAssembly = concatBytes(session.receiveAssembly, segment.payload);
|
||||
if ((segment.flags & TCP_FLAG.push) !== 0) {
|
||||
session.receivedMessages.push(session.receiveAssembly);
|
||||
session.receiveAssembly = new Uint8Array();
|
||||
}
|
||||
}
|
||||
this.sendTcpAcknowledgement(socketId, segment);
|
||||
}
|
||||
|
||||
private sendTcpAcknowledgement(socketId: Id, segment: TcpSegment): void {
|
||||
this.transmitTcpSegment(socketId, TCP_FLAG.acknowledgement, new Uint8Array(), nextTcpSequenceNumber(segment), 'TCP ACK');
|
||||
}
|
||||
|
||||
private transmitTcpSegment(
|
||||
socketId: Id,
|
||||
flags: number,
|
||||
payload: Uint8Array,
|
||||
acknowledgementNumber: number,
|
||||
detail: string,
|
||||
retransmission = false,
|
||||
): IPv4DeliveryResult | undefined {
|
||||
const state = this.stateValue;
|
||||
const socket = state.socketsById.get(socketId);
|
||||
const session = state.tcpSessionsById.get(socketId);
|
||||
if (socket === undefined || socket.remoteIPAddress === undefined || socket.remotePort === undefined || session === undefined) {
|
||||
return undefined;
|
||||
}
|
||||
let segment: TcpSegment;
|
||||
if (retransmission && session.lastSentSegment !== undefined) {
|
||||
segment = session.lastSentSegment;
|
||||
session.sendAttempts += 1;
|
||||
} else {
|
||||
segment = {
|
||||
sourcePort: socket.localPort,
|
||||
destinationPort: socket.remotePort,
|
||||
sequenceNumber: session.nextSendSequenceNumber,
|
||||
acknowledgementNumber,
|
||||
flags,
|
||||
payload,
|
||||
};
|
||||
session.lastSentSequenceNumber = segment.sequenceNumber;
|
||||
session.nextSendSequenceNumber = nextTcpSequenceNumber(segment);
|
||||
if (tcpSegmentRequiresAcknowledgement(segment)) {
|
||||
session.lastSentSegment = segment;
|
||||
session.sendAttempts = 1;
|
||||
}
|
||||
}
|
||||
const interfaces = this.networkInterfaces(socket.nodeId);
|
||||
let networkInterface =
|
||||
socket.localIPAddress === undefined ? undefined : interfaces.find((each) => each.ipAddress === socket.localIPAddress);
|
||||
if (networkInterface === undefined) {
|
||||
const route = this.routeForPacket(socket.nodeId, socket.remoteIPAddress);
|
||||
networkInterface = route.type === 'egress' ? route.interface : interfaces[0];
|
||||
}
|
||||
if (networkInterface === undefined) return undefined;
|
||||
const packet: IPv4Packet = {
|
||||
identity: this.allocatePacketIdentity(),
|
||||
senderIPAddress: networkInterface.ipAddress,
|
||||
receiverIPAddress: socket.remoteIPAddress,
|
||||
timeToLive: 64,
|
||||
protocol: 'tcp',
|
||||
payload: { kind: 'tcp', segment },
|
||||
};
|
||||
this.recordTrace({
|
||||
packetIdentity: packet.identity,
|
||||
nodeId: socket.nodeId,
|
||||
interfaceId: networkInterface.portId,
|
||||
direction: 'outbound',
|
||||
layer: 'transport',
|
||||
operation: retransmission ? 'retransmitted' : 'sent',
|
||||
afterHeaders: tcpHeaders(segment),
|
||||
detail: retransmission ? `${detail} retransmission attempt ${session.sendAttempts}` : detail,
|
||||
});
|
||||
const delivery = this.sendIPv4Packet(socket.nodeId, packet, networkInterface.portId);
|
||||
if (state.tcpSessionsById.get(socketId)?.lastSentSegment !== undefined) this.scheduleTcpTimeoutIfNeeded(socketId);
|
||||
return delivery;
|
||||
}
|
||||
|
||||
private scheduleTcpTimeoutIfNeeded(socketId: Id): void {
|
||||
const pending = this.stateValue.pendingEvents.some((event) => event.kind.kind === 'tcpTimeout' && event.kind.sessionId === socketId);
|
||||
if (pending) return;
|
||||
this.schedule(this.stateValue.currentTimeMilliseconds + TCP_DEFAULT_TIMEOUT_MILLISECONDS, { kind: 'tcpTimeout', sessionId: socketId });
|
||||
}
|
||||
|
||||
private handleTcpTimeout(socketId: Id): void {
|
||||
const socket = this.stateValue.socketsById.get(socketId);
|
||||
if (socket === undefined || socket.tcpState === undefined) return;
|
||||
if (socket.tcpState === 'TIME_WAIT') {
|
||||
this.updateTcpState(socketId, 'CLOSED');
|
||||
this.clearTcpPendingTransmission(socketId);
|
||||
return;
|
||||
}
|
||||
const session = this.stateValue.tcpSessionsById.get(socketId);
|
||||
const last = session?.lastSentSegment;
|
||||
if (session === undefined || last === undefined) return;
|
||||
if (session.sendAttempts >= TCP_MAXIMUM_SEND_ATTEMPTS) {
|
||||
this.clearTcpPendingTransmission(socketId);
|
||||
this.updateTcpState(socketId, 'CLOSED');
|
||||
return;
|
||||
}
|
||||
const synchronize = (last.flags & TCP_FLAG.synchronize) !== 0;
|
||||
const baseDetail = synchronize
|
||||
? (last.flags & TCP_FLAG.acknowledgement) === 0
|
||||
? 'TCP SYN'
|
||||
: 'TCP SYN+ACK'
|
||||
: (last.flags & TCP_FLAG.finish) !== 0
|
||||
? 'TCP FIN'
|
||||
: 'TCP data';
|
||||
this.transmitTcpSegment(socketId, last.flags, last.payload, last.acknowledgementNumber, baseDetail, true);
|
||||
}
|
||||
|
||||
private acknowledgeTcpIfExpected(socketId: Id, acknowledgementNumber: number): boolean {
|
||||
const session = this.stateValue.tcpSessionsById.get(socketId);
|
||||
if (session === undefined || session.lastSentSegment === undefined || acknowledgementNumber !== session.nextSendSequenceNumber) {
|
||||
return false;
|
||||
}
|
||||
session.lastAcknowledgedSequenceNumber = acknowledgementNumber;
|
||||
session.lastSentSegment = undefined;
|
||||
this.cancelTcpTimeouts(socketId);
|
||||
return true;
|
||||
}
|
||||
|
||||
private clearTcpPendingTransmission(socketId: Id): void {
|
||||
const session = this.stateValue.tcpSessionsById.get(socketId);
|
||||
if (session !== undefined) session.lastSentSegment = undefined;
|
||||
this.cancelTcpTimeouts(socketId);
|
||||
}
|
||||
|
||||
private cancelTcpTimeouts(socketId: Id): void {
|
||||
this.stateValue.pendingEvents = this.stateValue.pendingEvents.filter(
|
||||
(event) => !(event.kind.kind === 'tcpTimeout' && event.kind.sessionId === socketId),
|
||||
);
|
||||
}
|
||||
|
||||
private enterTcpTimeWait(socketId: Id): void {
|
||||
this.clearTcpPendingTransmission(socketId);
|
||||
this.updateTcpState(socketId, 'TIME_WAIT');
|
||||
this.schedule(this.stateValue.currentTimeMilliseconds + TCP_ROUND_TRIP_TIME_MILLISECONDS, { kind: 'tcpTimeout', sessionId: socketId });
|
||||
}
|
||||
|
||||
/**
|
||||
* Advances the clock to the socket's timeouts while it waits in one of
|
||||
* the states or for an acknowledgement; whether it advanced at all.
|
||||
*/
|
||||
private advanceTcpUntilSettled(
|
||||
socketId: Id,
|
||||
options: { waitingStates?: readonly TcpSocketState[]; waitingForAcknowledgement?: boolean },
|
||||
): boolean {
|
||||
let advancedToTimeout = false;
|
||||
while (this.stateValue.phase === 'running') {
|
||||
const socket = this.stateValue.socketsById.get(socketId);
|
||||
if (socket === undefined || socket.tcpState === undefined) return advancedToTimeout;
|
||||
const waitingOnState = options.waitingStates?.includes(socket.tcpState) === true;
|
||||
const waitingOnAck =
|
||||
options.waitingForAcknowledgement === true && this.stateValue.tcpSessionsById.get(socketId)?.lastSentSegment !== undefined;
|
||||
if (!waitingOnState && !waitingOnAck) return advancedToTimeout;
|
||||
const deadlines = this.stateValue.pendingEvents
|
||||
.filter((event) => event.kind.kind === 'tcpTimeout' && event.kind.sessionId === socketId)
|
||||
.map((event) => event.deadlineMilliseconds);
|
||||
if (deadlines.length === 0) return advancedToTimeout;
|
||||
advancedToTimeout = true;
|
||||
this.advance(Math.min(...deadlines));
|
||||
}
|
||||
return advancedToTimeout;
|
||||
}
|
||||
|
||||
private updateTcpState(socketId: Id, newState: TcpSocketState): void {
|
||||
const socket = this.stateValue.socketsById.get(socketId);
|
||||
const oldState = socket?.tcpState;
|
||||
if (socket === undefined || oldState === undefined || oldState === newState) return;
|
||||
socket.tcpState = newState;
|
||||
const session = this.stateValue.tcpSessionsById.get(socketId);
|
||||
if (session !== undefined) session.state = newState;
|
||||
this.recordTcpStateTransition(socketId, oldState, newState);
|
||||
}
|
||||
|
||||
private recordTcpStateTransition(socketId: Id, oldState: TcpSocketState, newState: TcpSocketState): void {
|
||||
const socket = this.stateValue.socketsById.get(socketId);
|
||||
if (socket === undefined) return;
|
||||
this.recordTrace({
|
||||
nodeId: socket.nodeId,
|
||||
direction: 'local',
|
||||
layer: 'transport',
|
||||
operation: 'accepted',
|
||||
beforeHeaders: [{ name: 'state', value: oldState }],
|
||||
afterHeaders: [{ name: 'state', value: newState }],
|
||||
detail: `TCP ${oldState} -> ${newState}`,
|
||||
});
|
||||
}
|
||||
|
||||
private resetTcpSendAttempts(socketId: Id): void {
|
||||
const session = this.stateValue.tcpSessionsById.get(socketId);
|
||||
if (session !== undefined) session.sendAttempts = 0;
|
||||
}
|
||||
|
||||
private reserveFreeTcpPort(nodeId: Id): number | undefined {
|
||||
const span = TCP_EPHEMERAL_PORT_UPPER_BOUND_EXCLUSIVE - TCP_EPHEMERAL_PORT_LOWER_BOUND;
|
||||
for (let attempt = 0; attempt < span; attempt++) {
|
||||
const port = TCP_EPHEMERAL_PORT_LOWER_BOUND + this.tcpPortRandom.nextInt(span);
|
||||
if (!this.isTcpPortReserved(nodeId, port)) return port;
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
/** Swift: (value mod 2^32) · 1,000,000, truncated to 32 bits. */
|
||||
private allocateTcpInitialSequenceNumber(): number {
|
||||
const value = BigInt(this.stateValue.nextTcpInitialSequenceValue);
|
||||
this.stateValue.nextTcpInitialSequenceValue += 1;
|
||||
return Number(BigInt.asUintN(32, (value % 4_294_967_296n) * 1_000_000n));
|
||||
}
|
||||
|
||||
private makeTcpSession(socketId: Id, socketState: TcpSocketState): TcpSession {
|
||||
const initialSequence = this.allocateTcpInitialSequenceNumber();
|
||||
return {
|
||||
id: socketId,
|
||||
socketId,
|
||||
state: socketState,
|
||||
sendAttempts: 0,
|
||||
nextSendSequenceNumber: initialSequence,
|
||||
lastSentSequenceNumber: initialSequence,
|
||||
remoteSequenceNumber: 0,
|
||||
lastAcknowledgedSequenceNumber: undefined,
|
||||
lastSentSegment: undefined,
|
||||
receiveAssembly: new Uint8Array(),
|
||||
receivedMessages: [],
|
||||
};
|
||||
}
|
||||
|
||||
// ── Later slices hook in here ────────────────────────────────────────────
|
||||
|
||||
private initializeRipState(): void {
|
||||
@@ -1783,7 +2467,8 @@ export class NetworkRuntimeEngine {
|
||||
interfaceId: networkInterface.portId,
|
||||
packet,
|
||||
});
|
||||
throw new NotPortedError('TCP delivery');
|
||||
this.deliverTcp(packet, payload.segment, nodeId, networkInterface);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2179,6 +2864,47 @@ export function ipv4Headers(packet: IPv4Packet): HeaderField[] {
|
||||
];
|
||||
}
|
||||
|
||||
function tcpSegmentRequiresAcknowledgement(segment: TcpSegment): boolean {
|
||||
return (
|
||||
(segment.flags & (TCP_FLAG.synchronize | TCP_FLAG.finish | TCP_FLAG.push)) !== 0 || segment.payload.length > 0
|
||||
);
|
||||
}
|
||||
|
||||
/** The sequence number after the segment: SYN and FIN count one, and each payload byte. */
|
||||
export function nextTcpSequenceNumber(segment: TcpSegment): number {
|
||||
let next = segment.sequenceNumber;
|
||||
if ((segment.flags & (TCP_FLAG.synchronize | TCP_FLAG.finish)) !== 0) next = uint32Add(next, 1);
|
||||
return uint32Add(next, segment.payload.length);
|
||||
}
|
||||
|
||||
function concatBytes(first: Uint8Array, second: Uint8Array): Uint8Array {
|
||||
const result = new Uint8Array(first.length + second.length);
|
||||
result.set(first);
|
||||
result.set(second, first.length);
|
||||
return result;
|
||||
}
|
||||
|
||||
export function tcpFlagDescription(flags: number): string {
|
||||
const names: string[] = [];
|
||||
if ((flags & TCP_FLAG.finish) !== 0) names.push('FIN');
|
||||
if ((flags & TCP_FLAG.synchronize) !== 0) names.push('SYN');
|
||||
if ((flags & TCP_FLAG.reset) !== 0) names.push('RST');
|
||||
if ((flags & TCP_FLAG.push) !== 0) names.push('PSH');
|
||||
if ((flags & TCP_FLAG.acknowledgement) !== 0) names.push('ACK');
|
||||
return names.join('+');
|
||||
}
|
||||
|
||||
export function tcpHeaders(segment: TcpSegment): HeaderField[] {
|
||||
return [
|
||||
{ name: 'sourcePort', value: String(segment.sourcePort) },
|
||||
{ name: 'destinationPort', value: String(segment.destinationPort) },
|
||||
{ name: 'sequenceNumber', value: String(segment.sequenceNumber) },
|
||||
{ name: 'acknowledgementNumber', value: String(segment.acknowledgementNumber) },
|
||||
{ name: 'flags', value: tcpFlagDescription(segment.flags) },
|
||||
{ name: 'payloadLength', value: String(segment.payload.length) },
|
||||
];
|
||||
}
|
||||
|
||||
export function udpHeaders(datagram: UdpDatagram): HeaderField[] {
|
||||
return [
|
||||
{ name: 'sourcePort', value: String(datagram.sourcePort) },
|
||||
|
||||
Reference in New Issue
Block a user