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

GameCult.Networking: an ended session fails its flush; a repeated Connect owes one Accept

Browse source 
FlushReliable throws for writes an ended session forgot. A server-mode
transport and the listener answer a Connect from the peer they already hold
with AnswerRepeatedConnect: the Accept still owed, or an Ack, queueing
nothing. The listener replaced the peer session on every Connect, reusing
sequences and never reporting PeerDisconnected; it now replaces only on a
different connect payload and reports the old peer, as the Rust hub does.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-29 20:41:23 +02:00
commit ca18dfac55
2 changed files with 166 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -1025,6 +1025,36 @@ namespace GameCult.Networking
return response;
}
/// <summary>
/// Answers a connect from a peer this session has already accepted. The packet's sequence is
/// remembered and nothing is queued, so a connect storm cannot grow the reliable queue. The reply
/// is the accept still awaiting acknowledgement, or an ack once it was acknowledged.
/// </summary>
public CultNetRudpPacket AnswerRepeatedConnect(CultNetRudpPacket packet, long nowMs = 0)
{
RequireConnection(packet);
if (packet.PacketType != CultNetRudpPacketType.Connect)
{
throw new InvalidOperationException($"Expected RUDP connect packet, got {packet.PacketType}.");
}
ApplyAcknowledgements(packet);
RememberReceived(packet.Sequence);
_lastReceivedAtMs = nowMs;
lock (_pendingReliableGate)
{
foreach (var pending in _pendingReliable.Values)
{
if (pending.Packet.PacketType != CultNetRudpPacketType.Accept)
continue;
pending.LastSentAtMs = nowMs;
return ClonePacket(pending.Packet);
}
}
return CreateAck();
}
/// <summary>
/// Creates a data packet.
/// </summary>
@ -1715,6 +1745,7 @@ namespace GameCult.Networking
public sealed class CultNetRudpSocketTransportConnection : IDisposable
{
private const int MinimumReceiveBufferBytes = 4 * 1024 * 1024;
private static readonly byte[] ReplacedPeerReason = Encoding.UTF8.GetBytes("replaced by a new Connect generation");
// One acknowledgement carries the preceding 32-packet receive mask.
// Pace at that transport window instead of sleeping after every fragment;
// a one-millisecond sleep can consume a full scheduler quantum on Windows.
@ -1810,6 +1841,7 @@ namespace GameCult.Networking
throw new InvalidOperationException("Only a client RUDP socket transport can initiate connect.");
}
DisconnectReason = null;
lock (_sessionGate)
SendPacket(_session.CreateConnect(NowMs(), payload ?? Array.Empty<byte>()));
}
@ -1890,6 +1922,12 @@ namespace GameCult.Networking
{
while (true)
{
// An ended session forgot its unacknowledged writes; reporting them
// flushed would be a lie.
var ended = DisconnectReason;
if (ended != null)
throw new InvalidOperationException(
"RUDP session ended before its reliable writes were acknowledged: " + Encoding.UTF8.GetString(ended));
lock (_sessionGate)
{
if (_session.OutstandingReliablePacketCount == 0)
@ -2186,7 +2224,17 @@ namespace GameCult.Networking
CultNetRudpPacket accept;
try
{
accept = _session.AcceptConnect(packet, NowMs());
// A connect from the peer this transport already accepted repeats: answer it
// with the accept already owed and queue nothing.
if (_session.Connected)
{
accept = _session.AnswerRepeatedConnect(packet, NowMs());
}
else
{
accept = _session.AcceptConnect(packet, NowMs());
DisconnectReason = null;
}
}
catch (InvalidOperationException)
{
@ -2390,12 +2438,14 @@ namespace GameCult.Networking
/// </summary>
public sealed class CultNetRudpSocketServerPeer
{
internal CultNetRudpSocketServerPeer(EndPoint remoteEndPoint, CultNetRudpSession session)
internal CultNetRudpSocketServerPeer(EndPoint remoteEndPoint, byte[] connectPayload, CultNetRudpSession session)
{
RemoteEndPoint = remoteEndPoint;
ConnectPayload = connectPayload;
Session = session;
}
internal byte[] ConnectPayload { get; }
internal CultNetRudpSession Session { get; }
internal object SessionGate { get; } = new object();
@ -2616,8 +2666,32 @@ namespace GameCult.Networking
var peerKey = RemoteKey(remote);
if (packet.PacketType == CultNetRudpPacketType.Connect)
{
var connectPayload = packet.Payload ?? Array.Empty<byte>();
_peers.TryGetValue(peerKey, out var admitted);
// The same payload from an admitted peer is a repeated Connect: answer it with the
// Accept already owed and queue nothing. A different payload is a new generation.
if (admitted != null && admitted.ConnectPayload.AsSpan().SequenceEqual(connectPayload))
{
lock (admitted.SessionGate)
{
CultNetRudpPacket reply;
try
{
reply = admitted.Session.AnswerRepeatedConnect(packet, NowMs());
}
catch (InvalidOperationException)
{
_stats.PacketsDropped++;
return true;
}
SendPacket(admitted.RemoteEndPoint, reply);
}
return true;
}
var peer = new CultNetRudpSocketServerPeer(
CloneEndPoint(remote),
connectPayload,
new CultNetRudpSession(new CultNetRudpSessionOptions
{
ConnectionId = _connectionId,
@ -2628,6 +2702,11 @@ namespace GameCult.Networking
_peers[peerKey] = peer;
lock (peer.SessionGate)
SendPacket(peer.RemoteEndPoint, peer.Session.AcceptConnect(packet, NowMs(), _acceptPayload));
if (admitted != null)
{
admitted.DisconnectReason = ReplacedPeerReason;
PeerDisconnected?.Invoke(admitted);
}
return true;
}

85
tests/GameCult.Networking.Tests/NetworkingTests.cs
View file

@ -1438,6 +1438,91 @@ namespace GameCult.Networking.Tests
Assert.That(DrainPackets(peerSocket), Is.Empty);
}
[Test]
public void RudpFlushReliable_FailsForAWriteTheEndedSessionForgot()
{
using var serverSocket = BindUdpSocket();
using var peerSocket = BindUdpSocket();
var serverEndPoint = serverSocket.LocalEndPoint!;
const uint connectionId = 0x10203058;
using var server = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions
{
RuntimeId = "csharp-rudp-server",
Socket = serverSocket,
Mode = CultNetRudpSocketMode.Server,
ConnectionId = connectionId
});
var peer = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = connectionId });
peerSocket.SendTo(CultNetRudpPacketCodec.Encode(peer.CreateConnect(0)), serverEndPoint);
server.ReceiveOnce();
peer.Receive(DrainPackets(peerSocket).Single(p => p.PacketType == CultNetRudpPacketType.Accept), 0);
server.Send("schema", Encoding.UTF8.GetBytes("never acknowledged"));
peerSocket.SendTo(CultNetRudpPacketCodec.Encode(PoisonedFrame(peer)), serverEndPoint);
Assert.That(server.ReceiveOnce(), Is.Null);
Assert.That(server.DisconnectReason, Is.Not.Null);
Assert.Throws<InvalidOperationException>(() => server.FlushReliable(TimeSpan.FromMilliseconds(200)));
}
[Test]
public void RudpServerMode_OwesOneReliableAcceptHoweverManyConnectsRepeat()
{
using var serverSocket = BindUdpSocket();
using var peerSocket = BindUdpSocket();
const uint connectionId = 0x10203059;
using var server = new CultNetRudpSocketTransportConnection(new CultNetRudpSocketTransportOptions
{
RuntimeId = "csharp-rudp-server",
Socket = serverSocket,
Mode = CultNetRudpSocketMode.Server,
ConnectionId = connectionId
});
var connect = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = connectionId }).CreateConnect(0);
for (var attempt = 0; attempt < 20; attempt++)
{
peerSocket.SendTo(CultNetRudpPacketCodec.Encode(connect), serverSocket.LocalEndPoint!);
server.ReceiveOnce();
}
var accepts = DrainPackets(peerSocket).Where(p => p.PacketType == CultNetRudpPacketType.Accept && p.Reliable);
Assert.That(accepts.Select(p => p.Sequence).Distinct().Count(), Is.EqualTo(1));
}
[Test]
public void RudpListener_OwesOneReliableAcceptPerGenerationAndReportsAReplacedPeer()
{
using var serverSocket = BindUdpSocket();
using var peerSocket = BindUdpSocket();
const uint connectionId = 0x1020305a;
using var server = new CultNetRudpSocketTransportServer(new CultNetRudpSocketTransportServerOptions
{
RuntimeId = "csharp-rudp-listener",
Socket = serverSocket,
ConnectionId = connectionId
});
var ended = new List<CultNetRudpSocketServerPeer>();
server.PeerDisconnected += ended.Add;
var connect = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = connectionId }).CreateConnect(0, Encoding.UTF8.GetBytes("join"));
for (var attempt = 0; attempt < 20; attempt++)
{
peerSocket.SendTo(CultNetRudpPacketCodec.Encode(connect), serverSocket.LocalEndPoint!);
server.ReceiveOnce();
}
var accepts = DrainPackets(peerSocket).Where(p => p.PacketType == CultNetRudpPacketType.Accept && p.Reliable);
Assert.That(accepts.Select(p => p.Sequence).Distinct().Count(), Is.EqualTo(1));
Assert.That(server.Peers, Has.Count.EqualTo(1));
Assert.That(ended, Is.Empty);
// A different payload from the same endpoint is a new generation, and the old one is reported.
var other = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = connectionId }).CreateConnect(0, Encoding.UTF8.GetBytes("rejoin"));
peerSocket.SendTo(CultNetRudpPacketCodec.Encode(other), serverSocket.LocalEndPoint!);
server.ReceiveOnce();
Assert.That(server.Peers, Has.Count.EqualTo(1));
Assert.That(ended, Has.Count.EqualTo(1));
}
[Test]
public void RudpConfiguration_RejectsLimitsThatCannotAdmitAConnect()
{