From ba873d04c4dc7b1cb5aa747c1acb98f2c1af9a9e Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 09:11:51 +0200 Subject: [PATCH] Ack Cut 1d (TypeScript): stale generations are duplicates, a stale Connect is answered not obeyed, sessions draw their own sequence A frame at or before the seeded watermark is a duplicate of an earlier generation and is acknowledged, never delivered again (serial compare within the receive window). A Connect that precedes the current generation's within the window is a stale copy of an earlier attempt: it is answered like a repeat (Accept owed or Ack), changes no state and does not refresh liveness. A Connect unanswered for 3 s is replaced by a fresh Connect with a newly drawn sequence and the same payload. In server mode a Connect from another endpoint is always a new generation. The initial sequence is unset by default and each session draws its own; the public random helper is removed. The cultmesh-ts document server acknowledges what the session received, not every reliable packet by name. Co-Authored-By: Claude Sonnet 5.5 --- packages/cultmesh-ts/src/index.ts | 2 +- .../test/provider-rudp-transport.test.ts | 2 +- packages/cultnet-ts/src/index.ts | 1 - packages/cultnet-ts/src/rudp.ts | 139 ++++++++--- packages/cultnet-ts/test/cultnet.test.ts | 4 +- .../test/rudp-connect-generation.test.ts | 224 +++++++++++++++++- 6 files changed, 328 insertions(+), 44 deletions(-) diff --git a/packages/cultmesh-ts/src/index.ts b/packages/cultmesh-ts/src/index.ts index 1213967d..b89cd823 100644 --- a/packages/cultmesh-ts/src/index.ts +++ b/packages/cultmesh-ts/src/index.ts @@ -5846,7 +5846,7 @@ export class CultMesh { } if (packet.reliable) { socket.send( - encodeRudpPacket(record.session.createAckFor(packet.sequence)), + encodeRudpPacket(record.session.createAckForReceived(packet.sequence)), record.remote.port, record.remote.address, ); diff --git a/packages/cultmesh-ts/test/provider-rudp-transport.test.ts b/packages/cultmesh-ts/test/provider-rudp-transport.test.ts index 4e17eee9..e35a42a8 100644 --- a/packages/cultmesh-ts/test/provider-rudp-transport.test.ts +++ b/packages/cultmesh-ts/test/provider-rudp-transport.test.ts @@ -338,7 +338,7 @@ test("a client restarted on the same address with the same Connect payload repla socket.send(encodeRudpPacket(packet), server.bind.port, "127.0.0.1"); const evidence = Buffer.from("same connect payload"); try { - const first = new CultNetRudpSession({ connectionId, initialSequence: 50 }); + const first = new CultNetRudpSession({ connectionId, initialSequence: 50_000 }); const connect = first.createConnect(Date.now(), evidence); toServer(connect); await waitFor(() => received.some(packet => packet.packetType === "accept"), "the first Accept"); diff --git a/packages/cultnet-ts/src/index.ts b/packages/cultnet-ts/src/index.ts index ed34de3c..8950f638 100644 --- a/packages/cultnet-ts/src/index.ts +++ b/packages/cultnet-ts/src/index.ts @@ -106,7 +106,6 @@ export { createRudpTransportProfile, decodeRudpPacket, encodeRudpPacket, - randomInitialSequence, CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS, CultNetRudpReconnectLoop, CultNetRudpSession, diff --git a/packages/cultnet-ts/src/rudp.ts b/packages/cultnet-ts/src/rudp.ts index ca928c24..bfe94973 100644 --- a/packages/cultnet-ts/src/rudp.ts +++ b/packages/cultnet-ts/src/rudp.ts @@ -19,13 +19,28 @@ const RUDP_FIXED_HEADER_BYTES = 36; const MAX_CHANNEL_ID_BYTES = 255; export const CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS = 32; const RUDP_RECEIVED_SEQUENCE_WINDOW = 4_096; +/// How long a client keeps retransmitting a Connect nobody answered before it +/// abandons that attempt for a fresh one. A Connect the server answers with an +/// Ack, never an Accept, is one the server judged stale (see `acceptConnect`); +/// retransmitting the same sequence would never change its mind. +const RUDP_CONNECT_ATTEMPT_MS = 3_000; -/// A fresh default initial sequence, drawn from a secure source in [1, 2^31). -/// The Connect's sequence is what tells a peer whether a Connect repeats one it -/// already accepted or starts a new session, so two sessions must not share one -/// by default. -export function randomInitialSequence(): number { - return randomInt(1, 2 ** 31); +/// A sequence drawn from a secure source in [floor, 2^31). The Connect's +/// sequence is what tells a peer whether a Connect repeats one it already +/// accepted or starts a new session, so two sessions must not share one by +/// default. A floor above the range draws nothing new: the caller keeps its own +/// sequence. +function drawSequence(floor: number): number { + const low = Math.max(1, floor); + return low >= 2 ** 31 ? low : randomInt(low, 2 ** 31); +} + +/// Whether `sequence` is at or before `mark` in serial order, within the +/// receive window. The compare is modular, so it holds across the wrap of the +/// 32-bit space; a sequence further back than the window is the duplicate +/// test's below-window clause, and one ahead of the mark is never before it. +function atOrBefore(sequence: number, mark: number): boolean { + return ((mark - sequence) >>> 0) < RUDP_RECEIVED_SEQUENCE_WINDOW; } function packetAcknowledges(packet: CultNetRudpPacket, sequence: number): boolean { @@ -174,6 +189,9 @@ export class CultNetRudpSession { #connectSequence: number | undefined; /// A Connect this side sent is unanswered. Only then is an Accept honoured. #awaitingAccept = false; + /// When the unanswered Connect first went out; `dueResends` abandons it for a + /// fresh attempt once `RUDP_CONNECT_ATTEMPT_MS` has passed. + #connectStartedAtMs = 0; readonly #maxPendingReliablePackets: number | undefined; #lastReceivedAtMs: number | undefined; #highestReceivedSequence: number | undefined; @@ -200,7 +218,7 @@ export class CultNetRudpSession { constructor(options: CultNetRudpSessionOptions) { this.connectionId = toUint32(options.connectionId, "connectionId"); - this.#nextSequence = toUint32(options.initialSequence ?? randomInitialSequence(), "initialSequence"); + this.#nextSequence = toUint32(options.initialSequence ?? drawSequence(1), "initialSequence"); if (this.#nextSequence === 0xffff_ffff) { throw new Error("RUDP initialSequence must leave room for a reliable packet."); } @@ -298,32 +316,57 @@ export class CultNetRudpSession { }); this.#connectSequence = packet.sequence; this.#awaitingAccept = true; + this.#connectStartedAtMs = nowMs; this.#trackReliable(packet, nowMs); return packet; } - /// Whether `packet` is a retransmit of the Connect that started this session's - /// current generation: the session is connected and the Connect carries that - /// Connect's sequence. Servers that keep one session per peer ask this to tell - /// a repeat from a new session; anything else that reaches `acceptConnect` - /// starts a new generation. + /// Whether `packet` is a Connect this session's current generation already + /// owns: a retransmit of the Connect that started it (the session is + /// connected and the sequence is that Connect's), or a stale copy of an + /// earlier attempt by the same client (a sequence before it, within the + /// receive window). Servers that keep one session per peer ask this to tell a + /// Connect that starts a new session from one that does not; anything else + /// that reaches `acceptConnect` starts a new generation. connectRepeats(packet: CultNetRudpPacket): boolean { return packet.packetType === "connect" && this.#connected - && this.#connectSequence === packet.sequence; + && this.#connectSequence !== undefined + && (this.#connectSequence === packet.sequence + || this.#connectIsStale(packet.sequence, this.#connectSequence)); } + /// A Connect that precedes the current generation's within the receive window + /// is the client's earlier attempt, delayed in the network: a client that + /// retried never sends a lower sequence again. Restarting on it would strand + /// the client, which honours only the Accept for its newest Connect. The rule + /// is TCP's answer to a delayed SYN (RFC 5961's challenge ACK): keep the + /// connection and answer with an Ack. A restarted client whose random initial + /// sequence lands in this window is answered the same way and abandons the + /// attempt for a fresh draw (`RUDP_CONNECT_ATTEMPT_MS`). + #connectIsStale(sequence: number, current: number): boolean { + return sequence !== current && atOrBefore(sequence, current); + } + + /// Answers a Connect. A repeat of the accepted one queues nothing, so a + /// Connect storm cannot grow the reliable queue: the reply is the Accept + /// still awaiting acknowledgement, or an Ack once it was acknowledged. A + /// stale copy of an earlier attempt gets the same reply and changes nothing + /// else: it is not evidence the peer is alive. Any other Connect ends the + /// current generation, forgets the peer and accepts a new one. acceptConnect(packet: CultNetRudpPacket, nowMs = 0, payload = new Uint8Array()): CultNetRudpPacket { this.#requireConnection(packet); if (packet.packetType !== "connect") { throw new Error(`Expected RUDP connect packet, got ${packet.packetType}.`); } - // A repeat of the accepted Connect queues nothing, so a Connect storm cannot - // grow the reliable queue. Any other Connect ends the current generation, - // forgets the peer and accepts a new one. if (this.connectRepeats(packet)) { - return this.#answerRepeatedConnect(packet, nowMs); + if (this.#connectSequence === packet.sequence) { + this.#applyAcknowledgements(packet); + this.#rememberReceived(packet.sequence); + this.#lastReceivedAtMs = nowMs; + } + return this.#pendingAcceptForResend(nowMs) ?? this.createAck(); } this.resetPeerState(); this.#ensureReliableCapacity(1); @@ -343,16 +386,12 @@ export class CultNetRudpSession { return response; } - /// The reply to a repeated Connect is the Accept still awaiting - /// acknowledgement, or an Ack once it was acknowledged. - #answerRepeatedConnect(packet: CultNetRudpPacket, nowMs: number): CultNetRudpPacket { - this.#applyAcknowledgements(packet); - this.#rememberReceived(packet.sequence); - this.#lastReceivedAtMs = nowMs; + /// The Accept still awaiting acknowledgement, resent. + #pendingAcceptForResend(nowMs: number): CultNetRudpPacket | undefined { const pendingAccept = [...this.#pendingReliable.values()] .find(pending => pending.packet.packetType === "accept"); if (!pendingAccept) { - return this.createAck(); + return undefined; } pendingAccept.lastSentAtMs = nowMs; return { @@ -503,11 +542,7 @@ export class CultNetRudpSession { return { delivered: [], readyToSend }; } - const isDuplicate = (packet.reliable ?? false) - && (this.#receivedSequences.has(packet.sequence) - || (this.#highestReceivedSequence !== undefined - && packet.sequence < this.#highestReceivedSequence - && this.#highestReceivedSequence - packet.sequence >= RUDP_RECEIVED_SEQUENCE_WINDOW)); + const isDuplicate = (packet.reliable ?? false) && this.#wasReceived(packet.sequence); if (packet.reliable ?? false) this.#rememberReceived(packet.sequence); if (isDuplicate) { return { delivered: [], readyToSend }; @@ -616,6 +651,10 @@ export class CultNetRudpSession { } dueResends(nowMs: number): CultNetRudpPacket[] { + const fresh = this.#abandonUnansweredConnect(nowMs); + if (fresh) { + return [fresh]; + } const due: CultNetRudpPacket[] = []; for (const pending of this.#pendingReliable.values()) { if (nowMs - pending.lastSentAtMs >= this.resendDelayMs) { @@ -626,6 +665,27 @@ export class CultNetRudpSession { return due.sort((left, right) => left.sequence - right.sequence); } + /// A Connect unanswered for `RUDP_CONNECT_ATTEMPT_MS` is replaced, not + /// retransmitted further, by a Connect with a newly drawn sequence and the + /// same payload. Nothing was ever sent in an unanswered generation, so no + /// sequence issued so far is owed; the draw stays above them, so a frame the + /// peer still remembers from an earlier generation of this session stays at or + /// before the new Connect. + #abandonUnansweredConnect(nowMs: number): CultNetRudpPacket | undefined { + if (!this.#awaitingAccept || nowMs - this.#connectStartedAtMs < RUDP_CONNECT_ATTEMPT_MS) { + return undefined; + } + const pending = this.#connectSequence === undefined + ? undefined + : this.#pendingReliable.get(this.#connectSequence); + if (!pending) { + return undefined; + } + const payload = new Uint8Array(pending.packet.payload ?? []); + this.#nextSequence = drawSequence(this.#nextSequence); + return this.createConnect(nowMs, payload); + } + #createPacket(packet: { packetType: CultNetRudpPacketType; channelId: string; @@ -713,13 +773,18 @@ export class CultNetRudpSession { this.#rememberReceived(sequence); } - /// True for a sequence in the window that was received, and for one below the - /// window, which `receive` treats as a duplicate of something received. + /// The one duplicate test: true for a sequence in the window that was + /// received, for one below the window, and for one at or before the + /// watermark. The watermark starts at the handshake's seed, so a frame the + /// peer sent in an earlier generation (every sequence it issued is below the + /// Connect that began this one) is a duplicate of something already delivered + /// and is acknowledged, never delivered again. #wasReceived(sequence: number): boolean { return this.#receivedSequences.has(sequence) || (this.#highestReceivedSequence !== undefined && sequence < this.#highestReceivedSequence - && this.#highestReceivedSequence - sequence >= RUDP_RECEIVED_SEQUENCE_WINDOW); + && this.#highestReceivedSequence - sequence >= RUDP_RECEIVED_SEQUENCE_WINDOW) + || (this.#receivedThrough !== undefined && atOrBefore(sequence, this.#receivedThrough)); } #rememberReceived(sequence: number): void { @@ -1064,6 +1129,10 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement return; } + // A Connect from another endpoint is a new client, whatever sequence it + // carries: the endpoint moves and the Connect starts a new generation. Only + // a Connect from the peer's own endpoint can be its repeat. + let connectFromNewEndpoint = false; if (!this.#remoteHost || this.#remotePort === undefined) { if (this.#mode === "server" && packet.packetType !== "connect") { this.#stats.packetsDropped += 1; @@ -1075,6 +1144,7 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement if (this.#mode === "server" && packet.packetType === "connect") { this.#remoteHost = remote.address; this.#remotePort = remote.port; + connectFromNewEndpoint = true; } else { this.#stats.packetsDropped += 1; return; @@ -1086,9 +1156,12 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement // repeat is answered with the Accept owed, anything else replaces the // peer and the old session's writes die with it. A timed-out session // already ended its generation; any other predecessor ends here. - if (!this.#session.connectRepeats(packet) && !this.#session.ended) { + if ((connectFromNewEndpoint || !this.#session.connectRepeats(packet)) && !this.#session.ended) { this.#endedReason = Buffer.from("replaced by a new Connect", "utf8"); } + if (connectFromNewEndpoint) { + this.#session.resetPeerState(); + } let accept: CultNetRudpPacket; try { accept = this.#session.acceptConnect(packet, Date.now()); diff --git a/packages/cultnet-ts/test/cultnet.test.ts b/packages/cultnet-ts/test/cultnet.test.ts index 3e40e1b6..4a0c3bbc 100644 --- a/packages/cultnet-ts/test/cultnet.test.ts +++ b/packages/cultnet-ts/test/cultnet.test.ts @@ -2593,7 +2593,7 @@ test("server-mode transport admits a restarted client and keeps its session when peerSocket.send(encodeRudpPacket(packet), udpPort(serverSocket), "127.0.0.1"); const reliable = { reliable: true, ordered: true }; try { - const first = new CultNetRudpSession({ connectionId, initialSequence: 50 }); + const first = new CultNetRudpSession({ connectionId, initialSequence: 50_000 }); const connect = first.createConnect(0, Buffer.from("same")); toServer(connect); await waitFor(() => received.some((p) => p.packetType === "accept"), "the first Accept"); @@ -2641,7 +2641,7 @@ test("RUDP operation service admits a restarted client on the same address", asy const connectionId = 0x43554c54; const port = Number(new URL(server.endpoint).port); const socket = await bindUdpSocket(); - let session = new CultNetRudpSession({ connectionId, initialSequence: 50 }); + let session = new CultNetRudpSession({ connectionId, initialSequence: 50_000 }); const answered: string[] = []; socket.on("message", (wire) => { const packet = decodeRudpPacket(wire); diff --git a/packages/cultnet-ts/test/rudp-connect-generation.test.ts b/packages/cultnet-ts/test/rudp-connect-generation.test.ts index 2170205a..47455165 100644 --- a/packages/cultnet-ts/test/rudp-connect-generation.test.ts +++ b/packages/cultnet-ts/test/rudp-connect-generation.test.ts @@ -2,7 +2,15 @@ // the handshake alone seeds the watermark that orders delivery. import assert from "node:assert/strict"; import test from "node:test"; -import { CultNetRudpSession, randomInitialSequence, type CultNetRudpPacket } from "../src"; +import { createSocket, type Socket } from "node:dgram"; +import { once } from "node:events"; +import { + CultNetRudpSession, + CultNetRudpSocketTransportConnection, + decodeRudpPacket, + encodeRudpPacket, + type CultNetRudpPacket, +} from "../src"; const connectionId = 0x10203080; const enc = (text: string) => new TextEncoder().encode(text); @@ -197,10 +205,214 @@ test("a peer Disconnect does not forget what the peer sent", () => { assert.deepEqual(server.receive(s2, 4).delivered, [], "the Disconnect forgot what was received"); }); -test("default initial sequences are drawn at random from one to two to the thirty first", () => { - const draws = new Set(Array.from({ length: 64 }, () => randomInitialSequence())); - assert.ok(draws.size > 32, `does not draw at random: ${[...draws]}`); - assert.ok([...draws].every((value) => value >= 1 && value < 2 ** 31), `left [1, 2^31): ${[...draws]}`); - const firsts = new Set(Array.from({ length: 64 }, () => new CultNetRudpSession({ connectionId }).createConnect(0).sequence)); +test("sessions from one options object each draw their own initial sequence from one to two to the thirty first", () => { + const options = { connectionId }; + const firsts = new Set(Array.from({ length: 64 }, () => new CultNetRudpSession(options).createConnect(0).sequence)); assert.ok(firsts.size > 32, `a default session does not draw its own sequence: ${[...firsts]}`); + assert.ok([...firsts].every((value) => value >= 1 && value < 2 ** 31), `left [1, 2^31): ${[...firsts]}`); +}); + +// Ack Cut 1d: a frame from an earlier generation is a duplicate, a delayed +// Connect from an earlier attempt is stale, a Connect from a new endpoint is a +// new client. + +const namesSequence = (ack: CultNetRudpPacket, sequence: number) => + ack.ack === sequence + || Array.from({ length: 32 }, (_, bit) => bit).some( + (bit) => (ack.ackMask & (1 << bit)) !== 0 && ack.ack - bit - 1 === sequence, + ); + +test("a frame from the client's earlier generation is acknowledged, not delivered again", () => { + const client = session(10); + const server = session(500); + handshake(client, server); + const old = send(client, "old"); + assert.deepEqual(names(server.receive(old, 1)), ["old"]); + + const connect = client.createConnect(2); + client.receive(server.acceptConnect(connect, 2), 2); + assert.ok(old.sequence < connect.sequence); + assert.deepEqual(server.receive(old, 3).delivered, [], "an earlier generation's frame was delivered again"); + assert.ok(namesSequence(server.createAckForReceived(old.sequence), old.sequence), "the duplicate was not acknowledged"); + assert.deepEqual(names(server.receive(send(client, "fresh"), 4)), ["fresh"]); +}); + +test("a frame from the server's earlier generation is acknowledged, not delivered again", () => { + const client = session(10); + const server = session(500); + handshake(client, server); + const old = send(server, "old"); + assert.deepEqual(names(client.receive(old, 1)), ["old"]); + + client.receive(server.acceptConnect(client.createConnect(2), 2), 2); + assert.deepEqual(client.receive(old, 3).delivered, [], "an earlier generation's frame was delivered again"); + assert.ok(namesSequence(client.createAckForReceived(old.sequence), old.sequence)); + assert.deepEqual(names(client.receive(send(server, "fresh"), 4)), ["fresh"]); +}); + +test("a duplicate below the receive window is acknowledged by name", () => { + const client = session(10); + const server = session(500); + handshake(client, server); + const first = send(client, "first"); + assert.deepEqual(names(server.receive(first, 1)), ["first"]); + for (let index = 0; index < 4_200; index += 1) { + server.receive(send(client, "x"), 1); + client.receive(server.createAck(), 1); + } + assert.deepEqual(server.receive(first, 2).delivered, []); + assert.equal(server.createAckForReceived(first.sequence).ack, first.sequence, "a duplicate below the window was not acknowledged by name"); +}); + +test("a delayed Connect from an earlier attempt does not strand the client", () => { + const client = session(10); + const server = session(500); + const earlier = client.createConnect(0); + const retried = client.createConnect(300); + const accept = server.acceptConnect(retried, 301); + client.receive(accept, 302); + server.receive(client.createAckForReceived(accept.sequence), 302); + assert.ok(client.connected); + const a = send(client, "a"); + assert.deepEqual(names(server.receive(a, 303)), ["a"]); + client.receive(server.createAckForReceived(a.sequence), 303); + + assert.equal(server.connectRepeats(earlier), true, "the delayed Connect starts nothing"); + const reply = server.acceptConnect(earlier, 304); + assert.equal(reply.packetType, "ack"); + client.receive(reply, 305); + assert.ok(client.connected); + + const b = send(client, "b"); + assert.deepEqual(names(server.receive(b, 306)), ["b"], "the delayed Connect reset the server"); + client.receive(server.createAckForReceived(b.sequence), 307); + assert.ok(!client.pendingReliableSequences.includes(b.sequence)); +}); + +test("a repeated Connect refreshes liveness and a stale one does not", () => { + const client = session(10); + const server = session(500); + const earlier = client.createConnect(0); + const current = client.createConnect(1); + server.acceptConnect(current, 0); + + server.acceptConnect(current, 900); + assert.equal(server.checkTimeout(1_000, 500), false, "a repeated Connect did not refresh liveness"); + + server.acceptConnect(earlier, 1_400); + assert.equal(server.checkTimeout(1_600, 500), true, "a stale Connect refreshed liveness"); +}); + +test("stale Connects are recognised by serial arithmetic", () => { + const server = session(500); + server.acceptConnect(session(3).createConnect(0), 0); + const connect = (sequence: number) => session(sequence).createConnect(0); + assert.equal(server.connectRepeats(connect(2 ** 32 - 2)), true, "just before the wrap"); + assert.equal(server.connectRepeats(connect(4)), false, "just after"); + assert.equal(server.connectRepeats(connect(3 + 4_096)), false, "far after"); + assert.equal(server.connectRepeats(connect(2 ** 32 - 4_092)), true, "the window's edge"); + assert.equal(server.connectRepeats(connect(2 ** 32 - 4_093)), false, "past the window"); +}); + +test("a restarted client whose first sequence is stale connects after redrawing", () => { + const server = session(500); + const old = session(1_000); + const oldAccept = server.acceptConnect(old.createConnect(0), 0); + old.receive(oldAccept, 0); + + const restarted = session(900); + const first = restarted.createConnect(0, enc("join")); + // The old client has not acknowledged its Accept yet, so that is the reply: + // it names the old Connect, not this one. + const early = server.acceptConnect(first, 1); + assert.equal(early.packetType, "accept"); + restarted.receive(early, 1); + assert.equal(restarted.connected, false, "an Accept for another Connect connected the client"); + server.receive(old.createAckForReceived(oldAccept.sequence), 1); + const reply = server.acceptConnect(first, 2); + assert.equal(reply.packetType, "ack"); + restarted.receive(reply, 2); + assert.equal(restarted.connected, false); + + const retransmitted = restarted.dueResends(1_000); + assert.equal(retransmitted.length, 1); + assert.equal(retransmitted[0]!.sequence, first.sequence, "the attempt is still young"); + + const fresh = restarted.dueResends(3_500); + assert.equal(fresh.length, 1); + assert.equal(fresh[0]!.packetType, "connect"); + assert.notEqual(fresh[0]!.sequence, first.sequence, "the same Connect was retransmitted for ever"); + assert.equal(dec(fresh[0]!.payload ?? new Uint8Array()), "join", "the fresh attempt lost the Connect payload"); + + const accept = server.acceptConnect(fresh[0]!, 3_500); + assert.equal(accept.packetType, "accept"); + restarted.receive(accept, 3_500); + assert.ok(restarted.connected); + assert.deepEqual(names(server.receive(send(restarted, "hello"), 3_501)), ["hello"]); +}); + +test("an answered Connect is never replaced by the attempt timeout", () => { + const client = session(10); + const server = session(500); + const { connect } = handshake(client, server); + const resent = client.dueResends(60_000); + assert.ok(resent.every((packet) => packet.packetType !== "connect" || packet.sequence === connect.sequence), "a connected client started a new attempt"); + assert.ok(client.connected); +}); + +async function bind(): Promise { + const socket = createSocket("udp4"); + socket.bind(0, "127.0.0.1"); + await once(socket, "listening"); + return socket; +} +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +test("server mode admits a pinned client that restarts on a new port", async () => { + const serverSocket = await bind(); + const a = await bind(); + const b = await bind(); + const server = new CultNetRudpSocketTransportConnection({ + runtimeId: "srv", socket: serverSocket, mode: "server", connectionId, resendPollMs: 20, resendDelayMs: 20, + }); + const frames: string[] = []; + server.on("frame", (frame: { payload: Uint8Array }) => frames.push(dec(frame.payload))); + const port = (serverSocket.address() as { port: number }).port; + const atA: CultNetRudpPacket[] = []; + const atB: CultNetRudpPacket[] = []; + a.on("message", (wire) => atA.push(decodeRudpPacket(wire))); + b.on("message", (wire) => atB.push(decodeRudpPacket(wire))); + try { + const first = session(1); + a.send(encodeRudpPacket(first.createConnect(0, enc("join"))), port, "127.0.0.1"); + await sleep(100); + const accept = atA.find((packet) => packet.packetType === "accept"); + assert.ok(accept); + first.receive(accept, 0); + a.send(encodeRudpPacket(first.createAckForReceived(accept.sequence)), port, "127.0.0.1"); + a.send(encodeRudpPacket(send(first, "hello")), port, "127.0.0.1"); + await sleep(100); + assert.deepEqual(frames, ["hello"]); + + const second = session(1); // the process restarted on a new port, pinned + b.send(encodeRudpPacket(second.createConnect(0, enc("join"))), port, "127.0.0.1"); + await sleep(100); + const acceptB = atB.find((packet) => packet.packetType === "accept"); + assert.ok(acceptB, `the restarted client on a new port was not admitted: ${JSON.stringify(atB.map((packet) => packet.packetType))}`); + second.receive(acceptB, 1); + assert.ok(second.connected); + b.send(encodeRudpPacket(send(second, "after")), port, "127.0.0.1"); + await sleep(100); + assert.deepEqual(frames, ["hello", "after"]); + } finally { + server.close(); + a.close(); + b.close(); + } +}); + +test("an accepted peer that goes silent times out", () => { + const server = session(500); + server.acceptConnect(session(10).createConnect(0), 0); + assert.equal(server.checkTimeout(100_000, 1_000), true); });