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 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
e02d027019
commit
ba873d04c4
6 changed files with 330 additions and 46 deletions
|
|
@ -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,
|
||||
);
|
||||
|
|
|
|||
|
|
@ -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");
|
||||
|
|
|
|||
|
|
@ -106,7 +106,6 @@ export {
|
|||
createRudpTransportProfile,
|
||||
decodeRudpPacket,
|
||||
encodeRudpPacket,
|
||||
randomInitialSequence,
|
||||
CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS,
|
||||
CultNetRudpReconnectLoop,
|
||||
CultNetRudpSession,
|
||||
|
|
|
|||
|
|
@ -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());
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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<Socket> {
|
||||
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);
|
||||
});
|
||||
|
|
|
|||
Loading…
Reference in a new issue