diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index f723e319..eeba171c 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -160,9 +160,15 @@ Current progress: the RUDP runtime yet; it is the binary packet language the runtimes must converge on before resend loops, windows, and timeout behavior are allowed to claim parity. +- TypeScript now owns the first deterministic RUDP reliability state machine: + connect/accept handshake, packet-level ack/ack-mask accounting, reliable + resend scheduling, duplicate suppression, and reliable ordered channel + delivery. It is in-memory and socket-free so the behavior can be ported before + any runtime binds it to UDP I/O. - The remaining parity work is to add the equivalent port to Kotlin, deepen Python's server-side use of its port, and move each runtime's existing - TCP/LiteNetLib/WebSocket bodies behind those ports, then implement `rudp`. + TCP/LiteNetLib/WebSocket bodies behind those ports, port the RUDP reliability + state machine into C#/Rust/Python, then bind it to UDP sockets. ## RUDP Packet Contract V0 diff --git a/packages/cultnet-ts/src/index.ts b/packages/cultnet-ts/src/index.ts index 56b6f413..7eea20a9 100644 --- a/packages/cultnet-ts/src/index.ts +++ b/packages/cultnet-ts/src/index.ts @@ -69,8 +69,12 @@ export { createRudpTransportProfile, decodeRudpPacket, encodeRudpPacket, + CultNetRudpSession, + type CultNetRudpDeliveredFrame, type CultNetRudpPacket, type CultNetRudpPacketType, + type CultNetRudpReceiveResult, + type CultNetRudpSessionOptions, type RudpTransportProfileOptions, } from "./rudp"; export { diff --git a/packages/cultnet-ts/src/rudp.ts b/packages/cultnet-ts/src/rudp.ts index 0290448f..f2e754d9 100644 --- a/packages/cultnet-ts/src/rudp.ts +++ b/packages/cultnet-ts/src/rudp.ts @@ -30,6 +30,28 @@ export interface CultNetRudpPacket { payload?: Uint8Array; } +export interface CultNetRudpDeliveredFrame { + channelId: string; + payload: Uint8Array; + sequence: number; +} + +export interface CultNetRudpSessionOptions { + connectionId: number; + initialSequence?: number; + resendDelayMs?: number; +} + +export interface CultNetRudpReceiveResult { + delivered: CultNetRudpDeliveredFrame[]; + reply?: CultNetRudpPacket; +} + +type PendingReliablePacket = { + packet: CultNetRudpPacket; + lastSentAtMs: number; +}; + export interface RudpTransportProfileOptions { transportId?: string; host?: string; @@ -52,6 +74,268 @@ const packetTypeFromCode = new Map( Object.entries(packetTypeToCode).map(([name, code]) => [code, name as CultNetRudpPacketType]), ); +export class CultNetRudpSession { + readonly connectionId: number; + readonly resendDelayMs: number; + #nextSequence: number; + #connected = false; + #highestReceivedSequence: number | undefined; + readonly #receivedSequences = new Set(); + readonly #pendingReliable = new Map(); + readonly #orderedNextSequenceByChannel = new Map(); + readonly #orderedBuffers = new Map>(); + + constructor(options: CultNetRudpSessionOptions) { + this.connectionId = toUint32(options.connectionId, "connectionId"); + this.#nextSequence = toUint32(options.initialSequence ?? 1, "initialSequence"); + this.resendDelayMs = options.resendDelayMs ?? 250; + } + + get connected(): boolean { + return this.#connected; + } + + get pendingReliableSequences(): number[] { + return [...this.#pendingReliable.keys()].sort((left, right) => left - right); + } + + createConnect(nowMs = 0, payload = new Uint8Array()): CultNetRudpPacket { + const packet = this.#createPacket({ + packetType: "connect", + channelId: "control", + reliable: true, + ordered: true, + payload, + }); + this.#trackReliable(packet, nowMs); + return packet; + } + + 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}.`); + } + + this.#rememberReceived(packet.sequence); + this.#connected = true; + const response = this.#createPacket({ + packetType: "accept", + channelId: "control", + reliable: true, + ordered: true, + payload, + }); + this.#trackReliable(response, nowMs); + return response; + } + + send( + channelId: string, + payload: Uint8Array, + options: { reliable?: boolean; ordered?: boolean; sequenced?: boolean; nowMs?: number } = {}, + ): CultNetRudpPacket { + if (!this.#connected) { + throw new Error("Cannot send RUDP data before the session is connected."); + } + + const packet = this.#createPacket({ + packetType: "data", + channelId, + payload, + reliable: options.reliable, + ordered: options.ordered, + sequenced: options.sequenced, + }); + if (packet.reliable) { + this.#trackReliable(packet, options.nowMs ?? 0); + } + return packet; + } + + receive(packet: CultNetRudpPacket, nowMs = 0): CultNetRudpReceiveResult { + this.#requireConnection(packet); + this.#applyAcknowledgements(packet); + + if (packet.packetType === "accept") { + this.#rememberReceived(packet.sequence); + this.#connected = true; + return { delivered: [] }; + } + + if (packet.packetType === "ping") { + this.#rememberReceived(packet.sequence); + return { + delivered: [], + reply: this.#createPacket({ + packetType: "pong", + channelId: "control", + payload: packet.payload ?? new Uint8Array(), + }), + }; + } + + if (packet.packetType === "ack" || packet.packetType === "pong") { + this.#rememberReceived(packet.sequence); + return { delivered: [] }; + } + + if (packet.packetType !== "data") { + return { delivered: [] }; + } + + const isDuplicate = this.#receivedSequences.has(packet.sequence); + this.#rememberReceived(packet.sequence); + if (isDuplicate) { + return { delivered: [] }; + } + + const frame: CultNetRudpDeliveredFrame = { + channelId: packet.channelId, + payload: packet.payload ?? new Uint8Array(), + sequence: packet.sequence, + }; + + if (!packet.ordered) { + return { delivered: [frame] }; + } + + return { delivered: this.#deliverOrdered(frame) }; + } + + createAck(): CultNetRudpPacket { + return this.#createPacket({ + packetType: "ack", + channelId: "control", + }); + } + + dueResends(nowMs: number): CultNetRudpPacket[] { + const due: CultNetRudpPacket[] = []; + for (const pending of this.#pendingReliable.values()) { + if (nowMs - pending.lastSentAtMs >= this.resendDelayMs) { + pending.lastSentAtMs = nowMs; + due.push({ ...pending.packet }); + } + } + return due.sort((left, right) => left.sequence - right.sequence); + } + + #createPacket(packet: { + packetType: CultNetRudpPacketType; + channelId: string; + payload?: Uint8Array; + reliable?: boolean; + ordered?: boolean; + sequenced?: boolean; + }): CultNetRudpPacket { + const sequence = this.#nextSequence; + this.#nextSequence = toUint32(this.#nextSequence + 1, "sequence"); + const { ack, ackMask } = this.#ackState(); + return { + packetType: packet.packetType, + connectionId: this.connectionId, + sequence, + ack, + ackMask, + channelId: packet.channelId, + reliable: packet.reliable ?? false, + ordered: packet.ordered ?? false, + sequenced: packet.sequenced ?? false, + payload: packet.payload ?? new Uint8Array(), + }; + } + + #trackReliable(packet: CultNetRudpPacket, nowMs: number): void { + this.#pendingReliable.set(packet.sequence, { + packet: { ...packet, payload: packet.payload ? new Uint8Array(packet.payload) : new Uint8Array() }, + lastSentAtMs: nowMs, + }); + } + + #applyAcknowledgements(packet: CultNetRudpPacket): void { + this.#pendingReliable.delete(packet.ack); + for (let bit = 0; bit < 32; bit += 1) { + if ((packet.ackMask & (1 << bit)) !== 0) { + this.#pendingReliable.delete(packet.ack - bit - 1); + } + } + } + + #rememberReceived(sequence: number): void { + this.#receivedSequences.add(sequence); + if (this.#highestReceivedSequence === undefined || sequence > this.#highestReceivedSequence) { + this.#highestReceivedSequence = sequence; + } + } + + #ackState(): { ack: number; ackMask: number } { + const ack = this.#highestReceivedSequence ?? 0; + let ackMask = 0; + for (let bit = 0; bit < 32; bit += 1) { + if (ack > bit && this.#receivedSequences.has(ack - bit - 1)) { + ackMask |= 1 << bit; + } + } + return { ack, ackMask: ackMask >>> 0 }; + } + + #deliverOrdered(frame: CultNetRudpDeliveredFrame): CultNetRudpDeliveredFrame[] { + const next = this.#orderedNextSequenceByChannel.get(frame.channelId); + if (next === undefined) { + this.#orderedNextSequenceByChannel.set(frame.channelId, frame.sequence + 1); + return [ + frame, + ...this.#drainOrdered(frame.channelId), + ]; + } + + if (frame.sequence < next) { + return []; + } + + if (frame.sequence > next) { + let buffer = this.#orderedBuffers.get(frame.channelId); + if (!buffer) { + buffer = new Map(); + this.#orderedBuffers.set(frame.channelId, buffer); + } + buffer.set(frame.sequence, frame); + return []; + } + + this.#orderedNextSequenceByChannel.set(frame.channelId, next + 1); + return [ + frame, + ...this.#drainOrdered(frame.channelId), + ]; + } + + #drainOrdered(channelId: string): CultNetRudpDeliveredFrame[] { + const delivered: CultNetRudpDeliveredFrame[] = []; + const buffer = this.#orderedBuffers.get(channelId); + if (!buffer) { + return delivered; + } + + let next = this.#orderedNextSequenceByChannel.get(channelId); + while (next !== undefined && buffer.has(next)) { + const frame = buffer.get(next)!; + buffer.delete(next); + delivered.push(frame); + next += 1; + this.#orderedNextSequenceByChannel.set(channelId, next); + } + return delivered; + } + + #requireConnection(packet: CultNetRudpPacket): void { + if (packet.connectionId !== this.connectionId) { + throw new Error(`RUDP packet connection id ${packet.connectionId} does not match ${this.connectionId}.`); + } + } +} + export function createRudpTransportProfile( runtimeId: string, options: RudpTransportProfileOptions = {}, diff --git a/packages/cultnet-ts/test/cultnet.test.ts b/packages/cultnet-ts/test/cultnet.test.ts index 9ed0a1fd..4f17b1a0 100644 --- a/packages/cultnet-ts/test/cultnet.test.ts +++ b/packages/cultnet-ts/test/cultnet.test.ts @@ -16,6 +16,7 @@ import { CultNetClientSecurityOptions, CultNetDocumentRegistry, CultNetPeer, + CultNetRudpSession, CultNetSchemaRegistry, CultNetSecret, CultNetServerSecurityOptions, @@ -250,6 +251,90 @@ test("rudp transport profile advertises state and realtime channel semantics", ( ); }); +test("rudp session handshake acks reliable connect and accept packets", () => { + const client = new CultNetRudpSession({ connectionId: 0x0a0b0c0d, initialSequence: 1, resendDelayMs: 50 }); + const server = new CultNetRudpSession({ connectionId: 0x0a0b0c0d, initialSequence: 100, resendDelayMs: 50 }); + + const connect = client.createConnect(0, Buffer.from("join", "utf8")); + assert.equal(connect.packetType, "connect"); + assert.equal(connect.sequence, 1); + assert.deepEqual(client.pendingReliableSequences, [1]); + + const accept = server.acceptConnect(connect, 10, Buffer.from("ok", "utf8")); + assert.equal(accept.packetType, "accept"); + assert.equal(accept.ack, 1); + assert.equal(server.connected, true); + assert.deepEqual(server.pendingReliableSequences, [100]); + + client.receive(accept, 20); + assert.equal(client.connected, true); + assert.deepEqual(client.pendingReliableSequences, []); + + const ack = client.createAck(); + assert.equal(ack.ack, 100); + server.receive(ack, 30); + assert.deepEqual(server.pendingReliableSequences, []); +}); + +test("rudp session computes ack masks and clears pending reliable packets", () => { + const sender = new CultNetRudpSession({ connectionId: 7, initialSequence: 10, resendDelayMs: 100 }); + const receiver = new CultNetRudpSession({ connectionId: 7, initialSequence: 200, resendDelayMs: 100 }); + sender.receive({ packetType: "accept", connectionId: 7, sequence: 1, ack: 0, ackMask: 0, channelId: "control" }); + receiver.receive({ packetType: "accept", connectionId: 7, sequence: 2, ack: 0, ackMask: 0, channelId: "control" }); + + const first = sender.send("schema", Buffer.from("first"), { reliable: true, ordered: true, nowMs: 0 }); + const second = sender.send("schema", Buffer.from("second"), { reliable: true, ordered: true, nowMs: 0 }); + const third = sender.send("schema", Buffer.from("third"), { reliable: true, ordered: true, nowMs: 0 }); + assert.deepEqual(sender.pendingReliableSequences, [10, 11, 12]); + + receiver.receive(first); + receiver.receive(third); + const ackWithGap = receiver.createAck(); + assert.equal(ackWithGap.ack, 12); + assert.equal(ackWithGap.ackMask, 0b10 | (1 << 9)); + sender.receive(ackWithGap); + assert.deepEqual(sender.pendingReliableSequences, [11]); + + receiver.receive(second); + const fullAck = receiver.createAck(); + assert.equal(fullAck.ack, 12); + assert.equal(fullAck.ackMask, 0b11 | (1 << 9)); + sender.receive(fullAck); + assert.deepEqual(sender.pendingReliableSequences, []); +}); + +test("rudp session schedules reliable resends until acked", () => { + const session = new CultNetRudpSession({ connectionId: 99, initialSequence: 1, resendDelayMs: 100 }); + session.receive({ packetType: "accept", connectionId: 99, sequence: 50, ack: 0, ackMask: 0, channelId: "control" }); + const sent = session.send("schema", Buffer.from("payload"), { reliable: true, ordered: true, nowMs: 10 }); + + assert.deepEqual(session.dueResends(90), []); + assert.deepEqual(session.dueResends(110).map((packet) => packet.sequence), [sent.sequence]); + assert.deepEqual(session.dueResends(150), []); + + session.receive({ packetType: "ack", connectionId: 99, sequence: 51, ack: sent.sequence, ackMask: 0, channelId: "control" }); + assert.deepEqual(session.dueResends(250), []); +}); + +test("rudp session suppresses duplicates and delivers reliable ordered payloads in sequence", () => { + const sender = new CultNetRudpSession({ connectionId: 123, initialSequence: 1 }); + const receiver = new CultNetRudpSession({ connectionId: 123, initialSequence: 100 }); + sender.receive({ packetType: "accept", connectionId: 123, sequence: 90, ack: 0, ackMask: 0, channelId: "control" }); + receiver.receive({ packetType: "accept", connectionId: 123, sequence: 91, ack: 0, ackMask: 0, channelId: "control" }); + + const first = sender.send("schema", Buffer.from("first"), { reliable: true, ordered: true }); + const second = sender.send("schema", Buffer.from("second"), { reliable: true, ordered: true }); + const third = sender.send("schema", Buffer.from("third"), { reliable: true, ordered: true }); + + assert.deepEqual(receiver.receive(first).delivered.map((frame) => Buffer.from(frame.payload).toString("utf8")), ["first"]); + assert.deepEqual(receiver.receive(third).delivered, []); + assert.deepEqual(receiver.receive(first).delivered, []); + assert.deepEqual(receiver.receive(second).delivered.map((frame) => Buffer.from(frame.payload).toString("utf8")), [ + "second", + "third", + ]); +}); + test("CultNet peer can speak through a transport connection", async () => { const { a, b } = createDuplexPair(); const leftTransport = new TcpFramedTransportConnection(a, createTcpFramedTransportProfile("left"));