From 08cf3b34b5a27be6999ecbe5ac77d5884d6be90e Mon Sep 17 00:00:00 2001 From: Libby Date: Sun, 8 Nov 2026 11:26:22 +0000 Subject: [PATCH 01/11] cultnet-ts test: the Odin presence publisher does not complete on the ack when the refusal arrives first Co-Authored-By: Claude Sonnet 5.5 --- .../test/idunn-runtime-authority.test.ts | 25 ++++++++++++++++++- 1 file changed, 24 insertions(+), 1 deletion(-) diff --git a/packages/cultnet-ts/test/idunn-runtime-authority.test.ts b/packages/cultnet-ts/test/idunn-runtime-authority.test.ts index 2b45e3b1..8b9a5629 100644 --- a/packages/cultnet-ts/test/idunn-runtime-authority.test.ts +++ b/packages/cultnet-ts/test/idunn-runtime-authority.test.ts @@ -279,6 +279,8 @@ async function startOdinPeer(context: TestContext, endpoint: string) { context.after(() => peer.close()); const state = { reject: false, + // How a refusal and the ack of the put reach the publisher: in one packet, or the ack first, or the refusal first. + order: "piggyback" as "piggyback" | "ack-then-refusal" | "refusal-then-ack", received: [] as Array<{ schemaVersion?: string; document: { payload: DocumentPayload } }>, responseAck: undefined as number | undefined, failure: undefined as Error | undefined, @@ -305,7 +307,15 @@ async function startOdinPeer(context: TestContext, endpoint: string) { }); assert.ok(response); state.responseAck = response.ack; - peer.send(encodeRudpPacket(response), remote.port, remote.address); + if (state.reject && state.order === "ack-then-refusal") { + peer.send(encodeRudpPacket(serverSession.createAckForReceived(packet.sequence)), remote.port, remote.address); + setTimeout(() => peer.send(encodeRudpPacket(response), remote.port, remote.address), 120); + } else if (state.reject && state.order === "refusal-then-ack") { + peer.send(encodeRudpPacket({ ...response, ack: 0, ackMask: 0 }), remote.port, remote.address); + setTimeout(() => peer.send(encodeRudpPacket(serverSession!.createAckForReceived(packet.sequence)), remote.port, remote.address), 120); + } else { + peer.send(encodeRudpPacket(response), remote.port, remote.address); + } } catch (error) { state.failure = error as Error; } @@ -332,6 +342,19 @@ test("the Odin publisher publishes the signer's document to the endpoint Expecte assert.equal(peer.failure, undefined, peer.failure?.message); }); +for (const order of ["refusal-then-ack"] as const) { + test(`the Odin publisher does not complete on the ack of a refused presence (${order})`, async (context) => { + const endpoint = expectedFacts("service").odinEndpoint!; + const peer = await startOdinPeer(context, endpoint); + peer.reject = true; + peer.order = order; + const signer = signerFor(context, "service"); + signer.reportHealth("active"); + const publisher = createIdunnRuntimePresencePublisher({ signer, endpoint }); + await assert.rejects(publisher.publish("denied presence"), /Odin rejected runtime presence: test admission denied/); + }); +} + test("the Odin publisher leaves a capability shortfall to the Rust correlation instead of refusing to publish", async (context) => { const endpoint = expectedFacts("service").odinEndpoint!; const peer = await startOdinPeer(context, endpoint); -- 2.47.3 From 7766348c75b06ed112623042f3dfb356de94b0de Mon Sep 17 00:00:00 2001 From: Libby Date: Sun, 8 Nov 2026 11:26:22 +0000 Subject: [PATCH 02/11] cultnet-ts test (FAILS, a finding): the Odin presence publisher completes on the ack when the refusal follows it Co-Authored-By: Claude Sonnet 5.5 --- packages/cultnet-ts/test/idunn-runtime-authority.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/cultnet-ts/test/idunn-runtime-authority.test.ts b/packages/cultnet-ts/test/idunn-runtime-authority.test.ts index 8b9a5629..581cf852 100644 --- a/packages/cultnet-ts/test/idunn-runtime-authority.test.ts +++ b/packages/cultnet-ts/test/idunn-runtime-authority.test.ts @@ -342,7 +342,7 @@ test("the Odin publisher publishes the signer's document to the endpoint Expecte assert.equal(peer.failure, undefined, peer.failure?.message); }); -for (const order of ["refusal-then-ack"] as const) { +for (const order of ["ack-then-refusal", "refusal-then-ack"] as const) { test(`the Odin publisher does not complete on the ack of a refused presence (${order})`, async (context) => { const endpoint = expectedFacts("service").odinEndpoint!; const peer = await startOdinPeer(context, endpoint); -- 2.47.3 From 4f90ea815688965b4294221c4cea1453b6743946 Mon Sep 17 00:00:00 2001 From: Libby Date: Sun, 8 Nov 2026 11:27:06 +0000 Subject: [PATCH 03/11] cultmesh-ts test: publishRudpDocumentAndWaitForReceipt reports a refused put as refused, whether the ack comes before or after the refusal Co-Authored-By: Claude Sonnet 5.5 --- .../test/client-write-completion.test.ts | 83 +++++++++++++++++++ 1 file changed, 83 insertions(+) create mode 100644 packages/cultmesh-ts/test/client-write-completion.test.ts diff --git a/packages/cultmesh-ts/test/client-write-completion.test.ts b/packages/cultmesh-ts/test/client-write-completion.test.ts new file mode 100644 index 00000000..be31e0c7 --- /dev/null +++ b/packages/cultmesh-ts/test/client-write-completion.test.ts @@ -0,0 +1,83 @@ +// Heimdall's invariant: no client may treat a reliable-transport ack as write admission. A put completes only +// on its result frame (write_not_authorized, or the success reply). Each test serves a client a peer that +// acknowledges the put and refuses it, the ack arriving before the refusal and after it. +import assert from "node:assert/strict"; +import { createSocket, type Socket } from "node:dgram"; +import test from "node:test"; +import { z } from "zod"; +import { defineDocumentType } from "@gamecult/cultcache-ts"; +import { + CultNetDocumentRegistry, + CultNetRudpSession, + cultNetWriteNotAuthorizedError, + decodeRudpPacket, + defineCultNetDocumentBinding, + encodeCultNetMessageForWire, + encodeRudpPacket, +} from "@gamecult/cultnet-ts"; +import { encodeWireValue } from "@gamecult/cultnet-ts/wire"; +import { CultMesh } from "../src/index"; + +const note = defineDocumentType({ + type: "cultmesh.completion_note", + schemaId: "cultmesh.completion_note.v0", + schema: z.object({ noteId: z.string(), text: z.string() }), + name: "noteId", +}); +const binding = defineCultNetDocumentBinding({ definition: note }); +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +type Order = "ack-then-refusal" | "refusal-then-ack"; + +/** A peer that acknowledges every reliable put and refuses it with write_not_authorized, in the given order. */ +async function refusingPeer(connectionId: number, order: Order) { + const socket: Socket = createSocket("udp4"); + await new Promise((resolve) => socket.bind(0, "127.0.0.1", resolve)); + const session = new CultNetRudpSession({ connectionId, resendDelayMs: 50 }); + const send = (packet: Parameters[0], port: number, address: string) => + socket.send(encodeRudpPacket(packet), port, address); + socket.on("message", (wire, remote) => { + const packet = decodeRudpPacket(wire); + const now = Date.now(); + if (packet.packetType === "connect") { + send(session.acceptConnect(packet, now), remote.port, remote.address); + return; + } + const result = session.receive(packet, now); + if (result.reply) send(result.reply, remote.port, remote.address); + for (const frame of result.delivered) { + if (frame.channelId !== "schema") continue; + const ack = session.createAckForReceived(packet.sequence); + const refusal = session.sendMany( + "schema", + encodeWireValue(encodeCultNetMessageForWire(cultNetWriteNotAuthorizedError("tests:completion:a"), "cultnet.schema.v0")), + { reliable: true, ordered: true, nowMs: now }, + ); + if (order === "ack-then-refusal") { + send(ack, remote.port, remote.address); + setTimeout(() => refusal.forEach((p) => send(p, remote.port, remote.address)), 120); + } else { + // The refusal carries no acknowledgement of its own; the ack follows it. + refusal.forEach((p) => send({ ...p, ack: 0, ackMask: 0 }, remote.port, remote.address)); + setTimeout(() => send(ack, remote.port, remote.address), 120); + } + } + if (packet.reliable) send(session.createAckForReceived(packet.sequence), remote.port, remote.address); + }); + return { endpoint: `rudp://127.0.0.1:${socket.address().port}`, close: () => socket.close() }; +} + +const orders: Order[] = ["ack-then-refusal", "refusal-then-ack"]; +let nextConnection = 0x10205000; + +for (const order of orders) { + test(`publishRudpDocumentAndWaitForReceipt reports a refused put as refused, never as success (${order})`, async () => { + const peer = await refusingPeer(nextConnection, order); + try { + await assert.rejects( + CultMesh.publishRudpDocumentAndWaitForReceipt("completion-receipt", nextConnection++, peer.endpoint, binding, "tests:completion:a", { noteId: "a", text: "x" }, binding, { resendDelayMs: 50, resendPollMs: 5, receiptTimeoutMs: 800 }), + ); + } finally { peer.close(); } + await sleep(10); + }); +} -- 2.47.3 From dfd090dc557d8c08b7cf9979cf9abef3755e426d Mon Sep 17 00:00:00 2001 From: Libby Date: Sun, 8 Nov 2026 11:27:07 +0000 Subject: [PATCH 04/11] cultmesh-ts test (FAILS, a finding): publishRudpDocumentOnce completes on the ack of a refused put Co-Authored-By: Claude Sonnet 5.5 --- .../cultmesh-ts/test/client-write-completion.test.ts | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/packages/cultmesh-ts/test/client-write-completion.test.ts b/packages/cultmesh-ts/test/client-write-completion.test.ts index be31e0c7..fa3db243 100644 --- a/packages/cultmesh-ts/test/client-write-completion.test.ts +++ b/packages/cultmesh-ts/test/client-write-completion.test.ts @@ -71,6 +71,16 @@ const orders: Order[] = ["ack-then-refusal", "refusal-then-ack"]; let nextConnection = 0x10205000; for (const order of orders) { + test(`publishRudpDocumentOnce does not complete on the ack of a refused put (${order})`, async () => { + const peer = await refusingPeer(nextConnection, order); + try { + await assert.rejects( + CultMesh.publishRudpDocumentOnce("completion-once", nextConnection++, peer.endpoint, binding, "tests:completion:a", { noteId: "a", text: "x" }, { resendDelayMs: 50, resendPollMs: 5, flushTimeoutMs: 2_000 }), + /./, + ); + } finally { peer.close(); } + }); + test(`publishRudpDocumentAndWaitForReceipt reports a refused put as refused, never as success (${order})`, async () => { const peer = await refusingPeer(nextConnection, order); try { -- 2.47.3 From 9e734447f63b970029e536d83693a264e4ff09b4 Mon Sep 17 00:00:00 2001 From: Libby Date: Sun, 8 Nov 2026 11:27:53 +0000 Subject: [PATCH 05/11] cultmesh-rs test (FAILS, findings): publish completes on the ack when the refusal follows it; the batch publish completes on the ack whatever the refusal Co-Authored-By: Claude Sonnet 5.5 --- packages/cultmesh-rs/src/lib.rs | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/packages/cultmesh-rs/src/lib.rs b/packages/cultmesh-rs/src/lib.rs index 1dec91f9..172a5668 100644 --- a/packages/cultmesh-rs/src/lib.rs +++ b/packages/cultmesh-rs/src/lib.rs @@ -2035,4 +2035,28 @@ mod tests { Ok(()) } + + #[test] + fn a_publish_whose_ack_arrives_before_the_refusal_is_refused() -> Result<()> { + let (target, stop, catalog) = acknowledging_refusing_catalog(false)?; + let result = publish_cultnet_message_to_rudp_catalog(&refused_note(), publish_options(target)); + stop.store(true, std::sync::atomic::Ordering::SeqCst); + catalog.join().expect("catalog thread should not panic")?; + let error = result.expect_err("a refused put is never reported as admitted"); + assert!(error.to_string().contains("refused the put"), "{error:#}"); + Ok(()) + } + + #[test] + fn a_batch_publish_is_not_complete_on_the_ack_of_refused_puts() -> Result<()> { + for refusal_first in [false, true] { + let (target, stop, catalog) = acknowledging_refusing_catalog(refusal_first)?; + let result = publish_cultnet_messages_to_rudp_catalog(&[refused_note()], publish_options(target)); + stop.store(true, std::sync::atomic::Ordering::SeqCst); + catalog.join().expect("catalog thread should not panic")?; + assert!(result.is_err(), "refusal_first={refusal_first}: a refused put was reported as published"); + } + Ok(()) + } + } -- 2.47.3 From c6da8fc54a9f904960e0a0965d0468220e878785 Mon Sep 17 00:00:00 2001 From: Libby Date: Sun, 8 Nov 2026 11:28:23 +0000 Subject: [PATCH 06/11] cultmesh-py test: a put the server refused is reported refused after its reliable ack arrived Co-Authored-By: Claude Sonnet 5.5 --- .../tests/test_client_write_completion.py | 52 +++++++++++++++++++ 1 file changed, 52 insertions(+) create mode 100644 packages/cultmesh-py/tests/test_client_write_completion.py diff --git a/packages/cultmesh-py/tests/test_client_write_completion.py b/packages/cultmesh-py/tests/test_client_write_completion.py new file mode 100644 index 00000000..0f9d0c70 --- /dev/null +++ b/packages/cultmesh-py/tests/test_client_write_completion.py @@ -0,0 +1,52 @@ +"""Heimdall's invariant: no client treats a reliable-transport ack as write admission. A put completes only on +its result frame (write_not_authorized, or the success reply). + +The Python write path is a subscription over RUDP: `send` queues the put, `read_next` returns the result frame +and raises the peer's refusal. The server here refuses the put; its reliable ack of the put arrives (the test +waits for it) and the client still reports the refusal, never success. The transport's `receive` yields data +frames only, so the order of the ack against the refusal cannot change what `read_next` reports. +""" + +from __future__ import annotations + +import unittest + +from cultcache_py import define_database_entry_type +from cultmesh_py import CultMesh, create_node +from cultmesh_py.server import CultMeshLocalServer +from cultnet_py import CultNetDatabaseSubscription, CultNetPeerError, create_rudp_schema_transport + +from test_remote_write_authority import KEY_A, NOTE, SCHEMA_ID, put_of + + +class ClientWriteCompletionTests(unittest.TestCase): + def test_a_put_the_server_refused_is_reported_refused_after_its_ack_arrived(self) -> None: + node = create_node(runtime_id="completion-server") + node.database.register_document(NOTE) + server = CultMeshLocalServer(node=node, port=0, write_authority=None) + server.start() + try: + subscription = CultNetDatabaseSubscription( + host="127.0.0.1", + port=server.port, + timeout_seconds=4.0, + subscription_id="completion", + schema_ids=[SCHEMA_ID], + include_snapshot=False, + create_transport=lambda: create_rudp_schema_transport( + host="127.0.0.1", port=server.port, connection_id=server.rudp_connection_id, timeout_seconds=4.0 + ), + ) + with subscription as sub: + sub.send(put_of(KEY_A, "peer")) + sub._transport.flush_reliable(2.0) # the put's reliable ack has arrived + with self.assertRaises(CultNetPeerError) as raised: + sub.read_next() + finally: + server.stop() + self.assertEqual(raised.exception.response["code"], "write_not_authorized") + self.assertIsNone(server.node.cache.get(NOTE, KEY_A)) + + +if __name__ == "__main__": + unittest.main() -- 2.47.3 From c1d24146dbc1c087de110e14173f2f3208c1fd2c Mon Sep 17 00:00:00 2001 From: Libby Date: Sun, 8 Nov 2026 11:29:50 +0000 Subject: [PATCH 07/11] GameCult.Networking test: the RUDP schema client delivers a refused put's result frame whatever the order of the ack Co-Authored-By: Claude Sonnet 5.5 --- .../CultNetClientWriteCompletionTests.cs | 100 ++++++++++++++++++ 1 file changed, 100 insertions(+) create mode 100644 tests/GameCult.Networking.Tests/CultNetClientWriteCompletionTests.cs diff --git a/tests/GameCult.Networking.Tests/CultNetClientWriteCompletionTests.cs b/tests/GameCult.Networking.Tests/CultNetClientWriteCompletionTests.cs new file mode 100644 index 00000000..8641a082 --- /dev/null +++ b/tests/GameCult.Networking.Tests/CultNetClientWriteCompletionTests.cs @@ -0,0 +1,100 @@ +#nullable enable +using System; +using System.Collections.Concurrent; +using System.Net; +using System.Net.Sockets; +using System.Threading; +using System.Threading.Tasks; +using NUnit.Framework; + +namespace GameCult.Networking.Tests +{ + /// + /// Heimdall's invariant: no client treats a reliable-transport ack as write admission. A put completes + /// only on its result frame (write_not_authorized, or the success reply). The C# schema client has no + /// completion call: SendCultNet returns once the put is queued, and the outcome reaches the caller only + /// as the result frame, through OnCultNet. These tests serve it a peer that acknowledges the put and + /// refuses it, the ack before the refusal and after it, and require the refusal to arrive as the outcome. + /// + public class CultNetClientWriteCompletionTests + { + private const uint ConnectionId = 0x10203197; + + // Acknowledges the first reliable schema frame it receives and answers it with write_not_authorized, + // the second packet 150 ms after the first. The refusal carries no acknowledgement of its own. + private static Task RefusingPeer(Socket socket, bool refusalFirst, CancellationToken stop) + { + return Task.Run(() => + { + var session = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = ConnectionId, ResendDelayMs = 25 }); + var buffer = new byte[65535]; + EndPoint from = new IPEndPoint(IPAddress.Any, 0); + var answered = false; + while (!stop.IsCancellationRequested) + { + if (!socket.Poll(10_000, SelectMode.SelectRead)) + continue; + var received = socket.ReceiveFrom(buffer, ref from); + var packet = CultNetRudpPacketCodec.Decode(buffer.AsSpan(0, received).ToArray()); + void Send(CultNetRudpPacket p) => socket.SendTo(CultNetRudpPacketCodec.Encode(p), from); + if (packet.PacketType == CultNetRudpPacketType.Connect) + { + Send(session.AcceptConnect(packet)); + continue; + } + var result = session.Receive(packet, 1); + if (result.Reply != null) + Send(result.Reply); + if (answered || result.Delivered.Count == 0) + continue; + answered = true; + var ack = session.CreateAckForReceived(packet.Sequence); + var refusal = session.Send("schema", CultNetSchemaMessageSerialization.Serialize(new CultNetErrorMessage + { + Error = "write_not_authorized: the catalog refused the document", + Code = "write_not_authorized" + }), new CultNetRudpSendOptions { Reliable = true, Ordered = true }); + refusal.Ack = 0; + refusal.AckMask = 0; + if (refusalFirst) + { + Send(refusal); + Thread.Sleep(150); + Send(ack); + } + else + { + Send(ack); + Thread.Sleep(150); + Send(refusal); + } + } + }); + } + + [TestCase(false, TestName = "AckThenRefusal")] + [TestCase(true, TestName = "RefusalThenAck")] + public async Task ARefusedPutIsReportedRefusedWhateverTheOrderOfItsAck(bool refusalFirst) + { + using var serverSocket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp); + serverSocket.Bind(new IPEndPoint(IPAddress.Loopback, 0)); + using var stop = new CancellationTokenSource(); + var peer = RefusingPeer(serverSocket, refusalFirst, stop.Token); + using var client = new RudpCultNetSchemaClient("completion", ConnectionId); + var outcomes = new BlockingCollection(); + client.OnCultNet(outcomes.Add); + client.Connect("127.0.0.1", ((IPEndPoint)serverSocket.LocalEndPoint!).Port); + var deadline = DateTime.UtcNow.AddSeconds(3); + while (!client.Connected && DateTime.UtcNow < deadline) + await Task.Delay(5); + Assert.That(client.Connected, "fixture: the client connected"); + + client.SendCultNet(new CultNetDocumentPutRawMessage { MessageId = "completion" }); + + Assert.That(outcomes.TryTake(out var refusal, 3000), Is.True, "the result frame reached the caller"); + Assert.That(refusal!.Code, Is.EqualTo("write_not_authorized")); + stop.Cancel(); + await peer; + } + } +} -- 2.47.3 From 7082c9488ba0aa5cbeaf966d343a81c87c75f765 Mon Sep 17 00:00:00 2001 From: Libby Date: Sun, 8 Nov 2026 11:34:55 +0000 Subject: [PATCH 08/11] cultmesh-ts: a put completes on its result frame, not on the transport ack publishRudpDocumentOnce resolved when peer.flush returned (ack only) and read no frame, so a refused put resolved. It now rejects with CultMeshRudpWriteRefusedError (the peer's typed cultnet.error.v0: write_not_authorized, message_invalid) whether the refusal arrives before the ack, after it, or within resultWindowMs (default 500 ms) after it; a success reply ends the wait early. publishRudpDocumentAndWaitForReceipt rejects with the same typed error instead of timing out. Co-Authored-By: Claude Sonnet 5.5 --- packages/cultmesh-ts/src/index.ts | 44 ++++++++++++++++++- .../test/client-write-completion.test.ts | 38 +++++++++++++--- 2 files changed, 74 insertions(+), 8 deletions(-) diff --git a/packages/cultmesh-ts/src/index.ts b/packages/cultmesh-ts/src/index.ts index b77edd9f..f065dd83 100644 --- a/packages/cultmesh-ts/src/index.ts +++ b/packages/cultmesh-ts/src/index.ts @@ -4183,7 +4183,29 @@ export interface CultMeshRudpDocumentServer { close(): void; } +/** + * A write the peer refused: the typed `cultnet.error.v0` it answered with (`write_not_authorized`, + * `message_invalid`, ...), whichever of the transport ack and the result frame came first. + */ +export class CultMeshRudpWriteRefusedError extends Error { + readonly code: string | undefined; + readonly response: Extract; + constructor(response: Extract) { + super(response.error); + this.name = "CultMeshRudpWriteRefusedError"; + this.code = response.code ?? undefined; + this.response = response; + } +} + export interface CultMeshRudpDocumentPublishOptions extends CultMeshRudpPeerOptions { + /** + * How long after the put's transport ack a refusal may still arrive. A reliable-transport ack is not + * admission; the outcome is the peer's result frame. A server that answers a put only on refusal sends no + * success frame, so the put is reported published once this window passes with no refusal; a success reply + * ends the wait early. Default 500 ms. + */ + resultWindowMs?: number; messageId?: string; sourceRuntimeId?: string; sourceAgentId?: string; @@ -6157,11 +6179,23 @@ export class CultMesh { requireNonEmpty(runtimeId, "runtimeId"); requireNonEmpty(recordKey, "recordKey"); const registry = new CultNetDocumentRegistry([binding]); + const messageId = options.messageId ?? `${runtimeId}:${binding.definition.type}:${recordKey}`; const peer = await CultMesh.createRudpPeer(runtimeId, connectionId, endpoint, options); try { + // A transport ack is not admission: the put completes on its result frame. A refusal rejects with the + // typed error whether it arrives before the ack, after it, or inside the result window. + let refusal: CultMeshRudpWriteRefusedError | undefined; + let succeeded = false; + let wake: (() => void) | undefined; + peer.on("message", (message: CultNetMessage) => { + if (message.schemaVersion === "cultnet.error.v0") refusal ??= new CultMeshRudpWriteRefusedError(message); + else if ("messageId" in message && message.messageId === messageId) succeeded = true; + else return; + wake?.(); + }); peer.send(registry.createRawDocumentPutMessage( binding, - options.messageId ?? `${runtimeId}:${binding.definition.type}:${recordKey}`, + messageId, recordKey, value, { @@ -6172,6 +6206,13 @@ export class CultMesh { }, )); await peer.flush(options.flushTimeoutMs ?? 30_000); + if (!refusal && !succeeded) { + await new Promise((resolve) => { + const timer = setTimeout(resolve, options.resultWindowMs ?? 500); + wake = () => { clearTimeout(timer); resolve(); }; + }); + } + if (refusal) throw refusal; } finally { peer.close(); } @@ -6190,6 +6231,7 @@ export class CultMesh { const receipt = new Promise>((resolve, reject) => { const timeout = setTimeout(() => { cleanup(); reject(new Error(`Timed out waiting for RUDP application receipt for ${recordKey}.`)); }, options.receiptTimeoutMs ?? 5_000); const onMessage = (message: CultNetMessage): void => { + if (message.schemaVersion === "cultnet.error.v0") { cleanup(); reject(new CultMeshRudpWriteRefusedError(message)); return; } if (message.schemaVersion !== "cultnet.document_put_raw.v0" || message.messageId !== messageId || message.document.schemaId !== receiptBinding.definition.schemaId) return; try { const decoded = decodeDocumentPayload(receiptBinding.definition, message.document.payload); diff --git a/packages/cultmesh-ts/test/client-write-completion.test.ts b/packages/cultmesh-ts/test/client-write-completion.test.ts index fa3db243..8dd122ed 100644 --- a/packages/cultmesh-ts/test/client-write-completion.test.ts +++ b/packages/cultmesh-ts/test/client-write-completion.test.ts @@ -16,7 +16,10 @@ import { encodeRudpPacket, } from "@gamecult/cultnet-ts"; import { encodeWireValue } from "@gamecult/cultnet-ts/wire"; -import { CultMesh } from "../src/index"; +import { CultMesh, CultMeshGrantAllWrites, CultMeshRudpWriteRefusedError } from "../src/index"; + +const isTheRefusal = (error: unknown): boolean => + error instanceof CultMeshRudpWriteRefusedError && error.code === "write_not_authorized"; const note = defineDocumentType({ type: "cultmesh.completion_note", @@ -34,8 +37,12 @@ async function refusingPeer(connectionId: number, order: Order) { const socket: Socket = createSocket("udp4"); await new Promise((resolve) => socket.bind(0, "127.0.0.1", resolve)); const session = new CultNetRudpSession({ connectionId, resendDelayMs: 50 }); - const send = (packet: Parameters[0], port: number, address: string) => - socket.send(encodeRudpPacket(packet), port, address); + const timers = new Set(); + let closed = false; + const send = (packet: Parameters[0], port: number, address: string) => { + if (!closed) socket.send(encodeRudpPacket(packet), port, address); + }; + const later = (action: () => void) => { const timer = setTimeout(() => { timers.delete(timer); action(); }, 120); timers.add(timer); }; socket.on("message", (wire, remote) => { const packet = decodeRudpPacket(wire); const now = Date.now(); @@ -55,16 +62,16 @@ async function refusingPeer(connectionId: number, order: Order) { ); if (order === "ack-then-refusal") { send(ack, remote.port, remote.address); - setTimeout(() => refusal.forEach((p) => send(p, remote.port, remote.address)), 120); + later(() => refusal.forEach((p) => send(p, remote.port, remote.address))); } else { // The refusal carries no acknowledgement of its own; the ack follows it. refusal.forEach((p) => send({ ...p, ack: 0, ackMask: 0 }, remote.port, remote.address)); - setTimeout(() => send(ack, remote.port, remote.address), 120); + later(() => send(ack, remote.port, remote.address)); } } if (packet.reliable) send(session.createAckForReceived(packet.sequence), remote.port, remote.address); }); - return { endpoint: `rudp://127.0.0.1:${socket.address().port}`, close: () => socket.close() }; + return { endpoint: `rudp://127.0.0.1:${socket.address().port}`, close: () => { closed = true; timers.forEach(clearTimeout); socket.close(); } }; } const orders: Order[] = ["ack-then-refusal", "refusal-then-ack"]; @@ -76,7 +83,7 @@ for (const order of orders) { try { await assert.rejects( CultMesh.publishRudpDocumentOnce("completion-once", nextConnection++, peer.endpoint, binding, "tests:completion:a", { noteId: "a", text: "x" }, { resendDelayMs: 50, resendPollMs: 5, flushTimeoutMs: 2_000 }), - /./, + isTheRefusal, ); } finally { peer.close(); } }); @@ -86,8 +93,25 @@ for (const order of orders) { try { await assert.rejects( CultMesh.publishRudpDocumentAndWaitForReceipt("completion-receipt", nextConnection++, peer.endpoint, binding, "tests:completion:a", { noteId: "a", text: "x" }, binding, { resendDelayMs: 50, resendPollMs: 5, receiptTimeoutMs: 800 }), + isTheRefusal, ); } finally { peer.close(); } await sleep(10); }); } + +test("publishRudpDocumentOnce still publishes a put the server admitted", async () => { + const connectionId = nextConnection++; + const puts: string[] = []; + const server = CultMesh.createRudpDocumentServer("completion-admit", connectionId, { + documents: new CultNetDocumentRegistry([binding]), + writeAuthority: new CultMeshGrantAllWrites(), + bindHost: "127.0.0.1", bindPort: 0, resendDelayMs: 50, resendPollMs: 5, + onDocumentPutRaw: (put) => { puts.push(put.recordKey); }, + }); + await server.start(); + try { + await CultMesh.publishRudpDocumentOnce("completion-admit-client", connectionId, `rudp://127.0.0.1:${server.bind.port}`, binding, "tests:completion:a", { noteId: "a", text: "x" }, { resendDelayMs: 50, resendPollMs: 5, resultWindowMs: 200 }); + assert.deepEqual(puts, ["tests:completion:a"]); + } finally { server.close(); } +}); -- 2.47.3 From 0b326fa5eee6ccd6a27c2011778d16efc87e867d Mon Sep 17 00:00:00 2001 From: Libby Date: Sun, 8 Nov 2026 11:36:23 +0000 Subject: [PATCH 09/11] WIP cultnet-ts: Odin presence publisher completes on the result frame, not the ack (passes alone; two publisher tests fail inside the parallel full run, unexplained) Co-Authored-By: Claude Sonnet 5.5 --- .../src/idunn-odin-presence-publisher.ts | 40 ++++++++++++------- packages/cultnet-ts/src/peer.ts | 5 +++ .../test/idunn-runtime-authority.test.ts | 4 +- 3 files changed, 33 insertions(+), 16 deletions(-) diff --git a/packages/cultnet-ts/src/idunn-odin-presence-publisher.ts b/packages/cultnet-ts/src/idunn-odin-presence-publisher.ts index 3713363d..209a2881 100644 --- a/packages/cultnet-ts/src/idunn-odin-presence-publisher.ts +++ b/packages/cultnet-ts/src/idunn-odin-presence-publisher.ts @@ -107,26 +107,36 @@ async function publishDocument( const wirePayload = encode(encodeCultNetMessageForWire(message as never, "cultnet.schema.v0")); const packets = session.sendMany("schema", wirePayload, { reliable: true, ordered: true, nowMs: Date.now() }); for (const packet of packets) await sendPacket(socket, endpoint, packet); - await receiveUntil( - receiver, - session, - endpoint, - () => session.outstandingReliablePacketCount === 0, - 2000, - (frame) => { - if (frame.channelId !== "schema") return; - const response = parseCultNetMessage(decode(frame.payload)); - if (response.schemaVersion === "cultnet.error.v0") { - throw new Error(`Odin rejected runtime presence: ${response.error}`); - } - }, - ); + // The transport ack is not admission: the put completes on Odin's result frame. A refusal rejects whether + // it arrives before the ack, after it, or inside the result window; a success reply ends the wait early. + let succeeded = false; + const onFrame = (frame: { channelId: string; payload: Uint8Array }) => { + if (frame.channelId !== "schema") return; + const response = parseCultNetMessage(decode(frame.payload)); + if (response.schemaVersion === "cultnet.error.v0") throw new IdunnPresenceRefusedError(response); + succeeded = true; + }; + await receiveUntil(receiver, session, endpoint, () => session.outstandingReliablePacketCount === 0, 2000, onFrame); + if (!succeeded) await receiveUntil(receiver, session, endpoint, () => succeeded, RESULT_WINDOW_MS, onFrame, true); } finally { receiver.close(); socket.close(); } } +/** How long after the transport ack Odin's refusal may still arrive; a success reply ends the wait early. */ +const RESULT_WINDOW_MS = 500; + +/** Odin refused the presence put: the typed `cultnet.error.v0` it answered with. */ +export class IdunnPresenceRefusedError extends Error { + readonly code: string | undefined; + constructor(readonly response: { error: string; code?: string | null }) { + super(`Odin rejected runtime presence: ${response.error}`); + this.name = "IdunnPresenceRefusedError"; + this.code = response.code ?? undefined; + } +} + type PacketReceiver = { socket: dgram.Socket; next(timeoutMs: number): Promise; @@ -190,6 +200,7 @@ async function receiveUntil( predicate: (packet: CultNetRudpPacket) => boolean, timeoutMs: number, onDelivered?: (frame: { channelId: string; payload: Uint8Array; sequence: number }) => void, + returnOnTimeout = false, ): Promise { const deadline = Date.now() + timeoutMs; while (Date.now() < deadline) { @@ -204,6 +215,7 @@ async function receiveUntil( } for (const packet of session.dueResends(Date.now())) await sendPacket(receiver.socket, endpoint, packet); } + if (returnOnTimeout) return; throw new Error(`Timed out waiting for CultNet RUDP response after ${timeoutMs} ms.`); } diff --git a/packages/cultnet-ts/src/peer.ts b/packages/cultnet-ts/src/peer.ts index 0e54ee58..504973af 100644 --- a/packages/cultnet-ts/src/peer.ts +++ b/packages/cultnet-ts/src/peer.ts @@ -95,6 +95,10 @@ export class CultNetPeer extends EventEmitter { return this.#transport?.profile; } + /** + * Fire-and-forget: an ack or a return is not admission; read the result frame (a write is refused with + * `cultnet.error.v0`, `write_not_authorized` or `message_invalid`, as a `message` event). + */ send(message: CultNetMessage): void { const wireValue = encodeCultNetMessageForWire(message, this.#wireContract); const payload = encode(wireValue); @@ -106,6 +110,7 @@ export class CultNetPeer extends EventEmitter { this.#stream?.write(encodeFrame(payload)); } + /** Resolves when the transport has acknowledged what was sent. An ack is not admission; read the result frame. */ async flush(timeoutMs?: number): Promise { await this.#transport?.flush?.(timeoutMs); } diff --git a/packages/cultnet-ts/test/idunn-runtime-authority.test.ts b/packages/cultnet-ts/test/idunn-runtime-authority.test.ts index 581cf852..00d7f35b 100644 --- a/packages/cultnet-ts/test/idunn-runtime-authority.test.ts +++ b/packages/cultnet-ts/test/idunn-runtime-authority.test.ts @@ -309,10 +309,10 @@ async function startOdinPeer(context: TestContext, endpoint: string) { state.responseAck = response.ack; if (state.reject && state.order === "ack-then-refusal") { peer.send(encodeRudpPacket(serverSession.createAckForReceived(packet.sequence)), remote.port, remote.address); - setTimeout(() => peer.send(encodeRudpPacket(response), remote.port, remote.address), 120); + setTimeout(() => { try { peer.send(encodeRudpPacket(response), remote.port, remote.address); } catch { /* the test ended */ } }, 120); } else if (state.reject && state.order === "refusal-then-ack") { peer.send(encodeRudpPacket({ ...response, ack: 0, ackMask: 0 }), remote.port, remote.address); - setTimeout(() => peer.send(encodeRudpPacket(serverSession!.createAckForReceived(packet.sequence)), remote.port, remote.address), 120); + setTimeout(() => { try { peer.send(encodeRudpPacket(serverSession!.createAckForReceived(packet.sequence)), remote.port, remote.address); } catch { /* the test ended */ } }, 120); } else { peer.send(encodeRudpPacket(response), remote.port, remote.address); } -- 2.47.3 From 531fce567d0704cca6592b22c2d919ef11cc789e Mon Sep 17 00:00:00 2001 From: Libby Date: Sun, 8 Nov 2026 13:14:49 +0000 Subject: [PATCH 10/11] cultnet-ts test: cancel the Odin stub's delayed send when the test ends The stub's 120 ms second packet fired after peer.close() and threw ERR_SOCKET_DGRAM_NOT_RUNNING as an uncaught exception, failing whichever publisher test was running when the timer landed (load-dependent). Timers are now cleared in the test's after hook instead of swallowing the error. Co-Authored-By: Claude Sonnet 5.5 --- packages/cultnet-ts/test/idunn-runtime-authority.test.ts | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/packages/cultnet-ts/test/idunn-runtime-authority.test.ts b/packages/cultnet-ts/test/idunn-runtime-authority.test.ts index 00d7f35b..2e05bbf4 100644 --- a/packages/cultnet-ts/test/idunn-runtime-authority.test.ts +++ b/packages/cultnet-ts/test/idunn-runtime-authority.test.ts @@ -276,7 +276,10 @@ async function startOdinPeer(context: TestContext, endpoint: string) { peer.once("error", reject); peer.bind(port, "127.0.0.1", resolve); }); - context.after(() => peer.close()); + // The second packet of a split delivery is a timer; the test's end cancels it rather than racing the closed socket. + const timers = new Set(); + const later = (send: () => void) => { timers.add(setTimeout(send, 120)); }; + context.after(() => { for (const timer of timers) clearTimeout(timer); peer.close(); }); const state = { reject: false, // How a refusal and the ack of the put reach the publisher: in one packet, or the ack first, or the refusal first. @@ -309,10 +312,10 @@ async function startOdinPeer(context: TestContext, endpoint: string) { state.responseAck = response.ack; if (state.reject && state.order === "ack-then-refusal") { peer.send(encodeRudpPacket(serverSession.createAckForReceived(packet.sequence)), remote.port, remote.address); - setTimeout(() => { try { peer.send(encodeRudpPacket(response), remote.port, remote.address); } catch { /* the test ended */ } }, 120); + later(() => peer.send(encodeRudpPacket(response), remote.port, remote.address)); } else if (state.reject && state.order === "refusal-then-ack") { peer.send(encodeRudpPacket({ ...response, ack: 0, ackMask: 0 }), remote.port, remote.address); - setTimeout(() => { try { peer.send(encodeRudpPacket(serverSession!.createAckForReceived(packet.sequence)), remote.port, remote.address); } catch { /* the test ended */ } }, 120); + later(() => peer.send(encodeRudpPacket(serverSession!.createAckForReceived(packet.sequence)), remote.port, remote.address)); } else { peer.send(encodeRudpPacket(response), remote.port, remote.address); } -- 2.47.3 From 675a6494c1acdd3c288a529b68e2a7dda38575b3 Mon Sep 17 00:00:00 2001 From: Libby Date: Sun, 8 Nov 2026 13:18:40 +0000 Subject: [PATCH 11/11] cultmesh-rs: single and batch publish complete on the result frame, not the transport ack One loop now serves both: every put acknowledged, then a refusal frame (typed cultnet.error.v0) fails the call in either order against the ack, and a success frame or the end of the new result_window option (default 500 ms) completes it. The batch no longer waits on flush_reliable alone, so it cannot hide a refusal. Raw send primitives (CultNetRudpSocketTransportConnection::send/send_reliable/ flush_reliable, tcp_framed send) carry the doc line: fire-and-forget, an ack or a return is not admission, read the result frame. Co-Authored-By: Claude Sonnet 5.5 --- packages/cultmesh-rs/src/lib.rs | 198 ++++++++++++++++----------- packages/cultnet-rs/src/rudp.rs | 6 + packages/cultnet-rs/src/transport.rs | 1 + 3 files changed, 128 insertions(+), 77 deletions(-) diff --git a/packages/cultmesh-rs/src/lib.rs b/packages/cultmesh-rs/src/lib.rs index 172a5668..33cb4858 100644 --- a/packages/cultmesh-rs/src/lib.rs +++ b/packages/cultmesh-rs/src/lib.rs @@ -82,7 +82,10 @@ pub struct CultMeshRudpDocumentPublishOptions { pub target: SocketAddr, pub runtime_id: String, pub connect_timeout: Duration, + /// How long the acknowledgement of every put may take. pub flush_timeout: Duration, + /// How long after the acknowledgement a refusal may still arrive; a success frame ends the wait early. + pub result_window: Duration, pub poll_interval: Duration, pub resend_delay_ms: u64, pub source_agent_id: Option, @@ -149,6 +152,7 @@ impl Default for CultMeshRudpDocumentPublishOptions { runtime_id: "cultmesh-rudp-document-publisher".to_string(), connect_timeout: Duration::from_secs(3), flush_timeout: Duration::from_millis(300), + result_window: Duration::from_millis(500), poll_interval: Duration::from_millis(10), resend_delay_ms: 50, source_agent_id: None, @@ -882,77 +886,30 @@ fn unix_millis() -> u128 { /// Publish one raw CultNet schema message to a RUDP catalog. /// -/// Each call owns a fresh connection epoch and returns only after the catalog -/// acknowledges reliable application admission. Against a Rust CultMesh -/// catalog, which never acknowledges a message it refuses, a refusal is -/// returned as an error, naming the refusal when it is one of the fixed texts -/// such a catalog sends. A catalog that acknowledges a message on receipt and -/// refuses it afterwards (the TypeScript document server does) is reported as -/// having admitted it: the acknowledgement arrives first. +/// A put completes on the catalog's result frame, never on the transport +/// acknowledgement: an ack proves delivery, not admission. A refusal (a +/// `cultnet.error.v0` frame) returns an error in either order against the ack, +/// naming the refusal when it is one of the fixed texts a catalog sends. A +/// catalog that admits a put sends no success frame, so after the ack the call +/// waits `result_window` for a refusal and reports the put as admitted when +/// none comes (interim until the catalog sends a success frame; a success +/// frame ends the wait early). Each call owns a fresh connection epoch. pub fn publish_cultnet_message_to_rudp_catalog( message: &cultnet_rs::CultNetMessage, options: CultMeshRudpDocumentPublishOptions, ) -> Result<()> { - if options.runtime_id.trim().is_empty() { - anyhow::bail!("runtime_id must be non-empty"); - } - let mut client = connect_rudp_client( - options.target, - &options.runtime_id, - options.connect_timeout, - options.poll_interval, - options.resend_delay_ms, - "cultmesh-rudp-document-publisher", - )?; - - let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?; - let receipt = client.send_reliable("schema", payload)?; - let flush_deadline = Instant::now() + options.flush_timeout; - while Instant::now() < flush_deadline { - if let Some(frame) = client.receive_once()? { - if frame.channel_id == "schema" { - if let Ok(cultnet_rs::CultNetMessage::Error { error, .. }) = - decode_cultnet_message_from_slice( - &frame.payload, - CultNetWireContract::CultNetSchemaV0, - ) - { - let _ = client.disconnect(b"document-refused".to_vec()); - anyhow::bail!( - "CultMesh RUDP catalog {} refused the put{}", - options.target, - recognised_refusal(&error) - ); - } - } - } - match client.reliable_send_status(&receipt) { - cultnet_rs::CultNetRudpReliableSendStatus::Acknowledged => { - let _ = client.disconnect(b"document-acknowledged".to_vec()); - return Ok(()); - } - cultnet_rs::CultNetRudpReliableSendStatus::Invalidated => { - anyhow::bail!("CultMesh RUDP document receipt was invalidated") - } - cultnet_rs::CultNetRudpReliableSendStatus::Pending => {} - } - client.poll_resends()?; - thread::sleep(options.poll_interval); - } - let _ = client.disconnect(b"document-unacknowledged".to_vec()); - anyhow::bail!( - "timed out waiting for reliable acknowledgement from CultMesh RUDP catalog {}", - options.target - ) + publish_cultnet_messages_to_rudp_catalog(std::slice::from_ref(message), options) } -/// Publish a batch of raw CultNet schema messages to a RUDP catalog. +/// Publish a batch of raw CultNet schema messages to a RUDP catalog on one +/// connection. /// -/// Unlike [`publish_cultnet_message_to_rudp_catalog`], which polls for -/// per-message application-level acknowledgement, this sends every message on -/// one connection and waits only for a transport-level `flush_reliable`. It -/// trades per-receipt confirmation for batch throughput; callers that need a -/// stronger delivery guarantee should use the singular publish path instead. +/// Completes under the same rule as [`publish_cultnet_message_to_rudp_catalog`]: +/// every message acknowledged, then no refusal frame within `result_window` +/// (or one success frame per message). A refusal of any message in the batch +/// fails the whole call; the batch cannot hide it. The catalog's answers do not +/// name the message they answer, so a failed batch does not say which message +/// was refused or whether the others were admitted. pub fn publish_cultnet_messages_to_rudp_catalog( messages: &[cultnet_rs::CultNetMessage], options: CultMeshRudpDocumentPublishOptions, @@ -969,11 +926,62 @@ pub fn publish_cultnet_messages_to_rudp_catalog( "cultmesh-rudp-document-publisher", )?; + let mut receipts = Vec::with_capacity(messages.len()); for message in messages { let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?; - client.send("schema", payload)?; + receipts.push(client.send_reliable("schema", payload)?); + } + let ack_deadline = Instant::now() + options.flush_timeout; + let mut acknowledged_at: Option = None; + let mut successes = 0; + loop { + while let Some(frame) = client.receive_once()? { + if frame.channel_id != "schema" { + continue; + } + match decode_cultnet_message_from_slice(&frame.payload, CultNetWireContract::CultNetSchemaV0) { + Ok(cultnet_rs::CultNetMessage::Error { error, .. }) => { + let _ = client.disconnect(b"document-refused".to_vec()); + anyhow::bail!( + "CultMesh RUDP catalog {} refused the put{}", + options.target, + recognised_refusal(&error) + ); + } + Ok(_) => successes += 1, + Err(_) => {} + } + } + if acknowledged_at.is_none() { + let mut all_acknowledged = true; + for receipt in &receipts { + match client.reliable_send_status(receipt) { + cultnet_rs::CultNetRudpReliableSendStatus::Acknowledged => {} + cultnet_rs::CultNetRudpReliableSendStatus::Invalidated => { + anyhow::bail!("CultMesh RUDP document receipt was invalidated") + } + cultnet_rs::CultNetRudpReliableSendStatus::Pending => all_acknowledged = false, + } + } + if all_acknowledged { + acknowledged_at = Some(Instant::now()); + } else if Instant::now() >= ack_deadline { + let _ = client.disconnect(b"document-unacknowledged".to_vec()); + anyhow::bail!( + "timed out waiting for reliable acknowledgement from CultMesh RUDP catalog {}", + options.target + ) + } + } + if let Some(acknowledged_at) = acknowledged_at { + if successes >= messages.len() || acknowledged_at.elapsed() >= options.result_window { + let _ = client.disconnect(b"document-completed".to_vec()); + return Ok(()); + } + } + client.poll_resends()?; + thread::sleep(options.poll_interval); } - client.flush_reliable(options.flush_timeout) } fn connect_rudp_client( @@ -1454,8 +1462,16 @@ mod tests { /// A catalog that acknowledges every reliable message it receives and refuses it with a typed /// write_not_authorized, the acknowledgement either before the refusal or after it (the second packet /// follows the first by 100 ms). The refusal carries no acknowledgement of its own. + /// What the stub catalog sends for a put, and in which order against the transport ack. + #[derive(Clone, Copy, PartialEq, Eq)] + enum CatalogReply { + RefusalThenAck, + AckThenRefusal, + AckThenSuccess, + } + fn acknowledging_refusing_catalog( - refusal_first: bool, + reply: CatalogReply, ) -> Result<( SocketAddr, std::sync::Arc, @@ -1503,10 +1519,17 @@ mod tests { } let ack = session.create_ack_for_received(packet.sequence); let refusal = encode_cultnet_message_to_vec( - &cultnet_rs::CultNetMessage::Error { - error: "write_not_authorized: the catalog refused the document".into(), - code: Some(cultnet_rs::CultNetErrorCode::WriteNotAuthorized), - details: None, + &if reply == CatalogReply::AckThenSuccess { + cultnet_rs::CultNetMessage::SnapshotResponseRaw { + message_id: "completion".into(), + documents: Vec::new(), + } + } else { + cultnet_rs::CultNetMessage::Error { + error: "write_not_authorized: the catalog refused the document".into(), + code: Some(cultnet_rs::CultNetErrorCode::WriteNotAuthorized), + details: None, + } }, CultNetWireContract::CultNetSchemaV0, )?; @@ -1528,7 +1551,7 @@ mod tests { } Ok(()) }; - if refusal_first { + if reply == CatalogReply::RefusalThenAck { send_refusals(&socket)?; thread::sleep(Duration::from_millis(100)); socket.send_to(&encode_rudp_packet(&ack)?, peer)?; @@ -2026,7 +2049,7 @@ mod tests { // acknowledges the put and refuses it, the refusal arriving before or after the ack. #[test] fn a_publish_whose_refusal_arrives_before_the_ack_is_refused() -> Result<()> { - let (target, stop, catalog) = acknowledging_refusing_catalog(true)?; + let (target, stop, catalog) = acknowledging_refusing_catalog(CatalogReply::RefusalThenAck)?; let result = publish_cultnet_message_to_rudp_catalog(&refused_note(), publish_options(target)); stop.store(true, std::sync::atomic::Ordering::SeqCst); catalog.join().expect("catalog thread should not panic")?; @@ -2038,7 +2061,7 @@ mod tests { #[test] fn a_publish_whose_ack_arrives_before_the_refusal_is_refused() -> Result<()> { - let (target, stop, catalog) = acknowledging_refusing_catalog(false)?; + let (target, stop, catalog) = acknowledging_refusing_catalog(CatalogReply::AckThenRefusal)?; let result = publish_cultnet_message_to_rudp_catalog(&refused_note(), publish_options(target)); stop.store(true, std::sync::atomic::Ordering::SeqCst); catalog.join().expect("catalog thread should not panic")?; @@ -2049,14 +2072,35 @@ mod tests { #[test] fn a_batch_publish_is_not_complete_on_the_ack_of_refused_puts() -> Result<()> { - for refusal_first in [false, true] { - let (target, stop, catalog) = acknowledging_refusing_catalog(refusal_first)?; + for reply in [CatalogReply::AckThenRefusal, CatalogReply::RefusalThenAck] { + let (target, stop, catalog) = acknowledging_refusing_catalog(reply)?; let result = publish_cultnet_messages_to_rudp_catalog(&[refused_note()], publish_options(target)); stop.store(true, std::sync::atomic::Ordering::SeqCst); catalog.join().expect("catalog thread should not panic")?; - assert!(result.is_err(), "refusal_first={refusal_first}: a refused put was reported as published"); + assert!(result.is_err(), "a refused put was reported as published"); } Ok(()) } + #[test] + fn a_success_frame_completes_the_publish_without_waiting_out_the_result_window() -> Result<()> { + for batch in [false, true] { + let (target, stop, catalog) = acknowledging_refusing_catalog(CatalogReply::AckThenSuccess)?; + let options = CultMeshRudpDocumentPublishOptions { + result_window: Duration::from_secs(30), + ..publish_options(target) + }; + let started = Instant::now(); + let result = if batch { + publish_cultnet_messages_to_rudp_catalog(&[refused_note()], options) + } else { + publish_cultnet_message_to_rudp_catalog(&refused_note(), options) + }; + stop.store(true, std::sync::atomic::Ordering::SeqCst); + catalog.join().expect("catalog thread should not panic")?; + result?; + assert!(started.elapsed() < Duration::from_secs(10), "batch={batch}: waited out the window"); + } + Ok(()) + } } diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index 37bf73d8..6a2aa20b 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -2198,6 +2198,7 @@ impl CultNetRudpSocketTransportConnection { self.send_packet(&packet) } + /// Fire-and-forget: an ack or a return is not admission; read the result frame. pub fn send(&mut self, channel_id: &str, payload: Vec) -> Result<()> { let options = channel_send_options( @@ -2216,6 +2217,8 @@ impl CultNetRudpSocketTransportConnection { /// Sends one frame on a reliable channel whose packets cannot expire and /// returns the transport receipt used to observe its exact acknowledgement. + /// + /// Fire-and-forget: an ack or a return is not admission; read the result frame. pub fn send_reliable( &mut self, channel_id: &str, @@ -2309,6 +2312,9 @@ impl CultNetRudpSocketTransportConnection { self.session.outstanding_reliable_packet_count() } + /// Waits until every reliable write is acknowledged by the transport. + /// + /// Fire-and-forget: an ack or a return is not admission; read the result frame. pub fn flush_reliable(&mut self, timeout: Duration) -> Result<()> { let original_timeout = self.socket.read_timeout()?; let poll_timeout = original_timeout diff --git a/packages/cultnet-rs/src/transport.rs b/packages/cultnet-rs/src/transport.rs index f2c196d4..57070dc3 100644 --- a/packages/cultnet-rs/src/transport.rs +++ b/packages/cultnet-rs/src/transport.rs @@ -252,6 +252,7 @@ impl TcpFramedTransportConnection where TStream: Write, { + /// Fire-and-forget: an ack or a return is not admission; read the result frame. pub fn send(&mut self, channel_id: &str, payload: &[u8]) -> Result<()> { if channel_id != "schema" { return Err(anyhow!( -- 2.47.3