Add TypeScript RUDP reliability core
This commit is contained in:
Meta Vens
parent
c72abaf904
commit
105f564e76
4 changed files with 380 additions and 1 deletions
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -69,8 +69,12 @@ export {
|
|||
createRudpTransportProfile,
|
||||
decodeRudpPacket,
|
||||
encodeRudpPacket,
|
||||
CultNetRudpSession,
|
||||
type CultNetRudpDeliveredFrame,
|
||||
type CultNetRudpPacket,
|
||||
type CultNetRudpPacketType,
|
||||
type CultNetRudpReceiveResult,
|
||||
type CultNetRudpSessionOptions,
|
||||
type RudpTransportProfileOptions,
|
||||
} from "./rudp";
|
||||
export {
|
||||
|
|
|
|||
|
|
@ -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<number, CultNetRudpPacketType>(
|
|||
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<number>();
|
||||
readonly #pendingReliable = new Map<number, PendingReliablePacket>();
|
||||
readonly #orderedNextSequenceByChannel = new Map<string, number>();
|
||||
readonly #orderedBuffers = new Map<string, Map<number, CultNetRudpDeliveredFrame>>();
|
||||
|
||||
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 = {},
|
||||
|
|
|
|||
|
|
@ -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"));
|
||||
|
|
|
|||
Loading…
Reference in a new issue