src/GameCult.Networking/CultNetRudpSchemaServer.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Runtime.CompilerServices;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Net;
@@ -113,9 +114,18 @@ namespace GameCult.Networking
         private readonly CultNetRudpSocketTransportServer _transport;
         private readonly ConcurrentDictionary<Type, Delegate> _handlers = new();
         private readonly ConcurrentDictionary<Delegate, Delegate> _schemaPeerAdapters = new();
+        private readonly ConditionalWeakTable<CultNetRudpSocketServerPeer, StrongBox<int>> _undecodableFramesInARow = new();
         private bool _disposed;
         private ILogger _logger = new NullLogger();
 
+        /// <summary>
+        /// How many consecutive undecodable frames a peer's session may send before it is ended. Fewer are
+        /// answered with a typed <c>message_invalid</c> and the session is kept; the frame that reaches this
+        /// number is answered too, then the session ends. Any frame that decodes resets the run. The same
+        /// number is the cap in every runtime.
+        /// </summary>
+        public const int MaxConsecutiveRefusedFrames = 8;
+
         /// <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.
@@ -317,11 +327,21 @@ namespace GameCult.Networking
             }
             catch (Exception ex)
             {
-                // The peer sent something no runtime can read: the refusal is the peer's.
+                // The peer sent something no runtime can read: the refusal is the peer's. It is answered
+                // message_invalid and the session kept, until MaxConsecutiveRefusedFrames in a row end it.
+                var run = _undecodableFramesInARow.GetOrCreateValue(delivered.Peer);
+                run.Value++;
+                _transport.SendSchemaMessage(delivered.Peer, CultNetErrorMessage.ForMessageInvalid("message", "CultNet RUDP schema server could not read the message."));
+                if (run.Value < MaxConsecutiveRefusedFrames)
+                {
+                    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)}");
                 return false;
             }
+            _undecodableFramesInARow.GetOrCreateValue(delivered.Peer).Value = 0;
 
             if (!_handlers.TryGetValue(message.GetType(), out var handler) || handler == null)
             {

tests/GameCult.Networking.Tests/CultNetRudpSchemaServerPeerInputTests.cs

@@ -14,8 +14,9 @@ namespace GameCult.Networking.Tests
 {
     /// <summary>
     /// A schema server's poll serves every peer whatever one peer sends. A message that cannot be
-    /// decoded or handled is that peer's failure: its session ends, the drop is counted and logged,
-    /// and the drain goes on to the other peers.
+    /// decoded is answered message_invalid and its session kept until MaxConsecutiveRefusedFrames in a
+    /// row end it; one that cannot be handled ends that peer's session. The fault is logged, and the
+    /// drain goes on to the other peers.
     /// </summary>
     public class CultNetRudpSchemaServerPeerInputTests
     {
@@ -161,30 +162,74 @@ namespace GameCult.Networking.Tests
         }
 
         [Test]
-        public async Task APeerWhoseMessageCannotBeDecodedHasItsSessionEnded()
+        public async Task APeerWhoseMessageCannotBeDecodedIsAnsweredAndKept()
         {
             using var scene = await BadThenGood(Malformed());
-            var dropped = scene.Server.Stats.PacketsDropped;
 
             try { await scene.Server.PollAvailableAsync(16); }
             catch (MessagePackSerializationException) { Assert.Fail("the poll threw for a peer's input"); }
 
-            // The frame was already acknowledged, so the session cannot be kept: the peer is told,
-            // the listener forgets it, and the drop is counted.
-            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 }));
-            Assert.That(scene.Server.Stats.PacketsDropped, Is.EqualTo(dropped + 1));
-            var goodbye = Drain(scene.Bad.Socket).Single(packet => packet.PacketType == CultNetRudpPacketType.Disconnect);
-            Assert.That(scene.Bad.Session.Receive(goodbye, 1).Disconnected, Is.True);
+            Assert.That(scene.Ended, Is.Empty);
+            Assert.That(scene.Server.Peers, Has.Count.EqualTo(2));
+            var refusal = ErrorsReceivedBy(scene.Bad).Single();
+            Assert.That(refusal.Code, Is.EqualTo("message_invalid"));
+            Assert.That(refusal.Details!.Field, Is.EqualTo("message"));
             Assert.That(Drain(scene.Good.Socket).Select(packet => packet.PacketType), Has.None.EqualTo(CultNetRudpPacketType.Disconnect));
             Assert.That(scene.Logger.Errors, Has.Count.EqualTo(1));
         }
 
-        // The peer's malformed message and a well-formed one after it arrive in one datagram's
-        // delivery: the second is held for order, and both are released together. The session ends
-        // on the first, so the second belongs to no session and is never handled.
+        private static List<CultNetErrorMessage> ErrorsReceivedBy(Client client)
+        {
+            var errors = new List<CultNetErrorMessage>();
+            foreach (var packet in Drain(client.Socket))
+            {
+                var received = client.Session.Receive(packet, 1);
+                foreach (var frame in received.Delivered)
+                    if (CultNetSchemaMessageSerialization.Deserialize(frame.Payload) is CultNetErrorMessage error)
+                        errors.Add(error);
+            }
+            return errors;
+        }
+
+        // The numbers are the shared cross-runtime cap, spelled out so changing the constant fails here.
         [Test]
-        public async Task ARefusedPeersFramesStillQueuedAreNotHandled()
+        public async Task EightConsecutiveUndecodableFramesEndTheSessionAfterTheEighthIsAnswered()
+        {
+            Assert.That(RudpCultNetSchemaServer.MaxConsecutiveRefusedFrames, Is.EqualTo(8));
+            using var scene = await BadThenGood(Malformed());
+            await scene.Server.PollAvailableAsync(16);
+            ErrorsReceivedBy(scene.Bad);
+            async Task Refuse(int count)
+            {
+                for (var i = 0; i < count; i++)
+                {
+                    SendSchema(scene.Server, scene.Bad, Malformed());
+                    await scene.Server.PollAvailableAsync(16);
+                }
+            }
+
+            await Refuse(6); // with the first, seven in a row
+            Assert.That(ErrorsReceivedBy(scene.Bad), Has.Count.EqualTo(6));
+            Assert.That(scene.Ended, Is.Empty);
+            SendSchema(scene.Server, scene.Bad, CultNetSchemaMessageSerialization.Serialize(
+                new CultNetShardCatalogRequestMessage { MessageId = "resets-the-run" }));
+            await scene.Server.PollAvailableAsync(16);
+            Assert.That(scene.Served.Select(s => s.MessageId), Does.Contain("resets-the-run"));
+            await Refuse(7);
+            Assert.That(ErrorsReceivedBy(scene.Bad), Has.Count.EqualTo(7));
+            Assert.That(scene.Ended, Is.Empty, "a decoded frame reset the run");
+
+            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.EqualTo(new[] { scene.Bad.EndPoint }));
+            Assert.That(scene.Server.Peers.Select(peer => peer.RemoteEndPoint), Is.EqualTo(new[] { scene.Good.EndPoint }));
+        }
+
+        // The peer's malformed message and a well-formed one after it arrive in one datagram's
+        // delivery: the second is held for order, and both are released together. The session is
+        // kept through the first, so the second is handled.
+        [Test]
+        public async Task AFrameQueuedBehindAnUndecodableOneIsStillHandled()
         {
             var logger = new CapturingLogger();
             using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions
@@ -205,8 +250,8 @@ namespace GameCult.Networking.Tests
 
             await server.PollAvailableAsync(16);
 
-            Assert.That(served, Is.Empty);
-            Assert.That(server.Peers, Is.Empty);
+            Assert.That(served, Is.EqualTo(new[] { "after-the-refused-one" }));
+            Assert.That(server.Peers, Has.Count.EqualTo(1));
             Assert.That(logger.Errors, Has.Count.EqualTo(1));
             bad.Socket.Dispose();
         }