diff --git a/src/GameCult.Networking/CultNetTransport.cs b/src/GameCult.Networking/CultNetTransport.cs index 03543ffe..307f7aa8 100644 --- a/src/GameCult.Networking/CultNetTransport.cs +++ b/src/GameCult.Networking/CultNetTransport.cs @@ -1025,6 +1025,36 @@ namespace GameCult.Networking return response; } + /// + /// Answers a connect from a peer this session has already accepted. The packet's sequence is + /// remembered and nothing is queued, so a connect storm cannot grow the reliable queue. The reply + /// is the accept still awaiting acknowledgement, or an ack once it was acknowledged. + /// + public CultNetRudpPacket AnswerRepeatedConnect(CultNetRudpPacket packet, long nowMs = 0) + { + RequireConnection(packet); + if (packet.PacketType != CultNetRudpPacketType.Connect) + { + throw new InvalidOperationException($"Expected RUDP connect packet, got {packet.PacketType}."); + } + + ApplyAcknowledgements(packet); + RememberReceived(packet.Sequence); + _lastReceivedAtMs = nowMs; + lock (_pendingReliableGate) + { + foreach (var pending in _pendingReliable.Values) + { + if (pending.Packet.PacketType != CultNetRudpPacketType.Accept) + continue; + pending.LastSentAtMs = nowMs; + return ClonePacket(pending.Packet); + } + } + + return CreateAck(); + } + /// /// Creates a data packet. /// @@ -1715,6 +1745,7 @@ namespace GameCult.Networking public sealed class CultNetRudpSocketTransportConnection : IDisposable { private const int MinimumReceiveBufferBytes = 4 * 1024 * 1024; + private static readonly byte[] ReplacedPeerReason = Encoding.UTF8.GetBytes("replaced by a new Connect generation"); // One acknowledgement carries the preceding 32-packet receive mask. // Pace at that transport window instead of sleeping after every fragment; // a one-millisecond sleep can consume a full scheduler quantum on Windows. @@ -1810,6 +1841,7 @@ namespace GameCult.Networking throw new InvalidOperationException("Only a client RUDP socket transport can initiate connect."); } + DisconnectReason = null; lock (_sessionGate) SendPacket(_session.CreateConnect(NowMs(), payload ?? Array.Empty())); } @@ -1890,6 +1922,12 @@ namespace GameCult.Networking { while (true) { + // An ended session forgot its unacknowledged writes; reporting them + // flushed would be a lie. + var ended = DisconnectReason; + if (ended != null) + throw new InvalidOperationException( + "RUDP session ended before its reliable writes were acknowledged: " + Encoding.UTF8.GetString(ended)); lock (_sessionGate) { if (_session.OutstandingReliablePacketCount == 0) @@ -2186,7 +2224,17 @@ namespace GameCult.Networking CultNetRudpPacket accept; try { - accept = _session.AcceptConnect(packet, NowMs()); + // A connect from the peer this transport already accepted repeats: answer it + // with the accept already owed and queue nothing. + if (_session.Connected) + { + accept = _session.AnswerRepeatedConnect(packet, NowMs()); + } + else + { + accept = _session.AcceptConnect(packet, NowMs()); + DisconnectReason = null; + } } catch (InvalidOperationException) { @@ -2390,12 +2438,14 @@ namespace GameCult.Networking /// public sealed class CultNetRudpSocketServerPeer { - internal CultNetRudpSocketServerPeer(EndPoint remoteEndPoint, CultNetRudpSession session) + internal CultNetRudpSocketServerPeer(EndPoint remoteEndPoint, byte[] connectPayload, CultNetRudpSession session) { RemoteEndPoint = remoteEndPoint; + ConnectPayload = connectPayload; Session = session; } + internal byte[] ConnectPayload { get; } internal CultNetRudpSession Session { get; } internal object SessionGate { get; } = new object(); @@ -2616,8 +2666,32 @@ namespace GameCult.Networking var peerKey = RemoteKey(remote); if (packet.PacketType == CultNetRudpPacketType.Connect) { + var connectPayload = packet.Payload ?? Array.Empty(); + _peers.TryGetValue(peerKey, out var admitted); + // The same payload from an admitted peer is a repeated Connect: answer it with the + // Accept already owed and queue nothing. A different payload is a new generation. + if (admitted != null && admitted.ConnectPayload.AsSpan().SequenceEqual(connectPayload)) + { + lock (admitted.SessionGate) + { + CultNetRudpPacket reply; + try + { + reply = admitted.Session.AnswerRepeatedConnect(packet, NowMs()); + } + catch (InvalidOperationException) + { + _stats.PacketsDropped++; + return true; + } + SendPacket(admitted.RemoteEndPoint, reply); + } + return true; + } + var peer = new CultNetRudpSocketServerPeer( CloneEndPoint(remote), + connectPayload, new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = _connectionId, @@ -2628,6 +2702,11 @@ namespace GameCult.Networking _peers[peerKey] = peer; lock (peer.SessionGate) SendPacket(peer.RemoteEndPoint, peer.Session.AcceptConnect(packet, NowMs(), _acceptPayload)); + if (admitted != null) + { + admitted.DisconnectReason = ReplacedPeerReason; + PeerDisconnected?.Invoke(admitted); + } return true; } diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index ef4d5a3d..ce39ea30 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -1438,6 +1438,91 @@ namespace GameCult.Networking.Tests Assert.That(DrainPackets(peerSocket), Is.Empty); } + [Test] + public void RudpFlushReliable_FailsForAWriteTheEndedSessionForgot() + { + using var serverSocket = BindUdpSocket(); + using var peerSocket = BindUdpSocket(); + var serverEndPoint = serverSocket.LocalEndPoint!; + const uint connectionId = 0x10203058; + using var server = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions + { + RuntimeId = "csharp-rudp-server", + Socket = serverSocket, + Mode = CultNetRudpSocketMode.Server, + ConnectionId = connectionId + }); + var peer = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = connectionId }); + + peerSocket.SendTo(CultNetRudpPacketCodec.Encode(peer.CreateConnect(0)), serverEndPoint); + server.ReceiveOnce(); + peer.Receive(DrainPackets(peerSocket).Single(p => p.PacketType == CultNetRudpPacketType.Accept), 0); + server.Send("schema", Encoding.UTF8.GetBytes("never acknowledged")); + + peerSocket.SendTo(CultNetRudpPacketCodec.Encode(PoisonedFrame(peer)), serverEndPoint); + Assert.That(server.ReceiveOnce(), Is.Null); + Assert.That(server.DisconnectReason, Is.Not.Null); + Assert.Throws(() => server.FlushReliable(TimeSpan.FromMilliseconds(200))); + } + + [Test] + public void RudpServerMode_OwesOneReliableAcceptHoweverManyConnectsRepeat() + { + using var serverSocket = BindUdpSocket(); + using var peerSocket = BindUdpSocket(); + const uint connectionId = 0x10203059; + using var server = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions + { + RuntimeId = "csharp-rudp-server", + Socket = serverSocket, + Mode = CultNetRudpSocketMode.Server, + ConnectionId = connectionId + }); + var connect = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = connectionId }).CreateConnect(0); + for (var attempt = 0; attempt < 20; attempt++) + { + peerSocket.SendTo(CultNetRudpPacketCodec.Encode(connect), serverSocket.LocalEndPoint!); + server.ReceiveOnce(); + } + + var accepts = DrainPackets(peerSocket).Where(p => p.PacketType == CultNetRudpPacketType.Accept && p.Reliable); + Assert.That(accepts.Select(p => p.Sequence).Distinct().Count(), Is.EqualTo(1)); + } + + [Test] + public void RudpListener_OwesOneReliableAcceptPerGenerationAndReportsAReplacedPeer() + { + using var serverSocket = BindUdpSocket(); + using var peerSocket = BindUdpSocket(); + const uint connectionId = 0x1020305a; + using var server = new CultNetRudpSocketTransportServer(new CultNetRudpSocketTransportServerOptions + { + RuntimeId = "csharp-rudp-listener", + Socket = serverSocket, + ConnectionId = connectionId + }); + var ended = new List(); + server.PeerDisconnected += ended.Add; + var connect = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = connectionId }).CreateConnect(0, Encoding.UTF8.GetBytes("join")); + for (var attempt = 0; attempt < 20; attempt++) + { + peerSocket.SendTo(CultNetRudpPacketCodec.Encode(connect), serverSocket.LocalEndPoint!); + server.ReceiveOnce(); + } + + var accepts = DrainPackets(peerSocket).Where(p => p.PacketType == CultNetRudpPacketType.Accept && p.Reliable); + Assert.That(accepts.Select(p => p.Sequence).Distinct().Count(), Is.EqualTo(1)); + Assert.That(server.Peers, Has.Count.EqualTo(1)); + Assert.That(ended, Is.Empty); + + // A different payload from the same endpoint is a new generation, and the old one is reported. + var other = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = connectionId }).CreateConnect(0, Encoding.UTF8.GetBytes("rejoin")); + peerSocket.SendTo(CultNetRudpPacketCodec.Encode(other), serverSocket.LocalEndPoint!); + server.ReceiveOnce(); + Assert.That(server.Peers, Has.Count.EqualTo(1)); + Assert.That(ended, Has.Count.EqualTo(1)); + } + [Test] public void RudpConfiguration_RejectsLimitsThatCannotAdmitAConnect() {