diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs index 5177119b..bebe023a 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs @@ -59,8 +59,9 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, private IReadOnlyDictionary Devices => Volatile.Read(ref _devices); /// - /// 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 (). Held only for synchronous swaps + /// and single health transitions — never across an await, and nothing is called under it. /// 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); } /// @@ -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; } + /// key carrying the lifecycle generation a failed batch started under. + internal const string BatchGenerationKey = "FocasDriver.BatchGeneration"; + + /// Test seam — the live lifecycle generation. + internal int LifecycleGeneration => CurrentGeneration; + /// /// 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, /// One link failure ends the device's batch (#740 review I1). Concurrency alone does not bound a /// batch: Task.WhenAll 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 BadCommunicationError, 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 BadCommunicationError, exactly as the coalescer's /// FailMembers 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. /// private async Task ReadDeviceAsync( DeviceState device, List indices, IReadOnlyList 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, } /// - /// 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 (measured from its last answer, or from loop start when /// it never answered), so one dropped tick on a slow control does not flap /hosts. /// diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.IntegrationTests/Series/WireBackendHealthAndLayoutTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.IntegrationTests/Series/WireBackendHealthAndLayoutTests.cs index 30d39394..c704b21f 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.IntegrationTests/Series/WireBackendHealthAndLayoutTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.IntegrationTests/Series/WireBackendHealthAndLayoutTests.cs @@ -88,7 +88,19 @@ public sealed class WireBackendHealthAndLayoutTests } } - public void Up() => Listen(Port); + /// + /// 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 , or hand it to another process — the latter + /// surfaces as a clear failure after the retries rather than a random flake. + /// + public void Up() + { + for (var attempt = 1; ; attempt++) + { + try { Listen(Port); return; } + catch (SocketException) when (attempt < 20) { Thread.Sleep(100); } + } + } public ValueTask DisposeAsync() { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasConcurrentDeviceReadTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasConcurrentDeviceReadTests.cs index 23e0e4e3..f4fa7d11 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasConcurrentDeviceReadTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasConcurrentDeviceReadTests.cs @@ -153,6 +153,108 @@ public sealed class FocasConcurrentDeviceReadTests await drv.ShutdownAsync(CancellationToken.None); } + /// Counts per-tag reads; one address can be made to time out (throw) or answer a status. + 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); + + /// + /// #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. + /// + [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); + } + + /// Positive control for N1: a TIMEOUT (no reply at all) still fails the rest of the batch without reading it. + [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); + } + + /// + /// 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. + /// + [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); + } + + /// Review round 2 m5 — a coalescer CONNECT failure fails the per-tag tags of the device without a second connect. + [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); + } + /// Results land at their own indices whatever order the devices finish in. [Fact] public async Task Results_keep_their_request_order_across_devices() diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasFixedTreeHealthTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasFixedTreeHealthTests.cs index fae11d58..ff5aef11 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasFixedTreeHealthTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasFixedTreeHealthTests.cs @@ -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); /// A CNC that can be switched off (transport failure) or made to refuse the dynamic read. 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, }; /// @@ -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); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasPollErrorHealthTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasPollErrorHealthTests.cs index f738c275..3e89a3ff 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasPollErrorHealthTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasPollErrorHealthTests.cs @@ -57,4 +57,32 @@ public sealed class FocasPollErrorHealthTests drv.GetHealth().State.ShouldBe(DriverState.Faulted); } + + /// + /// #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. + /// + [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); + } }