Net: heartbeat liveness, protocol version, message cap, replication reset

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 <noreply@anthropic.com>

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<int, Entity> _entities = [];
+    private bool _warnedVersion;
 
     /// <summary>Creates a replication client writing into <paramref name="store"/>.</summary>
     public ReplicationClient(ReplicationSchema schema, EntityStore store)
@@ -38,55 +40,98 @@ public sealed class ReplicationClient
         }
     }
 
+    /// <summary>
+    /// 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.
+    /// </summary>
+    public void Clear()
+    {
+        foreach (var entity in _entities.Values)
+        {
+            entity.DeleteEntity();
+        }
+
+        _entities.Clear();
+    }
+
     /// <summary>Applies one snapshot message to the local store.</summary>
     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)
+        {
+            // Структурно битый/усечённый снапшот — игнорируем остаток, соединение не роняем.
+        }
     }
 }

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;
+
+    /// <summary>
+    /// 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).
+    /// </summary>
+    internal const byte ProtocolVersion = 1;
+
     internal const byte OpUpsert = 0;
     internal const byte OpDespawn = 1;
 }

src/MrGameEng.Net/WebSocketClient.cs

@@ -101,6 +101,9 @@ public sealed class WebSocketClient : INetConnection, IDisposable
     /// <summary>Closes the connection.</summary>
     public void Dispose() => Close();
 
+    /// <summary>Largest reassembled message accepted from the server before the connection is dropped.</summary>
+    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)
                 {

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 <see cref="TryAcceptConnection"/> 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 <see cref="IdleTimeout"/>
+/// (detects half-open TCP — a peer that vanished without a close frame). A reassembled
+/// message is capped at <see cref="MaxMessageBytes"/> so a peer can't exhaust memory.
 /// </summary>
 public sealed class WebSocketServer : IDisposable
 {
+    /// <summary>Largest reassembled (possibly fragmented) message accepted from a peer.</summary>
+    public const int MaxMessageBytes = WebSocketProtocol.MaxPayloadBytes;
+
     /// <summary>The port the server listens on.</summary>
     public int Port { get; }
 
+    /// <summary>How often the server pings each connection to keep it alive and probe liveness.</summary>
+    public TimeSpan HeartbeatInterval { get; }
+
+    /// <summary>A connection with no traffic for longer than this is considered dead and closed.</summary>
+    public TimeSpan IdleTimeout { get; }
+
     /// <summary>Snapshot of currently open connections.</summary>
     public IReadOnlyList<INetConnection> Connections
     {
@@ -36,10 +48,22 @@ public sealed class WebSocketServer : IDisposable
     private int _nextConnectionId;
     private bool _started;
 
-    /// <summary>Creates a server for <paramref name="port"/> on all interfaces. Call <see cref="Start"/> to listen.</summary>
-    public WebSocketServer(int port)
+    /// <summary>
+    /// Creates a server for <paramref name="port"/> on all interfaces. Call <see cref="Start"/>
+    /// to listen. <paramref name="heartbeatInterval"/> (default 10 s) sets how often each
+    /// connection is pinged; <paramref name="idleTimeout"/> (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.
+    /// </summary>
+    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;
 
+        /// <summary>UTC of the last frame received from the peer — drives idle-timeout detection.</summary>
+        public DateTime LastActivityUtc =>
+            new(Volatile.Read(ref _lastActivityTicks), DateTimeKind.Utc);
+
         private readonly TcpClient _client;
         private readonly NetworkStream _stream;
         private readonly ConcurrentQueue<byte[]> _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);
 
+        /// <summary>Sends a heartbeat ping; a live peer answers with a pong, refreshing activity.</summary>
+        internal void SendPing() => SendControl(WebSocketOpcode.Ping, []);
+
         public void Send(ReadOnlySpan<byte> message)
         {
             if (_closed)
@@ -184,6 +260,27 @@ public sealed class WebSocketServer : IDisposable
             }
         }
 
+        private void SendControl(WebSocketOpcode opcode, ReadOnlySpan<byte> 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)
                     {

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<T> WaitFor<T>(Func<T?> 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;
+    }
+}

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<NetId>());
+        }
+    }
+
+    [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)