From 6edc16ed2003d717e70565ea385a38c3e3af33b8 Mon Sep 17 00:00:00 2001 From: Nomads_Reach <144523850+NomadsReach@users.noreply.github.com> Date: Sun, 16 Aug 2026 19:17:08 -0400 Subject: [PATCH] Add an end-to-end TCP acceptance harness (issue #15) (#20) Stand up the real AuthoritativeServer behind the real TCP transport on a loopback port and drive real SyntheticProtocolClient sockets through it, so the matrix is exercised over an actual connection rather than in-memory fakes. Covers the baseline-protocol, TCP-compatibility, and interest-management sections: welcome->hello->sessionReady handshake, unique/non-spoofable server-owned ids, malformed and oversized rejection before mutation, rate-limit tripping, transform/playerState/worldState relay, action events not replay-cached to late joiners, disconnect/reconnect leaving no stale session, same-cell relay vs distant interest filtering, multi-client no cross-cell spam, and handshake-timeout reaping. The harness runs for real wherever loopback TCP can bind (dev machines, the self-hosted runner) and skips cleanly otherwise. Wired into a self-hosted CI job gated by the runtime-policy guard. 60s idle-timeout, 32/64-client load / packet-loss / reconnect-churn, and GNS sections remain follow-ups. --- .github/workflows/acceptance.yml | 38 ++ .gitignore | 2 + server/AssemblyInfo.cs | 1 + server/CommonwealthOnline.Server.csproj | 1 + server/README.md | 9 + ...ommonwealthOnline.Server.Acceptance.csproj | 14 + server/acceptance/Program.cs | 485 ++++++++++++++++++ 7 files changed, 550 insertions(+) create mode 100644 .github/workflows/acceptance.yml create mode 100644 server/acceptance/CommonwealthOnline.Server.Acceptance.csproj create mode 100644 server/acceptance/Program.cs diff --git a/.github/workflows/acceptance.yml b/.github/workflows/acceptance.yml new file mode 100644 index 0000000..6405d9b --- /dev/null +++ b/.github/workflows/acceptance.yml @@ -0,0 +1,38 @@ +name: Acceptance (end-to-end TCP) + +# Issue #15 acceptance matrix, executed end-to-end: stands up the real +# AuthoritativeServer behind the real TCP transport on a loopback port and drives +# real client sockets through it. If the runner cannot bind a loopback listener, +# the harness prints SKIP and exits 0 (it still runs for real on dev machines and +# the NAS runner). + +on: + push: + paths: + - "server/**" + - ".github/workflows/acceptance.yml" + pull_request: + paths: + - "server/**" + - ".github/workflows/acceptance.yml" + workflow_dispatch: + +jobs: + acceptance: + name: End-to-end TCP acceptance + runs-on: [self-hosted, Linux, X64] + steps: + - uses: actions/checkout@v4 + + - name: Enforce repository runtime policy + run: bash server/scripts/verify-no-legacy-runtime.sh + + - uses: actions/setup-dotnet@v4 + env: + DOTNET_INSTALL_DIR: ${{ runner.tool_cache }}/dotnet + with: + dotnet-version: "8.0.x" + + - name: Run end-to-end acceptance harness + working-directory: server + run: dotnet run --project acceptance/CommonwealthOnline.Server.Acceptance.csproj -c Release diff --git a/.gitignore b/.gitignore index b7e8c2c..326a716 100644 --- a/.gitignore +++ b/.gitignore @@ -7,6 +7,8 @@ server/obj/ server/publish/ server/tests/bin/ server/tests/obj/ +server/acceptance/bin/ +server/acceptance/obj/ cmake-build-*/ *.exe *.dll diff --git a/server/AssemblyInfo.cs b/server/AssemblyInfo.cs index f325ae5..7f55642 100644 --- a/server/AssemblyInfo.cs +++ b/server/AssemblyInfo.cs @@ -1,3 +1,4 @@ using System.Runtime.CompilerServices; [assembly: InternalsVisibleTo("CommonwealthOnline.Server.Tests")] +[assembly: InternalsVisibleTo("CommonwealthOnline.Server.Acceptance")] diff --git a/server/CommonwealthOnline.Server.csproj b/server/CommonwealthOnline.Server.csproj index f1104f1..eab93c1 100644 --- a/server/CommonwealthOnline.Server.csproj +++ b/server/CommonwealthOnline.Server.csproj @@ -11,5 +11,6 @@ + diff --git a/server/README.md b/server/README.md index 2a80ca1..1f66ee1 100644 --- a/server/README.md +++ b/server/README.md @@ -26,6 +26,15 @@ dotnet run --project CommonwealthOnline.Server.csproj -- serve --config commonwe `start.bat` and `start.sh` prefer a published apphost, then a framework-dependent DLL, then `dotnet run` in a source checkout. +### Tests + +- `tests/` — fast in-memory unit/component tests over the authoritative core (run in the `CSharp Server Gate` CI). +- `acceptance/` — end-to-end acceptance harness (issue #15): stands up the real server behind the real TCP transport on a loopback port and drives real client sockets through the baseline-protocol, TCP-compatibility, and interest-management matrix sections. Runs in the `Acceptance (end-to-end TCP)` CI, or locally: + +```bash +dotnet run --project acceptance/CommonwealthOnline.Server.Acceptance.csproj -c Release +``` + ## Config Generate defaults: diff --git a/server/acceptance/CommonwealthOnline.Server.Acceptance.csproj b/server/acceptance/CommonwealthOnline.Server.Acceptance.csproj new file mode 100644 index 0000000..663cda0 --- /dev/null +++ b/server/acceptance/CommonwealthOnline.Server.Acceptance.csproj @@ -0,0 +1,14 @@ + + + Exe + net8.0 + enable + enable + latest + CommonwealthOnline.Server.Acceptance + CommonwealthOnline.Server.Acceptance + + + + + diff --git a/server/acceptance/Program.cs b/server/acceptance/Program.cs new file mode 100644 index 0000000..176c4f9 --- /dev/null +++ b/server/acceptance/Program.cs @@ -0,0 +1,485 @@ +using System.Net; +using System.Net.Sockets; +using System.Text; +using System.Text.Json.Nodes; +using CommonwealthOnline.Server; + +namespace CommonwealthOnline.Server.Acceptance; + +// End-to-end acceptance harness for issue #15. Unlike the in-memory unit tests, +// this stands up the real AuthoritativeServer behind the real TCP transport on a +// loopback port and drives real SyntheticProtocolClient sockets through it, so it +// exercises framing, admission, relay, interest filtering and session teardown +// over an actual connection. +// +// Covered here: the Baseline protocol, TCP compatibility, and interest-management +// sections of the matrix. GNS sections wait on #2 (real-bridge loopback); the +// 60s idle-timeout and 32/64-client load/fault runs are deferred follow-ups +// (see the summary printed at the end). +internal static class Program +{ + private static int _passed; + private static int _failed; + + public static async Task Main() + { + Console.WriteLine("Commonwealth Online - end-to-end acceptance harness (TCP)"); + + if (!CanBindLoopback()) + { + Console.WriteLine("SKIP: environment cannot bind a loopback TCP listener; end-to-end acceptance not executed here."); + return 0; + } + + await Run("baseline: two clients handshake welcome->hello->sessionReady", BaselineHandshake); + await Run("baseline: server-owned ids are unique and not client-spoofable", ServerOwnedIdsNotSpoofable); + await Run("baseline: malformed json rejected before gameplay mutation", MalformedJsonRejected); + await Run("baseline: oversized payload rejected before gameplay mutation", OversizeRejected); + await Run("baseline: sustained packet-rate abuse triggers rate limit", RateLimitTrips); + await Run("tcp: two-client transform/playerState/worldState smoke", TwoClientSmoke); + await Run("player-state: action events are not replay-cached to late joiners", ActionEventsNotReplayCached); + await Run("tcp: disconnect leaves no stale session; reconnect works", DisconnectNoStale); + await Run("interest: same-cell relays, distant is filtered", InterestSameCellVsDistant); + await Run("interest: multi-client across cells, no cross-cell transform spam", MultiClientNoCrossCellSpam); + await Run("baseline: handshake timeout closes an unactivated session", HandshakeTimeoutCloses); + + Console.WriteLine(); + Console.WriteLine($"acceptance: {_passed} passed, {_failed} failed"); + Console.WriteLine("deferred (follow-ups): 60s idle-timeout, 32/64-client load + packet-loss/reorder + reconnect-churn, and all GNS sections (blocked on #2)."); + return _failed == 0 ? 0 : 1; + } + + // ---- baseline protocol -------------------------------------------------- + + private static async Task BaselineHandshake() + { + await using var s = new TestServer(); + await using var a = await Connect(s.Port); + await using var b = await Connect(s.Port); + True(a.PlayerId != 0, "client a received a server-owned playerId"); + True(b.PlayerId != 0, "client b received a server-owned playerId"); + True(a.PlayerId != b.PlayerId, "player ids are unique"); + } + + private static async Task ServerOwnedIdsNotSpoofable() + { + await using var s = new TestServer(); + await using var a = await Connect(s.Port); + await using var b = await Connect(s.Port); + await Drain(a, 300); await Drain(b, 300); + + // a claims a bogus playerId inside its own transform; the server must relay + // b a transform stamped with a's real server-owned id, never the claim. + var spoof = TransformPacket("0000AAAA", "0000BBBB", 0, 0, "spawn"); + spoof["playerId"] = 999999; + await a.SendAsync(spoof); + + var relay = await ReceiveUntil(b, p => Type(p) == "transform", 3000); + NotNull(relay, "b received a's transform"); + Equal(a.PlayerId, UInt(relay!["playerId"]), "relayed transform carries the server-owned id"); + } + + private static async Task MalformedJsonRejected() + { + await using var s = new TestServer(); + using var raw = await RawClient.Connect(s.Port); + await raw.ExpectType("welcome", 2000); + await raw.WriteLine(Encoding.UTF8.GetBytes("{\"type\":\"hello\",\"protocolVersion\":2}")); + await raw.ExpectType("sessionReady", 2000); + var before = Stat(s.Server, "packetsRejected"); + var transformsBefore = Stat(s.Server, "transformPacketsReceived"); + + await raw.WriteLine(Encoding.UTF8.GetBytes("{ this is not valid json")); + await SpinUntil(() => Stat(s.Server, "packetsRejected") > before, 2000); + + True(Stat(s.Server, "packetsRejected") > before, "malformed json incremented packetsRejected"); + Equal(transformsBefore, Stat(s.Server, "transformPacketsReceived"), "no gameplay mutation from malformed json"); + } + + private static async Task OversizeRejected() + { + await using var s = new TestServer(); + using var raw = await RawClient.Connect(s.Port); + await raw.ExpectType("welcome", 2000); + await raw.WriteLine(Encoding.UTF8.GetBytes("{\"type\":\"hello\",\"protocolVersion\":2}")); + await raw.ExpectType("sessionReady", 2000); + var transformsBefore = Stat(s.Server, "transformPacketsReceived"); + + // A single line larger than the protocol maximum must terminate the session + // at the transport before it is ever decoded or applied. + await raw.WriteRaw(new byte[ProtocolConstants.MaxMessageBytes + 1024]); + + True(await raw.WaitClosed(4000), "server closed the connection on oversize input"); + Equal(transformsBefore, Stat(s.Server, "transformPacketsReceived"), "oversize input caused no gameplay mutation"); + } + + private static async Task RateLimitTrips() + { + await using var s = new TestServer(); + await using var a = await Connect(s.Port); + // Burst well past the 120 packets/second window in a single window. + for (var i = 0; i < 400; i++) + { + try { await a.SendAsync(new JsonObject { ["type"] = "keepAlive" }); } + catch { break; } + } + await SpinUntil(() => Stat(s.Server, "rateLimitedPackets") > 0, 3000); + True(Stat(s.Server, "rateLimitedPackets") > 0, "sustained burst tripped the rate limiter"); + } + + // ---- tcp compatibility -------------------------------------------------- + + private static async Task TwoClientSmoke() + { + await using var s = new TestServer(); + await using var a = await Connect(s.Port); // a connects first -> world-state host + await using var b = await Connect(s.Port); + await a.SendAsync(TransformPacket("0000CAFE", "0000F00D", 0, 0, "spawn")); + await b.SendAsync(TransformPacket("0000CAFE", "0000F00D", 0, 0, "spawn")); + await Drain(a, 400); await Drain(b, 400); + + // transform relay + await a.SendAsync(TransformPacket("0000CAFE", "0000F00D", 10, 0)); + NotNull(await ReceiveUntil(b, p => Type(p) == "transform" && UInt(p["playerId"]) == a.PlayerId, 3000), + "b received a's transform"); + + // durable playerState relay: an in-interest peer receives the durable + // fields (and, being in scope, the live action events too). + await a.SendAsync(new JsonObject + { + ["type"] = "playerState", + ["characterName"] = "Nomad", + ["actionEvents"] = new JsonArray(new JsonObject { ["sequence"] = 1, ["type"] = 3, ["eventName"] = "fireSingle" }) + }); + var ps = await ReceiveUntil(b, p => Type(p) == "playerState" && UInt(p["playerId"]) == a.PlayerId, 3000); + NotNull(ps, "b received a's playerState"); + Equal("Nomad", JsonHelpers.String(ps!["characterName"]), "playerState carried the character name"); + + // world-state from the host relays + await a.SendAsync(new JsonObject { ["type"] = "worldState", ["timeHours"] = 12.0 }); + NotNull(await ReceiveUntil(b, p => Type(p) == "worldState", 3000), "b received world state from the host"); + } + + private static async Task ActionEventsNotReplayCached() + { + await using var s = new TestServer(); + await using var a = await Connect(s.Port); + await a.SendAsync(TransformPacket("0000AC70", "0000AC70", 0, 0, "spawn")); + await a.SendAsync(new JsonObject + { + ["type"] = "playerState", + ["characterName"] = "Nomad", + ["actionEvents"] = new JsonArray(new JsonObject { ["sequence"] = 1, ["type"] = 3, ["eventName"] = "fireSingle" }) + }); + await Drain(a, 300); + + // a late joiner is replayed a's durable state, which must carry the durable + // fields but never the discrete action events. + await using var late = await Connect(s.Port); + var replay = await ReceiveUntil(late, p => Type(p) == "playerState" && UInt(p["playerId"]) == a.PlayerId, 3000); + NotNull(replay, "late joiner received a's durable player state"); + Equal("Nomad", JsonHelpers.String(replay!["characterName"]), "durable state replayed the character name"); + True(!replay.ContainsKey("actionEvents"), "discrete action events were not replay-cached to the late joiner"); + } + + private static async Task DisconnectNoStale() + { + await using var s = new TestServer(); + var a = await Connect(s.Port); + await using var b = await Connect(s.Port); + await SpinUntil(() => Stat(s.Server, "connectedClients") == 2, 3000); + Equal(2u, Stat(s.Server, "connectedClients"), "two clients active"); + + await a.DisposeAsync(); + await SpinUntil(() => Stat(s.Server, "connectedClients") == 1, 4000); + Equal(1u, Stat(s.Server, "connectedClients"), "disconnect removed the session"); + + await using var a2 = await Connect(s.Port); + await SpinUntil(() => Stat(s.Server, "connectedClients") == 2, 3000); + Equal(2u, Stat(s.Server, "connectedClients"), "reconnect restored two active sessions"); + True(a2.PlayerId != 0, "reconnected client received a fresh server-owned id"); + } + + // ---- interest management ------------------------------------------------ + + private static async Task InterestSameCellVsDistant() + { + await using var s = new TestServer(); + await using var a = await Connect(s.Port); + await using var b = await Connect(s.Port); + // both anchor in the same cell + await a.SendAsync(TransformPacket("0000BEEF", "0000BEEF", 0, 0, "spawn")); + await b.SendAsync(TransformPacket("0000BEEF", "0000BEEF", 0, 0, "spawn")); + await Drain(a, 400); await Drain(b, 400); + + await a.SendAsync(TransformPacket("0000BEEF", "0000BEEF", 5, 5)); + NotNull(await ReceiveUntil(b, p => Type(p) == "transform" && UInt(p["playerId"]) == a.PlayerId, 3000), + "same-cell transform relayed to b"); + + var filteredBefore = Stat(s.Server, "transformPacketsInterestFiltered"); + // a makes a valid transition to a distant, unrelated cell/worldspace; the + // move itself is legal (transition type) but b is no longer in interest. + await a.SendAsync(TransformPacket("00000111", "00000222", 5_000_000, 5_000_000, "cell_change")); + await Drain(a, 300); + True(await ExpectNone(b, p => Type(p) == "transform" && UInt(p["playerId"]) == a.PlayerId, 1200), + "distant transform was not relayed to b"); + await SpinUntil(() => Stat(s.Server, "transformPacketsInterestFiltered") > filteredBefore, 1500); + True(Stat(s.Server, "transformPacketsInterestFiltered") > filteredBefore, "distant transform counted as interest-filtered"); + } + + // The per-IP connect-attempt throttle admits 8 connections per 10s from one + // source address, so over a single loopback IP this case proves the no-cross- + // cell-spam property at 8 clients. Scaling to the matrix's 16/32/64-client load + // runs needs distinct source IPs (loopback aliasing) or a loopback connect- + // throttle exemption; that is the load-testing follow-up. + private const int MultiClientCount = 8; + + private static async Task MultiClientNoCrossCellSpam() + { + await using var s = new TestServer(); + var clients = new List(); + try + { + for (var i = 0; i < MultiClientCount; i++) + { + var c = await Connect(s.Port); + clients.Add(c); + // every client sits in its own cell AND worldspace, far apart + await c.SendAsync(TransformPacket((0x2000 + i).ToString("X8"), (0x9000 + i).ToString("X8"), i * 1_000_000.0, 0, "spawn")); + } + var ids = clients.Select(c => c.PlayerId).ToArray(); + Equal(MultiClientCount, ids.Distinct().Count(), "unique server-owned ids for every client"); + foreach (var c in clients) await Drain(c, 150); + + // each client moves; nobody should see anybody else's transform + for (var i = 0; i < clients.Count; i++) + await clients[i].SendAsync(TransformPacket((0x2000 + i).ToString("X8"), (0x9000 + i).ToString("X8"), i * 1_000_000.0 + 10, 0)); + + foreach (var c in clients) + True(await ExpectNone(c, p => Type(p) == "transform", 500), "client received no cross-cell transform spam"); + } + finally + { + foreach (var c in clients) await c.DisposeAsync(); + } + } + + // ---- session teardown --------------------------------------------------- + + private static async Task HandshakeTimeoutCloses() + { + await using var s = new TestServer(); + using var raw = await RawClient.Connect(s.Port); + await raw.ExpectType("welcome", 2000); + // deliberately never send hello; the reaper must close the pending session + Equal(0u, Stat(s.Server, "connectedClients"), "pending session is not counted as active"); + True(await raw.WaitClosed(14000), "server reaped the unactivated session after the handshake timeout"); + Equal(0u, Stat(s.Server, "connectedClients"), "still no active clients after reap"); + } + + // ---- infrastructure ----------------------------------------------------- + + private static async Task Run(string name, Func test) + { + try { await test(); _passed++; Console.WriteLine($"PASS {name}"); } + catch (Exception ex) { _failed++; Console.Error.WriteLine($"FAIL {name}: {ex.Message}"); } + } + + private static bool CanBindLoopback() + { + try { var p = FreePort(); return p > 0; } + catch (SocketException) { return false; } + } + + private static int FreePort() + { + var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + try { return ((IPEndPoint)listener.LocalEndpoint).Port; } + finally { listener.Stop(); } + } + + private static async Task Connect(int port) + { + var client = new SyntheticProtocolClient(); + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + await client.ConnectAsync("127.0.0.1", port, timeout.Token); + return client; + } + + private static JsonObject TransformPacket(string cell, string world, double x, double y, string movement = "normal") => new() + { + ["type"] = "transform", ["x"] = x, ["y"] = y, ["z"] = 0.0, ["angleZ"] = 0.0, + ["cellId"] = cell, ["worldspaceId"] = world, ["movementType"] = movement + }; + + private static string? Type(JsonObject packet) => JsonHelpers.String(packet["type"]); + private static uint UInt(JsonNode? node) => JsonHelpers.TryUInt32(node, 0, uint.MaxValue, out var v) ? v : 0; + private static uint Stat(AuthoritativeServer server, string name) => JsonHelpers.TryUInt32(server.GetCoreStats()[name], 0, uint.MaxValue, out var v) ? v : 0; + + private static async Task TryReceive(SyntheticProtocolClient client, int timeoutMs) + { + using var cts = new CancellationTokenSource(timeoutMs); + try { return await client.ReceiveAsync(cts.Token); } + catch (OperationCanceledException) { return null; } + catch (EndOfStreamException) { return null; } + catch (IOException) { return null; } + catch (InvalidDataException) { return null; } + } + + private static async Task ReceiveUntil(SyntheticProtocolClient client, Func predicate, int timeoutMs) + { + var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMs); + while (DateTime.UtcNow < deadline) + { + var remaining = (int)Math.Max(1, (deadline - DateTime.UtcNow).TotalMilliseconds); + var packet = await TryReceive(client, remaining); + if (packet is null) break; + if (predicate(packet)) return packet; + } + return null; + } + + private static async Task ExpectNone(SyntheticProtocolClient client, Func predicate, int windowMs) + { + var deadline = DateTime.UtcNow.AddMilliseconds(windowMs); + while (DateTime.UtcNow < deadline) + { + var remaining = (int)Math.Max(1, (deadline - DateTime.UtcNow).TotalMilliseconds); + var packet = await TryReceive(client, remaining); + if (packet is null) break; + if (predicate(packet)) return false; + } + return true; + } + + private static Task Drain(SyntheticProtocolClient client, int windowMs) => ExpectNone(client, _ => false, windowMs); + + private static async Task SpinUntil(Func condition, int timeoutMs) + { + var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMs); + while (DateTime.UtcNow < deadline) + { + if (condition()) return; + await Task.Delay(50); + } + } + + private static void True(bool value, string what) { if (!value) throw new Exception($"expected: {what}"); } + private static void NotNull(object? value, string what) { if (value is null) throw new Exception($"expected non-null: {what}"); } + private static void Equal(T expected, T actual, string what) + { + if (!EqualityComparer.Default.Equals(expected, actual)) throw new Exception($"{what}: expected {expected}, got {actual}"); + } + + private sealed class TestServer : IAsyncDisposable + { + private readonly TcpServerTransport _tcp; + private readonly string _dir; + + public TestServer(int maxPlayers = 32) + { + Port = FreePort(); + _dir = Path.Combine(Path.GetTempPath(), "co-accept-" + Guid.NewGuid().ToString("N")); + Directory.CreateDirectory(_dir); + var options = new ServerOptions + { + ConfigPath = Path.Combine(_dir, "commonwealth-server.json"), + Host = "127.0.0.1", Port = Port, AdminPort = FreePort(), MaxPlayers = maxPlayers + }; + Server = new AuthoritativeServer(options); + _tcp = new TcpServerTransport(options, Server); + _tcp.Start(); + } + + public int Port { get; } + public AuthoritativeServer Server { get; } + + public async ValueTask DisposeAsync() + { + await _tcp.DisposeAsync(); + await Server.DisposeAsync(); + try { Directory.Delete(_dir, true); } catch { } + } + } + + // Minimal raw client for cases the framed SyntheticProtocolClient can't express + // (oversize input, and connecting without completing the handshake). + private sealed class RawClient : IDisposable + { + private readonly TcpClient _client; + private readonly NetworkStream _stream; + + private RawClient(TcpClient client) { _client = client; _stream = client.GetStream(); } + + public static async Task Connect(int port) + { + var client = new TcpClient { NoDelay = true }; + await client.ConnectAsync(IPAddress.Loopback, port); + return new RawClient(client); + } + + public async Task WriteLine(byte[] payload) + { + await _stream.WriteAsync(payload); + await _stream.WriteAsync(new byte[] { (byte)'\n' }); + } + + public Task WriteRaw(byte[] bytes) => _stream.WriteAsync(bytes).AsTask(); + + public async Task ExpectType(string type, int timeoutMs) + { + var packet = await ReadLine(timeoutMs) ?? throw new Exception($"expected {type}, got connection close"); + var actual = JsonHelpers.String(packet["type"]); + if (actual != type) throw new Exception($"expected {type}, got {actual}"); + } + + public async Task ReadLine(int timeoutMs) + { + using var cts = new CancellationTokenSource(timeoutMs); + var buffer = new MemoryStream(); + var one = new byte[1]; + try + { + while (buffer.Length <= ProtocolConstants.MaxMessageBytes) + { + var n = await _stream.ReadAsync(one, cts.Token); + if (n == 0) return null; + if (one[0] == (byte)'\n') + { + var data = buffer.ToArray(); + if (data.Length > 0 && data[^1] == (byte)'\r') Array.Resize(ref data, data.Length - 1); + return JsonNode.Parse(data) as JsonObject; + } + buffer.WriteByte(one[0]); + } + } + catch (OperationCanceledException) { return null; } + catch (IOException) { return null; } + return null; + } + + public async Task WaitClosed(int timeoutMs) + { + using var cts = new CancellationTokenSource(timeoutMs); + var one = new byte[1]; + try + { + while (true) + { + var n = await _stream.ReadAsync(one, cts.Token); + if (n == 0) return true; // clean EOF -> server closed + } + } + catch (OperationCanceledException) { return false; } + catch (IOException) { return true; } // reset also counts as closed + catch (SocketException) { return true; } + } + + public void Dispose() + { + try { _stream.Dispose(); } catch { } + try { _client.Dispose(); } catch { } + } + } +}