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

Let the TS interop peer prove convergence and golden bytes for lane 3/4

Browse source 
dial gains --expect-sequence (stop once a frame with that sequence arrives,
rather than counting frames received — the number of frames a stalled
latest-only reader sees is not the interesting value, what it converges on
is) and --pause-after/--pause-ms, which simulate the stalled-consumer test
hook at the only level this subprocess can reach: not calling receiveFrame
for a while. The shared native pump keeps dispatching and coalescing into
this transport's own inbox regardless of whether the app reads, so not
reading is externally equivalent to not polling.

serve now also logs each broadcast frame's own TypeScript encoding as hex, so
the harness can compare it against the same frame's C#-side re-encoding
without importing the codec itself.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-23 15:09:37 +02:00
commit fa8a9f4e31
1 changed files with 32 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

36
packages/cultmesh-ts/test/interop/cultmesh-quic-peer.ts
View file

@ -29,7 +29,7 @@ import {
CultMeshQuicRealtimeProvider,
type CultMeshRealtimeTarget,
} from "../../src/realtime-quic";
import type { CultMeshRealtimeDelivery, CultMeshRealtimeFrame } from "../../src/realtime-wire";
import { encodeRealtimeFrame, type CultMeshRealtimeDelivery, type CultMeshRealtimeFrame } from "../../src/realtime-wire";
// `__dirname` at runtime is `dist-test/test/interop`; fixtures are binary and
// are never compiled/copied there, so they are read from their source
@ -123,7 +123,10 @@ async function serveAsync(args: Map<string, string>): Promise<void> {
payload: Buffer.from(`frame-${sequence}`, "utf8"),
};
await provider.broadcast(frame);
writeLog("quic-realtime-serve", { sent: sequence });
// Printed so the harness can compare this exact frame's TypeScript
// encoding against the same frame's C#-side re-encoding (golden bytes
// across processes), without the harness importing the codec itself.
writeLog("quic-realtime-serve", { sent: sequence, hex: Buffer.from(encodeRealtimeFrame(frame)).toString("hex") });
if (intervalMs > 0) await new Promise((resolve) => setTimeout(resolve, intervalMs));
}
} finally {
@ -147,7 +150,22 @@ async function main(): Promise<void> {
const args = parseArgs(rest);
const endpoint = requireArg(args, "endpoint");
const expect = optionalIntArg(args, "expect", 1);
// Alternative stop condition to --expect: keep receiving (through whatever
// the consumer's own coalescing inbox delivers) until a frame with this
// sequence arrives, then stop. Lets a caller prove convergence under
// latest-only coalescing, where the frame *count* a stalled reader sees is
// not the interesting number, but the final value it converges on is.
const expectSequenceRaw = args.get("expect-sequence");
const expectSequence = expectSequenceRaw !== undefined ? BigInt(expectSequenceRaw) : undefined;
const timeoutMs = optionalIntArg(args, "timeout-ms", 15_000);
// Simulates "a test hook on the runtime that stops calling next_event"
// (spec section 10, LatestOnly-under-a-stalled-consumer lane) at the level
// this subprocess can reach: after receiving --pause-after frames, stop
// calling receiveFrame for --pause-ms. The shared native pump keeps
// dispatching and coalescing into this transport's own inbox regardless
// (Cut 4), so not reading is externally equivalent to not polling.
const pauseAfter = optionalIntArg(args, "pause-after", -1);
const pauseMs = optionalIntArg(args, "pause-ms", 0);
const target: CultMeshRealtimeTarget = { verseId: "interop", authorityRuntimeId: "interop.csharp-provider" };
const connector = new CultMeshQuicRealtimeConnector();
@ -168,13 +186,23 @@ async function main(): Promise<void> {
}> = [];
try {
const deadline = Date.now() + timeoutMs;
while (frames.length < expect) {
let lastSequence: bigint | undefined;
while (
(expectSequence === undefined && frames.length < expect) ||
(expectSequence !== undefined && lastSequence !== expectSequence)
) {
if (pauseAfter >= 0 && frames.length === pauseAfter) {
writeLog("pausing", { afterFrames: frames.length, pauseMs });
await new Promise((resolve) => setTimeout(resolve, pauseMs));
writeLog("resumed", { afterFrames: frames.length });
}
const remaining = deadline - Date.now();
if (remaining <= 0) throw new Error(`Timed out after receiving ${frames.length} of ${expect} frames.`);
if (remaining <= 0) throw new Error(`Timed out after receiving ${frames.length} frames (last sequence ${lastSequence}).`);
const controller = new AbortController();
const timer = setTimeout(() => controller.abort(), remaining);
try {
const frame = await transport.receiveFrame(controller.signal);
lastSequence = frame.sequence;
frames.push({
channelId: frame.channelId,
schemaId: frame.schemaId,