From 2d76b8d1b6315ee85a4b6cb328b346183a57fe45 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 16:02:27 +0200 Subject: [PATCH] Pin the reset before a refusal's goodbye (cultmesh-ts, cultmesh-py) and C#'s once-guard The document servers' refusal tests sent an unreliable fragment, whose sequence the session never records, so dropping the reset before the goodbye survived. They now refuse a reliable fragment and check the goodbye does not acknowledge it. C#: a BeforeSend seam orders a caller's failing send between the polling thread's ack and its send, so both threads end the same session; the peer hears one goodbye. Co-Authored-By: Claude Opus 5.5 --- packages/cultmesh-py/tests/test_cultmesh.py | 31 ++++++++++++++++ .../rudp-document-server-send-failure.test.ts | 13 +++++-- src/GameCult.Networking/CultNetTransport.cs | 4 +++ .../CultNetRudpSendFailureTests.cs | 36 +++++++++++++++++++ 4 files changed, 82 insertions(+), 2 deletions(-) diff --git a/packages/cultmesh-py/tests/test_cultmesh.py b/packages/cultmesh-py/tests/test_cultmesh.py index 9b936ca1..4c6afabc 100644 --- a/packages/cultmesh-py/tests/test_cultmesh.py +++ b/packages/cultmesh-py/tests/test_cultmesh.py @@ -4868,6 +4868,37 @@ class CultMeshRudpSendFailureTests(unittest.TestCase): session.receive(accept, 0) return sock, session + def test_the_goodbye_for_a_refused_reliable_frame_does_not_acknowledge_it(self) -> None: + import dataclasses + + from cultnet_py import CultNetRudpSendOptions, encode_rudp_packet + + server = CultMesh.serve_node(CultMesh.create_node(runtime_id="mesh-server")) + try: + sock, session = self._raw_peer(server) + # A reliable fragment whose metadata the session refuses: fragment id 0 names no set. + # The session records its sequence before it refuses it. + (fragment,) = session.send_many("schema", b"x", CultNetRudpSendOptions(reliable=True, ordered=True)) + refused = dataclasses.replace(fragment, fragment_id=0, fragment_index=0, fragment_count=2) + sock.sendto(encode_rudp_packet(refused), ("127.0.0.1", server.port)) + goodbye = None + for _ in range(40): + try: + wire, _ = sock.recvfrom(65535) + except TimeoutError: + continue + packet = decode_rudp_packet(wire) + if packet.packet_type == CultNetRudpPacketType.DISCONNECT: + goodbye = packet + break + self.assertIsNotNone(goodbye) + self.assertEqual(goodbye.payload, b"session refused a packet") + distance = goodbye.ack - refused.sequence + acknowledged = distance == 0 or (0 < distance <= 32 and goodbye.ack_mask & (1 << (distance - 1)) != 0) + self.assertFalse(acknowledged, "the goodbye acknowledged the refused frame") + finally: + server.stop() + def test_responses_after_one_that_can_never_be_sent_are_not_sent(self) -> None: from cultnet_py import CultNetRudpSendOptions, encode_rudp_packet diff --git a/packages/cultmesh-ts/test/rudp-document-server-send-failure.test.ts b/packages/cultmesh-ts/test/rudp-document-server-send-failure.test.ts index 534a9440..64f19111 100644 --- a/packages/cultmesh-ts/test/rudp-document-server-send-failure.test.ts +++ b/packages/cultmesh-ts/test/rudp-document-server-send-failure.test.ts @@ -189,8 +189,9 @@ test("a packet the session refuses ends only that peer's session, and the peer i session.receive(arrived.find((packet) => packet.packetType === "accept")!, Date.now()); y = await served.peer("y"); - // A fragment whose metadata the session refuses: fragment id 0 names no set. - const [fragment] = session.sendMany("schema", new Uint8Array([1]), { reliable: false }); + // A reliable fragment whose metadata the session refuses: fragment id 0 names no set. + // The session records its sequence before it refuses it. + const [fragment] = session.sendMany("schema", new Uint8Array([1]), { reliable: true, ordered: true }); raw.send( encodeRudpPacket({ ...fragment!, fragmentId: 0, fragmentIndex: 0, fragmentCount: 2 }), served.server.bind.port, @@ -199,6 +200,14 @@ test("a packet the session refuses ends only that peer's session, and the peer i await waitFor(() => arrived.some((packet) => packet.packetType === "disconnect"), "the goodbye"); const goodbye = arrived.find((packet) => packet.packetType === "disconnect")!; assert.equal(Buffer.from(goodbye.payload ?? []).toString("utf8"), "session refused a packet"); + // The goodbye's ack field does not acknowledge the refused frame, or the peer would + // count as delivered a frame that was never handled. + const acknowledged = + goodbye.ack === fragment!.sequence || + (goodbye.ack > fragment!.sequence && + goodbye.ack - fragment!.sequence <= 32 && + (goodbye.ackMask & (1 << (goodbye.ack - fragment!.sequence - 1))) !== 0); + assert.equal(acknowledged, false, "the goodbye acknowledged the refused frame"); assert.deepEqual(served.closed, [(raw.address() as { port: number }).port]); y.send(put("y-after")); diff --git a/src/GameCult.Networking/CultNetTransport.cs b/src/GameCult.Networking/CultNetTransport.cs index dcfe3b6a..7b09fe6a 100644 --- a/src/GameCult.Networking/CultNetTransport.cs +++ b/src/GameCult.Networking/CultNetTransport.cs @@ -3209,6 +3209,7 @@ namespace GameCult.Networking private SocketException? SendPacket(EndPoint remoteEndPoint, CultNetRudpPacket packet) { var wire = CultNetRudpPacketCodec.Encode(packet); + BeforeSend?.Invoke(remoteEndPoint, packet); try { if (UnsendableAfter.TryGetValue(remoteEndPoint, out var remaining)) @@ -3246,6 +3247,9 @@ namespace GameCult.Networking // Peers whose next datagram after this many fails permanently, once. internal Dictionary UnsendableAfter { get; } = new Dictionary(); + // Runs before each datagram is sent, to order another thread's work between two sends. + internal Action? BeforeSend { get; set; } + private static string RemoteKey(EndPoint endpoint) { return endpoint.ToString() ?? string.Empty; diff --git a/tests/GameCult.Networking.Tests/CultNetRudpSendFailureTests.cs b/tests/GameCult.Networking.Tests/CultNetRudpSendFailureTests.cs index 70f9acc7..0ae010dc 100644 --- a/tests/GameCult.Networking.Tests/CultNetRudpSendFailureTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetRudpSendFailureTests.cs @@ -313,6 +313,42 @@ namespace GameCult.Networking.Tests } } + [Test] + public void ASessionEndedByACallerAndThePollingThreadTogetherSaysGoodbyeOnce() + { + var (server, listenerEndPoint, x, _) = TwoPeers(maxFragmentBytes: 1000); + using (server) + { + var ended = RecordDisconnects(server); + // The polling thread acknowledges X's frame outside the session's gate. Just before + // that ack goes out, a caller's send to X fails after its first fragment and ends the + // session; then the ack can never be sent either. + server.BeforeSend = (remote, packet) => + { + if (!remote.Equals(x.EndPoint) || packet.PacketType != CultNetRudpPacketType.Ack) + return; + server.BeforeSend = null; + server.UnsendableAfter[x.EndPoint] = 1; + var caller = new Thread(() => + { + try { server.SendSchema(x.Server, new byte[3000]); } + catch (SocketException) { } + }); + caller.Start(); + caller.Join(); + server.UnsendableAfter[x.EndPoint] = 0; + }; + x.Socket.SendTo(CultNetRudpPacketCodec.Encode(ReliableFrame(x, "from-x")), listenerEndPoint); + server.ReceiveOnce(); + server.PollResends(); + + var goodbyes = Drain(x.Socket).Where(p => p.PacketType == CultNetRudpPacketType.Disconnect).ToList(); + Assert.That(goodbyes, Has.Count.EqualTo(1)); + Assert.That(ended, Is.EqualTo(new[] { x.Server })); + Assert.That(IsUnsendableReason(x.Server.DisconnectReason), Is.True); + } + } + [Test] public void APermanentFailureInResendsEndsOnlyThatPeersSession() {