From ce63ff88dcf02e4e4932ae55998025104ea72e3d Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 20:19:29 +0200 Subject: [PATCH] C#: a handler's fault ends the session without blaming the peer The schema server's backstop ended every session with "session refused a packet", including when the fault was the server's own: for example a handler's response too large to send, or a refusal the handler did not type. Decoding and handling are now two catches: - A message that cannot be decoded is the peer's. The session ends with the refusal reason. - A handler fault ends the session with "server could not handle a message". The transport's EndRefusedPeer becomes EndPeerSession(peer, reason). It reports whether it ended the session, so the log says when a response that failed after its first fragment had already ended it over the send. Before, the log reported that it ended a session that was already gone. The comment said that ending the session matched cultmesh-py. It does not: Python silently skips a malformed frame. The honest anchor is the C# TCP schema server, whose ServePeerAsync ends the one peer whose message fails to decode or handle. The comment now says so. Pinned, each by a test that kills its mutant: - a handler fault's goodbye carries the server's reason, not the refusal (mutant: the refusal reason); - the post-fragment-0 path logs that the session had already ended (mutant: a constant outcome); - EndPeerSession says no goodbye to a replaced peer's endpoint (mutant: the Ended check dropped); - a refused peer's frames still queued are never handled (mutant: the queue purge dropped). Co-Authored-By: Claude Opus 5.5 --- .../CultNetRudpSchemaServer.cs | 59 ++++++++--- src/GameCult.Networking/CultNetTransport.cs | 32 +++--- .../CultNetRudpSchemaServerPeerInputTests.cs | 97 +++++++++++++++++++ .../CultNetRudpSendFailureTests.cs | 22 +++++ 4 files changed, 184 insertions(+), 26 deletions(-) diff --git a/src/GameCult.Networking/CultNetRudpSchemaServer.cs b/src/GameCult.Networking/CultNetRudpSchemaServer.cs index 3112f07b..de8d0154 100644 --- a/src/GameCult.Networking/CultNetRudpSchemaServer.cs +++ b/src/GameCult.Networking/CultNetRudpSchemaServer.cs @@ -3,6 +3,7 @@ using System.Collections.Concurrent; using System.Collections.Generic; using System.Net; using System.Net.Sockets; +using System.Text; using System.Threading.Tasks; using GameCult.Logging; @@ -169,6 +170,9 @@ namespace GameCult.Networking /// public CultNetTransportStats Stats => _transport.Stats; + // The listener this server drives, for tests that inject send failures through its seams. + internal CultNetRudpSocketTransportServer Transport => _transport; + /// /// Gets the currently tracked RUDP peers. /// @@ -283,27 +287,49 @@ namespace GameCult.Networking return new RudpCultNetSchemaPollResult(consumed, dispatched); } + /// + /// The goodbye's reason when a handler fails on a peer's message: the server could not handle + /// it, which says nothing about whether the peer's packet was well formed. + /// + internal static readonly byte[] HandlerFailedReason = Encoding.UTF8.GetBytes("server could not handle a message"); + + private static string SessionOutcome(bool ended) => + ended ? "that peer's session ended." : "that peer's session had already ended."; + private async Task DispatchAsync(CultNetRudpSocketServerFrame delivered) { if (!string.Equals(delivered.Frame.ChannelId, "schema", StringComparison.Ordinal)) return false; // 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. + // handled ends that peer's session, never the poll: the drain goes on serving every other + // peer. The transport already acknowledged the frame, so the session cannot be kept (a + // retransmit would never be handled). The peer is told, and the drop is counted and logged. + // The C# TCP schema server (CultNetTcpSchemaTransport.ServePeerAsync) likewise ends the one + // peer whose message fails to decode or handle. 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. + ICultNetSchemaMessage message; try { - var message = CultNetSchemaMessageSerialization.Deserialize(delivered.Frame.Payload); - if (!_handlers.TryGetValue(message.GetType(), out var handler) || handler == null) - { - return false; - } + message = CultNetSchemaMessageSerialization.Deserialize(delivered.Frame.Payload); + } + catch (Exception ex) + { + // The peer sent something no runtime can read: the refusal is the peer's. + 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; + } + if (!_handlers.TryGetValue(message.GetType(), out var handler) || handler == null) + { + return false; + } + + try + { var result = handler.DynamicInvoke(message, new RudpCultNetSchemaServerPeer(this, delivered.Peer)); if (result is Task task) { @@ -313,10 +339,13 @@ namespace GameCult.Networking catch (Exception ex) { // DynamicInvoke wraps a synchronously-thrown handler fault in TargetInvocationException; - // name the real fault, not that wrapper. + // name the real fault, not that wrapper. The fault is the server's (a response too large + // to send, a handler bug) or a refusal the handler did not type, so the goodbye does not + // say the peer's packet was refused. A response that failed after its first fragment + // already ended the session over the send, and the log says so. var reported = ex is System.Reflection.TargetInvocationException { InnerException: { } inner } ? inner : ex; - 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); + var ended = _transport.EndPeerSession(delivered.Peer, HandlerFailedReason); + Logger.LogError($"CultNet RUDP schema server's {message.GetType().Name} handler failed ({reported.GetType().Name}); {SessionOutcome(ended)}"); return false; } return true; diff --git a/src/GameCult.Networking/CultNetTransport.cs b/src/GameCult.Networking/CultNetTransport.cs index dd8f3992..a6102de3 100644 --- a/src/GameCult.Networking/CultNetTransport.cs +++ b/src/GameCult.Networking/CultNetTransport.cs @@ -1462,10 +1462,17 @@ namespace GameCult.Networking /// sequence, so the peer's sequences are forgotten before the goodbye is built: its ack field /// would otherwise acknowledge the very frame the session refused. /// - internal CultNetRudpPacket EndRefused() + internal CultNetRudpPacket EndRefused() => EndWith(RefusedPacketReason); + + /// + /// Ends a session whose peer's frame the application above it refused or failed on, after the + /// session acknowledged it. As with , the peer's sequences are forgotten + /// before the goodbye is built. + /// + internal CultNetRudpPacket EndWith(byte[] reason) { ResetPeerState(); - return CreateDisconnect(RefusedPacketReason); + return CreateDisconnect(reason); } /// @@ -3191,19 +3198,21 @@ namespace GameCult.Networking } // 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) + // decode or handle, with a goodbye carrying the application's reason. 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, since the endpoint may belong to a new session; the call then returns false. + internal bool EndPeerSession(CultNetRudpSocketServerPeer peer, byte[] reason) { if (peer == null) throw new ArgumentNullException(nameof(peer)); + if (reason == null) throw new ArgumentNullException(nameof(reason)); _stats.PacketsDropped++; lock (peer.SessionGate) { if (peer.Session.Ended) - return; - SendPacket(peer.RemoteEndPoint, peer.Session.EndRefused()); + return false; + SendPacket(peer.RemoteEndPoint, peer.Session.EndWith(reason)); } var kept = _deliveredFrames.Where(frame => !ReferenceEquals(frame.Peer, peer)).ToArray(); @@ -3213,10 +3222,11 @@ namespace GameCult.Networking var key = RemoteKey(peer.RemoteEndPoint); if (!_peers.TryGetValue(key, out var current) || !ReferenceEquals(current, peer)) - return; + return true; _peers.Remove(key); - peer.DisconnectReason = CultNetRudpSession.RefusedPacketReason; + peer.DisconnectReason = reason; PeerDisconnected?.Invoke(peer); + return true; } // Polling thread only. A peer a new Connect already replaced was reported then, and its key now diff --git a/tests/GameCult.Networking.Tests/CultNetRudpSchemaServerPeerInputTests.cs b/tests/GameCult.Networking.Tests/CultNetRudpSchemaServerPeerInputTests.cs index 74fc1d67..b8de674f 100644 --- a/tests/GameCult.Networking.Tests/CultNetRudpSchemaServerPeerInputTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetRudpSchemaServerPeerInputTests.cs @@ -4,6 +4,7 @@ using System.Collections.Generic; using System.Linq; using System.Net; using System.Net.Sockets; +using System.Text; using System.Threading.Tasks; using GameCult.Logging; using MessagePack; @@ -179,6 +180,102 @@ namespace GameCult.Networking.Tests 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. + [Test] + public async Task ARefusedPeersFramesStillQueuedAreNotHandled() + { + var logger = new CapturingLogger(); + using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions + { + RuntimeId = "csharp-peer-input", + ConnectionId = ConnectionId + }); + server.Logger = logger; + var served = new List(); + server.OnCultNet((message, _) => served.Add(message.MessageId)); + var bad = await Connect(server); + var options = new CultNetRudpSendOptions { Reliable = true, Ordered = true }; + var malformed = bad.Session.Send("schema", Malformed(), options); + var after = bad.Session.Send("schema", CultNetSchemaMessageSerialization.Serialize( + new CultNetShardCatalogRequestMessage { MessageId = "after-the-refused-one" }), options); + bad.Socket.SendTo(CultNetRudpPacketCodec.Encode(after), server.LocalEndPoint); + bad.Socket.SendTo(CultNetRudpPacketCodec.Encode(malformed), server.LocalEndPoint); + + await server.PollAvailableAsync(16); + + Assert.That(served, Is.Empty); + Assert.That(server.Peers, Is.Empty); + Assert.That(logger.Errors, Has.Count.EqualTo(1)); + bad.Socket.Dispose(); + } + + private static async Task<(RudpCultNetSchemaServer Server, Client Client, CapturingLogger Logger)> ServerWhoseHandlerAnswers( + byte[] response, int? maxFragmentBytes, int? unsendableAfter = null) + { + var logger = new CapturingLogger(); + var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions + { + RuntimeId = "csharp-peer-input", + ConnectionId = ConnectionId, + MaxFragmentBytes = maxFragmentBytes + }); + server.Logger = logger; + server.OnCultNet((_, peer) => + { + if (unsendableAfter is { } count) + server.Transport.UnsendableAfter[peer.RemoteEndPoint] = count; + server.Transport.SendSchema(peer.TransportPeer, response); + }); + var client = await Connect(server); + SendSchema(server, client, CultNetSchemaMessageSerialization.Serialize( + new CultNetShardCatalogRequestMessage { MessageId = "ask" })); + return (server, client, logger); + } + + // A handler whose own response can never be sent (larger than a datagram, unfragmented) fails + // on the server's account, not the peer's. The session still ends, but its goodbye does not say + // the peer's packet was refused. + [Test] + public async Task AHandlerFaultEndsTheSessionWithAReasonThatDoesNotBlameThePeer() + { + var (server, client, logger) = await ServerWhoseHandlerAnswers(new byte[70_000], maxFragmentBytes: null); + using (server) + { + await server.PollAvailableAsync(16); + + var goodbye = Drain(client.Socket).Single(packet => packet.PacketType == CultNetRudpPacketType.Disconnect); + Assert.That(Encoding.UTF8.GetString(goodbye.Payload!), Is.EqualTo("server could not handle a message")); + Assert.That(server.Peers, Is.Empty); + Assert.That(logger.Errors, Has.Count.EqualTo(1)); + Assert.That(logger.Errors[0], Does.Contain("handler failed (SocketException)")); + Assert.That(logger.Errors[0], Does.Contain("that peer's session ended.")); + } + client.Socket.Dispose(); + } + + // A response that fails after its first fragment has already ended the session over the send, + // with a goodbye that names the send failure. The backstop ends nothing more and says so. + [Test] + public async Task AHandlerWhoseResponseEndedTheSessionIsLoggedAsAlreadyEnded() + { + var (server, client, logger) = await ServerWhoseHandlerAnswers(new byte[3000], maxFragmentBytes: 1000, unsendableAfter: 1); + using (server) + { + await server.PollAvailableAsync(16); + await server.PollAvailableAsync(16); + + var goodbyes = Drain(client.Socket).Where(packet => packet.PacketType == CultNetRudpPacketType.Disconnect).ToList(); + Assert.That(goodbyes, Has.Count.EqualTo(1)); + Assert.That(Encoding.UTF8.GetString(goodbyes[0].Payload!), Does.StartWith("packet could not be sent: ")); + Assert.That(server.Peers, Is.Empty); + Assert.That(logger.Errors, Has.Count.EqualTo(1)); + Assert.That(logger.Errors[0], Does.Contain("that peer's session had already ended.")); + } + client.Socket.Dispose(); + } + [Test] public async Task TheDropIsLoggedByErrorTypeWithoutEchoingWhatThePeerSent() { diff --git a/tests/GameCult.Networking.Tests/CultNetRudpSendFailureTests.cs b/tests/GameCult.Networking.Tests/CultNetRudpSendFailureTests.cs index 7b327c02..81d255f7 100644 --- a/tests/GameCult.Networking.Tests/CultNetRudpSendFailureTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetRudpSendFailureTests.cs @@ -430,6 +430,28 @@ namespace GameCult.Networking.Tests } } + // An application ending a peer's session (a message it could not decode or handle) says no + // goodbye to a session that already ended: a new Connect replaced it, and the endpoint now + // belongs to the session that replaced it. + [Test] + public void EndingAReplacedPeersSessionSaysNoGoodbyeToTheSessionThatReplacedIt() + { + var (server, listenerEndPoint, x, _) = TwoPeers(); + using (server) + { + var (stale, replacing) = Reconnect(server, listenerEndPoint, x); + var ended = RecordDisconnects(server); + + Assert.That(server.EndPeerSession(stale, Encoding.UTF8.GetBytes("application reason")), Is.False); + + Assert.That(Drain(x.Socket).Select(p => p.PacketType), Has.None.EqualTo(CultNetRudpPacketType.Disconnect)); + Assert.That(x.Session.Connected, Is.True); + Assert.That(replacing.Connected, Is.True); + Assert.That(server.Peers, Has.Member(replacing)); + Assert.That(ended, Is.Empty); + } + } + [Test] public void ASessionEndedByACallerAndThePollingThreadTogetherSaysGoodbyeOnce() {