From 340df225d4fac5a0505e5b3f8d496c9d64b7994b Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Fri, 10 Jul 2026 06:56:49 +0100 Subject: [PATCH] Align TypeScript RUDP acknowledgement sequencing --- packages/cultnet-ts/src/rudp.ts | 67 +++++++++++++++++++++--- packages/cultnet-ts/test/cultnet.test.ts | 19 +++++++ 2 files changed, 78 insertions(+), 8 deletions(-) diff --git a/packages/cultnet-ts/src/rudp.ts b/packages/cultnet-ts/src/rudp.ts index 54b85e83..06c09bc3 100644 --- a/packages/cultnet-ts/src/rudp.ts +++ b/packages/cultnet-ts/src/rudp.ts @@ -177,6 +177,17 @@ export class CultNetRudpSession { return this.#lastReceivedAtMs; } + resetPeerState(): void { + this.#connected = false; + this.#lastReceivedAtMs = undefined; + this.#highestReceivedSequence = undefined; + this.#receivedSequences.clear(); + this.#pendingReliable.clear(); + this.#orderedNextSequenceByChannel.clear(); + this.#orderedBuffers.clear(); + this.#fragmentBuffers.clear(); + } + createConnect(nowMs = 0, payload = new Uint8Array()): CultNetRudpPacket { this.#ensureReliableCapacity(1); const packet = this.#createPacket({ @@ -196,8 +207,12 @@ export class CultNetRudpSession { throw new Error(`Expected RUDP connect packet, got ${packet.packetType}.`); } - this.#ensureReliableCapacity(1); + const wasConnected = this.#connected; + if (!wasConnected) { + this.#ensureReliableCapacity(1); + } this.#rememberReceived(packet.sequence); + this.#lastReceivedAtMs = nowMs; this.#connected = true; const response = this.#createPacket({ packetType: "accept", @@ -206,7 +221,9 @@ export class CultNetRudpSession { ordered: true, payload, }); - this.#trackReliable(response, nowMs); + if (!wasConnected) { + this.#trackReliable(response, nowMs); + } return response; } @@ -310,9 +327,9 @@ export class CultNetRudpSession { } if (packet.packetType === "ack" || packet.packetType === "pong") { - this.#rememberReceived(packet.sequence); + if (packet.packetType === "pong") this.#rememberReceived(packet.sequence); return { - delivered: [], + delivered: this.#drainUnblockedOrdered(), pong: packet.packetType === "pong", pongPayload: packet.packetType === "pong" ? packet.payload ?? new Uint8Array() : undefined, }; @@ -351,10 +368,22 @@ export class CultNetRudpSession { } createAck(): CultNetRudpPacket { - return this.#createPacket({ + const { ack, ackMask } = this.#ackState(); + return { packetType: "ack", + connectionId: this.connectionId, + sequence: 0, + ack, + ackMask, channelId: "control", - }); + reliable: false, + ordered: false, + sequenced: false, + fragmentId: 0, + fragmentIndex: 0, + fragmentCount: 0, + payload: new Uint8Array(), + }; } createPing(payload = new Uint8Array()): CultNetRudpPacket { @@ -606,6 +635,15 @@ export class CultNetRudpSession { return delivered; } + #drainUnblockedOrdered(): CultNetRudpDeliveredFrame[] { + const delivered: CultNetRudpDeliveredFrame[] = []; + for (const channelId of this.#orderedBuffers.keys()) { + this.#skipReceivedNonChannelSequences(channelId); + delivered.push(...this.#drainOrdered(channelId)); + } + return delivered; + } + #skipReceivedNonChannelSequences(channelId: string): void { let next = this.#orderedNextSequenceByChannel.get(channelId); while (next !== undefined && this.#receivedSequences.has(next) && !this.#orderedBuffers.get(channelId)?.has(next)) { @@ -727,6 +765,13 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement close(): void { clearInterval(this.#resendTimer); if (!this.#closed) { + if (this.#session.connected && this.#remoteHost && this.#remotePort !== undefined) { + try { + this.#sendPacket(this.#session.createDisconnect()); + } catch { + // Best-effort shutdown: close must still release the UDP socket. + } + } this.#closed = true; this.#socket.close(); } @@ -746,11 +791,17 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement this.#remoteHost = remote.address; this.#remotePort = remote.port; } else if (remote.address !== this.#remoteHost || remote.port !== this.#remotePort) { - return; + if (this.#mode === "server" && packet.packetType === "connect") { + this.#remoteHost = remote.address; + this.#remotePort = remote.port; + } else { + return; + } } try { if (this.#mode === "server" && packet.packetType === "connect") { + this.#session.resetPeerState(); this.#sendPacket(this.#session.acceptConnect(packet, Date.now())); return; } @@ -774,7 +825,7 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement this.emit("close"); return; } - if (packet.packetType === "accept" || result.delivered.length > 0) { + if (packet.packetType === "accept" || packet.packetType === "data" || result.delivered.length > 0) { this.#sendPacket(this.#session.createAck()); } } catch (error) { diff --git a/packages/cultnet-ts/test/cultnet.test.ts b/packages/cultnet-ts/test/cultnet.test.ts index 1d78072c..04107a0e 100644 --- a/packages/cultnet-ts/test/cultnet.test.ts +++ b/packages/cultnet-ts/test/cultnet.test.ts @@ -418,6 +418,7 @@ test("rudp session handshake acks reliable connect and accept packets", () => { assert.deepEqual(client.pendingReliableSequences, []); const ack = client.createAck(); + assert.equal(ack.sequence, 0); assert.equal(ack.ack, 100); server.receive(ack, 30); assert.deepEqual(server.pendingReliableSequences, []); @@ -533,6 +534,9 @@ test("rudp session skips received control packets while ordering schema payloads const control = sender.createAck(); const second = sender.send("schema", Buffer.from("second"), { reliable: true, ordered: true }); + assert.equal(control.sequence, 0); + assert.equal(second.sequence, first.sequence + 1); + assert.deepEqual(receiver.receive(first).delivered.map((frame) => Buffer.from(frame.payload).toString("utf8")), ["first"]); assert.deepEqual(receiver.receive(control).delivered, []); assert.deepEqual(receiver.receive(second).delivered.map((frame) => Buffer.from(frame.payload).toString("utf8")), ["second"]); @@ -1217,6 +1221,21 @@ test("CultNet raw replication preserves CultCache payload bytes for bit-compatib } }); +test("rudp sequence-neutral acknowledgements interoperate with ordered receivers", () => { + const sender = new CultNetRudpSession({ connectionId: 125, initialSequence: 1 }); + const receiver = new CultNetRudpSession({ connectionId: 125, initialSequence: 100 }); + sender.receive({ packetType: "accept", connectionId: 125, sequence: 90, ack: 0, ackMask: 0, channelId: "control" }); + receiver.receive({ packetType: "accept", connectionId: 125, sequence: 91, ack: 0, ackMask: 0, channelId: "control" }); + + const ack = sender.createAck(); + const request = sender.send("schema", Buffer.from("snapshot"), { reliable: true, ordered: true }); + + assert.equal(ack.sequence, 0); + assert.equal(request.sequence, 1); + assert.deepEqual(receiver.receive(ack).delivered, []); + assert.deepEqual(receiver.receive(request).delivered.map(frame => Buffer.from(frame.payload).toString("utf8")), ["snapshot"]); +}); + test("CultNet raw replication applies compatible foreign-schema snapshots", async () => { const tempDir = mkdtempSync(join(tmpdir(), "cultnetts-foreign-schema-raw-"));