GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Prove Python schema messages over RUDP

Browse source
This commit is contained in:
Meta Vens 2026-06-16 01:42:02 +01:00
commit 409a42e0f7
3 changed files with 311 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

9
docs/cultnet-transport-parity.md
View file

@ -181,6 +181,11 @@ Current progress:
UDP peers, C#/Kotlin/Python/Rust can dial a TypeScript UDP peer, and all sides
exchange reliable ordered `schema` frames through the shared handshake and
packet language.
- The same harness now proves schema-v0 MessagePack messages over RUDP in both
directions for C#, Kotlin, and Python peers against TypeScript. Those tests
send `cultnet.schema_catalog_request.v0` payloads through reliable ordered
`schema` frames and validate `cultnet.hello.v0` responses with advertised
RUDP transport profiles where the responding runtime owns them.
- The same harness now proves reliable ordered `schema` delivery survives one
dropped data packet in both directions between TypeScript and C#/Kotlin/Python/Rust.
The drops happen below the transport connection through a UDP bridge, so the
@ -234,8 +239,8 @@ Current progress:
creating a second authority path.
- The remaining parity work is to deepen Python's server-side use of its port
and move each runtime's existing TCP/LiteNetLib/WebSocket bodies behind those
ports, then broaden RUDP interop into reconnect policy and the wider
schema-message matrix.
ports, then broaden RUDP interop into reconnect policy, Rust schema-v0
MessagePack-over-RUDP coverage, and the wider schema-message matrix.
## RUDP Packet Contract V0

10
docs/runtime-parity-scope.md
View file

@ -31,7 +31,7 @@ CultMesh.
| Runtime | Role | Claimed parity | Not claimed |
| --- | --- | --- | --- |
| 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, 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, benchmark evidence against C# cache paths | Universal C# throughput, full C# daemon/server stack, and RUDP as the only public service pipe |
| 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, 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, 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 | Full CultMesh ergonomic facade 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, schema-v0 MessagePack over RUDP evidence | C# SoA store, full C# server daemon, or Unity hot-loop memory layout |
@ -42,8 +42,7 @@ CultMesh.
wire harness. It covers schema-v0 interop and RUDP exchange across C#,
TypeScript, Rust, Python, and Kotlin, including loss, reordering,
fragmentation, disconnect reason propagation, ping/pong, and schema-v0
MessagePack over RUDP for the runtimes currently wired into that message
lane.
MessagePack over RUDP for C#, Python, and Kotlin peers against TypeScript.
- `packages/cultcache-py/PARITY.md` is the Python-specific audit ledger. Its
"Still Not Claimed" section is part of the contract, not an apology.
- `docs/cultnet-transport-parity.md` is the transport ownership map. It tracks
@ -57,9 +56,8 @@ CultMesh.
- Broaden service adoption so local servers and daemon paths use the shared
transport port consistently instead of keeping raw TCP/LiteNetLib/WebSocket
ownership in individual bodies.
- Extend RUDP interop evidence through reconnect policy and the wider
schema-message matrix, not only the current schema-frame and selected
schema-v0 message paths.
- Extend RUDP interop evidence through reconnect policy, Rust schema-v0
MessagePack-over-RUDP coverage, and the wider schema-message matrix.
- 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

302
packages/cultnet-ts/test/interop/cultnet-interop.test.ts
View file

@ -237,6 +237,135 @@ transport.close()
raise SystemExit("timed out waiting for TypeScript RUDP frame")
`;
const pythonRudpMessagePeerScript = String.raw`
import msgpack
import socket
import time
from cultnet_py import (
CultNetRudpSocketMode,
CultNetRudpSocketTransportConnection,
CultNetRudpSocketTransportOptions,
create_rudp_transport_profile,
)
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind(("127.0.0.1", 0))
sock.settimeout(0.02)
transport = CultNetRudpSocketTransportConnection(CultNetRudpSocketTransportOptions(
runtime_id="python-rudp-message-interop",
socket=sock,
mode=CultNetRudpSocketMode.SERVER,
connection_id=0x55667789,
initial_sequence=100,
resend_delay_ms=25,
))
print(f"PORT {sock.getsockname()[1]}", flush=True)
deadline = time.time() + 5
while time.time() < deadline:
frame = transport.receive_once()
transport.poll_resends()
if frame is None:
time.sleep(0.005)
continue
if frame.channel_id != "schema":
transport.close()
raise SystemExit(f"unexpected RUDP channel: {frame.channel_id!r}")
message = msgpack.unpackb(frame.payload, raw=False)
if message.get("schemaVersion") != "cultnet.schema_catalog_request.v0":
transport.close()
raise SystemExit(f"unexpected schema message: {message!r}")
response = {
"schemaVersion": "cultnet.hello.v0",
"runtimeId": "python-rudp-message-interop",
"runtimeKind": "python",
"supportedDocumentTypes": [],
"supportedMutationContracts": [],
"supportedMessageVersions": ["cultnet.hello.v0", "cultnet.schema_catalog_request.v0"],
"transportProfiles": [create_rudp_transport_profile(
"python-rudp-message-interop",
host="127.0.0.1",
port=sock.getsockname()[1],
)],
"supportsSchemaCatalog": True,
}
transport.send("schema", msgpack.packb(response, use_bin_type=True))
print("OK", flush=True)
ack_deadline = time.time() + 0.25
while time.time() < ack_deadline:
transport.receive_once()
transport.poll_resends()
time.sleep(0.005)
transport.close()
raise SystemExit(0)
transport.close()
raise SystemExit("timed out waiting for RUDP schema-v0 message")
`;
const pythonRudpMessageClientScript = String.raw`
import msgpack
import socket
import sys
import time
from cultnet_py import (
CultNetRudpSocketMode,
CultNetRudpSocketTransportConnection,
CultNetRudpSocketTransportOptions,
)
remote_port = int(sys.argv[1])
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind(("127.0.0.1", 0))
sock.settimeout(0.02)
transport = CultNetRudpSocketTransportConnection(CultNetRudpSocketTransportOptions(
runtime_id="python-rudp-message-client-interop",
socket=sock,
mode=CultNetRudpSocketMode.CLIENT,
remote_addr=("127.0.0.1", remote_port),
connection_id=0x66554434,
initial_sequence=1,
resend_delay_ms=25,
))
transport.connect(b"python-message-join")
sent = False
deadline = time.time() + 5
while time.time() < deadline:
frame = transport.receive_once()
transport.poll_resends()
if transport.connected and not sent:
transport.send("schema", msgpack.packb({
"schemaVersion": "cultnet.schema_catalog_request.v0",
"messageId": "python-ts-schema-message",
"includeSchemaJson": False,
"schemaIds": [],
"kinds": ["wire_message"],
}, use_bin_type=True))
sent = True
if frame is None:
time.sleep(0.005)
continue
if frame.channel_id != "schema":
transport.close()
raise SystemExit(f"unexpected RUDP channel: {frame.channel_id!r}")
message = msgpack.unpackb(frame.payload, raw=False)
if message.get("schemaVersion") != "cultnet.hello.v0":
transport.close()
raise SystemExit(f"unexpected schema message: {message!r}")
if message.get("runtimeId") != "ts-python-rudp-message-server":
transport.close()
raise SystemExit(f"unexpected runtime id: {message!r}")
print("OK", flush=True)
transport.close()
raise SystemExit(0)
transport.close()
raise SystemExit("timed out waiting for TypeScript RUDP schema-v0 message")
`;
test("CultNet TS/Rust/C#/Python peers discover each other and exchange raw state over the shared schema-v0 lane", async (t) => {
await buildInteropPeers();
cleanInteropStores([
@ -834,6 +963,112 @@ test("CultNet Python and TypeScript exchange schema frames when Python dials a T
}
});
test("CultNet TypeScript and Python exchange schema-v0 MessagePack messages over RUDP", async () => {
const pythonPeer = spawnPythonRudpMessagePeer();
const clientSocket = await bindUdpSocket();
let client: CultNetRudpSocketTransportConnection | undefined;
try {
const pythonPort = await pythonPeer.ready;
client = new CultNetRudpSocketTransportConnection({
runtimeId: "ts-python-rudp-message-interop",
socket: clientSocket,
mode: "client",
remoteHost: "127.0.0.1",
remotePort: pythonPort,
connectionId: 0x55667789,
initialSequence: 1,
resendDelayMs: 25,
resendPollMs: 5,
});
const receivedFrame = once(client, "frame");
client.connect(Buffer.from("ts-python-message-join"));
await waitFor(() => client?.connected === true, "TypeScript RUDP client to complete Python message handshake");
client.send("schema", Buffer.from(encode({
schemaVersion: "cultnet.schema_catalog_request.v0",
messageId: "ts-python-schema-message",
includeSchemaJson: false,
schemaIds: [],
kinds: ["wire_message"],
})));
const [frame] = await withTimeout(receivedFrame, 2_000, "Python RUDP schema-v0 response frame");
assert.equal(frame.channelId, "schema");
const message = decode(frame.payload) as {
schemaVersion?: string;
runtimeId?: string;
supportsSchemaCatalog?: boolean;
transportProfiles?: Array<{ transports?: Array<{ protocol?: string; channels?: Array<{ channelId?: string }> }> }>;
};
assert.equal(message.schemaVersion, "cultnet.hello.v0");
assert.equal(message.runtimeId, "python-rudp-message-interop");
assert.equal(message.supportsSchemaCatalog, true);
assert.equal(message.transportProfiles?.[0]?.transports?.[0]?.protocol, "rudp");
assert.equal(message.transportProfiles?.[0]?.transports?.[0]?.channels?.[0]?.channelId, "schema");
const [exitCode] = await withTimeout(once(pythonPeer.child, "exit"), 2_000, "Python RUDP message peer exit");
assert.equal(exitCode, 0, pythonPeer.stderr.join(""));
} finally {
if (client) {
client.close();
} else {
clientSocket.close();
}
if (pythonPeer.child.exitCode === null && !pythonPeer.child.killed) {
pythonPeer.child.kill("SIGTERM");
await once(pythonPeer.child, "exit").catch(() => undefined);
}
}
});
test("CultNet Python and TypeScript exchange schema-v0 MessagePack messages over RUDP when Python dials", async () => {
const serverSocket = await bindUdpSocket();
const server = new CultNetRudpSocketTransportConnection({
runtimeId: "ts-python-rudp-message-server",
socket: serverSocket,
mode: "server",
connectionId: 0x66554434,
initialSequence: 100,
resendDelayMs: 25,
resendPollMs: 5,
});
const pythonClient = spawnPythonRudpMessageClient(udpSocketPort(serverSocket));
try {
const receivedFrame = once(server, "frame");
const [frame] = await withTimeout(receivedFrame, 2_000, "Python RUDP schema-v0 client frame");
assert.equal(frame.channelId, "schema");
const message = decode(frame.payload) as {
schemaVersion?: string;
messageId?: string;
kinds?: string[];
};
assert.equal(message.schemaVersion, "cultnet.schema_catalog_request.v0");
assert.equal(message.messageId, "python-ts-schema-message");
assert.deepEqual(message.kinds, ["wire_message"]);
server.send("schema", Buffer.from(encode({
schemaVersion: "cultnet.hello.v0",
runtimeId: "ts-python-rudp-message-server",
runtimeKind: "typescript",
supportedDocumentTypes: [],
supportedMutationContracts: [],
supportedMessageVersions: ["cultnet.hello.v0", "cultnet.schema_catalog_request.v0"],
transportProfiles: [],
supportsSchemaCatalog: true,
})));
const [exitCode] = await withTimeout(once(pythonClient.child, "exit"), 2_000, "Python RUDP message client exit");
assert.equal(exitCode, 0, pythonClient.stderr.join(""));
} finally {
server.close();
if (pythonClient.child.exitCode === null && !pythonClient.child.killed) {
pythonClient.child.kill("SIGTERM");
await once(pythonClient.child, "exit").catch(() => undefined);
}
}
});
test("CultNet TypeScript and Rust exchange schema frames over the shared RUDP socket transport", async () => {
await buildRustInteropPeer();
const rustPeer = spawnRustRudpServer();
@ -1046,7 +1281,7 @@ test("CultNet TypeScript and C# exchange schema-v0 MessagePack messages over RUD
}
});
test("CultNet C# and TypeScript exchange schema-v0 MessagePack messages when C# dials", async () => {
test("CultNet C# and TypeScript exchange schema-v0 MessagePack messages over RUDP when C# dials", async () => {
await buildCSharpInteropPeer();
const serverSocket = await bindUdpSocket();
const server = new CultNetRudpSocketTransportConnection({
@ -1230,7 +1465,7 @@ test("CultNet TypeScript and Kotlin exchange schema-v0 MessagePack messages over
}
});
test("CultNet Kotlin and TypeScript exchange schema-v0 MessagePack messages when Kotlin dials", async () => {
test("CultNet Kotlin and TypeScript exchange schema-v0 MessagePack messages over RUDP when Kotlin dials", async () => {
await buildKotlinInteropPeer();
const serverSocket = await bindUdpSocket();
const server = new CultNetRudpSocketTransportConnection({
@ -1873,6 +2108,54 @@ function spawnPythonRudpPeer(options: RudpServerSpawnOptions = {}): RunningPytho
return { child, ready, stderr };
}
function spawnPythonRudpMessagePeer(): RunningPythonRudpPeer {
const child = spawn(pythonCommand, ["-c", pythonRudpMessagePeerScript], {
cwd: cultcachePyRoot,
env: { ...process.env, PYTHONPATH: cultcachePySrc },
stdio: ["ignore", "pipe", "pipe"],
});
const stderr: string[] = [];
let stdoutBuffer = "";
const ready = new Promise<number>((resolveReady, rejectReady) => {
child.stdout.setEncoding("utf8");
child.stdout.on("data", (chunk: string) => {
stdoutBuffer += chunk;
while (true) {
const newline = stdoutBuffer.indexOf("\n");
if (newline === -1) {
break;
}
const line = stdoutBuffer.slice(0, newline).trim();
stdoutBuffer = stdoutBuffer.slice(newline + 1);
if (!line) {
continue;
}
const portMatch = /^PORT\s+(\d+)$/.exec(line);
if (portMatch) {
resolveReady(Number(portMatch[1]));
continue;
}
if (line !== "OK") {
rejectReady(new Error(`Python RUDP message peer emitted unexpected stdout: ${line}`));
}
}
});
child.stderr.setEncoding("utf8");
child.stderr.on("data", (chunk: string) => {
stderr.push(chunk);
});
child.once("exit", (code, signal) => {
rejectReady(new Error(`Python RUDP message peer exited before publishing a port (code=${code}, signal=${signal}).\n${stderr.join("")}`));
});
child.once("error", rejectReady);
});
return { child, ready, stderr };
}
function spawnRustRudpServer(options: RudpServerSpawnOptions = {}): RunningPythonRudpPeer {
const child = spawn(rustBinaryPath, [
"rudp-serve-once",
@ -2174,6 +2457,21 @@ function spawnPythonRudpClient(remotePort: number): Omit<RunningPythonRudpPeer,
return { child, stderr };
}
function spawnPythonRudpMessageClient(remotePort: number): Omit<RunningPythonRudpPeer, "ready"> {
const child = spawn(pythonCommand, ["-c", pythonRudpMessageClientScript, String(remotePort)], {
cwd: cultcachePyRoot,
env: { ...process.env, PYTHONPATH: cultcachePySrc },
stdio: ["ignore", "pipe", "pipe"],
});
const stderr: string[] = [];
child.stderr.setEncoding("utf8");
child.stderr.on("data", (chunk: string) => {
stderr.push(chunk);
});
return { child, stderr };
}
function spawnKotlinRudpClient(remotePort: number): Omit<RunningPythonRudpPeer, "ready"> {
const child = spawn(kotlinJavaCommand, [
"-cp", kotlinClasspath,