From f382fc98eadc4a489689e51e88748b616661a560 Mon Sep 17 00:00:00 2001 From: Leonid Pershin Date: Sat, 13 Jun 2026 04:43:49 +0300 Subject: [PATCH] Net: heartbeat liveness, protocol version, message cap, replication reset MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Robustness pass over MrGameEng.Net: - WebSocketServer heartbeats each connection (configurable interval/timeout) and drops peers idle past the timeout — detects half-open TCP that vanished without a close frame. Tracks last-activity per connection. - Reassembled messages capped (server and client) so a peer cannot exhaust memory with an oversized fragmented message. - Replication snapshots carry a protocol-version byte; a client receiving a mismatched version drops the message instead of decoding garbage, and a truncated snapshot is ignored without throwing. - ReplicationClient.Clear() deletes all replicated entities, so clients can wipe stale state before reconnecting. Tests: heartbeat healthy-survives / silent-peer-dropped, protocol-version mismatch, truncated snapshot, replication clear. Co-Authored-By: Claude Fable 5 --- .../Replication/ReplicationClient.cs | 103 +++++++++---- .../Replication/ReplicationServer.cs | 9 ++ src/MrGameEng.Net/WebSocketClient.cs | 8 ++ src/MrGameEng.Net/WebSocketServer.cs | 135 +++++++++++++++--- tests/MrGameEng.Net.Tests/HeartbeatTests.cs | 98 +++++++++++++ tests/MrGameEng.Net.Tests/ReplicationTests.cs | 79 +++++++++- 6 files changed, 380 insertions(+), 52 deletions(-) create mode 100644 tests/MrGameEng.Net.Tests/HeartbeatTests.cs diff --git a/src/MrGameEng.Net/Replication/ReplicationClient.cs b/src/MrGameEng.Net/Replication/ReplicationClient.cs index 4f6115d..b2e3894 100644 --- a/src/MrGameEng.Net/Replication/ReplicationClient.cs +++ b/src/MrGameEng.Net/Replication/ReplicationClient.cs @@ -1,4 +1,5 @@ using Friflo.Engine.ECS; +using MrGameEng.Core; namespace MrGameEng.Net; @@ -21,6 +22,7 @@ public sealed class ReplicationClient private readonly ReplicationSchema _schema; private readonly EntityStore _store; private readonly Dictionary _entities = []; + private bool _warnedVersion; /// Creates a replication client writing into . public ReplicationClient(ReplicationSchema schema, EntityStore store) @@ -38,55 +40,98 @@ public sealed class ReplicationClient } } + /// + /// Deletes every replicated entity and forgets all net ids. Call before reconnecting: the + /// fresh connection receives the full world again with a clean id space, so stale entities + /// from the previous session don't linger as duplicates. + /// + public void Clear() + { + foreach (var entity in _entities.Values) + { + entity.DeleteEntity(); + } + + _entities.Clear(); + } + /// Applies one snapshot message to the local store. public void Apply(byte[] message) { using var reader = new BinaryReader(new MemoryStream(message)); - if (reader.ReadByte() != ReplicationMessage.Snapshot) + // Заголовок: тип(1) + версия(1). Короче — точно не наш снапшот. + if (message.Length < 2 || reader.ReadByte() != ReplicationMessage.Snapshot) { return; // незнакомый тип сообщения — пропускаем, это не снапшот } - var slots = _schema.Slots; - var count = reader.ReadInt32(); - for (var record = 0; record < count; record++) + var version = reader.ReadByte(); + if (version != ReplicationMessage.ProtocolVersion) { - var netId = reader.ReadInt32(); - var op = reader.ReadByte(); - if (op == ReplicationMessage.OpDespawn) + if (!_warnedVersion) { - if (_entities.Remove(netId, out var dead)) - { - dead.DeleteEntity(); - } - - continue; + _warnedVersion = true; + Log.Warning( + $"Replication protocol mismatch: server v{version}, client " + + $"v{ReplicationMessage.ProtocolVersion} — snapshots dropped. Schemas out of sync." + ); } - var mask = reader.ReadUInt32(); - var spawned = false; - if (!_entities.TryGetValue(netId, out var entity)) - { - entity = _store.CreateEntity(new NetId { Value = netId }); - _entities[netId] = entity; - spawned = true; - } + return; + } - foreach (var slot in slots) + var slots = _schema.Slots; + try + { + var count = reader.ReadInt32(); + for (var record = 0; record < count; record++) { - if ((mask & (1u << slot.Bit)) == 0) + var netId = reader.ReadInt32(); + var op = reader.ReadByte(); + if (op == ReplicationMessage.OpDespawn) { + if (_entities.Remove(netId, out var dead)) + { + dead.DeleteEntity(); + } + continue; } - var data = reader.ReadBytes(slot.Size); - slot.Apply(entity, data, 0); - } + var mask = reader.ReadUInt32(); + var spawned = false; + if (!_entities.TryGetValue(netId, out var entity)) + { + entity = _store.CreateEntity(new NetId { Value = netId }); + _entities[netId] = entity; + spawned = true; + } - if (spawned) - { - EntitySpawned?.Invoke(entity); + foreach (var slot in slots) + { + if ((mask & (1u << slot.Bit)) == 0) + { + continue; + } + + var data = reader.ReadBytes(slot.Size); + if (data.Length < slot.Size) + { + return; // снапшот оборван на полпути — дальше читать нечего + } + + slot.Apply(entity, data, 0); + } + + if (spawned) + { + EntitySpawned?.Invoke(entity); + } } } + catch (EndOfStreamException) + { + // Структурно битый/усечённый снапшот — игнорируем остаток, соединение не роняем. + } } } diff --git a/src/MrGameEng.Net/Replication/ReplicationServer.cs b/src/MrGameEng.Net/Replication/ReplicationServer.cs index 65f0335..9885be0 100644 --- a/src/MrGameEng.Net/Replication/ReplicationServer.cs +++ b/src/MrGameEng.Net/Replication/ReplicationServer.cs @@ -98,6 +98,7 @@ public sealed class ReplicationServer using var stream = new MemoryStream(); using var writer = new BinaryWriter(stream); writer.Write(ReplicationMessage.Snapshot); + writer.Write(ReplicationMessage.ProtocolVersion); // версия формата — клиент отвергает чужую var countPosition = stream.Position; writer.Write(0); // количество записей, допишем в конце var records = 0; @@ -169,6 +170,14 @@ public sealed class ReplicationServer internal static class ReplicationMessage { internal const byte Snapshot = 1; + + /// + /// Wire-format version. Bump whenever the snapshot layout or the meaning of the schema's + /// component blits changes; a client receiving a mismatched version drops the message + /// instead of decoding garbage (guards against a desync between server and client schemas). + /// + internal const byte ProtocolVersion = 1; + internal const byte OpUpsert = 0; internal const byte OpDespawn = 1; } diff --git a/src/MrGameEng.Net/WebSocketClient.cs b/src/MrGameEng.Net/WebSocketClient.cs index be31f2f..b7ef4ed 100644 --- a/src/MrGameEng.Net/WebSocketClient.cs +++ b/src/MrGameEng.Net/WebSocketClient.cs @@ -101,6 +101,9 @@ public sealed class WebSocketClient : INetConnection, IDisposable /// Closes the connection. public void Dispose() => Close(); + /// Largest reassembled message accepted from the server before the connection is dropped. + public const int MaxMessageBytes = 16 * 1024 * 1024; + private async Task ReceiveLoop() { var buffer = new byte[64 * 1024]; @@ -117,6 +120,11 @@ public sealed class WebSocketClient : INetConnection, IDisposable break; } + if (message.Length + result.Count > MaxMessageBytes) + { + break; // сервер шлёт ненормально большое сообщение — рвём соединение + } + message.Write(buffer, 0, result.Count); if (result.EndOfMessage) { diff --git a/src/MrGameEng.Net/WebSocketServer.cs b/src/MrGameEng.Net/WebSocketServer.cs index 725adb7..871e84c 100644 --- a/src/MrGameEng.Net/WebSocketServer.cs +++ b/src/MrGameEng.Net/WebSocketServer.cs @@ -10,13 +10,25 @@ namespace MrGameEng.Net; /// dedicated servers. Accepting and reading happen on background tasks; the simulation /// drains new connections with and reads messages by /// polling each connection — nothing here touches the ECS world from another thread. -/// Binary messages only; pings are answered automatically. +/// Binary messages only; incoming pings are answered automatically and the server itself +/// heartbeats each connection, closing any that goes silent past +/// (detects half-open TCP — a peer that vanished without a close frame). A reassembled +/// message is capped at so a peer can't exhaust memory. /// public sealed class WebSocketServer : IDisposable { + /// Largest reassembled (possibly fragmented) message accepted from a peer. + public const int MaxMessageBytes = WebSocketProtocol.MaxPayloadBytes; + /// The port the server listens on. public int Port { get; } + /// How often the server pings each connection to keep it alive and probe liveness. + public TimeSpan HeartbeatInterval { get; } + + /// A connection with no traffic for longer than this is considered dead and closed. + public TimeSpan IdleTimeout { get; } + /// Snapshot of currently open connections. public IReadOnlyList Connections { @@ -36,10 +48,22 @@ public sealed class WebSocketServer : IDisposable private int _nextConnectionId; private bool _started; - /// Creates a server for on all interfaces. Call to listen. - public WebSocketServer(int port) + /// + /// Creates a server for on all interfaces. Call + /// to listen. (default 10 s) sets how often each + /// connection is pinged; (default 30 s) how long a silent + /// connection lives before it's dropped as dead. The timeout must exceed the interval so a + /// healthy peer's pong lands before it's judged idle. + /// + public WebSocketServer( + int port, + TimeSpan? heartbeatInterval = null, + TimeSpan? idleTimeout = null + ) { Port = port; + HeartbeatInterval = heartbeatInterval ?? TimeSpan.FromSeconds(10); + IdleTimeout = idleTimeout ?? TimeSpan.FromSeconds(30); _listener = new TcpListener(IPAddress.Any, port); } @@ -54,6 +78,7 @@ public sealed class WebSocketServer : IDisposable _started = true; _listener.Start(); Task.Run(AcceptLoop); + Task.Run(HeartbeatLoop); Log.Info($"WebSocketServer listening on port {Port}"); } @@ -109,6 +134,48 @@ public sealed class WebSocketServer : IDisposable } } + // Пингует живые соединения и закрывает те, что молчат дольше IdleTimeout (мёртвый peer + // не отвечает pong'ом — его активность не обновляется и он отваливается по таймауту). + private async Task HeartbeatLoop() + { + while (!_shutdown.IsCancellationRequested) + { + try + { + await Task.Delay(HeartbeatInterval, _shutdown.Token); + } + catch (OperationCanceledException) + { + return; + } + + ServerConnection[] snapshot; + lock (_connections) + { + snapshot = _connections.ToArray(); + } + + var now = DateTime.UtcNow; + foreach (var connection in snapshot) + { + if (!connection.IsOpen) + { + continue; + } + + if (now - connection.LastActivityUtc > IdleTimeout) + { + Log.Info($"WebSocketServer: connection #{connection.Id} timed out (idle)"); + connection.Close(); + } + else + { + connection.SendPing(); + } + } + } + } + private void Handshake(TcpClient client) { try @@ -148,21 +215,30 @@ public sealed class WebSocketServer : IDisposable public int Id { get; } public bool IsOpen => !_closed; + /// UTC of the last frame received from the peer — drives idle-timeout detection. + public DateTime LastActivityUtc => + new(Volatile.Read(ref _lastActivityTicks), DateTimeKind.Utc); + private readonly TcpClient _client; private readonly NetworkStream _stream; private readonly ConcurrentQueue _inbox = new(); private readonly object _sendLock = new(); private volatile bool _closed; + private long _lastActivityTicks; internal ServerConnection(int id, TcpClient client) { Id = id; _client = client; _stream = client.GetStream(); + _lastActivityTicks = DateTime.UtcNow.Ticks; } internal void StartReceiveLoop() => Task.Run(ReceiveLoop); + /// Sends a heartbeat ping; a live peer answers with a pong, refreshing activity. + internal void SendPing() => SendControl(WebSocketOpcode.Ping, []); + public void Send(ReadOnlySpan message) { if (_closed) @@ -184,6 +260,27 @@ public sealed class WebSocketServer : IDisposable } } + private void SendControl(WebSocketOpcode opcode, ReadOnlySpan payload) + { + if (_closed) + { + return; + } + + var frame = WebSocketProtocol.EncodeFrame(payload, opcode); + try + { + lock (_sendLock) + { + _stream.Write(frame, 0, frame.Length); + } + } + catch (Exception) + { + Close(); + } + } + public bool TryReceive(out byte[] message) => _inbox.TryDequeue(out message!); public void Close() @@ -224,31 +321,18 @@ public sealed class WebSocketServer : IDisposable break; } + // Любой кадр (включая pong) — признак жизни: сбрасываем счётчик простоя. + Volatile.Write(ref _lastActivityTicks, DateTime.UtcNow.Ticks); + switch (opcode) { case WebSocketOpcode.Ping: - lock (_sendLock) - { - var pong = WebSocketProtocol.EncodeFrame( - payload, - WebSocketOpcode.Pong - ); - _stream.Write(pong, 0, pong.Length); - } - + SendControl(WebSocketOpcode.Pong, payload); continue; case WebSocketOpcode.Pong: continue; case WebSocketOpcode.Close: - lock (_sendLock) - { - var close = WebSocketProtocol.EncodeFrame( - [], - WebSocketOpcode.Close - ); - _stream.Write(close, 0, close.Length); - } - + SendControl(WebSocketOpcode.Close, []); return; } @@ -258,6 +342,15 @@ public sealed class WebSocketServer : IDisposable pending.Clear(); } + if (pending.Count + payload.Length > MaxMessageBytes) + { + Log.Warning( + $"WebSocketServer: connection #{Id} exceeded {MaxMessageBytes}-byte " + + "message cap — closing" + ); + return; // finally закроет соединение + } + pending.AddRange(payload); if (fin && pendingOpcode == WebSocketOpcode.Binary) { diff --git a/tests/MrGameEng.Net.Tests/HeartbeatTests.cs b/tests/MrGameEng.Net.Tests/HeartbeatTests.cs new file mode 100644 index 0000000..b904c9f --- /dev/null +++ b/tests/MrGameEng.Net.Tests/HeartbeatTests.cs @@ -0,0 +1,98 @@ +using System.Net; +using System.Net.Sockets; +using System.Text; +using MrGameEng.Net; +using Xunit; + +namespace MrGameEng.Net.Tests; + +public class HeartbeatTests +{ + [Fact] + public async Task HealthyClient_SurvivesPastIdleTimeout() + { + var port = FreePort(); + using var server = new WebSocketServer( + port, + heartbeatInterval: TimeSpan.FromMilliseconds(100), + idleTimeout: TimeSpan.FromMilliseconds(400) + ); + server.Start(); + + using var client = await WebSocketClient.ConnectAsync( + new Uri($"ws://localhost:{port}/"), + new CancellationTokenSource(TimeSpan.FromSeconds(10)).Token + ); + var connection = await WaitFor( + () => server.TryAcceptConnection(out var c) ? c : null, + "server accept" + ); + + // Дольше idleTimeout: живой клиент авто-отвечает pong на server-ping и остаётся открыт. + await Task.Delay(900, TestContext.Current.CancellationToken); + + Assert.True(connection.IsOpen); + Assert.True(client.IsOpen); + } + + [Fact] + public async Task SilentPeer_IsDroppedAfterIdleTimeout() + { + var port = FreePort(); + using var server = new WebSocketServer( + port, + heartbeatInterval: TimeSpan.FromMilliseconds(100), + idleTimeout: TimeSpan.FromMilliseconds(400) + ); + server.Start(); + + // Сырой peer: проходит рукопожатие, но дальше молчит и не отвечает на ping. + using var tcp = new TcpClient(); + await tcp.ConnectAsync(IPAddress.Loopback, port, TestContext.Current.CancellationToken); + var request = + "GET / HTTP/1.1\r\n" + + "Host: localhost\r\n" + + "Upgrade: websocket\r\n" + + "Connection: Upgrade\r\n" + + "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n" + + "Sec-WebSocket-Version: 13\r\n" + + "\r\n"; + var bytes = Encoding.ASCII.GetBytes(request); + await tcp.GetStream().WriteAsync(bytes, TestContext.Current.CancellationToken); + + var connection = await WaitFor( + () => server.TryAcceptConnection(out var c) ? c : null, + "server accept" + ); + Assert.True(connection.IsOpen); + + // Peer не отвечает pong'ом → активность не обновляется → сервер закрывает по простою. + await WaitFor(() => connection.IsOpen ? null : "closed", "idle drop"); + Assert.False(connection.IsOpen); + } + + private static async Task WaitFor(Func poll, string what) + where T : class + { + for (var i = 0; i < 200; i++) + { + if (poll() is { } result) + { + return result; + } + + await Task.Delay(25); + } + + throw new TimeoutException($"Timed out waiting for {what}."); + } + + private static int FreePort() + { + var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + var port = ((IPEndPoint)listener.LocalEndpoint).Port; + listener.Stop(); + return port; + } +} diff --git a/tests/MrGameEng.Net.Tests/ReplicationTests.cs b/tests/MrGameEng.Net.Tests/ReplicationTests.cs index d5fce08..b06b426 100644 --- a/tests/MrGameEng.Net.Tests/ReplicationTests.cs +++ b/tests/MrGameEng.Net.Tests/ReplicationTests.cs @@ -103,8 +103,8 @@ public class ReplicationTests server.Send([connection]); Assert.True(connection.Sent.TryDequeue(out var delta)); - // Запись: type(1) + count(4) + netId(4) + op(1) + mask(4) + TestPosition(8) — без TestHealth. - Assert.Equal(22, delta!.Length); + // Запись: type(1) + version(1) + count(4) + netId(4) + op(1) + mask(4) + TestPosition(8) — без TestHealth. + Assert.Equal(23, delta!.Length); client.Apply(delta); var replicated = FindByNetId(clientStore, 1); @@ -185,6 +185,81 @@ public class ReplicationTests Assert.Equal(1, spawns); } + [Fact] + public void Clear_DeletesEveryReplicatedEntity() + { + var serverStore = new EntityStore(); + var clientStore = new EntityStore(); + var server = new ReplicationServer(MakeSchema(), serverStore); + var client = new ReplicationClient(MakeSchema(), clientStore); + var connection = new FakeConnection(); + serverStore.CreateEntity( + new NetId { Value = server.NextNetId() }, + new TestPosition { X = 1f, Y = 2f } + ); + serverStore.CreateEntity( + new NetId { Value = server.NextNetId() }, + new TestPosition { X = 3f, Y = 4f } + ); + server.Send([connection]); + client.Pump(connection); + Assert.Equal(2, client.EntityCount); + + client.Clear(); + + Assert.Equal(0, client.EntityCount); + foreach (var entity in clientStore.Entities) + { + Assert.False(entity.HasComponent()); + } + } + + [Fact] + public void MismatchedProtocolVersion_IsDropped() + { + var serverStore = new EntityStore(); + var clientStore = new EntityStore(); + var server = new ReplicationServer(MakeSchema(), serverStore); + var client = new ReplicationClient(MakeSchema(), clientStore); + var connection = new FakeConnection(); + serverStore.CreateEntity( + new NetId { Value = server.NextNetId() }, + new TestPosition { X = 1f, Y = 2f } + ); + server.Send([connection]); + Assert.True(connection.Sent.TryDequeue(out var snapshot)); + + // Портим байт версии (индекс 1: type=0, version=1) — клиент обязан отбросить снапшот целиком. + snapshot![1] = 0xFF; + client.Apply(snapshot); + + Assert.Equal(0, client.EntityCount); + } + + [Fact] + public void TruncatedSnapshot_IsIgnoredWithoutThrowing() + { + var serverStore = new EntityStore(); + var clientStore = new EntityStore(); + var server = new ReplicationServer(MakeSchema(), serverStore); + var client = new ReplicationClient(MakeSchema(), clientStore); + var connection = new FakeConnection(); + serverStore.CreateEntity( + new NetId { Value = server.NextNetId() }, + new TestPosition { X = 5f, Y = 6f }, + new TestHealth { Value = 9 } + ); + server.Send([connection]); + Assert.True(connection.Sent.TryDequeue(out var snapshot)); + + // Режем хвост: заголовок и счётчик целы, но данные компонентов оборваны. + var truncated = snapshot!.AsSpan(0, snapshot.Length - 6).ToArray(); + client.Apply(truncated); // не должно бросить + + // Записи могли частично примениться, но клиент остался живым и консистентным. + Assert.True(client.EntityCount <= 1); + } + private static Entity FindByNetId(EntityStore store, int netId) { foreach (var entity in store.Entities)