From 3805352dbbf4cd96a7e851248bb227a64d68bc19 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 16 Jun 2026 05:58:28 +0100 Subject: [PATCH] Add TypeScript RUDP reconnect loop --- docs/cultnet-transport-parity.md | 5 ++ docs/runtime-parity-scope.md | 7 +- packages/cultnet-ts/README.md | 3 + packages/cultnet-ts/src/index.ts | 3 + packages/cultnet-ts/src/rudp.ts | 107 +++++++++++++++++++++++ packages/cultnet-ts/test/cultnet.test.ts | 64 ++++++++++++++ 6 files changed, 186 insertions(+), 3 deletions(-) diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index 63c77330..0aec5081 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -238,6 +238,11 @@ Current progress: counter and duplicate backoff constants. It still owns LiteNetLib socket mechanics and UI-facing reconnect state; the controller owns retry attempt scheduling. +- TypeScript now has a `CultNetRudpReconnectLoop` that consumes the shared + controller while keeping socket construction caller-owned. A closed RUDP + transport schedules the next attempt with the portable policy, opens a fresh + caller-provided transport when the controller says it may retry, and resets + attempt state when the caller marks the connection established. - Kotlin now mirrors the C#/TypeScript/Python local authority lease gate: peer cards may advertise `authorityLeaseId`, but `CultMeshAuthorityLeaseCatalog` owns role/shard/time authorization so discovery contact cannot impersonate diff --git a/docs/runtime-parity-scope.md b/docs/runtime-parity-scope.md index 604f83fc..e453d38a 100644 --- a/docs/runtime-parity-scope.md +++ b/docs/runtime-parity-scope.md @@ -32,7 +32,7 @@ CultMesh. | --- | --- | --- | --- | | C# | Reference implementation and production game/server substrate | CultCache typed documents and SoA, CultNet schema-v0, LiteNetLib production path, shared transport profiles, shared RUDP socket transport, RUDP reconnect policy/controller, LiteNetLib client reconnect-controller adoption, CultMesh node/session/catalog/stream/simulation surfaces | No need to abandon LiteNetLib before the CultNet adapter fully owns production service adoption | | Python | Package runtime, local daemon endpoint, raw interop peer, typed state participant | CultCache/CultNet/CultMesh package surface, schema catalogs, shard catalogs/logs, local TCP CultMesh server, session/prediction/simulation fact helpers, authority leases, streams, RUDP codec/session/socket transport, RUDP reconnect policy/controller, schema-v0 MessagePack over RUDP evidence, benchmark evidence against C# cache paths | Universal C# throughput, full C# daemon/server stack, and RUDP as the only public service pipe | -| TypeScript | Node/browser-adjacent projection and interop runtime | CultCache parity tests, CultNet schema-v0 peer, transport profiles, TCP-framed transport port, shared RUDP socket transport, RUDP reconnect policy/controller, local CultMesh node/catalog/authority/stream ergonomics | Full C# game session/server stack | +| TypeScript | Node/browser-adjacent projection and interop runtime | CultCache parity tests, CultNet schema-v0 peer, transport profiles, TCP-framed transport port, shared RUDP socket transport, RUDP reconnect policy/controller, RUDP reconnect-loop helper, local CultMesh node/catalog/authority/stream ergonomics | Full C# game session/server stack | | Rust | Low-level cache/transport/runtime substrate | CultCache/CultNet interop lane, transport profiles, TCP-framed port, shared RUDP codec/session/socket transport, RUDP reconnect policy/controller, schema-v0 MessagePack over RUDP evidence, small CultMesh facade for peer cards, authority leases, RUDP endpoint parsing, and peer-card RUDP client construction | Full CultMesh game-session/server facade, stream catalog, or production game-session server | | Kotlin | JVM/Android client substrate | Typed MessagePack cache, schema/shard catalogs, authority leases, stream catalog, branded `CultMesh` facade, RUDP client/server sugar, RUDP reconnect policy/controller, schema-v0 MessagePack over RUDP evidence | C# SoA store, full C# server daemon, or Unity hot-loop memory layout | @@ -63,8 +63,9 @@ CultMesh. - Extend RUDP interop evidence through daemon/socket reconnect loops and the wider schema-message matrix. RUDP profiles now advertise the portable reconnect policy, runtime-local tests prove the shared controller across the - targeted runtimes, and the C# LiteNetLib client uses that controller; broader - service adoption remains a separate claim. + targeted runtimes, and the C# LiteNetLib client plus TypeScript RUDP + reconnect loop use that controller; broader service adoption remains a + separate claim. - Keep Kotlin ergonomic parity moving with Android/JVM-shaped APIs instead of backfilling C# server responsibilities it should not own. - Keep performance claims evidence-bound: C# owns SoA; Python has benchmark diff --git a/packages/cultnet-ts/README.md b/packages/cultnet-ts/README.md index 8dab828d..d9b96e6c 100644 --- a/packages/cultnet-ts/README.md +++ b/packages/cultnet-ts/README.md @@ -21,6 +21,9 @@ The library currently includes: - `CultCacheTS` replication helpers for document put/delete/snapshot flows - schema discovery catalog messages and a registry for asking what contracts a peer can actually exchange safely +- RUDP socket transport, reconnect policy helpers, and + `CultNetRudpReconnectLoop` for controller-owned retry scheduling around a + caller-owned socket factory - generated TypeScript contracts and runtime verifiers from canonical JSON schemas owned elsewhere in the swarm - an exact mirrored runtime validator and generated TS contract surface for diff --git a/packages/cultnet-ts/src/index.ts b/packages/cultnet-ts/src/index.ts index 0ffc98f4..6b9d1bb8 100644 --- a/packages/cultnet-ts/src/index.ts +++ b/packages/cultnet-ts/src/index.ts @@ -75,11 +75,14 @@ export { createRudpTransportProfile, decodeRudpPacket, encodeRudpPacket, + CultNetRudpReconnectLoop, CultNetRudpSession, CultNetRudpSocketTransportConnection, type CultNetRudpDeliveredFrame, type CultNetRudpPacket, type CultNetRudpPacketType, + type CultNetRudpReconnectLoopOptions, + type CultNetRudpReconnectTransport, type CultNetRudpReceiveResult, type CultNetRudpSocketTransportOptions, type CultNetRudpSessionOptions, diff --git a/packages/cultnet-ts/src/rudp.ts b/packages/cultnet-ts/src/rudp.ts index 21631a42..72707c99 100644 --- a/packages/cultnet-ts/src/rudp.ts +++ b/packages/cultnet-ts/src/rudp.ts @@ -3,7 +3,9 @@ import { type RemoteInfo, type Socket } from "node:dgram"; import type { CultNetTransportProfile } from "./contracts"; import { + CultNetReconnectController, createCultNetReconnectPolicy, + type CultNetReconnectDecision, type CultNetReconnectPolicy, type CultNetTransportConnection, type CultNetTransportFrame, @@ -109,6 +111,21 @@ export interface CultNetRudpSocketTransportOptions { reconnectPolicy?: CultNetReconnectPolicy; } +export interface CultNetRudpReconnectTransport extends EventEmitter { + connect(payload?: Uint8Array): void; + close(): void; +} + +export interface CultNetRudpReconnectLoopOptions { + reconnectPolicy?: CultNetReconnectPolicy; + createTransport: () => CultNetRudpReconnectTransport; + connectPayload?: Uint8Array; + nowMs?: () => number; + jitterMs?: () => number; + setTimer?: (callback: () => void, delayMs: number) => unknown; + clearTimer?: (handle: unknown) => void; +} + const packetTypeToCode: Record = { connect: 1, accept: 2, @@ -767,6 +784,96 @@ export class CultNetRudpSocketTransportConnection extends EventEmitter implement } } +export class CultNetRudpReconnectLoop extends EventEmitter { + readonly reconnectController: CultNetReconnectController; + readonly #createTransport: () => CultNetRudpReconnectTransport; + readonly #connectPayload: Uint8Array; + readonly #nowMs: () => number; + readonly #jitterMs: () => number; + readonly #setTimer: (callback: () => void, delayMs: number) => unknown; + readonly #clearTimer: (handle: unknown) => void; + #transport: CultNetRudpReconnectTransport | undefined; + #timer: unknown; + #stopped = true; + + constructor(options: CultNetRudpReconnectLoopOptions) { + super(); + this.reconnectController = new CultNetReconnectController(options.reconnectPolicy ?? createCultNetReconnectPolicy()); + this.#createTransport = options.createTransport; + this.#connectPayload = options.connectPayload ?? new Uint8Array(); + this.#nowMs = options.nowMs ?? Date.now; + this.#jitterMs = options.jitterMs ?? (() => 0); + this.#setTimer = options.setTimer ?? ((callback, delayMs) => setTimeout(callback, delayMs)); + this.#clearTimer = options.clearTimer ?? ((handle) => clearTimeout(handle as NodeJS.Timeout)); + } + + get transport(): CultNetRudpReconnectTransport | undefined { + return this.#transport; + } + + start(): CultNetRudpReconnectTransport { + this.#stopped = false; + this.reconnectController.reset(); + return this.#openTransport(); + } + + stop(): void { + this.#stopped = true; + if (this.#timer !== undefined) { + this.#clearTimer(this.#timer); + this.#timer = undefined; + } + const transport = this.#transport; + this.#transport = undefined; + transport?.close(); + this.reconnectController.reset(); + } + + markConnected(): void { + this.reconnectController.reset(); + } + + #openTransport(): CultNetRudpReconnectTransport { + const transport = this.#createTransport(); + this.#transport = transport; + transport.once("close", () => { + if (this.#transport === transport) { + this.#transport = undefined; + } + this.#scheduleReconnect(); + }); + transport.on("error", (error) => this.emit("error", error)); + this.emit("transport", transport); + try { + transport.connect(this.#connectPayload); + } catch (error) { + this.emit("error", error instanceof Error ? error : new Error(String(error))); + this.#scheduleReconnect(); + } + return transport; + } + + #scheduleReconnect(): void { + if (this.#stopped || this.#timer !== undefined) { + return; + } + const decision = this.reconnectController.recordFailure(this.#nowMs(), this.#jitterMs()); + this.emit("reconnectScheduled", decision satisfies CultNetReconnectDecision); + if (!decision.shouldRetry) { + this.emit("reconnectExhausted", decision); + return; + } + this.#timer = this.#setTimer(() => { + this.#timer = undefined; + if (this.#stopped || !this.reconnectController.canAttempt(this.#nowMs())) { + return; + } + this.emit("reconnecting", decision); + this.#openTransport(); + }, decision.delayMs); + } +} + 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 a133fd1a..c9a75b7f 100644 --- a/packages/cultnet-ts/test/cultnet.test.ts +++ b/packages/cultnet-ts/test/cultnet.test.ts @@ -1,5 +1,6 @@ import test from "node:test"; import assert from "node:assert/strict"; +import { EventEmitter } from "node:events"; import { Duplex } from "node:stream"; import dgram, { type Socket } from "node:dgram"; import { rmSync, mkdtempSync } from "node:fs"; @@ -17,6 +18,7 @@ import { CultNetClientSecurityOptions, CultNetDocumentRegistry, CultNetPeer, + CultNetRudpReconnectLoop, CultNetRudpSession, CultNetRudpSocketTransportConnection, CultNetSchemaRegistry, @@ -50,6 +52,22 @@ import { type InteropNote, } from "./interop/cultnet-interop-shared"; +class FakeRudpReconnectTransport extends EventEmitter { + public connectCalls = 0; + public closeCalls = 0; + public lastPayload: Uint8Array | undefined; + + public connect(payload = new Uint8Array()): void { + this.connectCalls += 1; + this.lastPayload = payload; + } + + public close(): void { + this.closeCalls += 1; + this.emit("close"); + } +} + class LinkedDuplex extends Duplex { peer?: LinkedDuplex; @@ -330,6 +348,52 @@ test("reconnect controller schedules attempts and reset with the shared policy", assert.equal(controller.canAttempt(99_000), true); }); +test("rudp reconnect loop consumes the shared reconnect controller", () => { + let nowMs = 10_000; + let capturedTimer: (() => void) | undefined; + const transports: FakeRudpReconnectTransport[] = []; + const loop = new CultNetRudpReconnectLoop({ + reconnectPolicy: createCultNetReconnectPolicy({ maxAttempts: 2 }), + connectPayload: Buffer.from("join", "utf8"), + createTransport: () => { + const transport = new FakeRudpReconnectTransport(); + transports.push(transport); + return transport; + }, + nowMs: () => nowMs, + jitterMs: () => 17, + setTimer: (callback, delayMs) => { + assert.equal(delayMs, 1_017); + capturedTimer = callback; + return "timer"; + }, + clearTimer: () => { + capturedTimer = undefined; + }, + }); + + const first = loop.start() as FakeRudpReconnectTransport; + assert.equal(first.connectCalls, 1); + assert.equal(Buffer.from(first.lastPayload ?? []).toString("utf8"), "join"); + + first.emit("close"); + assert.equal(loop.reconnectController.attempt, 1); + assert.equal(loop.reconnectController.nextAttemptAtMs, 11_017); + assert.equal(typeof capturedTimer, "function"); + + nowMs = 11_017; + capturedTimer?.(); + assert.equal(transports.length, 2); + assert.equal(transports[1]?.connectCalls, 1); + + loop.markConnected(); + assert.equal(loop.reconnectController.attempt, 0); + + loop.stop(); + assert.equal(transports[1]?.closeCalls, 1); + assert.equal(loop.transport, undefined); +}); + 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 });