fix(focas): a returned status never ends a device's batch; round-2 minors (#740 review round 2)

N1: linkDown is set only when NO reply arrived, i.e. a timeout, a transport exception or a connect failure. A status the CNC returned proves it answered. EW_REJECT and every unlisted rc map to BadCommunicationError, so a deterministic rc on one tag used to fail every later tag of a reachable control on every poll. The comms-classed status still degrades health. The coalescer already treated a returned refusal this way; it is unchanged. Tests: a returned EW_REJECT on tag 1 leaves tags 2 and 3 read and Good (red before). A timeout still fails the rest without reading them (positive control). Minors: - m1: the _lifecycleLock doc now covers the health transitions it also guards. - m2: the NoteFixedTreeFailed summary says a refusal credits the answer. - m3: ReadAsync stamps a failed batch's exception with the generation the batch started under, and HandlePollError degrades that generation only. - m4: the fixed-tree "blip" absence check runs unconditionally under a wide budget, and asserts that both ticks landed inside it. - m5: direct unit cases for the coalescer link-down paths. A transport fault on the first range fails the second range and the per-tag tag without reading them. A connect failure costs one connect per batch. CncRelay.Up() retries the re-bind. Refs #740 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
2026-10-22 14:55:37 -04:00
parent 95b75c2e58
commit 09e0fe1841
5 changed files with 190 additions and 22 deletions
@@ -59,8 +59,9 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
    private IReadOnlyDictionary<string, DeviceState> Devices => Volatile.Read(ref _devices);

    /// <summary>
-    ///     Guards the publish/retire of the device table against an overlapping lifecycle call. Held only for
-    ///     synchronous swaps — never across an await, and nothing a loop runs ever takes it.
+    ///     Guards the publish/retire of the device table against an overlapping lifecycle call, and makes each
+    ///     generation-checked health transition atomic (<see cref="UpdateHealth"/>). Held only for synchronous swaps
+    ///     and single health transitions — never across an await, and nothing is called under it.
    /// </summary>
    private readonly object _lifecycleLock = new();

@@ -200,7 +201,9 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
    internal void HandlePollError(Exception ex)
    {
        _logger.LogWarning(ex, "FOCAS poll reader failed. Driver={DriverInstanceId}", _driverInstanceId);
-        DegradeHealth(CurrentGeneration, ex.Message);
+        // The generation the failed batch started under, when ReadAsync stamped it; otherwise the live one.
+        var generation = ex.Data.Contains(BatchGenerationKey) && ex.Data[BatchGenerationKey] is int g ? g : CurrentGeneration;
+        DegradeHealth(generation, ex.Message);
    }

    /// <inheritdoc />
@@ -696,19 +699,35 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
        }

        if (byDevice is null) return results;
-        if (byDevice.Count == 1)
+        try
        {
-            foreach (var (device, indices) in byDevice)
-                await ReadDeviceAsync(device, indices, fullReferences, results, now, generation, cancellationToken).ConfigureAwait(false);
+            if (byDevice.Count == 1)
+            {
+                foreach (var (device, indices) in byDevice)
+                    await ReadDeviceAsync(device, indices, fullReferences, results, now, generation, cancellationToken).ConfigureAwait(false);
+            }
+            else
+            {
+                await Task.WhenAll(byDevice.Select(kv =>
+                    ReadDeviceAsync(kv.Key, kv.Value, fullReferences, results, now, generation, cancellationToken))).ConfigureAwait(false);
+            }
        }
-        else
+        catch (Exception ex) when (ex is not OperationCanceledException)
        {
-            await Task.WhenAll(byDevice.Select(kv =>
-                ReadDeviceAsync(kv.Key, kv.Value, fullReferences, results, now, generation, cancellationToken))).ConfigureAwait(false);
+            // Stamp the generation this batch STARTED under (review round 2, m3), so the poll engine's onError
+            // (HandlePollError) cannot degrade a generation that re-initialised after the batch began.
+            ex.Data[BatchGenerationKey] = generation;
+            throw;
        }
        return results;
    }

+    /// <summary><see cref="Exception.Data"/> key carrying the lifecycle generation a failed batch started under.</summary>
+    internal const string BatchGenerationKey = "FocasDriver.BatchGeneration";
+
+    /// <summary>Test seam — the live lifecycle generation.</summary>
+    internal int LifecycleGeneration => CurrentGeneration;
+
    /// <summary>
    ///     Reads one device's share of a batch, in request order: the PMC coalescing pre-pass first, then every
    ///     index it did not serve through the per-tag path. Each result lands at its own index, so concurrent
@@ -718,10 +737,11 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
    ///     <b>One link failure ends the device's batch (#740 review I1).</b> Concurrency alone does not bound a
    ///     batch: <c>Task.WhenAll</c> waits for the slowest device, and a CNC that accepts TCP and then stalls costs
    ///     the full call timeout on EVERY tag — so a 10-tag stalled device held the healthy devices' values back by
-    ///     ten timeouts. After the first timeout / transport failure (or a comms-class status) on this device, its
-    ///     remaining tags in the batch fail fast with <c>BadCommunicationError</c>, exactly as the coalescer's
+    ///     ten timeouts. After the first timeout / transport failure / connect failure on this device — i.e. NO
+    ///     reply arrived — its remaining tags in the batch fail fast with <c>BadCommunicationError</c>, exactly as the coalescer's
    ///     <c>FailMembers</c> does for a merged range. Fixed-tree tags are still served from the cache: they need no
-    ///     link. A CNC REFUSAL does not end the batch — the link answered.
+    ///     link. Any status the CNC RETURNED — a refusal, or a comms-classed rc such as EW_REJECT — does not end the
+    ///     batch: a reply proves the link answered.
    /// </remarks>
    private async Task ReadDeviceAsync(
        DeviceState device, List<int> indices, IReadOnlyList<string> fullReferences,
@@ -792,8 +812,11 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
                }
                else if (IsCommsStatus(status))
                {
+                    // Degrade, but do NOT end the batch (#740 review round 2, N1): this status came back in a reply,
+                    // so the CNC answered — EW_REJECT and every unlisted rc map to BadCommunicationError. Ending the
+                    // batch here starved every later tag of a reachable control forever (authored order is fixed).
+                    // A session that is really dead throws or times out on the next tag, which does end it.
                    DegradeHealth(generation, $"FOCAS status 0x{status:X8} reading {reference}");
-                    linkDown = true;
                }
                else
                {
@@ -2059,8 +2082,8 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery,
    }

    /// <summary>
-    ///     A failed fixed-tree tick. A CNC REFUSAL (non-OK rc, transport fine) is not a comms loss and clears any
-    ///     failure without crediting a read; a transport failure degrades the driver only once the device has been
+    ///     A failed fixed-tree tick. A CNC REFUSAL (non-OK rc, transport fine) is not a comms loss: it clears any
+    ///     failure AND credits the answer to LastSuccessfulRead (the CNC replied); a transport failure degrades the driver only once the device has been
    ///     silent past <see cref="FixedTreeStaleBudget"/> (measured from its last answer, or from loop start when
    ///     it never answered), so one dropped tick on a slow control does not flap <c>/hosts</c>.
    /// </summary>
@@ -88,7 +88,19 @@ public sealed class WireBackendHealthAndLayoutTests
            }
        }

-        public void Up() => Listen(Port);
+        /// <summary>
+        ///     Listens again on the SAME port (the driver's host string names it). The re-bind is retried briefly: the
+        ///     OS can hold the port for a moment after <see cref="Down"/>, or hand it to another process — the latter
+        ///     surfaces as a clear failure after the retries rather than a random flake.
+        /// </summary>
+        public void Up()
+        {
+            for (var attempt = 1; ; attempt++)
+            {
+                try { Listen(Port); return; }
+                catch (SocketException) when (attempt < 20) { Thread.Sleep(100); }
+            }
+        }

        public ValueTask DisposeAsync()
        {
@@ -153,6 +153,108 @@ public sealed class FocasConcurrentDeviceReadTests
        await drv.ShutdownAsync(CancellationToken.None);
    }

+    /// <summary>Counts per-tag reads; one address can be made to time out (throw) or answer a status.</summary>
+    private sealed class CountingClient : FakeFocasClient
+    {
+        public int Reads;
+        public string? TimeoutAt;
+
+        public override async Task<(object? value, uint status)> ReadAsync(
+            FocasAddress address, FocasDataType type, CancellationToken ct)
+        {
+            Interlocked.Increment(ref Reads);
+            if (address.Canonical == TimeoutAt)
+            {
+                await Task.Delay(Timeout.Infinite, ct).ConfigureAwait(false);
+            }
+            return await base.ReadAsync(address, type, ct).ConfigureAwait(false);
+        }
+    }
+
+    private static FocasDriver OneDevice(FakeFocasClientFactory factory, bool coalesce, params string[] addresses) => new(
+        new FocasDriverOptions
+        {
+            Devices = [new FocasDeviceOptions(HostA, DeviceName: "A")],
+            RawTags = [.. addresses.Select(a => FocasRawTags.Entry(new FocasTagDefinition($"raw/{a}", "A", a, FocasDataType.Byte)))],
+            Probe = new FocasProbeOptions { Enabled = false },
+            PmcCoalesce = new FocasPmcCoalesceOptions { Enabled = coalesce },
+            Timeout = TimeSpan.FromMilliseconds(200),
+        }, "focas-linkdown", factory);
+
+    /// <summary>
+    ///     #740 review round 2 N1 — a status the CNC RETURNED (EW_REJECT and every unlisted rc map to
+    ///     BadCommunicationError) proves the device answered, so it must not end the device's batch: the tags after it
+    ///     still read. Before the fix, a deterministic rc on the first tag starved every later tag forever.
+    /// </summary>
+    [Fact]
+    public async Task A_returned_comms_status_does_not_fail_the_rest_of_the_batch()
+    {
+        var client = new CountingClient { Values = { ["R101"] = (sbyte)1, ["R102"] = (sbyte)2 } };
+        client.ReadStatuses["R100"] = FocasStatusMapper.MapFocasReturn(13); // EW_REJECT ⇒ BadCommunicationError
+        var drv = OneDevice(new FakeFocasClientFactory { Customise = () => client }, coalesce: false, "R100", "R101", "R102");
+        await drv.InitializeAsync("{}", CancellationToken.None);
+
+        var snaps = await drv.ReadAsync(["raw/R100", "raw/R101", "raw/R102"], CancellationToken.None);
+
+        snaps[0].StatusCode.ShouldBe(FocasStatusMapper.BadCommunicationError);
+        snaps[1].StatusCode.ShouldBe(FocasStatusMapper.Good);
+        snaps[2].Value.ShouldBe((sbyte)2);
+        client.Reads.ShouldBe(3, "every tag was really read");
+        await drv.ShutdownAsync(CancellationToken.None);
+    }
+
+    /// <summary>Positive control for N1: a TIMEOUT (no reply at all) still fails the rest of the batch without reading it.</summary>
+    [Fact]
+    public async Task A_timeout_still_fails_the_rest_of_the_batch_fast()
+    {
+        var client = new CountingClient { TimeoutAt = "R100", Values = { ["R101"] = (sbyte)1 } };
+        var drv = OneDevice(new FakeFocasClientFactory { Customise = () => client }, coalesce: false, "R100", "R101", "R102");
+        await drv.InitializeAsync("{}", CancellationToken.None);
+
+        var snaps = await drv.ReadAsync(["raw/R100", "raw/R101", "raw/R102"], CancellationToken.None);
+
+        snaps.ShouldAllBe(s => s.StatusCode == FocasStatusMapper.BadCommunicationError);
+        client.Reads.ShouldBe(1, "only the tag that timed out touched the wire");
+        await drv.ShutdownAsync(CancellationToken.None);
+    }
+
+    /// <summary>
+    ///     Review round 2 m5 — the coalescer's link-down paths: a transport fault on the first merged range fails the
+    ///     second range WITHOUT a second range read, and the per-tag tags of the same device without a per-tag read.
+    /// </summary>
+    [Fact]
+    public async Task Coalescer_transport_fault_fails_later_ranges_and_per_tag_tags_without_reading_them()
+    {
+        var client = new CountingClient { ThrowOnPmcRange = true };
+        // Two ranges far apart (gap > MaxGapBytes) + one lone tag that takes the per-tag path.
+        var drv = OneDevice(new FakeFocasClientFactory { Customise = () => client }, coalesce: true,
+            "R100", "R101", "R500", "R501", "D900");
+        await drv.InitializeAsync("{}", CancellationToken.None);
+
+        var snaps = await drv.ReadAsync(["raw/R100", "raw/R101", "raw/R500", "raw/R501", "raw/D900"], CancellationToken.None);
+
+        snaps.ShouldAllBe(s => s.StatusCode == FocasStatusMapper.BadCommunicationError);
+        client.PmcRangeReads.Count.ShouldBe(1, "the second range must not pay another transport attempt");
+        client.Reads.ShouldBe(0, "the per-tag tag must not pay one either");
+        await drv.ShutdownAsync(CancellationToken.None);
+    }
+
+    /// <summary>Review round 2 m5 — a coalescer CONNECT failure fails the per-tag tags of the device without a second connect.</summary>
+    [Fact]
+    public async Task Coalescer_connect_failure_fails_per_tag_tags_without_reconnecting()
+    {
+        var factory = new FakeFocasClientFactory { Customise = () => new CountingClient { ThrowOnConnect = true } };
+        var drv = OneDevice(factory, coalesce: true, "R100", "R101", "D900");
+        await drv.InitializeAsync("{}", CancellationToken.None);
+
+        var snaps = await drv.ReadAsync(["raw/R100", "raw/R101", "raw/D900"], CancellationToken.None);
+
+        snaps.ShouldAllBe(s => s.StatusCode == FocasStatusMapper.BadCommunicationError);
+        factory.Clients.Count.ShouldBe(1, "one connect attempt per device per batch");
+        factory.Clients.Sum(c => c.ConnectCount).ShouldBe(1);
+        await drv.ShutdownAsync(CancellationToken.None);
+    }
+
    /// <summary>Results land at their own indices whatever order the devices finish in.</summary>
    [Fact]
    public async Task Results_keep_their_request_order_across_devices()
@@ -18,6 +18,7 @@ public sealed class FocasFixedTreeHealthTests
 {
    private const string Host = "focas://10.0.0.5:8193";
    private static readonly TimeSpan Stale = TimeSpan.FromMilliseconds(300);
+    private static readonly TimeSpan BlipBudget = TimeSpan.FromMilliseconds(1500);

    /// <summary>A CNC that can be switched off (transport failure) or made to refuse the dynamic read.</summary>
    private sealed class CncSwitch
@@ -58,7 +59,7 @@ public sealed class FocasFixedTreeHealthTests
        }
    }

-    private static FocasDriver FixedOnlyDriver(CncSwitch sw) => new(new FocasDriverOptions
+    private static FocasDriver FixedOnlyDriver(CncSwitch sw, TimeSpan? budget = null) => new(new FocasDriverOptions
    {
        Devices = [new FocasDeviceOptions(Host)],
        RawTags = [FocasRawTags.Entry(new FocasTagDefinition("raw/X", Host, "FIXED:Axes/X/AbsolutePosition", FocasDataType.Float64))],
@@ -71,7 +72,7 @@ public sealed class FocasFixedTreeHealthTests
        },
    }, "focas-fixed-health", new FakeFocasClientFactory { Customise = () => new SwitchableCnc(sw) })
    {
-        FixedTreeStaleBudgetOverride = Stale,
+        FixedTreeStaleBudgetOverride = budget ?? Stale,
    };

    /// <summary>
@@ -83,7 +84,9 @@ public sealed class FocasFixedTreeHealthTests
    public async Task Fixed_only_driver_degrades_when_its_cnc_disappears_and_recovers_when_it_returns()
    {
        var sw = new CncSwitch();
-        await using var drv = FixedOnlyDriver(sw);
+        // A wide budget (review round 2 m4): two 20 ms ticks always land well inside it, so the "blip does not
+        // degrade" absence check below runs unconditionally instead of being skipped on a slow agent.
+        await using var drv = FixedOnlyDriver(sw, BlipBudget);
        await drv.InitializeAsync("{}", CancellationToken.None);
        await WaitUntil(() => drv.GetDeviceState(Host)?.LastFixedPollUtc is not null);

@@ -96,9 +99,9 @@ public sealed class FocasFixedTreeHealthTests
        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");
+        // … inside the budget (asserted, not assumed), and the driver is still Healthy.
+        (DateTime.UtcNow - lastAnswer).ShouldBeLessThan(BlipBudget, "the two failed ticks must land inside the budget");
+        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;
        var failuresAtDegrade = Volatile.Read(ref sw.Failures);
@@ -57,4 +57,32 @@ public sealed class FocasPollErrorHealthTests

        drv.GetHealth().State.ShouldBe(DriverState.Faulted);
    }
+
+    /// <summary>
+    ///     #740 review round 2 m3 — a poll failure carries the generation its batch STARTED under, so a batch that
+    ///     began before a re-initialise cannot degrade the new generation. Positive control: a failure stamped with
+    ///     the live generation (or unstamped) does degrade.
+    /// </summary>
+    [Fact]
+    public async Task PollReaderFailure_from_a_previous_generation_does_not_degrade_the_new_one()
+    {
+        var drv = new FocasDriver(new FocasDriverOptions
+        {
+            Devices = [new FocasDeviceOptions("focas://10.0.0.5:8193")],
+            Probe = new FocasProbeOptions { Enabled = false },
+        }, "focas-1", new FakeFocasClientFactory());
+        await drv.InitializeAsync("{}", CancellationToken.None);
+        var oldGeneration = drv.LifecycleGeneration;
+        await drv.InitializeAsync("{}", CancellationToken.None);
+
+        var stale = new InvalidOperationException("old batch boom");
+        stale.Data[FocasDriver.BatchGenerationKey] = oldGeneration;
+        drv.HandlePollError(stale);
+        drv.GetHealth().State.ShouldBe(DriverState.Healthy);
+
+        var live = new InvalidOperationException("live batch boom");
+        live.Data[FocasDriver.BatchGenerationKey] = drv.LifecycleGeneration;
+        drv.HandlePollError(live);
+        drv.GetHealth().State.ShouldBe(DriverState.Degraded);
+    }
 }