diff --git a/src/GameCult.Networking/CultNetRudpSchemaServer.cs b/src/GameCult.Networking/CultNetRudpSchemaServer.cs
index 444e97e9..3378036a 100644
--- a/src/GameCult.Networking/CultNetRudpSchemaServer.cs
+++ b/src/GameCult.Networking/CultNetRudpSchemaServer.cs
@@ -126,6 +126,12 @@ namespace GameCult.Networking
///
public const int MaxConsecutiveRefusedFrames = 8;
+ ///
+ /// How long a session whose capped frame was answered waits for the peer to acknowledge the answer
+ /// before it ends anyway.
+ ///
+ internal const long CappedAnswerGraceMs = 2000;
+
///
/// Gets or sets the logger used by the server. R-AM: the dispatch backstop in DispatchAsync
/// logs every untyped fault through this instead of swallowing it silently.
@@ -310,6 +316,9 @@ namespace GameCult.Networking
{
if (!string.Equals(delivered.Frame.ChannelId, "schema", StringComparison.Ordinal))
return false;
+ // The capped frame was answered: nothing after it is served while the answer is delivered.
+ if (delivered.Peer.EndingReason != null)
+ return false;
// A peer's message that cannot be decoded (malformed bytes, an unknown schemaVersion) or
// handled ends that peer's session, never the poll: the drain goes on serving every other
@@ -337,8 +346,10 @@ namespace GameCult.Networking
Logger.LogError($"CultNet RUDP schema server could not decode a peer's message ({ex.GetType().Name}); it was refused and that peer's session kept.");
return false;
}
- 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)}");
+ // The answer was sent reliably: the session ends once the peer has acknowledged it (a lost
+ // datagram is resent first), or after the grace.
+ _transport.EndPeerSessionWhenAnswered(delivered.Peer, CultNetRudpSession.RefusedPacketReason, CappedAnswerGraceMs);
+ Logger.LogError($"CultNet RUDP schema server could not decode a peer's message ({ex.GetType().Name}); that peer's session ends once its answer is acknowledged.");
return false;
}
_undecodableFramesInARow.GetOrCreateValue(delivered.Peer).Value = 0;
diff --git a/src/GameCult.Networking/CultNetTransport.cs b/src/GameCult.Networking/CultNetTransport.cs
index ac2c2952..da465326 100644
--- a/src/GameCult.Networking/CultNetTransport.cs
+++ b/src/GameCult.Networking/CultNetTransport.cs
@@ -2759,6 +2759,11 @@ namespace GameCult.Networking
// under SessionGate by whichever thread's send found the failure; the polling thread reads
// it when it retires the peer.
internal byte[]? UnsendableReason { get; set; }
+
+ // Set once the peer's capped frame has been answered: the session ends with this reason when the
+ // peer has acknowledged every answer, or at the deadline. Under SessionGate.
+ internal byte[]? EndingReason { get; set; }
+ internal long EndingDeadlineMs { get; set; }
}
///
@@ -2928,6 +2933,7 @@ namespace GameCult.Networking
public bool TryReceiveOnce(out CultNetRudpSocketServerFrame? delivered)
{
RetireUnsendablePeers();
+ EndAnsweredPeers();
if (_deliveredFrames.Count > 0)
{
delivered = _deliveredFrames.Dequeue();
@@ -3146,6 +3152,7 @@ namespace GameCult.Networking
public void PollResends()
{
RetireUnsendablePeers();
+ EndAnsweredPeers();
if (_socket.Poll(0, SelectMode.SelectRead))
return;
foreach (var peer in _peers.Values.ToArray())
@@ -3244,6 +3251,44 @@ namespace GameCult.Networking
return true;
}
+ // Polling thread only. The application answered a peer's capped frame: the answer is reliable, so a
+ // lost datagram is resent, and the session ends with `reason` once the peer has acknowledged every
+ // answer or `graceMs` have passed (EndAnsweredPeers). Frames of the peer still queued are not delivered.
+ internal void EndPeerSessionWhenAnswered(CultNetRudpSocketServerPeer peer, byte[] reason, long graceMs)
+ {
+ if (peer == null) throw new ArgumentNullException(nameof(peer));
+ if (reason == null) throw new ArgumentNullException(nameof(reason));
+ lock (peer.SessionGate)
+ {
+ if (peer.Session.Ended || peer.EndingReason != null)
+ return;
+ peer.EndingReason = reason;
+ peer.EndingDeadlineMs = NowMs() + graceMs;
+ }
+ var kept = _deliveredFrames.Where(frame => !ReferenceEquals(frame.Peer, peer)).ToArray();
+ _deliveredFrames.Clear();
+ foreach (var frame in kept)
+ _deliveredFrames.Enqueue(frame);
+ }
+
+ // Polling thread only.
+ private void EndAnsweredPeers()
+ {
+ foreach (var peer in _peers.Values.ToArray())
+ {
+ byte[]? reason;
+ lock (peer.SessionGate)
+ {
+ reason = peer.EndingReason != null
+ && (peer.Session.OutstandingReliablePacketCount == 0 || NowMs() >= peer.EndingDeadlineMs)
+ ? peer.EndingReason
+ : null;
+ }
+ if (reason != null)
+ EndPeerSession(peer, reason);
+ }
+ }
+
// 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
index a07b1771..e7900e7c 100644
--- a/tests/GameCult.Networking.Tests/CultNetRudpSchemaServerPeerInputTests.cs
+++ b/tests/GameCult.Networking.Tests/CultNetRudpSchemaServerPeerInputTests.cs
@@ -191,6 +191,95 @@ namespace GameCult.Networking.Tests
return errors;
}
+ // The bad peer acknowledges what it received and the server polls on.
+ private static async Task Acknowledge(Scene scene)
+ {
+ scene.Bad.Socket.SendTo(CultNetRudpPacketCodec.Encode(scene.Bad.Session.CreateAck()), scene.Server.LocalEndPoint);
+ await scene.Server.PollAvailableAsync(16);
+ }
+
+ // The bad peer has sent seven undecodable frames (BadThenGood sent the first), each answered.
+ private static async Task AtTheCap()
+ {
+ var scene = await BadThenGood(Malformed());
+ await scene.Server.PollAvailableAsync(16);
+ ErrorsReceivedBy(scene.Bad);
+ for (var i = 0; i < 6; i++)
+ {
+ SendSchema(scene.Server, scene.Bad, Malformed());
+ await scene.Server.PollAvailableAsync(16);
+ ErrorsReceivedBy(scene.Bad);
+ }
+ Assert.That(scene.Ended, Is.Empty);
+ return scene;
+ }
+
+ // The capped frame's answer is lost on the wire: the server resends it, and the goodbye follows only
+ // once the peer has acknowledged it.
+ [Test]
+ public async Task ALostCappedAnswerIsResentAndTheGoodbyeWaitsForItsAcknowledgement()
+ {
+ using var scene = await AtTheCap();
+ SendSchema(scene.Server, scene.Bad, Malformed());
+ await scene.Server.PollAvailableAsync(16);
+
+ var lost = Drain(scene.Bad.Socket);
+ Assert.That(lost.Where(p => p.PacketType == CultNetRudpPacketType.Data), Is.Not.Empty, "fixture: the answer was sent and is lost here");
+ Assert.That(lost.Select(p => p.PacketType), Has.None.EqualTo(CultNetRudpPacketType.Disconnect), "no goodbye before the answer is acknowledged");
+ Assert.That(scene.Ended, Is.Empty);
+
+ var resent = new List();
+ for (var attempt = 0; attempt < 40 && resent.Count == 0; attempt++)
+ {
+ await Task.Delay(50);
+ scene.Server.PollResends();
+ resent = Drain(scene.Bad.Socket);
+ }
+ Assert.That(resent.Select(p => p.PacketType), Has.None.EqualTo(CultNetRudpPacketType.Disconnect), "still no goodbye");
+ var answers = new List();
+ foreach (var packet in resent)
+ foreach (var frame in scene.Bad.Session.Receive(packet, 1).Delivered)
+ if (CultNetSchemaMessageSerialization.Deserialize(frame.Payload) is CultNetErrorMessage error)
+ answers.Add(error);
+ Assert.That(answers.Select(e => e.Code), Is.EqualTo(new[] { "message_invalid" }), "the lost answer arrived by resend");
+ Assert.That(scene.Ended, Is.Empty);
+
+ await Acknowledge(scene);
+ Assert.That(scene.Ended, Is.EqualTo(new[] { scene.Bad.EndPoint }), "acknowledged: the session ends");
+ Assert.That(Drain(scene.Bad.Socket).Select(p => p.PacketType), Has.Some.EqualTo(CultNetRudpPacketType.Disconnect));
+ }
+
+ // A peer that never acknowledges the answer is not kept for ever: the session ends after the grace.
+ [Test]
+ public async Task ACappedSessionWhoseAnswerIsNeverAcknowledgedEndsAfterTheGrace()
+ {
+ using var scene = await AtTheCap();
+ SendSchema(scene.Server, scene.Bad, Malformed());
+ await scene.Server.PollAvailableAsync(16);
+ Drain(scene.Bad.Socket);
+
+ await Task.Delay(1500); // the grace is two seconds
+ scene.Server.PollResends();
+ Assert.That(scene.Ended, Is.Empty, "inside the grace");
+ await Task.Delay(700);
+ scene.Server.PollResends();
+ Assert.That(scene.Ended, Is.EqualTo(new[] { scene.Bad.EndPoint }), "the grace ran out");
+ }
+
+ // Nothing behind the capped frame is served while its answer is delivered.
+ [Test]
+ public async Task NothingBehindTheCappedFrameIsServed()
+ {
+ using var scene = await AtTheCap();
+ SendSchema(scene.Server, scene.Bad, Malformed());
+ SendSchema(scene.Server, scene.Bad, CultNetSchemaMessageSerialization.Serialize(
+ new CultNetShardCatalogRequestMessage { MessageId = "behind-the-cap" }));
+ await scene.Server.PollAvailableAsync(16);
+ await scene.Server.PollAvailableAsync(16);
+
+ Assert.That(scene.Served.Select(s => s.MessageId), Does.Not.Contain("behind-the-cap"));
+ }
+
// The numbers are the shared cross-runtime cap, spelled out so changing the constant fails here.
[Test]
public async Task EightConsecutiveUndecodableFramesEndTheSessionAfterTheEighthIsAnswered()
@@ -221,6 +310,8 @@ namespace GameCult.Networking.Tests
await Refuse(1);
Assert.That(ErrorsReceivedBy(scene.Bad).Select(e => e.Code), Is.EqualTo(new[] { "message_invalid" }), "the eighth is answered too");
+ Assert.That(scene.Ended, Is.Empty, "the session ends once the answer is acknowledged");
+ await Acknowledge(scene);
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 }));
}