diff --git a/src/GameCult.Networking/CultNetTransport.cs b/src/GameCult.Networking/CultNetTransport.cs index 83006d2f..f646a7a2 100644 --- a/src/GameCult.Networking/CultNetTransport.cs +++ b/src/GameCult.Networking/CultNetTransport.cs @@ -1459,6 +1459,53 @@ namespace GameCult.Networking return CreateDisconnect(RefusedPacketReason); } + /// + /// Whether a failed send can never succeed for the datagram as built, so resending it is + /// pointless: it is too large for a datagram (EMSGSIZE, WSAEMSGSIZE), its address is + /// malformed for the socket (EINVAL, WSAEINVAL), or belongs to another address family + /// (EAFNOSUPPORT, WSAEAFNOSUPPORT). Everything else (no route, full buffers, firewall drops, + /// a refusal reported for an earlier datagram, an interrupted call) is a property of the path + /// or the moment and may pass, so it is a lost datagram, not a reason to end a session. + /// + internal static bool IsPermanentSendError(SocketException error) => + error.SocketErrorCode == SocketError.MessageSize || + error.SocketErrorCode == SocketError.InvalidArgument || + error.SocketErrorCode == SocketError.AddressFamilyNotSupported; + + internal const string UnsendablePacketReason = "packet could not be sent"; + + /// + /// Ends a session that owes its peer a packet that can never be sent as built. Resending it + /// would fail forever, and dropping it would leave a reliable sequence the peer waits on for + /// good, so the session cannot be kept. The goodbye's reason names the error. + /// + internal CultNetRudpPacket EndUnsendable(SocketException error) + { + ResetPeerState(); + return CreateDisconnect(Encoding.UTF8.GetBytes(UnsendablePacketReason + ": " + error.Message)); + } + + internal uint NextSequence => _nextSequence; + + /// + /// Undoes a send that never reached the wire: every packet it created at or after + /// , pending or queued, leaves no trace and its sequences are + /// reissued, so the peer never sees a gap. Only valid while nothing else has been sent since. + /// + internal void WithdrawSend(uint firstSequence) + { + lock (_pendingReliableGate) + { + foreach (var sequence in _pendingReliable.Keys.Where(sequence => sequence >= firstSequence).ToArray()) + _pendingReliable.Remove(sequence); + var kept = _queuedReliable.Where(packet => packet.Sequence < firstSequence).ToArray(); + _queuedReliable.Clear(); + foreach (var packet in kept) + _queuedReliable.Enqueue(packet); + } + _nextSequence = firstSequence; + } + /// /// Creates a packet carrying a transport-level disconnect reason. /// @@ -2733,8 +2780,25 @@ namespace GameCult.Networking public void Send(CultNetRudpSocketServerPeer peer, string channelId, byte[] payload) { if (peer == null) throw new ArgumentNullException(nameof(peer)); + SocketException? unsendable = null; + var failedIndex = 0; lock (peer.SessionGate) - SendPackets(peer, peer.Session.SendMany(channelId, payload, ChannelSendOptions(channelId), _maxFragmentBytes)); + { + var firstSequence = peer.Session.NextSequence; + var packets = peer.Session.SendMany(channelId, payload, ChannelSendOptions(channelId), _maxFragmentBytes); + unsendable = SendPackets(peer, packets, out failedIndex); + // A caller-directed send that can never succeed is the caller's error and queues + // nothing. Only when earlier fragments are already on the wire has the peer seen + // part of it, and then the session cannot be kept. + if (unsendable != null && failedIndex == 0) + peer.Session.WithdrawSend(firstSequence); + } + if (unsendable != null) + { + if (failedIndex > 0) + EndUnsendablePeer(peer, unsendable); + throw unsendable; + } _stats.FramesSent++; } @@ -2844,10 +2908,13 @@ namespace GameCult.Networking // a new generation. if (admitted != null) { + var repeats = false; + SocketException? unsendableReply = null; lock (admitted.SessionGate) { if (admitted.Session.ConnectRepeats(packet)) { + repeats = true; CultNetRudpPacket reply; try { @@ -2858,10 +2925,15 @@ namespace GameCult.Networking _stats.PacketsDropped++; return true; } - SendPacket(admitted.RemoteEndPoint, reply); - return true; + unsendableReply = SendPacket(admitted.RemoteEndPoint, reply); } } + if (repeats) + { + if (unsendableReply != null) + EndUnsendablePeer(admitted, unsendableReply); + return true; + } } var peer = new CultNetRudpSocketServerPeer( @@ -2874,9 +2946,16 @@ namespace GameCult.Networking ResendDelayMs = _resendDelayMs, MaxPendingReliablePackets = _maxPendingReliablePackets })); - _peers[peerKey] = peer; + SocketException? unsendableAccept; lock (peer.SessionGate) - SendPacket(peer.RemoteEndPoint, peer.Session.AcceptConnect(packet, NowMs(), _acceptPayload)); + unsendableAccept = SendPacket(peer.RemoteEndPoint, peer.Session.AcceptConnect(packet, NowMs(), _acceptPayload)); + if (unsendableAccept != null) + { + // The peer cannot be answered, so no session starts. + _stats.PacketsDropped++; + return true; + } + _peers[peerKey] = peer; if (admitted != null) { admitted.DisconnectReason = ReplacedPeerReason; @@ -2893,6 +2972,7 @@ namespace GameCult.Networking CultNetRudpReceiveResult? outcome; CultNetRudpPacket? acknowledgement = null; + SocketException? unsendable = null; lock (existingPeer.SessionGate) { try @@ -2915,8 +2995,8 @@ namespace GameCult.Networking else { if (outcome.Reply != null) - SendPacket(existingPeer.RemoteEndPoint, outcome.Reply); - SendPackets(existingPeer, outcome.ReadyToSend); + unsendable = SendPacket(existingPeer.RemoteEndPoint, outcome.Reply); + unsendable ??= SendPackets(existingPeer, outcome.ReadyToSend, out _); if (packet.PacketType == CultNetRudpPacketType.Data) acknowledgement = existingPeer.Session.CreateAckForReceived(packet.Sequence); } @@ -2953,8 +3033,10 @@ namespace GameCult.Networking } delivered = _deliveredFrames.Count > 0 ? _deliveredFrames.Dequeue() : null; - if (acknowledgement != null) - SendPacket(existingPeer.RemoteEndPoint, acknowledgement); + if (acknowledgement != null && unsendable == null) + unsendable = SendPacket(existingPeer.RemoteEndPoint, acknowledgement); + if (unsendable != null) + EndUnsendablePeer(existingPeer, unsendable); return true; } @@ -2968,19 +3050,48 @@ namespace GameCult.Networking return; foreach (var peer in _peers.Values.ToArray()) { + SocketException? unsendable; lock (peer.SessionGate) - SendPackets(peer, peer.Session.DueResends(NowMs())); + unsendable = SendPackets(peer, peer.Session.DueResends(NowMs()), out _); + if (unsendable != null) + EndUnsendablePeer(peer, unsendable); } } - private void SendPackets(CultNetRudpSocketServerPeer peer, IReadOnlyList packets) + // Stops at the first packet that can never be sent as built and returns its error and + // index; a transient failure is a lost datagram and the loop goes on. + private SocketException? SendPackets(CultNetRudpSocketServerPeer peer, IReadOnlyList packets, out int failedIndex) { + failedIndex = 0; for (var index = 0; index < packets.Count; index++) { - SendPacket(peer.RemoteEndPoint, packets[index]); + var unsendable = SendPacket(peer.RemoteEndPoint, packets[index]); + if (unsendable != null) + { + failedIndex = index; + return unsendable; + } if ((index + 1) % WireBurstPackets == 0 && index + 1 < packets.Count) Thread.Yield(); } + return null; + } + + // An admitted peer owes a packet that can never be sent as built. Inside a poll that ends the + // peer's session, never the poll: the peer is told, and PeerDisconnected carries a reason that + // names the error. + private void EndUnsendablePeer(CultNetRudpSocketServerPeer peer, SocketException error) + { + var key = RemoteKey(peer.RemoteEndPoint); + if (!_peers.TryGetValue(key, out var current) || !ReferenceEquals(current, peer)) + return; + _peers.Remove(key); + CultNetRudpPacket goodbye; + lock (peer.SessionGate) + goodbye = peer.Session.EndUnsendable(error); + peer.DisconnectReason = goodbye.Payload; + SendPacket(peer.RemoteEndPoint, goodbye); + PeerDisconnected?.Invoke(peer); } /// @@ -2995,30 +3106,51 @@ namespace GameCult.Networking _socket.Dispose(); } - // A failed send is that peer's lost datagram: counted, never thrown. A reliable packet - // stays pending and is resent; an unreliable one was allowed to be lost. The peer's - // session ends by the rules that already end sessions. Only an encode failure, the - // listener's own bug, throws. - private void SendPacket(EndPoint remoteEndPoint, CultNetRudpPacket packet) + // A transient send failure is that peer's lost datagram: counted, never thrown. A reliable + // packet stays pending and is resent; an unreliable one was allowed to be lost. A permanent + // failure (CultNetRudpSession.IsPermanentSendError) is returned: the datagram can never be + // sent as built, and the caller decides between refusing the send and ending the peer's + // session. Only an encode failure, the listener's own bug, throws. + private SocketException? SendPacket(EndPoint remoteEndPoint, CultNetRudpPacket packet) { 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) + { + UnsendableAfter[remoteEndPoint] = remaining - 1; + } + else + { + UnsendableAfter.Remove(remoteEndPoint); + throw new SocketException((int)SocketError.MessageSize); + } + } var sent = _socket.SendTo(wire, remoteEndPoint); _stats.BytesSent += sent; } + catch (SocketException error) when (CultNetRudpSession.IsPermanentSendError(error)) + { + return error; + } catch (SocketException) { _stats.SendFailures++; } + return null; } - // Test seam: peers whose every datagram fails to send, standing in for an unroutable or - // full path that a loopback peer cannot be made to have. + // Test seams, standing in for paths a loopback peer cannot be made to have. Peers whose every + // datagram fails to send transiently: internal HashSet FailingSendPeers { get; } = new HashSet(); + // Peers whose next datagram after this many fails permanently, once. + internal Dictionary UnsendableAfter { get; } = new Dictionary(); + 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 359b842f..ba6cba1f 100644 --- a/tests/GameCult.Networking.Tests/CultNetRudpSendFailureTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetRudpSendFailureTests.cs @@ -60,7 +60,7 @@ namespace GameCult.Networking.Tests return peer; } - private static (CultNetRudpSocketTransportServer Server, EndPoint ServerEndPoint, Peer X, Peer Y) TwoPeers() + private static (CultNetRudpSocketTransportServer Server, EndPoint ServerEndPoint, Peer X, Peer Y) TwoPeers(int? maxFragmentBytes = null) { var serverSocket = Bind(); var server = new CultNetRudpSocketTransportServer(new CultNetRudpSocketTransportServerOptions @@ -68,7 +68,8 @@ namespace GameCult.Networking.Tests RuntimeId = "csharp-send-failure", Socket = serverSocket, ConnectionId = ConnectionId, - ResendDelayMs = 5 + ResendDelayMs = 5, + MaxFragmentBytes = maxFragmentBytes }); var x = Connect(server, serverSocket.LocalEndPoint!); var y = Connect(server, serverSocket.LocalEndPoint!); @@ -130,5 +131,196 @@ namespace GameCult.Networking.Tests Assert.That(server.Peers, Has.Count.EqualTo(2)); } } + + // A permanent send failure is one that can never succeed for the datagram as built. A + // caller-directed send throws it and queues nothing; inside a poll it ends that peer's + // session and never the poll. + + private static bool IsUnsendableReason(byte[]? reason) + { + var text = Encoding.UTF8.GetString(reason ?? Array.Empty()); + return text.StartsWith("packet could not be sent: ", StringComparison.Ordinal) + && text.Length > "packet could not be sent: ".Length; + } + + private static List RecordDisconnects(CultNetRudpSocketTransportServer server) + { + var ended = new List(); + server.PeerDisconnected += ended.Add; + return ended; + } + + [Test] + public void AnOversizedSendThrowsTheErrorAndQueuesNothing() + { + var (server, _, x, y) = TwoPeers(); + using (server) + { + var error = Assert.Throws(() => server.SendSchema(x.Server, new byte[70_000]))!; + Assert.That(CultNetRudpSession.IsPermanentSendError(error), Is.True, error.SocketErrorCode.ToString()); + Assert.That(server.Stats.SendFailures, Is.Zero); + Assert.That(server.Stats.FramesSent, Is.Zero); + Assert.That(server.Peers, Has.Count.EqualTo(2), "the caller's error ends no session"); + Assert.That(x.Server.Session.OutstandingReliablePacketCount, Is.Zero, "nothing is pending or queued to resend"); + + // The refused send consumed no sequence: the next frame is delivered in order instead + // of waiting behind a gap that never fills. + server.SendSchema(x.Server, "next"); + var data = Drain(x.Socket).Single(p => p.PacketType == CultNetRudpPacketType.Data); + var delivered = x.Session.Receive(data, 1).Delivered; + Assert.That(delivered.Select(frame => Encoding.UTF8.GetString(frame.Payload)), Is.EqualTo(new[] { "next" })); + Assert.That(y.Server.Session.OutstandingReliablePacketCount, Is.Zero); + } + } + + [Test] + public void ASendThatFailsPermanentlyAfterAFragmentLeftEndsTheSession() + { + var (server, _, x, _) = TwoPeers(maxFragmentBytes: 1000); + using (server) + { + var ended = RecordDisconnects(server); + server.UnsendableAfter[x.EndPoint] = 1; + + var error = Assert.Throws(() => server.SendSchema(x.Server, new byte[3000]))!; + Assert.That(CultNetRudpSession.IsPermanentSendError(error), Is.True); + Assert.That(server.Peers, Has.Count.EqualTo(1)); + Assert.That(ended, Is.EqualTo(new[] { x.Server })); + Assert.That(IsUnsendableReason(x.Server.DisconnectReason), Is.True); + } + } + + [Test] + public void APermanentFailureInResendsEndsOnlyThatPeersSession() + { + var (server, _, x, y) = TwoPeers(); + using (server) + { + var ended = RecordDisconnects(server); + server.FailingSendPeers.Add(x.EndPoint); + server.FailingSendPeers.Add(y.EndPoint); + server.SendSchema(x.Server, "for-x"); + server.SendSchema(y.Server, "for-y"); + + // X's datagram can now never be sent as built; Y's path recovers. X is visited first. + server.FailingSendPeers.Clear(); + server.UnsendableAfter[x.EndPoint] = 0; + Thread.Sleep(30); + Assert.DoesNotThrow(() => server.PollResends()); + + Assert.That(server.Peers, Is.EqualTo(new[] { y.Server })); + Assert.That(ended, Is.EqualTo(new[] { x.Server })); + Assert.That(IsUnsendableReason(x.Server.DisconnectReason), Is.True); + var resent = Drain(y.Socket).Where(p => p.PacketType == CultNetRudpPacketType.Data).ToList(); + Assert.That(resent, Has.Count.EqualTo(1)); + Assert.That(Encoding.UTF8.GetString(resent[0].Payload!), Is.EqualTo("for-y")); + var goodbye = Drain(x.Socket).Single(p => p.PacketType == CultNetRudpPacketType.Disconnect); + Assert.That(IsUnsendableReason(goodbye.Payload), Is.True); + } + } + + [Test] + public void APermanentFailureAcknowledgingAFrameEndsTheSessionAfterTheFrameIsDelivered() + { + var (server, listenerEndPoint, x, _) = TwoPeers(); + using (server) + { + var ended = RecordDisconnects(server); + server.UnsendableAfter[x.EndPoint] = 0; + x.Socket.SendTo(CultNetRudpPacketCodec.Encode(ReliableFrame(x, "from-x")), listenerEndPoint); + + CultNetRudpSocketServerFrame? frame = null; + Assert.DoesNotThrow(() => frame = server.ReceiveOnce()); + + Assert.That(frame, Is.Not.Null, "the frame is delivered before the session ends"); + Assert.That(Encoding.UTF8.GetString(frame!.Frame.Payload), Is.EqualTo("from-x")); + Assert.That(ended, Is.EqualTo(new[] { x.Server })); + Assert.That(IsUnsendableReason(x.Server.DisconnectReason), Is.True); + Assert.That(server.Peers, Has.Count.EqualTo(1)); + Assert.That(server.Stats.SendFailures, Is.Zero); + } + } + + [Test] + public void APermanentFailureAnsweringARepeatedConnectEndsTheSession() + { + var serverSocket = Bind(); + using var server = new CultNetRudpSocketTransportServer(new CultNetRudpSocketTransportServerOptions + { + RuntimeId = "csharp-send-failure", + Socket = serverSocket, + ConnectionId = ConnectionId, + ResendDelayMs = 5 + }); + var ended = RecordDisconnects(server); + var peer = Bind(); + var connect = CultNetRudpPacketCodec.Encode( + new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = ConnectionId }).CreateConnect(0)); + peer.SendTo(connect, serverSocket.LocalEndPoint!); + server.ReceiveOnce(); + Assert.That(server.Peers, Has.Count.EqualTo(1)); + var admitted = server.Peers.Single(); + + server.UnsendableAfter[peer.LocalEndPoint!] = 0; + peer.SendTo(connect, serverSocket.LocalEndPoint!); + Assert.DoesNotThrow(() => server.ReceiveOnce()); + + Assert.That(server.Peers, Is.Empty); + Assert.That(ended, Is.EqualTo(new[] { admitted })); + Assert.That(IsUnsendableReason(admitted.DisconnectReason), Is.True); + } + + [Test] + public void AConnectWhoseAcceptCanNeverBeSentStartsNoSession() + { + var serverSocket = Bind(); + using var server = new CultNetRudpSocketTransportServer(new CultNetRudpSocketTransportServerOptions + { + RuntimeId = "csharp-send-failure", + Socket = serverSocket, + ConnectionId = ConnectionId, + ResendDelayMs = 5 + }); + var peer = Bind(); + server.UnsendableAfter[peer.LocalEndPoint!] = 0; + peer.SendTo( + CultNetRudpPacketCodec.Encode(new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = ConnectionId }).CreateConnect(0)), + serverSocket.LocalEndPoint!); + + Assert.DoesNotThrow(() => server.ReceiveOnce()); + + Assert.That(server.Peers, Is.Empty); + Assert.That(server.Stats.PacketsDropped, Is.EqualTo(1)); + } + + [Test] + public void AHealthySendCountsItsBytesAndNoFailure() + { + var (server, _, x, _) = TwoPeers(); + using (server) + { + var before = server.Stats.BytesSent; + server.SendSchema(x.Server, "hello"); + var received = Drain(x.Socket).Sum(p => CultNetRudpPacketCodec.Encode(p).Length); + Assert.That(received, Is.GreaterThan("hello".Length)); + Assert.That(server.Stats.BytesSent - before, Is.EqualTo(received)); + Assert.That(server.Stats.SendFailures, Is.Zero); + } + } + + [Test] + public void OnlyADatagramThatCanNeverBeSentIsAPermanentFailure() + { + foreach (var permanent in new[] { SocketError.MessageSize, SocketError.InvalidArgument, SocketError.AddressFamilyNotSupported }) + Assert.That(CultNetRudpSession.IsPermanentSendError(new SocketException((int)permanent)), Is.True, permanent.ToString()); + foreach (var transient in new[] + { + SocketError.HostUnreachable, SocketError.NetworkUnreachable, SocketError.NoBufferSpaceAvailable, + SocketError.WouldBlock, SocketError.ConnectionReset, SocketError.ConnectionRefused, + SocketError.AccessDenied, SocketError.NetworkDown, SocketError.AddressNotAvailable, + SocketError.Interrupted, SocketError.TimedOut, SocketError.Shutdown + }) + Assert.That(CultNetRudpSession.IsPermanentSendError(new SocketException((int)transient)), Is.False, transient.ToString()); + } } }