diff --git a/src/GameCult.Networking/CultNetRudpSchemaServer.cs b/src/GameCult.Networking/CultNetRudpSchemaServer.cs
index e19fccc3..3112f07b 100644
--- a/src/GameCult.Networking/CultNetRudpSchemaServer.cs
+++ b/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;
diff --git a/src/GameCult.Networking/CultNetTransport.cs b/src/GameCult.Networking/CultNetTransport.cs
index 24839d41..ba63cf46 100644
--- a/src/GameCult.Networking/CultNetTransport.cs
+++ b/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()
diff --git a/tests/GameCult.Networking.Tests/CultNetRudpSchemaServerPeerInputTests.cs b/tests/GameCult.Networking.Tests/CultNetRudpSchemaServerPeerInputTests.cs
new file mode 100644
index 00000000..74fc1d67
--- /dev/null
+++ b/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
+{
+ ///
+ /// 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.
+ ///
+ 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 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 Drain(Socket socket)
+ {
+ var packets = new List();
+ 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 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 { ["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 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 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((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));
+ }
+ }
+}
diff --git a/tests/GameCult.Networking.Tests/CultNetSelectionFixBatchTests.cs b/tests/GameCult.Networking.Tests/CultNetSelectionFixBatchTests.cs
index 9b6959c0..e4c31d64 100644
--- a/tests/GameCult.Networking.Tests/CultNetSelectionFixBatchTests.cs
+++ b/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