fix(focas): a stalled device fails fast for the rest of its batch; generation-stamp health (#740 review round 1)

I1: Task.WhenAll waits for the slowest device, so concurrency alone still let a device that accepts TCP and then stalls cost (its tag count x call timeout) per batch. After the first timeout, transport failure or comms-class status on a device, its remaining tags in the batch now fail BadCommunicationError immediately, the same as the coalescer's FailMembers, which now also stops after a failed range. FIXED: tags are still served from the cache. A malformed address is BadNodeIdUnknown and does not count as a link failure. Test: a 10-tag stalled device beside a healthy one completes in under 3 call timeouts with the healthy value Good. The P9 test now also asserts per-client read counts and that two reads overlapped across devices. Minors: - 1: a superseded init releases its table only if it was never published (flag). - 2: a fixed-tree refusal credits LastSuccessfulRead; a value-path refusal after a comms fault restores Healthy. - 5: alarm/fixed/refusal health is reset inside the publish lock, after the generation check. - 6: FocasFixedTreeHealthTests wait on served-failure/refusal counters, not wall-clock. - 8: data-path health writes (reads, coalescer, writes, poll errors, refusals) carry the generation they started under and are dropped once a lifecycle call has moved on. Each write is a check-and-set under the lifecycle lock. Also adds a unit case where 20 vendor records with type 0 and empty messages stay vendor. Refs #740 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
2026-10-22 14:44:42 -04:00
parent 8ffa6178b5
commit fd64fed217
5 changed files with 335 additions and 86 deletions
@@ -84,6 +84,24 @@ public sealed class FocasAlarmLayoutTests
        alarms[19].Message.ShouldBe("OVERTRAVEL +19");
    }

+    /// <summary>
+    ///     Review I2 edge: vendor records with type 0 and EMPTY messages also satisfy the legacy invariants (the
+    ///     int32 read of <c>type&lt;&lt;16 | axis</c> is just the axis, and the "length" int32 is the empty message's
+    ///     zero bytes) — both readings are plausible, so the vendor default must hold and the numbers decode.
+    /// </summary>
+    [Fact]
+    public void Twenty_vendor_records_with_type_zero_and_empty_messages_stay_vendor()
+    {
+        var payload = Vendor([.. Enumerable.Range(0, 20).Select(i => (2000 + i, (short)0, (short)(i % 3), ""))]);
+        payload.Length.ShouldBe(1520);
+
+        var alarms = FocasWireClient.ParseAlarms(payload, 32);
+
+        alarms.Count.ShouldBe(20);
+        alarms.Select(a => a.AlarmNumber).ShouldBe(Enumerable.Range(2000, 20));
+        alarms.ShouldAllBe(a => a.Type == 0 && a.Message == "");
+    }
+
    /// <summary>The unambiguous lengths are unchanged.</summary>
    [Fact]
    public void Unambiguous_lengths_are_unchanged()
@@ -20,11 +20,12 @@ public sealed class FocasConcurrentDeviceReadTests
    ///     STARTED — so a sequential walk (A first) can never finish A, while a concurrent one does. Also counts
    ///     the peak number of reads in flight on any one client.
    /// </summary>
-    private sealed class OrderingClient(TaskCompletionSource bStarted) : FakeFocasClient
+    private sealed class OrderingClient(TaskCompletionSource bStarted, int[] global) : FakeFocasClient
    {
        private string _host = "";
        private int _inFlight;
        public int PeakInFlight;
+        public int Reads;

        public override Task ConnectAsync(FocasHostAddress address, TimeSpan timeout, CancellationToken ct)
        {
@@ -37,6 +38,9 @@ public sealed class FocasConcurrentDeviceReadTests
        {
            var now = Interlocked.Increment(ref _inFlight);
            InterlockedMax(ref PeakInFlight, now);
+            Interlocked.Increment(ref Reads);
+            // global[0] = reads in flight across EVERY client, global[1] = its peak.
+            InterlockedMax(ref global[1], Interlocked.Increment(ref global[0]));
            try
            {
                if (_host.Contains("10.0.0.6", StringComparison.Ordinal)) bStarted.TrySetResult();
@@ -44,7 +48,11 @@ public sealed class FocasConcurrentDeviceReadTests
                await Task.Delay(20, ct).ConfigureAwait(false);
                return ((sbyte)1, FocasStatusMapper.Good);
            }
-            finally { Interlocked.Decrement(ref _inFlight); }
+            finally
+            {
+                Interlocked.Decrement(ref _inFlight);
+                Interlocked.Decrement(ref global[0]);
+            }
        }

        private static void InterlockedMax(ref int target, int value)
@@ -60,7 +68,8 @@ public sealed class FocasConcurrentDeviceReadTests
    public async Task Devices_are_read_concurrently_and_each_stays_serialised()
    {
        var bStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
-        var factory = new FakeFocasClientFactory { Customise = () => new OrderingClient(bStarted) };
+        var global = new int[2];
+        var factory = new FakeFocasClientFactory { Customise = () => new OrderingClient(bStarted, global) };
        var drv = new FocasDriver(new FocasDriverOptions
        {
            Devices = [new FocasDeviceOptions(HostA, DeviceName: "A"), new FocasDeviceOptions(HostB, DeviceName: "B")],
@@ -81,8 +90,66 @@ public sealed class FocasConcurrentDeviceReadTests
        snaps.ShouldAllBe(s => s.StatusCode == FocasStatusMapper.Good,
            "device A's read can only finish once device B's has started — a sequential walk times A out");
        factory.Clients.Count.ShouldBe(2);
-        factory.Clients.Cast<OrderingClient>().ShouldAllBe(c => c.PeakInFlight == 1,
+        var clients = factory.Clients.Cast<OrderingClient>().ToList();
+        // Every client really served reads (2 on A, 1 on B), so "peak 1" is measured, not vacuous …
+        clients.Select(c => c.Reads).OrderBy(n => n).ShouldBe([1, 2]);
+        clients.ShouldAllBe(c => c.PeakInFlight == 1,
            "one device's socket is still request→response: never two reads in flight on it");
+        // … and the two devices overlapped: two reads were in flight at once across the clients.
+        global[1].ShouldBe(2, "the devices must be read at the same time, not one after the other");
+        await drv.ShutdownAsync(CancellationToken.None);
+    }
+
+    /// <summary>A device that accepted TCP and then stopped answering: every read hangs until the call timeout.</summary>
+    private sealed class StalledClient : FakeFocasClient
+    {
+        public int Reads;
+
+        public override async Task<(object? value, uint status)> ReadAsync(
+            FocasAddress address, FocasDataType type, CancellationToken ct)
+        {
+            Interlocked.Increment(ref Reads);
+            await Task.Delay(Timeout.Infinite, ct).ConfigureAwait(false);
+            return (null, FocasStatusMapper.Good);
+        }
+    }
+
+    /// <summary>
+    ///     #740 review I1 — <c>Task.WhenAll</c> waits for the slowest device, so concurrency alone still made a
+    ///     stalled device cost (its tag count × the call timeout) per batch, and the healthy device's values waited
+    ///     that long. After the first timeout on a device its remaining tags in the batch fail fast: the batch is
+    ///     bounded by about ONE call timeout and the healthy device's values are Good.
+    /// </summary>
+    [Fact]
+    public async Task A_stalled_device_costs_one_call_timeout_per_batch_not_one_per_tag()
+    {
+        var callTimeout = TimeSpan.FromMilliseconds(200);
+        var factory = new FakeFocasClientFactory();
+        factory.Customise = () => factory.Clients.Count == 0 ? new StalledClient() : new FakeFocasClient { Values = { ["R100"] = (sbyte)9 } };
+        var stalledTags = Enumerable.Range(0, 10)
+            .Select(i => FocasRawTags.Entry(new FocasTagDefinition($"raw/A{i}", "A", $"R{100 + i}", FocasDataType.Byte)));
+        var drv = new FocasDriver(new FocasDriverOptions
+        {
+            Devices = [new FocasDeviceOptions(HostA, DeviceName: "A"), new FocasDeviceOptions(HostB, DeviceName: "B")],
+            RawTags = [.. stalledTags, FocasRawTags.Entry(new FocasTagDefinition("raw/B1", "B", "R100", FocasDataType.Byte))],
+            Probe = new FocasProbeOptions { Enabled = false },
+            PmcCoalesce = new FocasPmcCoalesceOptions { Enabled = false },
+            Timeout = callTimeout,
+        }, "focas-stall", factory);
+        await drv.InitializeAsync("{}", CancellationToken.None);
+        // Connect device A first so it gets the stalled client (connect order = client order).
+        await drv.ReadAsync(["raw/A0"], CancellationToken.None);
+
+        var refs = Enumerable.Range(0, 10).Select(i => $"raw/A{i}").Append("raw/B1").ToArray();
+        var sw = System.Diagnostics.Stopwatch.StartNew();
+        var snaps = await drv.ReadAsync(refs, CancellationToken.None);
+        sw.Stop();
+
+        sw.Elapsed.ShouldBeLessThan(callTimeout * 3, "one stalled device must cost ~one call timeout, not ten");
+        snaps[10].StatusCode.ShouldBe(FocasStatusMapper.Good);
+        snaps[10].Value.ShouldBe((sbyte)9);
+        snaps.Take(10).ShouldAllBe(s => s.StatusCode == FocasStatusMapper.BadCommunicationError);
+        ((StalledClient)factory.Clients[0]).Reads.ShouldBe(2, "one read in the warm-up batch, one in the measured batch");
        await drv.ShutdownAsync(CancellationToken.None);
    }

@@ -24,6 +24,10 @@ public sealed class FocasFixedTreeHealthTests
    {
        public volatile bool Down;
        public volatile bool Refuse;
+        /// <summary>Transport failures served so far — what the tests wait on instead of wall-clock delays.</summary>
+        public int Failures;
+        /// <summary>Refusals served so far.</summary>
+        public int Refusals;
    }

    private sealed class SwitchableCnc(CncSwitch sw) : FakeFocasClient
@@ -31,13 +35,25 @@ public sealed class FocasFixedTreeHealthTests
        private bool Down => sw.Down;
        private bool Refuse => sw.Refuse;

-        public override Task<FocasSysInfo> GetSysInfoAsync(CancellationToken ct) =>
-            Down ? throw new IOException("socket closed") : base.GetSysInfoAsync(ct);
+        public override Task<FocasSysInfo> GetSysInfoAsync(CancellationToken ct)
+        {
+            if (!Down) return base.GetSysInfoAsync(ct);
+            Interlocked.Increment(ref sw.Failures);
+            throw new IOException("socket closed");
+        }

        public override Task<FocasDynamicSnapshot> ReadDynamicAsync(int axisIndex, CancellationToken ct)
        {
-            if (Down) throw new IOException("socket closed");
-            if (Refuse) throw new FocasWireException("cnc_rddynamic2 refused", rc: 6);
+            if (Down)
+            {
+                Interlocked.Increment(ref sw.Failures);
+                throw new IOException("socket closed");
+            }
+            if (Refuse)
+            {
+                Interlocked.Increment(ref sw.Refusals);
+                throw new FocasWireException("cnc_rddynamic2 refused", rc: 6);
+            }
            return base.ReadDynamicAsync(axisIndex, ct);
        }
    }
@@ -71,18 +87,22 @@ public sealed class FocasFixedTreeHealthTests
        await drv.InitializeAsync("{}", CancellationToken.None);
        await WaitUntil(() => drv.GetDeviceState(Host)?.LastFixedPollUtc is not null);

-        var first = drv.GetHealth().LastSuccessfulRead;
-        await Task.Delay(100);
-        var healthy = drv.GetHealth();
-        healthy.State.ShouldBe(DriverState.Healthy);
-        healthy.LastSuccessfulRead!.Value.ShouldBeGreaterThan(first!.Value, "an answered axis pass is a successful read");
+        var first = drv.GetHealth().LastSuccessfulRead!.Value;
+        // An answered axis pass is a successful read: wait for it to move rather than sleeping.
+        await WaitUntil(() => drv.GetHealth().LastSuccessfulRead > first);
+        drv.GetHealth().State.ShouldBe(DriverState.Healthy);

        sw.Down = true;
-        await Task.Delay(100);
-        drv.GetHealth().State.ShouldBe(DriverState.Healthy, "a blip inside the staleness budget does not degrade");
+        var lastAnswer = drv.GetDeviceState(Host)!.LastFixedPollUtc!.Value;
+        // Two failed ticks have demonstrably happened (positive evidence for the absence check below) …
+        await WaitUntil(() => Volatile.Read(ref sw.Failures) >= 2);
+        // … and, if they landed inside the budget, the driver is still Healthy.
+        if (DateTime.UtcNow - lastAnswer < Stale)
+            drv.GetHealth().State.ShouldBe(DriverState.Healthy, "a blip inside the staleness budget does not degrade");
        await WaitUntil(() => drv.GetHealth().State == DriverState.Degraded);
        var frozen = drv.GetHealth().LastSuccessfulRead;
-        await Task.Delay(150);
+        var failuresAtDegrade = Volatile.Read(ref sw.Failures);
+        await WaitUntil(() => Volatile.Read(ref sw.Failures) >= failuresAtDegrade + 5);
        drv.GetHealth().State.ShouldBe(DriverState.Degraded);
        drv.GetHealth().LastSuccessfulRead.ShouldBe(frozen, "LastSuccessfulRead must stay frozen while the CNC is silent");

@@ -116,8 +136,15 @@ public sealed class FocasFixedTreeHealthTests
        await WaitUntil(() => drv.GetDeviceState(Host)?.LastFixedPollUtc is not null);

        sw.Refuse = true;
-        await Task.Delay(Stale * 3);
+        var started = DateTime.UtcNow;
+        await WaitUntil(() => Volatile.Read(ref sw.Refusals) >= 1);
+        var atFirstRefusal = drv.GetHealth().LastSuccessfulRead!.Value;
+        // Refused ticks really happened, for longer than the staleness budget …
+        await WaitUntil(() => Volatile.Read(ref sw.Refusals) >= 5 && DateTime.UtcNow - started > Stale * 2);
+        // … and the driver stayed Healthy, with each refusal crediting the answer (review minor 2).
        drv.GetHealth().State.ShouldBe(DriverState.Healthy);
+        drv.GetHealth().LastSuccessfulRead!.Value.ShouldBeGreaterThan(atFirstRefusal,
+            "a refusal is an answer from the CNC — LastSuccessfulRead advances");
        sw.Refuse = false;
    }

@@ -112,6 +112,78 @@ public sealed class FocasReadHealthContractTests
        await drv.ShutdownAsync(CancellationToken.None);
    }

+    /// <summary>
+    ///     Review minor 2 — a device that recovers from a comms fault and then only REFUSES (every tag it serves is
+    ///     refused) is answering, so it is restored to Healthy instead of staying Degraded forever.
+    /// </summary>
+    [Fact]
+    public async Task Refusal_after_a_comms_fault_restores_healthy()
+    {
+        var (drv, client, _) = NewDriver(FocasStatusMapper.BadCommunicationError);
+        await drv.InitializeAsync("{}", CancellationToken.None);
+        await drv.ReadAsync(["raw/R100"], CancellationToken.None);
+        drv.GetHealth().State.ShouldBe(DriverState.Degraded, "positive control: the comms fault degraded it");
+
+        client.ReadStatuses["R100"] = FocasStatusMapper.BadOutOfRange;
+        await drv.ReadAsync(["raw/R100"], CancellationToken.None);
+
+        drv.GetHealth().State.ShouldBe(DriverState.Healthy);
+        await drv.ShutdownAsync(CancellationToken.None);
+    }
+
+    /// <summary>A read whose CNC answer is held until released, then fails at the transport.</summary>
+    private sealed class HeldReadClient(TaskCompletionSource entered, Task release) : FakeFocasClient
+    {
+        public volatile bool Hold = true;
+
+        public override async Task<(object? value, uint status)> ReadAsync(
+            FocasAddress address, FocasDataType type, CancellationToken ct)
+        {
+            if (!Hold) return await base.ReadAsync(address, type, ct).ConfigureAwait(false);
+            entered.TrySetResult();
+            await release.ConfigureAwait(false);
+            throw new IOException("socket reset");
+        }
+    }
+
+    /// <summary>
+    ///     Review minor 8 — a read that started against the PREVIOUS table and fails after a re-initialise must not
+    ///     write its Degraded onto the new generation's health. Positive control: the same failure under the live
+    ///     generation does degrade.
+    /// </summary>
+    [Fact]
+    public async Task A_read_from_the_previous_generation_cannot_degrade_the_new_one()
+    {
+        var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var clients = new List<HeldReadClient>();
+        var drv = new FocasDriver(new FocasDriverOptions
+        {
+            Devices = [new FocasDeviceOptions(Host)],
+            RawTags = [FocasRawTags.Entry(new FocasTagDefinition("raw/R100", Host, "R100", FocasDataType.Byte))],
+            Probe = new FocasProbeOptions { Enabled = false },
+            Timeout = TimeSpan.FromSeconds(10),
+        }, "focas-gen", new FakeFocasClientFactory
+        {
+            Customise = () => { var c = new HeldReadClient(entered, release.Task); lock (clients) clients.Add(c); return c; },
+        });
+        await drv.InitializeAsync("{}", CancellationToken.None);
+
+        var stale = drv.ReadAsync(["raw/R100"], CancellationToken.None);
+        await entered.Task.WaitAsync(TimeSpan.FromSeconds(5));
+        await drv.InitializeAsync("{}", CancellationToken.None); // a new generation, Healthy
+        release.SetResult();
+        (await stale.WaitAsync(TimeSpan.FromSeconds(5)))[0].StatusCode.ShouldBe(FocasStatusMapper.BadCommunicationError);
+
+        drv.GetHealth().State.ShouldBe(DriverState.Healthy, "the old generation's failure must not land on the new one");
+
+        // Positive control: a failure under the live generation degrades.
+        var live = drv.ReadAsync(["raw/R100"], CancellationToken.None);
+        (await live.WaitAsync(TimeSpan.FromSeconds(5)))[0].StatusCode.ShouldBe(FocasStatusMapper.BadCommunicationError);
+        drv.GetHealth().State.ShouldBe(DriverState.Degraded);
+        await drv.ShutdownAsync(CancellationToken.None);
+    }
+
    /// <summary>A write the driver refuses (NotSupportedException) is not a comms fault either.</summary>
    [Fact]
    public async Task Write_refusal_does_not_degrade()