Ack Cut 1b tests: queued writes, flush after a reconnect, the refusal goodbye

The lost write in every ending test now leaves part of a fragmented write
queued. A flush that starts after a reconnect, or after a server-mode
transport accepts a new Connect, must succeed. The three stray-packet tests
that read the peer's pending set after a Disconnect encoded the old rule
(the peer's owed write survives the goodbye); they now assert on the
goodbye's ack field and mask, which is where the refused frame must not be
acknowledged.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

packages/cultnet-py/tests/test_cultnet.py

@@ -94,6 +94,17 @@ from cultnet_py import (
 )
 
 
+def acknowledges(packet, sequence: int) -> bool:
+    """Whether the packet's ack field and mask cover the sequence."""
+    if packet.ack == sequence:
+        return True
+    return (
+        packet.ack > sequence
+        and packet.ack - sequence <= 32
+        and (packet.ack_mask & (1 << (packet.ack - sequence - 1))) != 0
+    )
+
+
 def bind_udp_socket() -> socket.socket:
     sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
     sock.bind(("127.0.0.1", 0))
@@ -1063,14 +1074,13 @@ class CultNetTests(unittest.TestCase):
             # The peer is told, and the goodbye does not acknowledge the refused frame.
             (goodbye,) = [p for p in drain_peer() if p.packet_type == CultNetRudpPacketType.DISCONNECT]
             self.assertTrue(peer.receive(goodbye, 1).disconnected)
-            self.assertIn(poison.sequence, peer.pending_reliable_sequences)
+            self.assertFalse(acknowledges(goodbye, poison.sequence), "the goodbye acknowledged the refused frame")
 
             # A retransmit is dropped, not acknowledged.
             to_server(poison)
             self.assertIsNone(server.receive_once())
             self.assertEqual(server.stats.packets_dropped, 2)
             self.assertEqual(drain_peer(), [])
-            self.assertIn(poison.sequence, peer.pending_reliable_sequences)
 
             # A Connect claims the endpoint again.
             again = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=connection_id))

packages/cultnet-py/tests/test_rudp_session_ending.py

@@ -37,11 +37,26 @@ def client_with_a_lost_write() -> tuple[CultNetRudpSession, CultNetRudpSession]:
     server = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=CONNECTION_ID, initial_sequence=500))
     client.receive(server.accept_connect(client.create_connect(0), 0), 0)
     client.send("schema", b"owed to the old session", CultNetRudpSendOptions(reliable=True, ordered=True))
-    assert client.outstanding_reliable_packet_count == 1
+    # A fragmented write larger than the send window leaves part of it queued.
+    client.send_many(
+        "schema", bytes(40 * 8), CultNetRudpSendOptions(reliable=True, ordered=True), max_fragment_bytes=8
+    )
+    assert client.outstanding_reliable_packet_count == 41
+    assert client.queued_reliable_packet_count > 0
     return client, server
 
 
-def connected_client() -> tuple[CultNetRudpSocketTransportConnection, socket.socket]:
+def drain(sock: socket.socket) -> list:
+    packets = []
+    while True:
+        try:
+            wire, _ = sock.recvfrom(65535)
+        except TimeoutError:
+            return packets
+        packets.append(decode_rudp_packet(wire))
+
+
+def connected_client() -> tuple[CultNetRudpSocketTransportConnection, socket.socket, tuple]:
     peer_socket = bind_udp_socket()
     client = CultNetRudpSocketTransportConnection(
         CultNetRudpSocketTransportOptions(
@@ -59,7 +74,7 @@ def connected_client() -> tuple[CultNetRudpSocketTransportConnection, socket.soc
     peer_socket.sendto(encode_rudp_packet(accept), client_addr)
     client.receive_once()
     assert client.connected
-    return client, peer_socket
+    return client, peer_socket, client_addr
 
 
 class CultNetRudpSessionEndingTests(unittest.TestCase):
@@ -88,7 +103,7 @@ class CultNetRudpSessionEndingTests(unittest.TestCase):
                     )
 
     def test_a_flush_fails_after_a_local_disconnect_or_a_timeout(self) -> None:
-        client, peer_socket = connected_client()
+        client, peer_socket, _ = connected_client()
         try:
             client.send("schema", b"owed")
             client.disconnect(b"bye")
@@ -98,7 +113,7 @@ class CultNetRudpSessionEndingTests(unittest.TestCase):
             client.close()
             peer_socket.close()
 
-        client, peer_socket = connected_client()
+        client, peer_socket, _ = connected_client()
         try:
             client.send("schema", b"owed")
             time.sleep(0.005)
@@ -110,7 +125,7 @@ class CultNetRudpSessionEndingTests(unittest.TestCase):
             peer_socket.close()
 
     def test_a_flush_that_started_before_the_end_fails_even_after_a_later_connect(self) -> None:
-        client, peer_socket = connected_client()
+        client, peer_socket, _ = connected_client()
         outcome: list[BaseException | None] = []
 
         def flush() -> None:
@@ -135,5 +150,56 @@ class CultNetRudpSessionEndingTests(unittest.TestCase):
             peer_socket.close()
 
 
+    def test_a_flush_started_after_a_reconnect_waits_on_the_new_session(self) -> None:
+        client, peer_socket, client_addr = connected_client()
+        try:
+            client.send("schema", b"owed to the old session")
+            client.disconnect(b"bye")
+            client.connect(b"hello again")
+            connect = [p for p in drain(peer_socket) if p.packet_type == CultNetRudpPacketType.CONNECT][-1]
+            accept = CultNetRudpSession(
+                CultNetRudpSessionOptions(connection_id=CONNECTION_ID, initial_sequence=700)
+            ).accept_connect(connect, 0)
+            peer_socket.sendto(encode_rudp_packet(accept), client_addr)
+            client.receive_once()
+
+            client.flush_reliable(0.5)
+        finally:
+            client.close()
+            peer_socket.close()
+
+    def test_a_flush_started_after_the_server_accepts_a_new_connect_waits_on_the_new_session(self) -> None:
+        server_socket = bind_udp_socket()
+        peer_socket = bind_udp_socket()
+        server = CultNetRudpSocketTransportConnection(
+            CultNetRudpSocketTransportOptions(
+                runtime_id="python-rudp-server",
+                socket=server_socket,
+                mode=CultNetRudpSocketMode.SERVER,
+                connection_id=CONNECTION_ID,
+            )
+        )
+        to_server = lambda packet: peer_socket.sendto(encode_rudp_packet(packet), server_socket.getsockname())
+        try:
+            first = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=CONNECTION_ID))
+            to_server(first.create_connect(0))
+            server.receive_once()
+            to_server(first.create_disconnect())
+            server.receive_once()
+
+            following = CultNetRudpSession(CultNetRudpSessionOptions(connection_id=CONNECTION_ID))
+            to_server(following.create_connect(0))
+            server.receive_once()
+            accept = [p for p in drain(peer_socket) if p.packet_type == CultNetRudpPacketType.ACCEPT][-1]
+            following.receive(accept, 0)
+            to_server(following.create_ack())
+            server.receive_once()
+
+            server.flush_reliable(0.5)
+        finally:
+            server.close()
+            peer_socket.close()
+
+
 if __name__ == "__main__":
     unittest.main()

packages/cultnet-rs/tests/rudp_stray_packets.rs

@@ -766,7 +766,19 @@ fn client_with_a_lost_write() -> Result<(CultNetRudpSession, CultNetRudpSession)
             ..Default::default()
         },
     )?;
-    assert_eq!(client.outstanding_reliable_packet_count(), 1);
+    // A fragmented write larger than the send window leaves part of it queued.
+    client.send_many(
+        "schema",
+        vec![7u8; 40 * 8],
+        CultNetRudpSendOptions {
+            reliable: true,
+            ordered: true,
+            ..Default::default()
+        },
+        Some(8),
+    )?;
+    assert_eq!(client.outstanding_reliable_packet_count(), 41);
+    assert!(client.queued_reliable_packet_count() > 0);
     Ok((client, server))
 }
 
@@ -873,3 +885,56 @@ fn a_flush_fails_after_a_local_disconnect_or_a_timeout() -> Result<()> {
     assert!(error.to_string().contains("ended before its reliable writes were acknowledged"), "{error}");
     Ok(())
 }
+
+/// A flush that starts after a reconnect waits on the new session: the end of
+/// the old one does not fail it.
+#[test]
+fn a_flush_started_after_a_reconnect_waits_on_the_new_session() -> Result<()> {
+    let server = socket()?;
+    let server_addr = server.local_addr()?;
+    let (mut client, _peer, client_addr) = connected_client(&server, server_addr)?;
+    client.send_reliable("schema", b"owed to the old session".to_vec())?;
+    client.disconnect(b"bye".to_vec())?;
+    client.connect(b"hello again".to_vec())?;
+    let connect = drain(&server)?
+        .into_iter()
+        .rfind(|packet| packet.packet_type == CultNetRudpPacketType::Connect)
+        .expect("the reconnect reached the peer");
+    let accept = raw_session(CONNECTION_ID).accept_connect(&connect, 0, Vec::new())?;
+    send_to(&server, client_addr, &accept)?;
+    let _ = client.receive_once()?;
+
+    client.flush_reliable(Duration::from_millis(500))?;
+    Ok(())
+}
+
+/// A server-mode transport that accepts a new Connect after its peer left has a
+/// live session again, and a flush started in it waits on that session.
+#[test]
+fn a_flush_started_after_the_server_accepts_a_new_connect_waits_on_the_new_session() -> Result<()> {
+    let server_socket = socket()?;
+    let server_addr = server_socket.local_addr()?;
+    let mut server = CultNetRudpSocketTransportConnection::new(
+        CultNetRudpSocketTransportOptions::server("server", server_socket, CONNECTION_ID),
+    )?;
+    let peer = socket()?;
+    let mut first = raw_session(CONNECTION_ID);
+    send_to(&peer, server_addr, &first.create_connect(0, b"peer".to_vec())?)?;
+    let _ = server.receive_once()?;
+    send_to(&peer, server_addr, &first.create_disconnect(b"bye".to_vec()))?;
+    let _ = server.receive_once()?;
+
+    let mut next = raw_session(CONNECTION_ID);
+    send_to(&peer, server_addr, &next.create_connect(0, b"peer".to_vec())?)?;
+    let _ = server.receive_once()?;
+    let accept = drain(&peer)?
+        .into_iter()
+        .rfind(|packet| packet.packet_type == CultNetRudpPacketType::Accept)
+        .expect("the new Connect was accepted");
+    next.receive(&accept, 0)?;
+    send_to(&peer, server_addr, &next.create_ack_for_received(accept.sequence))?;
+    let _ = server.receive_once()?;
+
+    server.flush_reliable(Duration::from_millis(500))?;
+    Ok(())
+}

packages/cultnet-ts/test/cultnet.test.ts

@@ -166,6 +166,14 @@ function createDuplexPair(): { a: Duplex; b: Duplex } {
   return { a, b };
 }
 
+/// Whether the packet's ack field and mask cover the sequence.
+function acknowledges(packet: CultNetRudpPacket, sequence: number): boolean {
+  if (packet.ack === sequence) return true;
+  return packet.ack > sequence
+    && packet.ack - sequence <= 32
+    && (packet.ackMask & (1 << (packet.ack - sequence - 1))) !== 0;
+}
+
 async function bindUdpSocket(): Promise<Socket> {
   const socket = dgram.createSocket("udp4");
   await new Promise<void>((resolve) => socket.bind(0, "127.0.0.1", resolve));
@@ -1975,7 +1983,7 @@ test("rudp socket transport ends the session that refuses a reliable frame and n
     // neither in the goodbye nor after it retransmits.
     const goodbye = received.find((p) => p.packetType === "disconnect")!;
     assert.equal(peer.receive(goodbye, 1).disconnected, true);
-    assert.ok(peer.pendingReliableSequences.includes(poison.sequence), "the refused frame was acknowledged");
+    assert.equal(acknowledges(goodbye, poison.sequence), false, "the goodbye acknowledged the refused frame");
     const before = received.length;
     toServer(poison);
     await waitFor(() => server.stats.packetsDropped === 2, "the retransmit dropped");
@@ -2489,3 +2497,84 @@ test("operation service sweeps a session that ran a handler once it idles out",
     await server.close();
   }
 });
+
+test("a flush started after a reconnect waits on the new session", async () => {
+  const peerSocket = await bindUdpSocket();
+  const clientSocket = await bindUdpSocket();
+  const connectionId = 0x1020305b;
+  const client = new CultNetRudpSocketTransportConnection({
+    runtimeId: "rudp-client",
+    socket: clientSocket,
+    mode: "client",
+    remoteHost: "127.0.0.1",
+    remotePort: udpPort(peerSocket),
+    connectionId,
+    resendDelayMs: 1000,
+    resendPollMs: 1000,
+  });
+  const received: CultNetRudpPacket[] = [];
+  let clientPort = 0;
+  peerSocket.on("message", (wire, remote) => {
+    clientPort = remote.port;
+    received.push(decodeRudpPacket(wire));
+  });
+  try {
+    client.connect();
+    await waitFor(() => received.some((p) => p.packetType === "connect"), "the Connect");
+    const first = new CultNetRudpSession({ connectionId });
+    peerSocket.send(encodeRudpPacket(first.acceptConnect(received[0]!, 0)), clientPort, "127.0.0.1");
+    await waitFor(() => client.connected, "the client connected");
+    client.send("schema", Buffer.from("owed to the old session"));
+    client.checkTimeout(50, Date.now() + 10_000);
+    assert.equal(client.connected, false);
+
+    client.connect();
+    await waitFor(() => received.filter((p) => p.packetType === "connect").length === 2, "the reconnect");
+    const second = new CultNetRudpSession({ connectionId });
+    const reconnect = received.filter((p) => p.packetType === "connect")[1]!;
+    peerSocket.send(encodeRudpPacket(second.acceptConnect(reconnect, 0)), clientPort, "127.0.0.1");
+    await waitFor(() => client.connected, "the client reconnected");
+    await client.flush(500);
+  } finally {
+    peerSocket.close();
+    client.close();
+  }
+});
+
+test("a flush started after the server accepts a new Connect waits on the new session", async () => {
+  const serverSocket = await bindUdpSocket();
+  const peerSocket = await bindUdpSocket();
+  const connectionId = 0x1020305c;
+  const server = new CultNetRudpSocketTransportConnection({
+    runtimeId: "rudp-server",
+    socket: serverSocket,
+    mode: "server",
+    connectionId,
+    resendDelayMs: 1000,
+    resendPollMs: 1000,
+  });
+  const received: CultNetRudpPacket[] = [];
+  peerSocket.on("message", (wire) => received.push(decodeRudpPacket(wire)));
+  const toServer = (packet: CultNetRudpPacket) =>
+    peerSocket.send(encodeRudpPacket(packet), udpPort(serverSocket), "127.0.0.1");
+  try {
+    const first = new CultNetRudpSession({ connectionId });
+    toServer(first.createConnect(0));
+    await waitFor(() => received.some((p) => p.packetType === "accept"), "the first Accept");
+    toServer(first.createDisconnect(Buffer.from("bye")));
+    await waitFor(() => !server.connected, "the first session ended");
+
+    const next = new CultNetRudpSession({ connectionId });
+    toServer(next.createConnect(0));
+    await waitFor(() => received.filter((p) => p.packetType === "accept").length === 2, "the second Accept");
+    const accept = received.filter((p) => p.packetType === "accept")[1]!;
+    next.receive(accept, 0);
+    toServer(next.createAckForReceived(accept.sequence));
+    await waitFor(() => server.outstandingReliablePacketCount === 0, "the second Accept acknowledged");
+
+    await server.flush(500);
+  } finally {
+    peerSocket.close();
+    server.close();
+  }
+});

packages/cultnet-ts/test/rudp-session-ending.test.ts

@@ -13,7 +13,10 @@ function clientWithALostWrite(): { client: CultNetRudpSession; server: CultNetRu
   const server = new CultNetRudpSession({ connectionId, initialSequence: 500 });
   client.receive(server.acceptConnect(client.createConnect(0), 0), 0);
   client.send("schema", enc("owed to the old session"), { reliable: true, ordered: true });
-  assert.equal(client.outstandingReliablePacketCount, 1);
+  // A fragmented write larger than the send window leaves part of it queued.
+  client.sendMany("schema", new Uint8Array(40 * 8), { reliable: true, ordered: true, maxFragmentBytes: 8 });
+  assert.equal(client.outstandingReliablePacketCount, 41);
+  assert.ok(client.queuedReliablePacketCount > 0);
   return { client, server };
 }
 

tests/GameCult.Networking.Tests/CultNetRudpSessionEndingTests.cs

@@ -33,7 +33,10 @@ namespace GameCult.Networking.Tests
             var server = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = ConnectionId, InitialSequence = 500 });
             client.Receive(server.AcceptConnect(client.CreateConnect(0), 0), 0);
             client.Send("schema", Encoding.UTF8.GetBytes("owed to the old session"), new CultNetRudpSendOptions { Reliable = true, Ordered = true });
-            Assert.That(client.OutstandingReliablePacketCount, Is.EqualTo(1));
+            // A fragmented write larger than the send window leaves part of it queued.
+            client.SendMany("schema", new byte[40 * 8], new CultNetRudpSendOptions { Reliable = true, Ordered = true }, maxFragmentBytes: 8);
+            Assert.That(client.OutstandingReliablePacketCount, Is.EqualTo(41));
+            Assert.That(client.QueuedReliablePacketCount, Is.GreaterThan(0));
             return (client, server);
         }
 
@@ -65,7 +68,7 @@ namespace GameCult.Networking.Tests
             }
         }
 
-        private static (CultNetRudpSocketTransportConnection Client, Socket Peer, CultNetRudpSession PeerSession) ConnectedClient()
+        private static (CultNetRudpSocketTransportConnection Client, Socket Peer, EndPoint ClientEndPoint) ConnectedClient()
         {
             var peerSocket = Bind();
             var client = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions
@@ -85,7 +88,7 @@ namespace GameCult.Networking.Tests
             peerSocket.SendTo(CultNetRudpPacketCodec.Encode(accept), from);
             client.ReceiveOnce();
             Assert.That(client.Connected, Is.True);
-            return (client, peerSocket, peer);
+            return (client, peerSocket, from);
         }
 
         [Test]
@@ -128,5 +131,66 @@ namespace GameCult.Networking.Tests
                 Assert.That(error.InnerException, Is.InstanceOf<InvalidOperationException>());
             }
         }
+
+        private static List<CultNetRudpPacket> Drain(Socket socket)
+        {
+            var packets = new List<CultNetRudpPacket>();
+            var buffer = new byte[65535];
+            EndPoint from = new IPEndPoint(IPAddress.Any, 0);
+            while (socket.Poll(20_000, SelectMode.SelectRead))
+            {
+                var received = socket.ReceiveFrom(buffer, ref from);
+                packets.Add(CultNetRudpPacketCodec.Decode(buffer.AsSpan(0, received).ToArray()));
+            }
+            return packets;
+        }
+
+        [Test]
+        public void AFlushStartedAfterAReconnectWaitsOnTheNewSession()
+        {
+            var (client, peerSocket, clientEndPoint) = ConnectedClient();
+            using (client)
+            using (peerSocket)
+            {
+                client.Send("schema", Encoding.UTF8.GetBytes("owed to the old session"));
+                client.Disconnect();
+                client.Connect("hello again");
+                var connect = Drain(peerSocket).Last(packet => packet.PacketType == CultNetRudpPacketType.Connect);
+                var accept = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = ConnectionId, InitialSequence = 700 }).AcceptConnect(connect, 0);
+                peerSocket.SendTo(CultNetRudpPacketCodec.Encode(accept), clientEndPoint);
+                client.ReceiveOnce();
+
+                Assert.DoesNotThrow(() => client.FlushReliable(TimeSpan.FromMilliseconds(500)));
+            }
+        }
+
+        [Test]
+        public void AFlushStartedAfterTheServerAcceptsANewConnectWaitsOnTheNewSession()
+        {
+            using var serverSocket = Bind();
+            using var peerSocket = Bind();
+            using var server = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions
+            {
+                RuntimeId = "csharp-rudp-server",
+                Socket = serverSocket,
+                Mode = CultNetRudpSocketMode.Server,
+                ConnectionId = ConnectionId
+            });
+            var first = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = ConnectionId });
+            peerSocket.SendTo(CultNetRudpPacketCodec.Encode(first.CreateConnect(0)), serverSocket.LocalEndPoint!);
+            server.ReceiveOnce();
+            peerSocket.SendTo(CultNetRudpPacketCodec.Encode(first.CreateDisconnect()), serverSocket.LocalEndPoint!);
+            server.ReceiveOnce();
+
+            var next = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = ConnectionId });
+            peerSocket.SendTo(CultNetRudpPacketCodec.Encode(next.CreateConnect(0)), serverSocket.LocalEndPoint!);
+            server.ReceiveOnce();
+            var accept = Drain(peerSocket).Last(packet => packet.PacketType == CultNetRudpPacketType.Accept);
+            next.Receive(accept, 0);
+            peerSocket.SendTo(CultNetRudpPacketCodec.Encode(next.CreateAck()), serverSocket.LocalEndPoint!);
+            server.ReceiveOnce();
+
+            Assert.DoesNotThrow(() => server.FlushReliable(TimeSpan.FromMilliseconds(500)));
+        }
     }
 }

tests/GameCult.Networking.Tests/NetworkingTests.cs

@@ -1269,6 +1269,16 @@ namespace GameCult.Networking.Tests
             Assert.That(Encoding.UTF8.GetString(ReceiveRudpFrame(client).Payload), Is.EqualTo("server still here"));
         }
 
+        /// <summary>Whether the packet's ack field and mask cover the sequence.</summary>
+        private static bool Acknowledges(CultNetRudpPacket packet, uint sequence)
+        {
+            if (packet.Ack == sequence)
+                return true;
+            return packet.Ack > sequence
+                && packet.Ack - sequence <= 32
+                && (packet.AckMask & (1u << (int)(packet.Ack - sequence - 1))) != 0;
+        }
+
         private static List<CultNetRudpPacket> DrainPackets(Socket socket)
         {
             var packets = new List<CultNetRudpPacket>();
@@ -1397,7 +1407,7 @@ namespace GameCult.Networking.Tests
             var goodbye = DrainPackets(peerSocket).Single();
             Assert.That(goodbye.PacketType, Is.EqualTo(CultNetRudpPacketType.Disconnect));
             Assert.That(peer.Receive(goodbye, 1).Disconnected, Is.True);
-            Assert.That(peer.PendingReliableSequences, Does.Contain(poison.Sequence));
+            Assert.That(Acknowledges(goodbye, poison.Sequence), Is.False, "the goodbye acknowledged the refused frame");
 
             // A retransmit is dropped, not acknowledged.
             peerSocket.SendTo(CultNetRudpPacketCodec.Encode(poison), serverEndPoint);
@@ -1444,7 +1454,7 @@ namespace GameCult.Networking.Tests
             var goodbye = DrainPackets(peerSocket).Single();
             Assert.That(goodbye.PacketType, Is.EqualTo(CultNetRudpPacketType.Disconnect));
             Assert.That(peer.Receive(goodbye, 1).Disconnected, Is.True);
-            Assert.That(peer.PendingReliableSequences, Does.Contain(poison.Sequence));
+            Assert.That(Acknowledges(goodbye, poison.Sequence), Is.False, "the goodbye acknowledged the refused frame");
 
             peerSocket.SendTo(CultNetRudpPacketCodec.Encode(poison), serverEndPoint);
             Assert.That(server.ReceiveOnce(), Is.Null);