diff --git a/packages/cultmesh-py/tests/test_cultmesh.py b/packages/cultmesh-py/tests/test_cultmesh.py index 18243923..22a3f692 100644 --- a/packages/cultmesh-py/tests/test_cultmesh.py +++ b/packages/cultmesh-py/tests/test_cultmesh.py @@ -4651,6 +4651,106 @@ class CultMeshRudpSendFailureTests(unittest.TestCase): finally: server.stop() + def test_a_permanent_failure_in_a_resend_ends_only_that_peers_session(self) -> None: + server = CultMesh.serve_node(CultMesh.create_node(runtime_id="mesh-server")) + original_sendto = socket.socket.sendto + try: + x, x_addr = self._client(server, "peer-x") + y, _ = self._client(server, "peer-y") + mode = ["transient"] + + def sendto(sock: socket.socket, data: bytes, *args: Any) -> int: + if args and tuple(args[-1]) == tuple(x_addr) and mode[0] == "transient": + raise OSError(errno.ENOBUFS, "injected transient failure") + if args and tuple(args[-1]) == tuple(x_addr) and mode[0] == "permanent": + mode[0] = "healthy" + raise OSError(errno.EMSGSIZE, "Message too long") + return original_sendto(sock, data, *args) + + with patch.object(socket.socket, "sendto", sendto): + # X's response is lost, so a resend is pending; then it can never be sent as built. + x.send("schema", hello(runtime_id="probe").to_bytes()) + self._wait_for(lambda: server.rudp_send_failures > 0) + self.assertFalse(x.disconnect_reason, "a transient failure ends no session") + mode[0] = "permanent" + for _ in range(200): + x.receive_once() + if x.disconnect_reason is not None: + break + time.sleep(0.005) + self.assertIsNotNone(x.disconnect_reason) + self.assertTrue(x.disconnect_reason.startswith(b"packet could not be sent: ")) + self.assertTrue(server._rudp_thread.is_alive()) + self.assertIsNotNone(self._served(y)) + finally: + server.stop() + + def test_a_permanent_failure_acknowledging_a_frame_ends_the_session(self) -> None: + import msgpack # type: ignore + + server = CultMesh.serve_node(CultMesh.create_node(runtime_id="mesh-server")) + original_sendto = socket.socket.sendto + try: + x, x_addr = self._client(server, "peer-x") + failures = [1] + + def sendto(sock: socket.socket, data: bytes, *args: Any) -> int: + if args and tuple(args[-1]) == tuple(x_addr) and failures[0] > 0: + failures[0] -= 1 + raise OSError(errno.EMSGSIZE, "Message too long") + return original_sendto(sock, data, *args) + + with patch.object(socket.socket, "sendto", sendto): + # Not a message the server answers, so its ack is the only datagram owed. + x.send("schema", msgpack.packb(1)) + for _ in range(200): + x.receive_once() + if x.disconnect_reason is not None: + break + time.sleep(0.005) + self.assertIsNotNone(x.disconnect_reason) + self.assertTrue(x.disconnect_reason.startswith(b"packet could not be sent: ")) + self.assertTrue(server._rudp_thread.is_alive()) + finally: + server.stop() + + def test_a_connect_whose_accept_can_never_be_sent_starts_no_session(self) -> None: + server = CultMesh.serve_node(CultMesh.create_node(runtime_id="mesh-server")) + original_sendto = socket.socket.sendto + try: + y, _ = self._client(server, "peer-y") + z_socket = bind_udp_socket() + z_addr = z_socket.getsockname()[:2] + failures = [1] + + def sendto(sock: socket.socket, data: bytes, *args: Any) -> int: + if args and tuple(args[-1]) == tuple(z_addr) and failures[0] > 0: + failures[0] -= 1 + raise OSError(errno.EMSGSIZE, "Message too long") + return original_sendto(sock, data, *args) + + with patch.object(socket.socket, "sendto", sendto): + z = CultNetRudpSocketTransportConnection( + CultNetRudpSocketTransportOptions( + runtime_id="peer-z", + socket=z_socket, + mode=CultNetRudpSocketMode.CLIENT, + connection_id=server.rudp_connection_id, + remote_addr=("127.0.0.1", server.port), + resend_delay_ms=25, + ) + ) + z.connect(b"join") + # A session that started would resend its Accept, and the resend would get through. + time.sleep(0.5) + z_socket.settimeout(0.05) + with self.assertRaises(TimeoutError): + z_socket.recvfrom(65535) + self.assertTrue(server._rudp_thread.is_alive()) + self.assertIsNotNone(self._served(y)) + finally: + server.stop() + def _wait_for(self, predicate: Callable[[], bool]) -> None: deadline = time.monotonic() + 2.0 while time.monotonic() < deadline: diff --git a/packages/cultmesh-rs/src/rudp_document_server.rs b/packages/cultmesh-rs/src/rudp_document_server.rs index 518a44eb..c0cd200f 100644 --- a/packages/cultmesh-rs/src/rudp_document_server.rs +++ b/packages/cultmesh-rs/src/rudp_document_server.rs @@ -724,9 +724,6 @@ where fn send_datagram(&mut self, wire: &[u8], remote_addr: SocketAddr) -> std::io::Result { #[cfg(test)] { - if self.failing_peers.contains(&remote_addr) { - return Err(std::io::Error::other("injected send failure")); - } match self.unsendable_after.get(&remote_addr).copied() { Some(0) => { self.unsendable_after.remove(&remote_addr); @@ -737,6 +734,9 @@ where } None => {} } + if self.failing_peers.contains(&remote_addr) { + return Err(std::io::Error::other("injected send failure")); + } } self.socket.send_to(wire, remote_addr) } diff --git a/packages/cultmesh-rs/src/rudp_document_server/send_failure_tests.rs b/packages/cultmesh-rs/src/rudp_document_server/send_failure_tests.rs index 3554ab3c..7320c2bf 100644 --- a/packages/cultmesh-rs/src/rudp_document_server/send_failure_tests.rs +++ b/packages/cultmesh-rs/src/rudp_document_server/send_failure_tests.rs @@ -344,3 +344,25 @@ fn healthy_sends_are_not_failures() { assert_eq!(server.send_failures(), 0); assert_eq!(server.session_count(), 2); } + +#[test] +fn a_session_ended_in_the_resend_loop_is_sent_nothing_more_from_it() { + let (mut server, clock, _received, (mut x, x_addr), (_y, _)) = two_peers(); + server.failing_peers.insert(x_addr); + send(&mut x, &snapshot("one")); + send(&mut x, &snapshot("two")); + for _ in 0..10 { + server.poll_once().unwrap(); + } + let before = server.send_failures(); + + server.unsendable_after.insert(x_addr, 0); + clock.0.store(1_200, Ordering::SeqCst); + let maintenance = server.maintain().unwrap(); + + // Both responses are due. The first can never be sent and ends the session; + // the goodbye is the only other datagram attempted at X. + assert_eq!(maintenance.packets_resent, 2); + assert_eq!(server.send_failures() - before, 1); + assert_eq!(server.session_count(), 1); +} diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index 1873c1fb..a9263d06 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -2032,9 +2032,6 @@ impl CultNetRudpServerHub { fn send_datagram(&mut self, wire: &[u8], remote_addr: SocketAddr) -> std::io::Result { #[cfg(test)] { - if self.failing_peers.contains(&remote_addr) { - return Err(std::io::Error::other("injected send failure")); - } match self.unsendable_after.get(&remote_addr).copied() { Some(0) => { self.unsendable_after.remove(&remote_addr); @@ -2045,6 +2042,9 @@ impl CultNetRudpServerHub { } None => {} } + if self.failing_peers.contains(&remote_addr) { + return Err(std::io::Error::other("injected send failure")); + } } self.socket.send_to(wire, remote_addr) } diff --git a/packages/cultnet-rs/src/rudp/send_failure_tests.rs b/packages/cultnet-rs/src/rudp/send_failure_tests.rs index 1c2cb3f2..49a80f80 100644 --- a/packages/cultnet-rs/src/rudp/send_failure_tests.rs +++ b/packages/cultnet-rs/src/rudp/send_failure_tests.rs @@ -162,6 +162,10 @@ fn one_peer( #[test] fn an_oversized_send_returns_the_error_and_queues_nothing() { let (mut hub, mut peer, session) = one_peer(None); + // The hub's own Accept is still awaiting its ack; the refused send adds nothing to it. + let outstanding = hub.peers[&session.remote_addr] + .session + .outstanding_reliable_packet_count(); let error = hub .send(&session, "schema", vec![7; 70_000]) @@ -176,7 +180,7 @@ fn an_oversized_send_returns_the_error_and_queues_nothing() { let remote = session.remote_addr; assert_eq!( hub.peers[&remote].session.outstanding_reliable_packet_count(), - 0, + outstanding, "nothing is pending or queued to resend" ); @@ -213,7 +217,7 @@ fn a_send_that_fails_permanently_after_a_fragment_left_ends_the_session() { #[test] fn a_permanent_failure_in_resends_ends_only_that_peers_session() { - let (mut hub, (_x, x_session), (mut y, y_session)) = two_peers(); + let (mut hub, (mut x, x_session), (mut y, y_session)) = two_peers(); // Both peers' first datagram is lost, so both have a reliable resend due. hub.failing_peers.insert(x_session.remote_addr); hub.failing_peers.insert(y_session.remote_addr); @@ -240,6 +244,16 @@ fn a_permanent_failure_in_resends_ends_only_that_peers_session() { } } assert_eq!(delivered.expect("Y receives its resend").payload, b"for-y"); + // The goodbye reaches X and names the error. + for _ in 0..20 { + x.receive_once().unwrap(); + if x.disconnect_reason().is_some() { + break; + } + } + assert!(is_unsendable_reason( + x.disconnect_reason().expect("X is told its session ended") + )); } #[test] @@ -369,3 +383,34 @@ fn only_a_datagram_that_can_never_be_sent_is_a_permanent_failure() { } } } + +#[test] +fn a_permanent_failure_sending_a_disconnect_is_returned_to_the_caller() { + let (mut hub, _peer, session) = one_peer(None); + hub.unsendable_after.insert(session.remote_addr, 0); + + let error = hub + .disconnect(&session, b"bye".to_vec()) + .expect_err("the goodbye can never be sent"); + assert!(is_permanent_send_error( + error.downcast_ref::().unwrap() + )); +} + +#[test] +fn a_peer_ended_in_the_resend_loop_is_sent_nothing_more_from_it() { + let (mut hub, (_x, x_session), (_y, _y_session)) = two_peers(); + hub.failing_peers.insert(x_session.remote_addr); + hub.send(&x_session, "schema", b"one".to_vec()).unwrap(); + hub.send(&x_session, "schema", b"two".to_vec()).unwrap(); + assert_eq!(hub.stats().send_failures, 2); + + hub.unsendable_after.insert(x_session.remote_addr, 0); + std::thread::sleep(Duration::from_millis(40)); + hub.poll_resends().unwrap(); + + // Both resends are due. The first can never be sent and ends the session; + // the goodbye is the only other datagram attempted at X. + assert_eq!(hub.stats().send_failures, 3); + assert_eq!(hub.sessions().len(), 1); +} diff --git a/src/GameCult.Networking/CultNetTransport.cs b/src/GameCult.Networking/CultNetTransport.cs index f646a7a2..710227be 100644 --- a/src/GameCult.Networking/CultNetTransport.cs +++ b/src/GameCult.Networking/CultNetTransport.cs @@ -3116,8 +3116,6 @@ namespace GameCult.Networking var wire = CultNetRudpPacketCodec.Encode(packet); try { - if (FailingSendPeers.Contains(remoteEndPoint)) - throw new SocketException((int)SocketError.HostUnreachable); if (UnsendableAfter.TryGetValue(remoteEndPoint, out var remaining)) { if (remaining > 0) @@ -3130,6 +3128,8 @@ namespace GameCult.Networking throw new SocketException((int)SocketError.MessageSize); } } + if (FailingSendPeers.Contains(remoteEndPoint)) + throw new SocketException((int)SocketError.HostUnreachable); var sent = _socket.SendTo(wire, remoteEndPoint); _stats.BytesSent += sent; } diff --git a/tests/GameCult.Networking.Tests/CultNetRudpSendFailureTests.cs b/tests/GameCult.Networking.Tests/CultNetRudpSendFailureTests.cs index ba6cba1f..b43ce5fa 100644 --- a/tests/GameCult.Networking.Tests/CultNetRudpSendFailureTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetRudpSendFailureTests.cs @@ -293,6 +293,28 @@ namespace GameCult.Networking.Tests Assert.That(server.Stats.PacketsDropped, Is.EqualTo(1)); } + [Test] + public void APeerEndedInTheResendLoopIsSentNothingMoreFromIt() + { + var (server, _, x, _) = TwoPeers(); + using (server) + { + server.FailingSendPeers.Add(x.EndPoint); + server.SendSchema(x.Server, "one"); + server.SendSchema(x.Server, "two"); + Assert.That(server.Stats.SendFailures, Is.EqualTo(2)); + + server.UnsendableAfter[x.EndPoint] = 0; + Thread.Sleep(30); + server.PollResends(); + + // Both resends are due. The first can never be sent and ends the session; the + // goodbye is the only other datagram attempted at X. + Assert.That(server.Stats.SendFailures, Is.EqualTo(3)); + Assert.That(server.Peers, Has.Count.EqualTo(1)); + } + } + [Test] public void AHealthySendCountsItsBytesAndNoFailure() {