Merge pull request 'GameCult.Networking: resend the capped frame's answer until acknowledged, then end the session' (#49) from libby/cs-capped-answer-reliable into main

src/GameCult.Networking/CultNetRudpSchemaServer.cs

@@ -126,6 +126,12 @@ namespace GameCult.Networking
         /// </summary>
         public const int MaxConsecutiveRefusedFrames = 8;
 
+        /// <summary>
+        /// How long a session whose capped frame was answered waits for the peer to acknowledge the answer
+        /// before it ends anyway.
+        /// </summary>
+        internal const long CappedAnswerGraceMs = 2000;
+
         /// <summary>
         /// Gets or sets the logger used by the server. R-AM: the dispatch backstop in DispatchAsync
         /// logs every untyped fault through this instead of swallowing it silently.
@@ -310,6 +316,9 @@ namespace GameCult.Networking
         {
             if (!string.Equals(delivered.Frame.ChannelId, "schema", StringComparison.Ordinal))
                 return false;
+            // The capped frame was answered: nothing after it is served while the answer is delivered.
+            if (delivered.Peer.EndingReason != null)
+                return false;
 
             // A peer's message that cannot be decoded (malformed bytes, an unknown schemaVersion) or
             // handled ends that peer's session, never the poll: the drain goes on serving every other
@@ -337,8 +346,10 @@ namespace GameCult.Networking
                     Logger.LogError($"CultNet RUDP schema server could not decode a peer's message ({ex.GetType().Name}); it was refused and that peer's session kept.");
                     return false;
                 }
-                var ended = _transport.EndPeerSession(delivered.Peer, CultNetRudpSession.RefusedPacketReason);
-                Logger.LogError($"CultNet RUDP schema server could not decode a peer's message ({ex.GetType().Name}); {SessionOutcome(ended)}");
+                // The answer was sent reliably: the session ends once the peer has acknowledged it (a lost
+                // datagram is resent first), or after the grace.
+                _transport.EndPeerSessionWhenAnswered(delivered.Peer, CultNetRudpSession.RefusedPacketReason, CappedAnswerGraceMs);
+                Logger.LogError($"CultNet RUDP schema server could not decode a peer's message ({ex.GetType().Name}); that peer's session ends once its answer is acknowledged.");
                 return false;
             }
             _undecodableFramesInARow.GetOrCreateValue(delivered.Peer).Value = 0;

src/GameCult.Networking/CultNetTransport.cs

@@ -2759,6 +2759,11 @@ namespace GameCult.Networking
         // under SessionGate by whichever thread's send found the failure; the polling thread reads
         // it when it retires the peer.
         internal byte[]? UnsendableReason { get; set; }
+
+        // Set once the peer's capped frame has been answered: the session ends with this reason when the
+        // peer has acknowledged every answer, or at the deadline. Under SessionGate.
+        internal byte[]? EndingReason { get; set; }
+        internal long EndingDeadlineMs { get; set; }
     }
 
     /// <summary>
@@ -2928,6 +2933,7 @@ namespace GameCult.Networking
         public bool TryReceiveOnce(out CultNetRudpSocketServerFrame? delivered)
         {
             RetireUnsendablePeers();
+            EndAnsweredPeers();
             if (_deliveredFrames.Count > 0)
             {
                 delivered = _deliveredFrames.Dequeue();
@@ -3146,6 +3152,7 @@ namespace GameCult.Networking
         public void PollResends()
         {
             RetireUnsendablePeers();
+            EndAnsweredPeers();
             if (_socket.Poll(0, SelectMode.SelectRead))
                 return;
             foreach (var peer in _peers.Values.ToArray())
@@ -3244,6 +3251,44 @@ namespace GameCult.Networking
             return true;
         }
 
+        // Polling thread only. The application answered a peer's capped frame: the answer is reliable, so a
+        // lost datagram is resent, and the session ends with `reason` once the peer has acknowledged every
+        // answer or `graceMs` have passed (EndAnsweredPeers). Frames of the peer still queued are not delivered.
+        internal void EndPeerSessionWhenAnswered(CultNetRudpSocketServerPeer peer, byte[] reason, long graceMs)
+        {
+            if (peer == null) throw new ArgumentNullException(nameof(peer));
+            if (reason == null) throw new ArgumentNullException(nameof(reason));
+            lock (peer.SessionGate)
+            {
+                if (peer.Session.Ended || peer.EndingReason != null)
+                    return;
+                peer.EndingReason = reason;
+                peer.EndingDeadlineMs = NowMs() + graceMs;
+            }
+            var kept = _deliveredFrames.Where(frame => !ReferenceEquals(frame.Peer, peer)).ToArray();
+            _deliveredFrames.Clear();
+            foreach (var frame in kept)
+                _deliveredFrames.Enqueue(frame);
+        }
+
+        // Polling thread only.
+        private void EndAnsweredPeers()
+        {
+            foreach (var peer in _peers.Values.ToArray())
+            {
+                byte[]? reason;
+                lock (peer.SessionGate)
+                {
+                    reason = peer.EndingReason != null
+                        && (peer.Session.OutstandingReliablePacketCount == 0 || NowMs() >= peer.EndingDeadlineMs)
+                        ? peer.EndingReason
+                        : null;
+                }
+                if (reason != null)
+                    EndPeerSession(peer, reason);
+            }
+        }
+
         // Polling thread only. A peer a new Connect already replaced was reported then, and its key now
         // names the peer that replaced it.
         private void RetireUnsendablePeers()

tests/GameCult.Networking.Tests/CultNetRudpSchemaServerPeerInputTests.cs

@@ -191,6 +191,95 @@ namespace GameCult.Networking.Tests
             return errors;
         }
 
+        // The bad peer acknowledges what it received and the server polls on.
+        private static async Task Acknowledge(Scene scene)
+        {
+            scene.Bad.Socket.SendTo(CultNetRudpPacketCodec.Encode(scene.Bad.Session.CreateAck()), scene.Server.LocalEndPoint);
+            await scene.Server.PollAvailableAsync(16);
+        }
+
+        // The bad peer has sent seven undecodable frames (BadThenGood sent the first), each answered.
+        private static async Task<Scene> AtTheCap()
+        {
+            var scene = await BadThenGood(Malformed());
+            await scene.Server.PollAvailableAsync(16);
+            ErrorsReceivedBy(scene.Bad);
+            for (var i = 0; i < 6; i++)
+            {
+                SendSchema(scene.Server, scene.Bad, Malformed());
+                await scene.Server.PollAvailableAsync(16);
+                ErrorsReceivedBy(scene.Bad);
+            }
+            Assert.That(scene.Ended, Is.Empty);
+            return scene;
+        }
+
+        // The capped frame's answer is lost on the wire: the server resends it, and the goodbye follows only
+        // once the peer has acknowledged it.
+        [Test]
+        public async Task ALostCappedAnswerIsResentAndTheGoodbyeWaitsForItsAcknowledgement()
+        {
+            using var scene = await AtTheCap();
+            SendSchema(scene.Server, scene.Bad, Malformed());
+            await scene.Server.PollAvailableAsync(16);
+
+            var lost = Drain(scene.Bad.Socket);
+            Assert.That(lost.Where(p => p.PacketType == CultNetRudpPacketType.Data), Is.Not.Empty, "fixture: the answer was sent and is lost here");
+            Assert.That(lost.Select(p => p.PacketType), Has.None.EqualTo(CultNetRudpPacketType.Disconnect), "no goodbye before the answer is acknowledged");
+            Assert.That(scene.Ended, Is.Empty);
+
+            var resent = new List<CultNetRudpPacket>();
+            for (var attempt = 0; attempt < 40 && resent.Count == 0; attempt++)
+            {
+                await Task.Delay(50);
+                scene.Server.PollResends();
+                resent = Drain(scene.Bad.Socket);
+            }
+            Assert.That(resent.Select(p => p.PacketType), Has.None.EqualTo(CultNetRudpPacketType.Disconnect), "still no goodbye");
+            var answers = new List<CultNetErrorMessage>();
+            foreach (var packet in resent)
+                foreach (var frame in scene.Bad.Session.Receive(packet, 1).Delivered)
+                    if (CultNetSchemaMessageSerialization.Deserialize(frame.Payload) is CultNetErrorMessage error)
+                        answers.Add(error);
+            Assert.That(answers.Select(e => e.Code), Is.EqualTo(new[] { "message_invalid" }), "the lost answer arrived by resend");
+            Assert.That(scene.Ended, Is.Empty);
+
+            await Acknowledge(scene);
+            Assert.That(scene.Ended, Is.EqualTo(new[] { scene.Bad.EndPoint }), "acknowledged: the session ends");
+            Assert.That(Drain(scene.Bad.Socket).Select(p => p.PacketType), Has.Some.EqualTo(CultNetRudpPacketType.Disconnect));
+        }
+
+        // A peer that never acknowledges the answer is not kept for ever: the session ends after the grace.
+        [Test]
+        public async Task ACappedSessionWhoseAnswerIsNeverAcknowledgedEndsAfterTheGrace()
+        {
+            using var scene = await AtTheCap();
+            SendSchema(scene.Server, scene.Bad, Malformed());
+            await scene.Server.PollAvailableAsync(16);
+            Drain(scene.Bad.Socket);
+
+            await Task.Delay(1500); // the grace is two seconds
+            scene.Server.PollResends();
+            Assert.That(scene.Ended, Is.Empty, "inside the grace");
+            await Task.Delay(700);
+            scene.Server.PollResends();
+            Assert.That(scene.Ended, Is.EqualTo(new[] { scene.Bad.EndPoint }), "the grace ran out");
+        }
+
+        // Nothing behind the capped frame is served while its answer is delivered.
+        [Test]
+        public async Task NothingBehindTheCappedFrameIsServed()
+        {
+            using var scene = await AtTheCap();
+            SendSchema(scene.Server, scene.Bad, Malformed());
+            SendSchema(scene.Server, scene.Bad, CultNetSchemaMessageSerialization.Serialize(
+                new CultNetShardCatalogRequestMessage { MessageId = "behind-the-cap" }));
+            await scene.Server.PollAvailableAsync(16);
+            await scene.Server.PollAvailableAsync(16);
+
+            Assert.That(scene.Served.Select(s => s.MessageId), Does.Not.Contain("behind-the-cap"));
+        }
+
         // The numbers are the shared cross-runtime cap, spelled out so changing the constant fails here.
         [Test]
         public async Task EightConsecutiveUndecodableFramesEndTheSessionAfterTheEighthIsAnswered()
@@ -221,6 +310,8 @@ namespace GameCult.Networking.Tests
 
             await Refuse(1);
             Assert.That(ErrorsReceivedBy(scene.Bad).Select(e => e.Code), Is.EqualTo(new[] { "message_invalid" }), "the eighth is answered too");
+            Assert.That(scene.Ended, Is.Empty, "the session ends once the answer is acknowledged");
+            await Acknowledge(scene);
             Assert.That(scene.Ended, Is.EqualTo(new[] { scene.Bad.EndPoint }));
             Assert.That(scene.Server.Peers.Select(peer => peer.RemoteEndPoint), Is.EqualTo(new[] { scene.Good.EndPoint }));
         }