GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Ack Cut D D2 (C#): a send that can never succeed throws from a caller-directed send and ends only that peer's session inside a poll

Browse source 
The listener swallowed every send failure as a counted loss. A send now splits
into permanent (MessageSize, InvalidArgument, AddressFamilyNotSupported: the
datagram can never be sent as built) and transient (everything else, still a
counted loss). SendSchema throws the SocketException and queues nothing, with
the packets it created withdrawn and their sequences reissued; a permanent
failure after an earlier fragment left ends the session. Inside a poll the
failure ends that peer's session, PeerDisconnected carries a reason naming the
error, and the poll is never failed. Tests pin each path, the classification,
BytesSent after a healthy send and a zero failure count.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 12:19:46 +02:00
commit 6cf92329b6
2 changed files with 345 additions and 21 deletions
Download patch file Download diff file Expand all files Collapse all files

170
src/GameCult.Networking/CultNetTransport.cs
View file

@ -1459,6 +1459,53 @@ namespace GameCult.Networking
return CreateDisconnect(RefusedPacketReason);
}
/// <summary>
/// Whether a failed send can never succeed for the datagram as built, so resending it is
/// pointless: it is too large for a datagram (EMSGSIZE, WSAEMSGSIZE), its address is
/// malformed for the socket (EINVAL, WSAEINVAL), or belongs to another address family
/// (EAFNOSUPPORT, WSAEAFNOSUPPORT). Everything else (no route, full buffers, firewall drops,
/// a refusal reported for an earlier datagram, an interrupted call) is a property of the path
/// or the moment and may pass, so it is a lost datagram, not a reason to end a session.
/// </summary>
internal static bool IsPermanentSendError(SocketException error) =>
error.SocketErrorCode == SocketError.MessageSize ||
error.SocketErrorCode == SocketError.InvalidArgument ||
error.SocketErrorCode == SocketError.AddressFamilyNotSupported;
internal const string UnsendablePacketReason = "packet could not be sent";
/// <summary>
/// Ends a session that owes its peer a packet that can never be sent as built. Resending it
/// would fail forever, and dropping it would leave a reliable sequence the peer waits on for
/// good, so the session cannot be kept. The goodbye's reason names the error.
/// </summary>
internal CultNetRudpPacket EndUnsendable(SocketException error)
{
ResetPeerState();
return CreateDisconnect(Encoding.UTF8.GetBytes(UnsendablePacketReason + ": " + error.Message));
}
internal uint NextSequence => _nextSequence;
/// <summary>
/// Undoes a send that never reached the wire: every packet it created at or after
/// <paramref name="firstSequence"/>, pending or queued, leaves no trace and its sequences are
/// reissued, so the peer never sees a gap. Only valid while nothing else has been sent since.
/// </summary>
internal void WithdrawSend(uint firstSequence)
{
lock (_pendingReliableGate)
{
foreach (var sequence in _pendingReliable.Keys.Where(sequence => sequence >= firstSequence).ToArray())
_pendingReliable.Remove(sequence);
var kept = _queuedReliable.Where(packet => packet.Sequence < firstSequence).ToArray();
_queuedReliable.Clear();
foreach (var packet in kept)
_queuedReliable.Enqueue(packet);
}
_nextSequence = firstSequence;
}
/// <summary>
/// Creates a packet carrying a transport-level disconnect reason.
/// </summary>
@ -2733,8 +2780,25 @@ namespace GameCult.Networking
public void Send(CultNetRudpSocketServerPeer peer, string channelId, byte[] payload)
{
if (peer == null) throw new ArgumentNullException(nameof(peer));
SocketException? unsendable = null;
var failedIndex = 0;
lock (peer.SessionGate)
SendPackets(peer, peer.Session.SendMany(channelId, payload, ChannelSendOptions(channelId), _maxFragmentBytes));
{
var firstSequence = peer.Session.NextSequence;
var packets = peer.Session.SendMany(channelId, payload, ChannelSendOptions(channelId), _maxFragmentBytes);
unsendable = SendPackets(peer, packets, out failedIndex);
// A caller-directed send that can never succeed is the caller's error and queues
// nothing. Only when earlier fragments are already on the wire has the peer seen
// part of it, and then the session cannot be kept.
if (unsendable != null && failedIndex == 0)
peer.Session.WithdrawSend(firstSequence);
}
if (unsendable != null)
{
if (failedIndex > 0)
EndUnsendablePeer(peer, unsendable);
throw unsendable;
}
_stats.FramesSent++;
}
@ -2844,10 +2908,13 @@ namespace GameCult.Networking
// a new generation.
if (admitted != null)
{
var repeats = false;
SocketException? unsendableReply = null;
lock (admitted.SessionGate)
{
if (admitted.Session.ConnectRepeats(packet))
{
repeats = true;
CultNetRudpPacket reply;
try
{
@ -2858,10 +2925,15 @@ namespace GameCult.Networking
_stats.PacketsDropped++;
return true;
}
SendPacket(admitted.RemoteEndPoint, reply);
return true;
unsendableReply = SendPacket(admitted.RemoteEndPoint, reply);
}
}
if (repeats)
{
if (unsendableReply != null)
EndUnsendablePeer(admitted, unsendableReply);
return true;
}
}
var peer = new CultNetRudpSocketServerPeer(
@ -2874,9 +2946,16 @@ namespace GameCult.Networking
ResendDelayMs = _resendDelayMs,
MaxPendingReliablePackets = _maxPendingReliablePackets
}));
_peers[peerKey] = peer;
SocketException? unsendableAccept;
lock (peer.SessionGate)
SendPacket(peer.RemoteEndPoint, peer.Session.AcceptConnect(packet, NowMs(), _acceptPayload));
unsendableAccept = SendPacket(peer.RemoteEndPoint, peer.Session.AcceptConnect(packet, NowMs(), _acceptPayload));
if (unsendableAccept != null)
{
// The peer cannot be answered, so no session starts.
_stats.PacketsDropped++;
return true;
}
_peers[peerKey] = peer;
if (admitted != null)
{
admitted.DisconnectReason = ReplacedPeerReason;
@ -2893,6 +2972,7 @@ namespace GameCult.Networking
CultNetRudpReceiveResult? outcome;
CultNetRudpPacket? acknowledgement = null;
SocketException? unsendable = null;
lock (existingPeer.SessionGate)
{
try
@ -2915,8 +2995,8 @@ namespace GameCult.Networking
else
{
if (outcome.Reply != null)
SendPacket(existingPeer.RemoteEndPoint, outcome.Reply);
SendPackets(existingPeer, outcome.ReadyToSend);
unsendable = SendPacket(existingPeer.RemoteEndPoint, outcome.Reply);
unsendable ??= SendPackets(existingPeer, outcome.ReadyToSend, out _);
if (packet.PacketType == CultNetRudpPacketType.Data)
acknowledgement = existingPeer.Session.CreateAckForReceived(packet.Sequence);
}
@ -2953,8 +3033,10 @@ namespace GameCult.Networking
}
delivered = _deliveredFrames.Count > 0 ? _deliveredFrames.Dequeue() : null;
if (acknowledgement != null)
SendPacket(existingPeer.RemoteEndPoint, acknowledgement);
if (acknowledgement != null && unsendable == null)
unsendable = SendPacket(existingPeer.RemoteEndPoint, acknowledgement);
if (unsendable != null)
EndUnsendablePeer(existingPeer, unsendable);
return true;
}
@ -2968,19 +3050,48 @@ namespace GameCult.Networking
return;
foreach (var peer in _peers.Values.ToArray())
{
SocketException? unsendable;
lock (peer.SessionGate)
SendPackets(peer, peer.Session.DueResends(NowMs()));
unsendable = SendPackets(peer, peer.Session.DueResends(NowMs()), out _);
if (unsendable != null)
EndUnsendablePeer(peer, unsendable);
}
}
private void SendPackets(CultNetRudpSocketServerPeer peer, IReadOnlyList<CultNetRudpPacket> packets)
// Stops at the first packet that can never be sent as built and returns its error and
// index; a transient failure is a lost datagram and the loop goes on.
private SocketException? SendPackets(CultNetRudpSocketServerPeer peer, IReadOnlyList<CultNetRudpPacket> packets, out int failedIndex)
{
failedIndex = 0;
for (var index = 0; index < packets.Count; index++)
{
SendPacket(peer.RemoteEndPoint, packets[index]);
var unsendable = SendPacket(peer.RemoteEndPoint, packets[index]);
if (unsendable != null)
{
failedIndex = index;
return unsendable;
}
if ((index + 1) % WireBurstPackets == 0 && index + 1 < packets.Count)
Thread.Yield();
}
return null;
}
// An admitted peer owes a packet that can never be sent as built. Inside a poll that ends the
// peer's session, never the poll: the peer is told, and PeerDisconnected carries a reason that
// names the error.
private void EndUnsendablePeer(CultNetRudpSocketServerPeer peer, SocketException error)
{
var key = RemoteKey(peer.RemoteEndPoint);
if (!_peers.TryGetValue(key, out var current) || !ReferenceEquals(current, peer))
return;
_peers.Remove(key);
CultNetRudpPacket goodbye;
lock (peer.SessionGate)
goodbye = peer.Session.EndUnsendable(error);
peer.DisconnectReason = goodbye.Payload;
SendPacket(peer.RemoteEndPoint, goodbye);
PeerDisconnected?.Invoke(peer);
}
/// <inheritdoc />
@ -2995,30 +3106,51 @@ namespace GameCult.Networking
_socket.Dispose();
}
// A failed send is that peer's lost datagram: counted, never thrown. A reliable packet
// stays pending and is resent; an unreliable one was allowed to be lost. The peer's
// session ends by the rules that already end sessions. Only an encode failure, the
// listener's own bug, throws.
private void SendPacket(EndPoint remoteEndPoint, CultNetRudpPacket packet)
// A transient send failure is that peer's lost datagram: counted, never thrown. A reliable
// packet stays pending and is resent; an unreliable one was allowed to be lost. A permanent
// failure (CultNetRudpSession.IsPermanentSendError) is returned: the datagram can never be
// sent as built, and the caller decides between refusing the send and ending the peer's
// session. Only an encode failure, the listener's own bug, throws.
private SocketException? SendPacket(EndPoint remoteEndPoint, CultNetRudpPacket packet)
{
var wire = CultNetRudpPacketCodec.Encode(packet);
try
{
if (FailingSendPeers.Contains(remoteEndPoint))
throw new SocketException((int)SocketError.HostUnreachable);
if (UnsendableAfter.TryGetValue(remoteEndPoint, out var remaining))
{
if (remaining > 0)
{
UnsendableAfter[remoteEndPoint] = remaining - 1;
}
else
{
UnsendableAfter.Remove(remoteEndPoint);
throw new SocketException((int)SocketError.MessageSize);
}
}
var sent = _socket.SendTo(wire, remoteEndPoint);
_stats.BytesSent += sent;
}
catch (SocketException error) when (CultNetRudpSession.IsPermanentSendError(error))
{
return error;
}
catch (SocketException)
{
_stats.SendFailures++;
}
return null;
}
// Test seam: peers whose every datagram fails to send, standing in for an unroutable or
// full path that a loopback peer cannot be made to have.
// Test seams, standing in for paths a loopback peer cannot be made to have. Peers whose every
// datagram fails to send transiently:
internal HashSet<EndPoint> FailingSendPeers { get; } = new HashSet<EndPoint>();
// Peers whose next datagram after this many fails permanently, once.
internal Dictionary<EndPoint, int> UnsendableAfter { get; } = new Dictionary<EndPoint, int>();
private static string RemoteKey(EndPoint endpoint)
{
return endpoint.ToString() ?? string.Empty;

196
tests/GameCult.Networking.Tests/CultNetRudpSendFailureTests.cs
View file

@ -60,7 +60,7 @@ namespace GameCult.Networking.Tests
return peer;
}
private static (CultNetRudpSocketTransportServer Server, EndPoint ServerEndPoint, Peer X, Peer Y) TwoPeers()
private static (CultNetRudpSocketTransportServer Server, EndPoint ServerEndPoint, Peer X, Peer Y) TwoPeers(int? maxFragmentBytes = null)
{
var serverSocket = Bind();
var server = new CultNetRudpSocketTransportServer(new CultNetRudpSocketTransportServerOptions
@ -68,7 +68,8 @@ namespace GameCult.Networking.Tests
RuntimeId = "csharp-send-failure",
Socket = serverSocket,
ConnectionId = ConnectionId,
ResendDelayMs = 5
ResendDelayMs = 5,
MaxFragmentBytes = maxFragmentBytes
});
var x = Connect(server, serverSocket.LocalEndPoint!);
var y = Connect(server, serverSocket.LocalEndPoint!);
@ -130,5 +131,196 @@ namespace GameCult.Networking.Tests
Assert.That(server.Peers, Has.Count.EqualTo(2));
}
}
// A permanent send failure is one that can never succeed for the datagram as built. A
// caller-directed send throws it and queues nothing; inside a poll it ends that peer's
// session and never the poll.
private static bool IsUnsendableReason(byte[]? reason)
{
var text = Encoding.UTF8.GetString(reason ?? Array.Empty<byte>());
return text.StartsWith("packet could not be sent: ", StringComparison.Ordinal)
&& text.Length > "packet could not be sent: ".Length;
}
private static List<CultNetRudpSocketServerPeer> RecordDisconnects(CultNetRudpSocketTransportServer server)
{
var ended = new List<CultNetRudpSocketServerPeer>();
server.PeerDisconnected += ended.Add;
return ended;
}
[Test]
public void AnOversizedSendThrowsTheErrorAndQueuesNothing()
{
var (server, _, x, y) = TwoPeers();
using (server)
{
var error = Assert.Throws<SocketException>(() => server.SendSchema(x.Server, new byte[70_000]))!;
Assert.That(CultNetRudpSession.IsPermanentSendError(error), Is.True, error.SocketErrorCode.ToString());
Assert.That(server.Stats.SendFailures, Is.Zero);
Assert.That(server.Stats.FramesSent, Is.Zero);
Assert.That(server.Peers, Has.Count.EqualTo(2), "the caller's error ends no session");
Assert.That(x.Server.Session.OutstandingReliablePacketCount, Is.Zero, "nothing is pending or queued to resend");
// The refused send consumed no sequence: the next frame is delivered in order instead
// of waiting behind a gap that never fills.
server.SendSchema(x.Server, "next");
var data = Drain(x.Socket).Single(p => p.PacketType == CultNetRudpPacketType.Data);
var delivered = x.Session.Receive(data, 1).Delivered;
Assert.That(delivered.Select(frame => Encoding.UTF8.GetString(frame.Payload)), Is.EqualTo(new[] { "next" }));
Assert.That(y.Server.Session.OutstandingReliablePacketCount, Is.Zero);
}
}
[Test]
public void ASendThatFailsPermanentlyAfterAFragmentLeftEndsTheSession()
{
var (server, _, x, _) = TwoPeers(maxFragmentBytes: 1000);
using (server)
{
var ended = RecordDisconnects(server);
server.UnsendableAfter[x.EndPoint] = 1;
var error = Assert.Throws<SocketException>(() => server.SendSchema(x.Server, new byte[3000]))!;
Assert.That(CultNetRudpSession.IsPermanentSendError(error), Is.True);
Assert.That(server.Peers, Has.Count.EqualTo(1));
Assert.That(ended, Is.EqualTo(new[] { x.Server }));
Assert.That(IsUnsendableReason(x.Server.DisconnectReason), Is.True);
}
}
[Test]
public void APermanentFailureInResendsEndsOnlyThatPeersSession()
{
var (server, _, x, y) = TwoPeers();
using (server)
{
var ended = RecordDisconnects(server);
server.FailingSendPeers.Add(x.EndPoint);
server.FailingSendPeers.Add(y.EndPoint);
server.SendSchema(x.Server, "for-x");
server.SendSchema(y.Server, "for-y");
// X's datagram can now never be sent as built; Y's path recovers. X is visited first.
server.FailingSendPeers.Clear();
server.UnsendableAfter[x.EndPoint] = 0;
Thread.Sleep(30);
Assert.DoesNotThrow(() => server.PollResends());
Assert.That(server.Peers, Is.EqualTo(new[] { y.Server }));
Assert.That(ended, Is.EqualTo(new[] { x.Server }));
Assert.That(IsUnsendableReason(x.Server.DisconnectReason), Is.True);
var resent = Drain(y.Socket).Where(p => p.PacketType == CultNetRudpPacketType.Data).ToList();
Assert.That(resent, Has.Count.EqualTo(1));
Assert.That(Encoding.UTF8.GetString(resent[0].Payload!), Is.EqualTo("for-y"));
var goodbye = Drain(x.Socket).Single(p => p.PacketType == CultNetRudpPacketType.Disconnect);
Assert.That(IsUnsendableReason(goodbye.Payload), Is.True);
}
}
[Test]
public void APermanentFailureAcknowledgingAFrameEndsTheSessionAfterTheFrameIsDelivered()
{
var (server, listenerEndPoint, x, _) = TwoPeers();
using (server)
{
var ended = RecordDisconnects(server);
server.UnsendableAfter[x.EndPoint] = 0;
x.Socket.SendTo(CultNetRudpPacketCodec.Encode(ReliableFrame(x, "from-x")), listenerEndPoint);
CultNetRudpSocketServerFrame? frame = null;
Assert.DoesNotThrow(() => frame = server.ReceiveOnce());
Assert.That(frame, Is.Not.Null, "the frame is delivered before the session ends");
Assert.That(Encoding.UTF8.GetString(frame!.Frame.Payload), Is.EqualTo("from-x"));
Assert.That(ended, Is.EqualTo(new[] { x.Server }));
Assert.That(IsUnsendableReason(x.Server.DisconnectReason), Is.True);
Assert.That(server.Peers, Has.Count.EqualTo(1));
Assert.That(server.Stats.SendFailures, Is.Zero);
}
}
[Test]
public void APermanentFailureAnsweringARepeatedConnectEndsTheSession()
{
var serverSocket = Bind();
using var server = new CultNetRudpSocketTransportServer(new CultNetRudpSocketTransportServerOptions
{
RuntimeId = "csharp-send-failure",
Socket = serverSocket,
ConnectionId = ConnectionId,
ResendDelayMs = 5
});
var ended = RecordDisconnects(server);
var peer = Bind();
var connect = CultNetRudpPacketCodec.Encode(
new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = ConnectionId }).CreateConnect(0));
peer.SendTo(connect, serverSocket.LocalEndPoint!);
server.ReceiveOnce();
Assert.That(server.Peers, Has.Count.EqualTo(1));
var admitted = server.Peers.Single();
server.UnsendableAfter[peer.LocalEndPoint!] = 0;
peer.SendTo(connect, serverSocket.LocalEndPoint!);
Assert.DoesNotThrow(() => server.ReceiveOnce());
Assert.That(server.Peers, Is.Empty);
Assert.That(ended, Is.EqualTo(new[] { admitted }));
Assert.That(IsUnsendableReason(admitted.DisconnectReason), Is.True);
}
[Test]
public void AConnectWhoseAcceptCanNeverBeSentStartsNoSession()
{
var serverSocket = Bind();
using var server = new CultNetRudpSocketTransportServer(new CultNetRudpSocketTransportServerOptions
{
RuntimeId = "csharp-send-failure",
Socket = serverSocket,
ConnectionId = ConnectionId,
ResendDelayMs = 5
});
var peer = Bind();
server.UnsendableAfter[peer.LocalEndPoint!] = 0;
peer.SendTo(
CultNetRudpPacketCodec.Encode(new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = ConnectionId }).CreateConnect(0)),
serverSocket.LocalEndPoint!);
Assert.DoesNotThrow(() => server.ReceiveOnce());
Assert.That(server.Peers, Is.Empty);
Assert.That(server.Stats.PacketsDropped, Is.EqualTo(1));
}
[Test]
public void AHealthySendCountsItsBytesAndNoFailure()
{
var (server, _, x, _) = TwoPeers();
using (server)
{
var before = server.Stats.BytesSent;
server.SendSchema(x.Server, "hello");
var received = Drain(x.Socket).Sum(p => CultNetRudpPacketCodec.Encode(p).Length);
Assert.That(received, Is.GreaterThan("hello".Length));
Assert.That(server.Stats.BytesSent - before, Is.EqualTo(received));
Assert.That(server.Stats.SendFailures, Is.Zero);
}
}
[Test]
public void OnlyADatagramThatCanNeverBeSentIsAPermanentFailure()
{
foreach (var permanent in new[] { SocketError.MessageSize, SocketError.InvalidArgument, SocketError.AddressFamilyNotSupported })
Assert.That(CultNetRudpSession.IsPermanentSendError(new SocketException((int)permanent)), Is.True, permanent.ToString());
foreach (var transient in new[]
{
SocketError.HostUnreachable, SocketError.NetworkUnreachable, SocketError.NoBufferSpaceAvailable,
SocketError.WouldBlock, SocketError.ConnectionReset, SocketError.ConnectionRefused,
SocketError.AccessDenied, SocketError.NetworkDown, SocketError.AddressNotAvailable,
SocketError.Interrupted, SocketError.TimedOut, SocketError.Shutdown
})
Assert.That(CultNetRudpSession.IsPermanentSendError(new SocketException((int)transient)), Is.False, transient.ToString());
}
}
}