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 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 20:19:29 +02:00
commit ce63ff88dc
4 changed files with 184 additions and 26 deletions

View file

@ -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
/// </summary>
public CultNetTransportStats Stats => _transport.Stats;
// The listener this server drives, for tests that inject send failures through its seams.
internal CultNetRudpSocketTransportServer Transport => _transport;
/// <summary>
/// Gets the currently tracked RUDP peers.
/// </summary>
@ -283,27 +287,49 @@ namespace GameCult.Networking
return new RudpCultNetSchemaPollResult(consumed, dispatched);
}
/// <summary>
/// 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.
/// </summary>
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<bool> 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;

View file

@ -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.
/// </summary>
internal CultNetRudpPacket EndRefused()
internal CultNetRudpPacket EndRefused() => EndWith(RefusedPacketReason);
/// <summary>
/// Ends a session whose peer's frame the application above it refused or failed on, after the
/// session acknowledged it. As with <see cref="EndRefused"/>, the peer's sequences are forgotten
/// before the goodbye is built.
/// </summary>
internal CultNetRudpPacket EndWith(byte[] reason)
{
ResetPeerState();
return CreateDisconnect(RefusedPacketReason);
return CreateDisconnect(reason);
}
/// <summary>
@ -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

View file

@ -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<string>();
server.OnCultNet<CultNetShardCatalogRequestMessage>((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<CultNetShardCatalogRequestMessage>((_, 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()
{

View file

@ -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()
{