DCT_WIFI/DCT_test/Verify-Transport.cs

using System;
using System.Collections.Generic;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Reflection;
using System.Threading;
using System.Threading.Tasks;

namespace DCT_test
{
    public static class TransportRegression
    {
        private static void Check(bool value, string message) { if (!value) throw new Exception(message); }
        private static DctPacket Receive(NetworkStream stream)
        {
            var frame = new List<byte>();
            while (true)
            {
                int value = stream.ReadByte();
                if (value < 0) throw new EndOfStreamException();
                frame.Add((byte)value);
                var parsed = DctPacket.ParseStream(frame);
                if (parsed.Count > 0) return parsed[0];
            }
        }
        private static DctPacket Request(NetworkStream stream, byte command, params byte[] payload)
        {
            byte[] bytes = DctPacket.Create(command, payload).Encode();
            stream.Write(bytes, 0, bytes.Length);
            return Receive(stream);
        }
        private static void VerifySlowReaderRecovery()
        {
            var listener = new TcpListener(IPAddress.Loopback, 0);
            listener.Start();
            ClientSession session = null;
            using (var peer = new TcpClient())
            {
                peer.ReceiveBufferSize = 65536;
                try
                {
                    peer.Connect(IPAddress.Loopback, ((IPEndPoint)listener.LocalEndpoint).Port);
                    var sender = listener.AcceptTcpClient();
                    sender.SendBufferSize = 4096;
                    session = new ClientSession(sender, "slow-reader");
                    int failures = 0;
                    session.SendFailed += (s, e) => { Interlocked.Increment(ref failures); Console.WriteLine(e); };
                    session.SetStreaming(true);
                    var bytes = new byte[4 * 1024 * 1024];
                    for (int i = 0; i < bytes.Length; i++) bytes[i] = (byte)(i % 251);
                    for (int chunk = 0; chunk < 16; chunk++)
                        Check(session.TryQueueTelemetry(bytes), "Queue slow-reader payload");
                    Thread.Sleep(1800);
                    Check(failures == 0 && session.SentPackets < 16, "A blocked write must survive the former 1s timeout; failures=" + failures + ", sent=" + session.SentPackets);
                    var stream = peer.GetStream();
                    stream.ReadTimeout = 5000;
                    var buffer = new byte[65536];
                    int received = 0;
                    while (received < bytes.Length * 16)
                    {
                        int n = stream.Read(buffer, 0, Math.Min(buffer.Length, bytes.Length * 16 - received));
                        Check(n > 0, "Connection must remain open after reader resumes");
                        for (int i = 0; i < n; i++) Check(buffer[i] == bytes[(received + i) % bytes.Length], "Byte order preserved after stall");
                        received += n;
                    }
                    Check(SpinWait.SpinUntil(() => session.SentPackets == 16, 2000), "Pending write must complete after recovery");
                    Check(session.SendControl(new byte[] { 0x20 }, 2000), "Control write after recovery");
                    Check(stream.ReadByte() == 0x20 && failures == 0, "Control traffic remains usable");
                    Console.WriteLine("PASS: 1.8s stalled reader recovers without disconnect, loss or duplication");
                }
                finally { if (session != null) session.Dispose(); listener.Stop(); }
            }
        }

        private static void VerifyQueuedControlSurvivesStall()
        {
            var listener = new TcpListener(IPAddress.Loopback, 0);
            listener.Start();
            ClientSession session = null;
            try
            {
                using (var peer = new TcpClient())
                {
                    peer.ReceiveBufferSize = 4096;
                    peer.Connect(IPAddress.Loopback, ((IPEndPoint)listener.LocalEndpoint).Port);
                    var sender = listener.AcceptTcpClient();
                    sender.SendBufferSize = 4096;
                    session = new ClientSession(sender, "control-stall");
                    session.SetStreaming(true);
                    var block = new byte[1024 * 1024];
                    session.TryQueueTelemetry(block);
                    Thread.Sleep(150);
                    var completed = new TaskCompletionSource<bool>();
                    Check(session.QueueControl(new byte[] { 0x77 }, sent => completed.TrySetResult(sent)), "Queue control");
                    Thread.Sleep(2800);
                    Check(!completed.Task.IsCompleted, "Control must remain pending behind stalled write");
                    var stream = peer.GetStream();
                    stream.ReadTimeout = 5000;
                    var buffer = new byte[65536];
                    int total = 0;
                    while (total < block.Length)
                    {
                        int count = stream.Read(buffer, 0, Math.Min(buffer.Length, block.Length - total));
                        Check(count > 0, "Stalled stream remains connected");
                        total += count;
                    }
                    Check(stream.ReadByte() == 0x77 && completed.Task.Wait(2000) && completed.Task.Result,
                        "Control must be delivered after the former 2.5s expiry");
                    int previousGeneration = session.TelemetryGeneration;
                    session.SetStreaming(false);
                    session.SetStreaming(true);
                    Check(!session.SendLogAsync(0xA3, new byte[] { 1 }, previousGeneration, CancellationToken.None).Result.HasValue,
                        "Previous run must not leak after restart");
                    session.Dispose();
                    Check(!session.SendLogAsync(0xA3, new byte[] { 1 }, session.TelemetryGeneration, CancellationToken.None).Result.HasValue,
                        "Disconnected file send must not wait forever");
                }
            }
            finally { if (session != null) session.Dispose(); listener.Stop(); }
            Console.WriteLine("PASS: delayed control delivery, old-run isolation, disconnect cancellation");
        }

        public static void Run()
        {
            VerifyApListener();
            VerifySlowReaderRecovery();
            VerifyQueuedControlSurvivesStall();
            var clock = new LogPlaybackClock();
            clock.Reset(0);
            clock.Advance(100, 1, false);
            Check(clock.DelayUntil(300, 2) == 100, "2x playback interval");
            clock.Advance(1100, 2, true);
            Check(clock.PositionMilliseconds == 100, "Pause must freeze source time");
            clock.Advance(1200, 2, false);
            Check(clock.PositionMilliseconds == 300, "Resume at changed speed");
            clock.Reset(1200);
            Check(clock.PositionMilliseconds == 0, "Restart/loop must reset source time");

            string fixture = Path.Combine(Path.GetTempPath(), Guid.NewGuid().ToString("N") + ".txt");
            try
            {
                string hex = BitConverter.ToString(DctPacket.Create(0xA3, new byte[] { 1, 164, 1 }).Encode()).Replace('-', ' ');
                File.WriteAllLines(fixture, new[] { "[Recv][23:59:59.900] " + hex,
                    "[Recv][00:00:00.100] " + hex, "[Recv][00:00:03.100] " + hex });
                var log = new LogFileReader().Load(fixture);
                Check(log.Success && log.Packets.Count == 3, "Midnight log parsing");
                Check(log.Packets[1].LogElapsedMilliseconds == 200, "Midnight rollover");
                Check(log.Packets[2].LogElapsedMilliseconds == 3200, "Original log gap preserved");
            }
            finally { File.Delete(fixture); }

            var session = new ClientSession(new TcpClient(), "window-test");
            session.SetStreaming(true);
            for (int i = 0; i < ClientSession.TelemetryAckWindowSize; i++)
                Check(session.ReserveTelemetry(0xA3), "Reserve window");
            var blocked = Task.Run(() => session.ReserveTelemetry(0xA3));
            Check(!blocked.Wait(50), "Full ACK window must backpressure producer");
            session.Acknowledge(0x23);
            Check(blocked.Wait(1000) && blocked.Result, "ACK must release producer");
            blocked = Task.Run(() => session.ReserveTelemetry(0xA3));
            session.SetStreaming(false);
            Check(blocked.Wait(1000) && !blocked.Result, "STOP must release producer without sending");
            session.Client.Close();

            // A stalled ACK window must recover by retiring stale reservations;
            // it is never allowed to throw and close the transport session.
            var stalled = new ClientSession(new TcpClient(), "stall-test");
            stalled.SetStreaming(true);
            for (int i = 0; i < ClientSession.TelemetryAckWindowSize; i++)
                Check(stalled.ReserveTelemetry(0xA3), "Stall window fill");
            Thread.Sleep(3200);
            Check(stalled.ReserveTelemetry(0xA3), "Stalled ACKs must not disconnect or throw");
            stalled.SetStreaming(false);
            stalled.ResetTelemetryAcks();
            stalled.Client.Close();

            var server = new DctTcpServer();
            Check(server.Start(0), "Test protocol listener");
            var listener = (TcpListener)typeof(DctTcpServer).GetField("_listener", BindingFlags.Instance | BindingFlags.NonPublic).GetValue(server);
            try
            {
                using (var client = new TcpClient())
                {
                    client.Connect(IPAddress.Loopback, ((IPEndPoint)listener.LocalEndpoint).Port);
                    var stream = client.GetStream();
                    stream.ReadTimeout = 3000;
                    Check(Request(stream, 0x20, 3).Payload[0] == 2, "START before handshake must fail");
                    Request(stream, 0x00); Request(stream, 0x01); Request(stream, 0x03); Request(stream, 0x04);
                    Check(Request(stream, 0x20, 3).Payload[0] == 2, "START before configuration must fail");
                    Request(stream, 0x09, 6, 0); Request(stream, 0x10, 0, 0);
                    Request(stream, 0x11, 0, 0); Request(stream, 0x13, 0, 0);
                    Check(Request(stream, 0x20, 3).Payload[0] == 1, "Configured START must succeed");
                    Check(SpinWait.SpinUntil(() => { foreach (var c in server.ActiveClients) if (c.Streaming) return true; return false; }, 1000), "Streaming activation");
                    server.BroadcastPacket(DctPacket.Create(0xA3, new byte[] { 1, 164, 7 }));
                    Check(Receive(stream).CommandId == 0xA3, "Telemetry after START");
                    byte[] ack = DctPacket.Create(0x23, new byte[] { 0 }).Encode(); // Valid NACK must retire pending frame too.
                    stream.Write(ack, 0, ack.Length);
                    // Lossless file replay must wait beyond the old 250 ms
                    // admission budget, not advance to the next record.
                    var replay = Task.Run(async () =>
                    {
                        for (int i = 0; i < 16; i++)
                            await server.BroadcastLogPacketAsync(DctPacket.Create(0xA3,
                                new byte[] { 1, 164, (byte)i }), i, CancellationToken.None);
                    });
                    Thread.Sleep(650);
                    Check(!replay.IsCompleted, "File replay must backpressure rather than skip records");
                    for (int i = 0; i < 16; i++)
                    {
                        var fileFrame = Receive(stream);
                        Check(fileFrame.CommandId == 0xA3 && fileFrame.Payload[2] == i,
                            "File record missing, duplicated or reordered: " + i);
                        stream.Write(ack, 0, ack.Length);
                    }
                    Check(replay.Wait(3000), "File replay resumes after ACKs");
                    Check(Request(stream, 0x21).CommandId == 0xA1, "STOP after backpressured file replay");
                    Check(Request(stream, 0x20, 1).Payload[0] == 1, "Restart after file replay");

                    // Do not ACK the next frame for longer than the old 3 s
                    // watchdog.  The server must retire the stale reservation,
                    // send the next frame, and keep the TCP session alive.
                    server.BroadcastPacket(DctPacket.Create(0xA3, new byte[] { 1, 164, 9 }));
                    Check(Receive(stream).CommandId == 0xA3, "Telemetry before delayed ACK");
                    Thread.Sleep(3200);
                    server.BroadcastPacket(DctPacket.Create(0xA3, new byte[] { 1, 164, 10 }));
                    Check(Receive(stream).CommandId == 0xA3, "Telemetry after delayed ACK");
                    Check(client.Client.Connected, "Delayed ACK must not close TCP session");
                    stream.Write(ack, 0, ack.Length);
                    Check(Request(stream, 0x21).CommandId == 0xA1, "STOP ACK");
                    server.BroadcastPacket(DctPacket.Create(0xA3, new byte[] { 1, 164, 8 }));
                    Check(!stream.DataAvailable, "No telemetry after STOP ACK");
                    Check(Request(stream, 0x20, 3).Payload[0] == 1, "Restart without reconnect");
                    Check(SpinWait.SpinUntil(() => { foreach (var c in server.ActiveClients) if (c.Streaming) return true; return false; }, 1000),
                        "Streaming reactivation on the existing session");
                    server.BroadcastPacket(DctPacket.Create(0xA3, new byte[] { 1, 164, 11 }));
                    var restartedFrame = Receive(stream);
                    Check(restartedFrame.CommandId == 0xA3 && restartedFrame.Payload[2] == 11,
                        "Telemetry must resume on the same TCP connection after STOP then START");
                    stream.Write(ack, 0, ack.Length);
                }

                // PC A2L-only setup flow: no MDB 0x10/0x13 is sent.
                // XCP USE -> IF_DATA -> MEASUREMENTS followed by common CAN
                // configuration must therefore be sufficient for START.
                using (var xcpClient = new TcpClient())
                {
                    xcpClient.Connect(IPAddress.Loopback, ((IPEndPoint)listener.LocalEndpoint).Port);
                    var stream = xcpClient.GetStream();
                    stream.ReadTimeout = 3000;
                    Request(stream, 0x00); Request(stream, 0x01); Request(stream, 0x03); Request(stream, 0x04);
                    Check(Request(stream, 0x40, 0x01).Payload[0] == 1, "XCP USE must succeed");
                    Check(Request(stream, 0x42, 0x00, 0x01, 0x02, 0x00, 0x10, 0x00, 0x70).Payload[0] == 0,
                        "XCP measurements before IF_DATA must fail");
                    Check(Request(stream, 0x41, 0x00, 0x90, 0x06, 0x00, 0x00, 0x93, 0x06, 0x00, 0x00).Payload[0] == 1,
                        "XCP IF_DATA must succeed");
                    Check(Request(stream, 0x42, 0x00, 0x01, 0x02, 0x00, 0x10, 0x00, 0x70).Payload[0] == 1,
                        "XCP measurements must succeed after IF_DATA");
                    Request(stream, 0x09, 0x06, 0x00);
                    Request(stream, 0x11, 0x00, 0x00);
                    Check(Request(stream, 0x20, 0x01).Payload[0] == 1,
                        "PC A2L-only configuration must allow START without MDB commands");
                    Check(Request(stream, 0x21).CommandId == 0xA1, "XCP STOP ACK");
                }
            }
            finally { server.Stop(); }
            Console.WriteLine("PASS: log timestamps/midnight/pause/speed/reset, ACK backpressure, MDB and PC A2L-only start/stop TCP exchange");
        }

        private static void VerifyApListener()
        {
            var server = new DctTcpServer();
            Check(!server.StartAp("invalid", 6000), "Invalid AP address must fail");
            Check(!server.StartAp("::1", 6000), "IPv6 AP address must fail");
            Check(!server.StartAp("0.0.0.0", 6000), "Wildcard AP address must fail");
            Check(!server.StartAp("127.0.0.1", 0) && !server.StartAp("127.0.0.1", 65536), "Out of range AP ports must fail");
            var start = typeof(DctTcpServer).GetMethod("Start", BindingFlags.Instance | BindingFlags.NonPublic,
                null, new[] { typeof(int), typeof(IPAddress), typeof(bool) }, null);
            var occupied = new TcpListener(IPAddress.Loopback, 0);
            occupied.Start();
            int occupiedPort = ((IPEndPoint)occupied.LocalEndpoint).Port;
            try {
                Check(!server.StartAp("127.0.0.1", occupiedPort), "AP occupied port must fail");
                Check(!server.IsRunning, "Failed AP listener must stay stopped");
            } finally { occupied.Stop(); }
            Check((bool)start.Invoke(server, new object[] { 0, IPAddress.Loopback, false }), "AP scoped listener after failure");
            try {
                var listener = (TcpListener)typeof(DctTcpServer).GetField("_listener", BindingFlags.Instance | BindingFlags.NonPublic).GetValue(server);
                var endpoint = (IPEndPoint)listener.LocalEndpoint;
                Check(endpoint.Address.Equals(IPAddress.Loopback), "AP listener must bind the requested address");
                using (var client = new TcpClient()) {
                    client.Connect(endpoint);
                    var stream = client.GetStream();
                    stream.ReadTimeout = 3000;
                    Check(Request(stream, 0x00).CommandId == 0x80, "AP handshake ENABLE");
                    Check(Request(stream, 0x01).CommandId == 0x81, "AP handshake VERSION");
                    Request(stream, 0x03); Request(stream, 0x04);
                    Request(stream, 0x09, 6, 0); Request(stream, 0x10, 0, 0);
                    Request(stream, 0x11, 0, 0); Request(stream, 0x13, 0, 0);
                    Check(Request(stream, 0x20, 1).Payload[0] == 1, "AP configured START");
                    Check(SpinWait.SpinUntil(() => { foreach (var c in server.ActiveClients) if (c.Streaming) return true; return false; }, 1000), "AP streaming activation");
                    server.BroadcastPacket(DctPacket.Create(0xA3, new byte[] { 1, 164, 7 }));
                    Check(Receive(stream).CommandId == 0xA3, "AP telemetry");
                    Check(Request(stream, 0x21).CommandId == 0xA1, "AP STOP ACK");
                }
            } finally { server.Stop(); }
            Check(server.Start(0), "Direct listener must still start after AP stop");
            server.Stop();
            Console.WriteLine("PASS: AP scoped TCP handshake/configuration/telemetry/STOP, bind failure cleanup, Direct restart");
        }
    }
}