C#: a peer's undecodable message ends that peer's session, never the poll

RudpCultNetSchemaServer decoded each schema frame outside its per-message
backstop, so a malformed payload (0xc1) or a well-formed message with an
unknown schemaVersion threw MessagePackSerializationException out of
PollAvailableAsync and PollOnceAsync. The drain aborted: a good peer's
request in the same drain went unserved, a consumer loop that did not
catch died, and the bad frame was already acknowledged while its peer
stayed connected.

Decoding now runs inside the backstop. A peer whose message cannot be
decoded or handled has its session ended: the transport's new
EndRefusedPeer sends the refusal goodbye under the gate, counts the drop
in PacketsDropped, drops that peer's frames still queued, forgets the
peer and raises PeerDisconnected. A session that already ended (replaced
by a new Connect, or ended by a send failure) gets no goodbye, since the
endpoint may belong to a new session. Other peers in the drain are still
served. Ending the session matches cultmesh-py's server, where one
peer's bad message ends that peer; runtime parity is the reason.

The log names the fault's type only and never its message, because the
decoder's message quotes the peer's input (the unknown schemaVersion).
The dispatch-backstop test now asserts that rule instead of the handler
message it used to require.

Tests, each failing on the pre-fix code: a malformed payload and an
unknown schemaVersion each leave the good peer served without Poll
throwing (also through PollOnceAsync); the bad peer's session ends,
counted and told; the log carries the type and not the canary value.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/GameCult.Networking/CultNetRudpSchemaServer.cs

@@ -288,23 +288,22 @@ namespace GameCult.Networking
             if (!string.Equals(delivered.Frame.ChannelId, "schema", StringComparison.Ordinal))
                 return false;
 
-            var message = CultNetSchemaMessageSerialization.Deserialize(delivered.Frame.Payload);
-            if (!_handlers.TryGetValue(message.GetType(), out var handler) || handler == null)
-            {
-                return false;
-            }
-
-            // R-AG: a handler is expected to catch its own typed refusals (HandleSnapshotRequestV1Async
-            // catches the three selection exceptions and answers cursor_invalid/selection_invalid/
-            // reference_outside_target on the wire, R-AM) - this catch is the backstop for whatever a
-            // handler does not, so an untyped exception from a malformed/hostile message can never
-            // reach the poll loop and stall the server for every other peer. It still answers nothing
-            // on the wire (an exception this generic carries no typed refusal to report) and drops the
-            // one malformed dispatch, exactly as an unregistered handler already does just above -
-            // R-AM: but it always logs. A backstop that swallows silently is a compensator, and an
-            // untested, unlogged one hid the SM-6 ArgumentNullException regression for a whole batch.
+            // A peer's message that cannot be decoded (malformed bytes, an unknown schemaVersion) or
+            // handled is that peer's failure, never the poll's: the drain goes on serving every other
+            // peer. The transport already acknowledged the frame, so the peer's session cannot be kept
+            // (a retransmit would never be handled). It ends, the peer is told, and the drop is counted
+            // and logged, as cultmesh-py's server does. A handler is still expected to answer its own
+            // typed refusals on the wire (HandleSnapshotRequestV1Async does, R-AM); this is the backstop
+            // for whatever it does not. The log names the fault's type only: its message can quote
+            // what the peer sent.
             try
             {
+                var message = CultNetSchemaMessageSerialization.Deserialize(delivered.Frame.Payload);
+                if (!_handlers.TryGetValue(message.GetType(), out var handler) || handler == null)
+                {
+                    return false;
+                }
+
                 var result = handler.DynamicInvoke(message, new RudpCultNetSchemaServerPeer(this, delivered.Peer));
                 if (result is Task task)
                 {
@@ -313,11 +312,11 @@ namespace GameCult.Networking
             }
             catch (Exception ex)
             {
-                // DynamicInvoke wraps a synchronously-thrown handler fault in TargetInvocationException,
-                // whose own .Message is the generic "Exception has been thrown by the target of an
-                // invocation." - log the real fault, not that wrapper.
+                // DynamicInvoke wraps a synchronously-thrown handler fault in TargetInvocationException;
+                // name the real fault, not that wrapper.
                 var reported = ex is System.Reflection.TargetInvocationException { InnerException: { } inner } ? inner : ex;
-                Logger.LogError($"CultNet RUDP schema dispatch failed for {message.GetType().Name}: {reported.Message}");
+                Logger.LogError($"CultNet RUDP schema server could not decode or handle a peer's message ({reported.GetType().Name}); that peer's session ended.");
+                _transport.EndRefusedPeer(delivered.Peer);
                 return false;
             }
             return true;

src/GameCult.Networking/CultNetTransport.cs

@@ -3188,6 +3188,35 @@ namespace GameCult.Networking
             _unsendablePeers.Enqueue(peer);
         }
 
+        // Polling thread only. Ends the session of a peer whose delivered frame the application could not
+        // decode or handle. The frame is already acknowledged, so the session cannot be kept: a
+        // retransmit would never be handled. The peer is told, the drop is counted, and the peer's
+        // frames still queued are not delivered. A session that already ended (a new Connect replaced
+        // it, or a send failure ended it) owes no goodbye; the endpoint may belong to a new session.
+        internal void EndRefusedPeer(CultNetRudpSocketServerPeer peer)
+        {
+            if (peer == null) throw new ArgumentNullException(nameof(peer));
+            _stats.PacketsDropped++;
+            lock (peer.SessionGate)
+            {
+                if (peer.Session.Ended)
+                    return;
+                SendPacket(peer.RemoteEndPoint, peer.Session.EndRefused());
+            }
+
+            var kept = _deliveredFrames.Where(frame => !ReferenceEquals(frame.Peer, peer)).ToArray();
+            _deliveredFrames.Clear();
+            foreach (var frame in kept)
+                _deliveredFrames.Enqueue(frame);
+
+            var key = RemoteKey(peer.RemoteEndPoint);
+            if (!_peers.TryGetValue(key, out var current) || !ReferenceEquals(current, peer))
+                return;
+            _peers.Remove(key);
+            peer.DisconnectReason = CultNetRudpSession.RefusedPacketReason;
+            PeerDisconnected?.Invoke(peer);
+        }
+
         // 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

@@ -0,0 +1,196 @@
+#nullable enable
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Net;
+using System.Net.Sockets;
+using System.Threading.Tasks;
+using GameCult.Logging;
+using MessagePack;
+using NUnit.Framework;
+
+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.
+    /// </summary>
+    public class CultNetRudpSchemaServerPeerInputTests
+    {
+        private const uint ConnectionId = 0x10203096;
+        private const string Canary = "canary-5f1c9e0b";
+
+        private sealed class Client
+        {
+            public Socket Socket = null!;
+            public CultNetRudpSession Session = null!;
+            public EndPoint EndPoint => Socket.LocalEndPoint!;
+        }
+
+        private sealed class CapturingLogger : ILogger
+        {
+            public List<string> Errors { get; } = new();
+            public void LogInfo(string message) { }
+            public void LogWarning(string message) { }
+            public void LogError(string message) => Errors.Add(message);
+            public void LogDebug(string message) { }
+        }
+
+        private static Socket Bind()
+        {
+            var socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
+            socket.Bind(new IPEndPoint(IPAddress.Loopback, 0));
+            return socket;
+        }
+
+        private static List<CultNetRudpPacket> Drain(Socket socket)
+        {
+            var packets = new List<CultNetRudpPacket>();
+            var buffer = new byte[65535];
+            EndPoint from = new IPEndPoint(IPAddress.Any, 0);
+            while (socket.Poll(20_000, SelectMode.SelectRead))
+            {
+                var received = socket.ReceiveFrom(buffer, ref from);
+                packets.Add(CultNetRudpPacketCodec.Decode(buffer.AsSpan(0, received).ToArray()));
+            }
+            return packets;
+        }
+
+        private static async Task<Client> Connect(RudpCultNetSchemaServer server)
+        {
+            var client = new Client
+            {
+                Socket = Bind(),
+                Session = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = ConnectionId })
+            };
+            client.Socket.SendTo(CultNetRudpPacketCodec.Encode(client.Session.CreateConnect(0)), server.LocalEndPoint);
+            await server.PollAvailableAsync(8);
+            client.Session.Receive(Drain(client.Socket).Single(p => p.PacketType == CultNetRudpPacketType.Accept), 0);
+            return client;
+        }
+
+        private static void SendSchema(RudpCultNetSchemaServer server, Client client, byte[] payload) =>
+            client.Socket.SendTo(
+                CultNetRudpPacketCodec.Encode(client.Session.Send("schema", payload, new CultNetRudpSendOptions { Reliable = true, Ordered = true })),
+                server.LocalEndPoint);
+
+        private static byte[] Malformed() => new byte[] { 0xc1 };
+
+        // Well-formed MessagePack naming a schema version no runtime knows. The decoder's own error
+        // quotes that version.
+        private static byte[] UnknownSchemaVersion() =>
+            MessagePackSerializer.Serialize(
+                new Dictionary<string, object> { ["schemaVersion"] = "cultnet." + Canary + ".v0", ["messageId"] = Canary },
+                CultNetSchemaMessageSerialization.Options);
+
+        private sealed class Scene : IDisposable
+        {
+            public RudpCultNetSchemaServer Server = null!;
+            public Client Bad = null!;
+            public Client Good = null!;
+            public CapturingLogger Logger = new();
+            public List<(string MessageId, EndPoint From)> Served = new();
+            public List<EndPoint> Ended = new();
+
+            public void Dispose()
+            {
+                Server.Dispose();
+                Bad.Socket.Dispose();
+                Good.Socket.Dispose();
+            }
+        }
+
+        // The bad peer's frame is read first, so the drain reaches the good peer only by getting past it.
+        private static async Task<Scene> BadThenGood(byte[] badPayload)
+        {
+            var scene = new Scene
+            {
+                Server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions
+                {
+                    RuntimeId = "csharp-peer-input",
+                    ConnectionId = ConnectionId
+                })
+            };
+            scene.Server.Logger = scene.Logger;
+            scene.Server.OnCultNet<CultNetShardCatalogRequestMessage>((message, peer) =>
+                scene.Served.Add((message.MessageId, peer.RemoteEndPoint)));
+            scene.Server.PeerDisconnected += peer => scene.Ended.Add(((RudpCultNetSchemaServerPeer)peer).RemoteEndPoint);
+            scene.Bad = await Connect(scene.Server);
+            scene.Good = await Connect(scene.Server);
+
+            SendSchema(scene.Server, scene.Bad, badPayload);
+            SendSchema(scene.Server, scene.Good, CultNetSchemaMessageSerialization.Serialize(
+                new CultNetShardCatalogRequestMessage { MessageId = "good-request" }));
+            return scene;
+        }
+
+        [Test]
+        public async Task AMalformedPayloadDoesNotStopTheDrainServingAnotherPeer()
+        {
+            using var scene = await BadThenGood(Malformed());
+
+            Assert.DoesNotThrowAsync(async () => await scene.Server.PollAvailableAsync(16));
+
+            Assert.That(scene.Served, Is.EqualTo(new[] { ("good-request", scene.Good.EndPoint) }));
+        }
+
+        [Test]
+        public async Task AnUnknownSchemaVersionDoesNotStopTheDrainServingAnotherPeer()
+        {
+            using var scene = await BadThenGood(UnknownSchemaVersion());
+
+            Assert.DoesNotThrowAsync(async () => await scene.Server.PollAvailableAsync(16));
+
+            Assert.That(scene.Served, Is.EqualTo(new[] { ("good-request", scene.Good.EndPoint) }));
+        }
+
+        [Test]
+        public async Task PollOnceDoesNotThrowForAPeersUndecodableMessage()
+        {
+            using var scene = await BadThenGood(Malformed());
+
+            Assert.DoesNotThrowAsync(async () =>
+            {
+                for (var attempt = 0; attempt < 16; attempt++)
+                    await scene.Server.PollOnceAsync();
+            });
+
+            Assert.That(scene.Served, Is.EqualTo(new[] { ("good-request", scene.Good.EndPoint) }));
+        }
+
+        [Test]
+        public async Task APeerWhoseMessageCannotBeDecodedHasItsSessionEnded()
+        {
+            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(Drain(scene.Good.Socket).Select(packet => packet.PacketType), Has.None.EqualTo(CultNetRudpPacketType.Disconnect));
+            Assert.That(scene.Logger.Errors, Has.Count.EqualTo(1));
+        }
+
+        [Test]
+        public async Task TheDropIsLoggedByErrorTypeWithoutEchoingWhatThePeerSent()
+        {
+            using var scene = await BadThenGood(UnknownSchemaVersion());
+
+            try { await scene.Server.PollAvailableAsync(16); }
+            catch (MessagePackSerializationException) { Assert.Fail("the poll threw for a peer's input"); }
+
+            // The decoder's own error message quotes the unknown version, which is the peer's input.
+            Assert.That(scene.Logger.Errors, Has.Count.EqualTo(1));
+            Assert.That(scene.Logger.Errors[0], Does.Contain(nameof(MessagePackSerializationException)));
+            Assert.That(scene.Logger.Errors[0], Does.Not.Contain(Canary));
+        }
+    }
+}

tests/GameCult.Networking.Tests/CultNetSelectionFixBatchTests.cs

@@ -796,8 +796,11 @@ namespace GameCult.Networking.Tests
                 serverThread.Join();
             }
 
-            Assert.That(logger.Errors, Has.Some.Contains("dispatch-backstop-probe"),
-                "the backstop must log the handler's own exception message, not swallow it silently");
+            // The log names the fault's type. The fault's own message can quote what the peer sent,
+            // so it is never logged.
+            Assert.That(logger.Errors, Has.Some.Contains(nameof(InvalidOperationException)),
+                "the backstop must log the handler's fault, not swallow it silently");
+            Assert.That(logger.Errors, Has.None.Contains("dispatch-backstop-probe"));
         }
 
         private sealed class CapturingLogger : ILogger