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 <noreply@anthropic.com>

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
 

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"));

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<EndPoint, int> UnsendableAfter { get; } = new Dictionary<EndPoint, int>();
 
+        // Runs before each datagram is sent, to order another thread's work between two sends.
+        internal Action<EndPoint, CultNetRudpPacket>? BeforeSend { get; set; }
+
         private static string RemoteKey(EndPoint endpoint)
         {
             return endpoint.ToString() ?? string.Empty;

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()
         {