diff --git a/packages/cultmesh-ts/test/provider-rudp-transport.test.ts b/packages/cultmesh-ts/test/provider-rudp-transport.test.ts index e35a42a8..f1dd5d2d 100644 --- a/packages/cultmesh-ts/test/provider-rudp-transport.test.ts +++ b/packages/cultmesh-ts/test/provider-rudp-transport.test.ts @@ -360,6 +360,41 @@ test("a client restarted on the same address with the same Connect payload repla } }); +test("the RUDP document server does not acknowledge by name a reliable packet its session did not receive", async () => { + const connectionId = 0x43554c55; + const server = CultMesh.createRudpDocumentServer("odin-ack-test", connectionId, { + bindPort: 0, + documents: new CultNetDocumentRegistry(), + onError: () => undefined, + }); + await server.start(); + const socket = createSocket("udp4"); + const received: CultNetRudpPacket[] = []; + socket.on("message", wire => received.push(decodeRudpPacket(wire))); + const toServer = (packet: CultNetRudpPacket) => + socket.send(encodeRudpPacket(packet), server.bind.port, "127.0.0.1"); + try { + const client = new CultNetRudpSession({ connectionId, initialSequence: 50_000 }); + const connect = client.createConnect(Date.now()); + toServer(connect); + await waitFor(() => received.some(packet => packet.packetType === "accept"), "the Accept"); + received.length = 0; + + // A reliable packet the session ignores (a client does not send an Accept). + // The session never received it, so the Ack must not name it. + const ignored: CultNetRudpPacket = { ...client.createAck(), packetType: "accept", reliable: true, sequence: connect.sequence + 5 }; + toServer(ignored); + await waitFor(() => received.some(packet => packet.packetType === "ack"), "the acknowledgement"); + const ack = received.find(packet => packet.packetType === "ack")!; + const names = ack.ack === ignored.sequence + || Array.from({ length: 32 }, (_, bit) => bit).some(bit => (ack.ackMask & (1 << bit)) !== 0 && ack.ack - bit - 1 === ignored.sequence); + assert.equal(names, false, "the server acknowledged by name a packet its session refused"); + } finally { + socket.close(); + server.close(); + } +}); + function deferred(): { promise: Promise; resolve(value: T): void } { let resolve!: (value: T) => void; return { promise: new Promise(done => resolve = done), resolve }; diff --git a/packages/cultnet-ts/src/rudp.ts b/packages/cultnet-ts/src/rudp.ts index bfe94973..e32bb2d7 100644 --- a/packages/cultnet-ts/src/rudp.ts +++ b/packages/cultnet-ts/src/rudp.ts @@ -25,14 +25,11 @@ const RUDP_RECEIVED_SEQUENCE_WINDOW = 4_096; /// retransmitting the same sequence would never change its mind. const RUDP_CONNECT_ATTEMPT_MS = 3_000; -/// 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); +/// A 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. +function drawSequence(): number { + return randomInt(1, 2 ** 31); } /// Whether `sequence` is at or before `mark` in serial order, within the @@ -189,6 +186,9 @@ export class CultNetRudpSession { #connectSequence: number | undefined; /// A Connect this side sent is unanswered. Only then is an Accept honoured. #awaitingAccept = false; + /// The payload of the Connect this side sent, kept so a fresh attempt can + /// always be built whatever became of the pending packet. + #connectPayload = new Uint8Array(); /// When the unanswered Connect first went out; `dueResends` abandons it for a /// fresh attempt once `RUDP_CONNECT_ATTEMPT_MS` has passed. #connectStartedAtMs = 0; @@ -218,7 +218,7 @@ export class CultNetRudpSession { constructor(options: CultNetRudpSessionOptions) { this.connectionId = toUint32(options.connectionId, "connectionId"); - this.#nextSequence = toUint32(options.initialSequence ?? drawSequence(1), "initialSequence"); + this.#nextSequence = toUint32(options.initialSequence ?? drawSequence(), "initialSequence"); if (this.#nextSequence === 0xffff_ffff) { throw new Error("RUDP initialSequence must leave room for a reliable packet."); } @@ -317,6 +317,7 @@ export class CultNetRudpSession { this.#connectSequence = packet.sequence; this.#awaitingAccept = true; this.#connectStartedAtMs = nowMs; + this.#connectPayload = new Uint8Array(payload); this.#trackReliable(packet, nowMs); return packet; } @@ -489,7 +490,12 @@ export class CultNetRudpSession { if (packet.packetType === "accept" && !honoursAccept) { return { delivered: [], readyToSend: [] }; } - this.#applyAcknowledgements(packet); + // While this side's Connect awaits its Accept, only the Accept it honours + // retires it: an Ack that names the Connect (a server's reply to a repeat or + // a stale copy) says the server did not start a session. + if (honoursAccept || !this.#awaitingAccept) { + this.#applyAcknowledgements(packet); + } const readyToSend = this.#promoteQueuedReliable(nowMs); this.#lastReceivedAtMs = nowMs; @@ -666,24 +672,24 @@ export class CultNetRudpSession { } /// 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. + /// retransmitted further, by a Connect with the same payload and the + /// abandoned sequence plus the receive window less one. Nothing was ever sent + /// in an unanswered generation, so no sequence issued so far is owed. The jump + /// keeps a late copy of the abandoned Connect inside the new one's stale + /// window, so a server that took the new Connect answers the copy with an Ack + /// instead of restarting; and it leaves the stale window of whatever + /// generation the server holds, unless that generation sits exactly at the + /// jump, and then the next attempt leaves it. The first Connect of a session + /// is the only one drawn at random. #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) { + if (this.#connectSequence === undefined) { return undefined; } - const payload = new Uint8Array(pending.packet.payload ?? []); - this.#nextSequence = drawSequence(this.#nextSequence); - return this.createConnect(nowMs, payload); + this.#nextSequence = Math.min(this.#connectSequence + RUDP_RECEIVED_SEQUENCE_WINDOW - 1, 0xffff_fffe); + return this.createConnect(nowMs, this.#connectPayload); } #createPacket(packet: { diff --git a/packages/cultnet-ts/test/rudp-connect-generation.test.ts b/packages/cultnet-ts/test/rudp-connect-generation.test.ts index 47455165..cea61a6d 100644 --- a/packages/cultnet-ts/test/rudp-connect-generation.test.ts +++ b/packages/cultnet-ts/test/rudp-connect-generation.test.ts @@ -416,3 +416,120 @@ test("an accepted peer that goes silent times out", () => { server.acceptConnect(session(10).createConnect(0), 0); assert.equal(server.checkTimeout(100_000, 1_000), true); }); + +// Ack Cut 1d, batch 2: the connect-attempt timeout. + +test("an Ack naming the pending Connect does not retire it", () => { + const server = session(500); + const old = session(1); + const oldAccept = server.acceptConnect(old.createConnect(0), 0); + old.receive(oldAccept, 0); + server.receive(old.createAckForReceived(oldAccept.sequence), 0); + + const restarted = session(1); + const first = restarted.createConnect(0, enc("join")); + const reply = server.acceptConnect(first, 1); + assert.equal(reply.packetType, "ack"); + assert.ok(namesSequence(reply, first.sequence), "the Ack must name the pending Connect for this test to bite"); + restarted.receive(reply, 1); + assert.equal(restarted.connected, false); + assert.deepEqual(restarted.pendingReliableSequences, [first.sequence], "an Ack retired the Connect"); + + const fresh = restarted.dueResends(3_000); + assert.equal(fresh.length, 1, "the client waits for ever with nothing to resend"); + assert.equal(dec(fresh[0]!.payload ?? new Uint8Array()), "join"); + const accept = server.acceptConnect(fresh[0]!, 3_000); + assert.equal(accept.packetType, "accept"); + restarted.receive(accept, 3_000); + assert.ok(restarted.connected); +}); + +test("a late copy of an abandoned Connect is stale once its replacement is accepted", () => { + const server = session(500); + const client = session(10); + const abandoned = client.createConnect(0); + const fresh = client.dueResends(3_000); + assert.equal(fresh.length, 1); + assert.equal(fresh[0]!.sequence, abandoned.sequence + 4_095); + const accept = server.acceptConnect(fresh[0]!, 3_001); + client.receive(accept, 3_002); + server.receive(client.createAckForReceived(accept.sequence), 3_002); + assert.ok(client.connected); + const a = send(client, "a"); + assert.deepEqual(names(server.receive(a, 3_003)), ["a"]); + client.receive(server.createAckForReceived(a.sequence), 3_003); + + assert.equal(server.connectRepeats(abandoned), true, "the late copy would restart the server"); + const reply = server.acceptConnect(abandoned, 3_004); + assert.equal(reply.packetType, "ack"); + client.receive(reply, 3_005); + assert.ok(client.connected); + assert.deepEqual(names(server.receive(send(client, "b"), 3_006)), ["b"]); +}); + +test("a server sitting at the jump is left by the next attempt", () => { + const server = session(500); + const old = session(5_095); + const oldAccept = server.acceptConnect(old.createConnect(0), 0); + old.receive(oldAccept, 0); + server.receive(old.createAckForReceived(oldAccept.sequence), 0); + + const restarted = session(1_000); + const first = restarted.createConnect(0, enc("join")); + restarted.receive(server.acceptConnect(first, 1), 1); + assert.equal(restarted.connected, false); + + const second = restarted.dueResends(3_000)[0]!; + assert.equal(second.sequence, 5_095, "the jump lands on the server's generation"); + const reply = server.acceptConnect(second, 3_001); + assert.equal(reply.packetType, "ack"); + restarted.receive(reply, 3_001); + assert.equal(restarted.connected, false); + + const third = restarted.dueResends(6_000)[0]!; + assert.equal(third.sequence, 5_095 + 4_095); + const accept = server.acceptConnect(third, 6_001); + assert.equal(accept.packetType, "accept"); + restarted.receive(accept, 6_001); + assert.ok(restarted.connected); +}); + +test("server mode admits a pinned client that restarts on the same address", async () => { + const serverSocket = await bind(); + const peer = await bind(); + const server = new CultNetRudpSocketTransportConnection({ + runtimeId: "srv", socket: serverSocket, mode: "server", connectionId, resendPollMs: 20, resendDelayMs: 20, + }); + const port = (serverSocket.address() as { port: number }).port; + const received: CultNetRudpPacket[] = []; + peer.on("message", (wire) => received.push(decodeRudpPacket(wire))); + const take = async (type: string) => { + for (let index = 0; index < 50; index += 1) { + const found = received.findIndex((packet) => packet.packetType === type); + if (found >= 0) return received.splice(found, 1)[0]!; + await sleep(10); + } + throw new Error(`no ${type} arrived`); + }; + try { + const first = session(1); + peer.send(encodeRudpPacket(first.createConnect(0, enc("join"))), port, "127.0.0.1"); + const accept = await take("accept"); + first.receive(accept, 0); + peer.send(encodeRudpPacket(first.createAckForReceived(accept.sequence)), port, "127.0.0.1"); + await sleep(50); + received.length = 0; + + const restarted = session(1); + peer.send(encodeRudpPacket(restarted.createConnect(0, enc("join"))), port, "127.0.0.1"); + restarted.receive(await take("ack"), 1); + assert.equal(restarted.connected, false); + + peer.send(encodeRudpPacket(restarted.dueResends(3_000)[0]!), port, "127.0.0.1"); + restarted.receive(await take("accept"), 3_001); + assert.ok(restarted.connected, "the restarted pinned client was never admitted"); + } finally { + server.close(); + peer.close(); + } +});