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