From fd64fed21784253cd3de0c1851e6905894a373be Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 14:44:42 -0400 Subject: [PATCH] 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 --- .../FocasDriver.cs | 203 ++++++++++++------ .../FocasAlarmLayoutTests.cs | 18 ++ .../FocasConcurrentDeviceReadTests.cs | 75 ++++++- .../FocasFixedTreeHealthTests.cs | 53 +++-- .../FocasReadHealthContractTests.cs | 72 +++++++ 5 files changed, 335 insertions(+), 86 deletions(-) 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 c5bb0dda..5177119b 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs @@ -200,10 +200,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, internal void HandlePollError(Exception ex) { _logger.LogWarning(ex, "FOCAS poll reader failed. Driver={DriverInstanceId}", _driverInstanceId); - var current = Volatile.Read(ref _health); - if (current.State != DriverState.Faulted) - Volatile.Write(ref _health, - new DriverHealth(DriverState.Degraded, current.LastSuccessfulRead, ex.Message)); + DegradeHealth(CurrentGeneration, ex.Message); } /// @@ -234,10 +231,11 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, // shutdown) is superseded by that shutdown's bump, and checks for it before publishing anything. var generation = Interlocked.Increment(ref _lifecycleGeneration); await ReleaseDeviceStatesAsync().ConfigureAwait(false); - ResetAlarmHealth(); - _refusalWarned.Clear(); WriteHealthIfCurrent(generation, new DriverHealth(DriverState.Initializing, null, null)); var devices = new Dictionary(StringComparer.OrdinalIgnoreCase); + // Set once this attempt's table is live: from then on it is the NEXT lifecycle call's to release, never + // this attempt's (review minor 1 — a reference compare could release a table twice). + var published = false; try { // adopt a changed config. Options and the backend client factory are re-derived by ONE @@ -361,6 +359,11 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, // leaves the previous snapshot standing rather than a truncated one that would read as // "we now skip fewer tags". _skippedTags.Commit(); + // The per-path health of the previous table, reset only now that this attempt is known to be the + // live one (review minor 5): a superseded attempt must not wipe the health of the call that + // superseded it. + ResetAlarmHealth(); + _refusalWarned.Clear(); Volatile.Write(ref _parsedAddressesByTagName, parsedAddresses); Volatile.Write(ref _tagsByRawPath, tagsByRawPath); Volatile.Write(ref _alarmTags, alarmTags); @@ -370,6 +373,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, _fixedTreeActive = _options.FixedTree.Enabled || parsedAddresses.Values.Any(a => a.Kind == FocasAreaKind.Fixed); Volatile.Write(ref _devices, devices); + published = true; StartDeviceLoops(devices.Values); } @@ -390,7 +394,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, { // Superseded: whatever this attempt built but never published is released here (a published table // is the superseding call's to release); health belongs to that call too. - if (!ReferenceEquals(Devices, devices)) await ReleaseTableAsync(devices.Values).ConfigureAwait(false); + if (!published) await ReleaseTableAsync(devices.Values).ConfigureAwait(false); throw; } catch (Exception ex) @@ -647,7 +651,9 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, ArgumentNullException.ThrowIfNull(fullReferences); var now = DateTime.UtcNow; var results = new DataValueSnapshot[fullReferences.Count]; - // One snapshot of each swapped table for the whole call (see _devices). + // The generation FIRST, then one snapshot of each swapped table for the whole call (see _devices): health + // written by this batch is dropped if a lifecycle call has moved on meanwhile. + var generation = CurrentGeneration; var devices = Devices; var alarmTags = Volatile.Read(ref _alarmTags); var fixedTreeActive = _fixedTreeActive; @@ -693,12 +699,12 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, if (byDevice.Count == 1) { foreach (var (device, indices) in byDevice) - await ReadDeviceAsync(device, indices, fullReferences, results, now, cancellationToken).ConfigureAwait(false); + 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, cancellationToken))).ConfigureAwait(false); + ReadDeviceAsync(kv.Key, kv.Value, fullReferences, results, now, generation, cancellationToken))).ConfigureAwait(false); } return results; } @@ -708,16 +714,25 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, /// index it did not serve through the per-tag path. Each result lands at its own index, so concurrent /// device reads never touch the same slot. /// + /// + /// 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 + /// 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. + /// private async Task ReadDeviceAsync( DeviceState device, List indices, IReadOnlyList fullReferences, - DataValueSnapshot[] results, DateTime now, CancellationToken cancellationToken) + DataValueSnapshot[] results, DateTime now, int generation, CancellationToken cancellationToken) { // Merge adjacent / near PMC reads of the device into fewer pmc_rdpmcrng calls. Indices it serves are // marked here and skipped by the per-tag loop below; everything else (including every single-tag range // and any range the CNC rejected) takes the unchanged per-tag path. - var served = _options.PmcCoalesce.Enabled - ? await CoalescePmcReadsAsync(device, indices, fullReferences, results, now, cancellationToken).ConfigureAwait(false) - : null; + var (served, linkDown) = _options.PmcCoalesce.Enabled + ? await CoalescePmcReadsAsync(device, indices, fullReferences, results, now, generation, cancellationToken).ConfigureAwait(false) + : (null, false); foreach (var i in indices) { @@ -729,29 +744,43 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, continue; } + FocasAddress parsed; + try { parsed = ResolveParsedAddress(def); } + catch (InvalidOperationException) + { + // A malformed address is a config fault of this tag, not of the link. + results[i] = new DataValueSnapshot(null, FocasStatusMapper.BadNodeIdUnknown, null, now); + continue; + } + + // A FIXED: tag names a field of this device's fixed-tree snapshot, which the + // fixed-tree loop already maintains — there is no area read to issue, and no + // connection to force. A null here means the path resolves to nothing on THIS + // device (an unsupported capability, a renamed axis, a typo), which is a genuine + // BadNodeIdUnknown; a device that has not answered yet reports a comms fault + // instead, because the address may well be fine. + if (parsed.Kind == FocasAreaKind.Fixed) + { + results[i] = TryReadFixedTreeOnDevice(device, parsed.FixedPath!, now) + ?? new DataValueSnapshot( + null, + device.LastFixedPollUtc is null + ? FocasStatusMapper.BadCommunicationError + : FocasStatusMapper.BadNodeIdUnknown, + null, + now); + continue; + } + + if (linkDown) + { + // Already failed (and health already degraded) this batch — don't pay another timeout. + results[i] = new DataValueSnapshot(null, FocasStatusMapper.BadCommunicationError, null, now); + continue; + } + try { - var parsed = ResolveParsedAddress(def); - - // A FIXED: tag names a field of this device's fixed-tree snapshot, which the - // fixed-tree loop already maintains — there is no area read to issue, and no - // connection to force. A null here means the path resolves to nothing on THIS - // device (an unsupported capability, a renamed axis, a typo), which is a genuine - // BadNodeIdUnknown; a device that has not answered yet reports a comms fault - // instead, because the address may well be fine. - if (parsed.Kind == FocasAreaKind.Fixed) - { - results[i] = TryReadFixedTreeOnDevice(device, parsed.FixedPath!, now) - ?? new DataValueSnapshot( - null, - device.LastFixedPollUtc is null - ? FocasStatusMapper.BadCommunicationError - : FocasStatusMapper.BadNodeIdUnknown, - null, - now); - continue; - } - var client = await EnsureConnectedAsync(device, cancellationToken).ConfigureAwait(false); var (value, status) = await client.ReadAsync(parsed, def.DataType, cancellationToken).ConfigureAwait(false); @@ -759,17 +788,16 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, if (status == FocasStatusMapper.Good) { _refusalWarned.TryRemove(reference, out _); - Volatile.Write(ref _health, new DriverHealth(DriverState.Healthy, now, null)); + CreditHealth(generation, now); } else if (IsCommsStatus(status)) { - Volatile.Write(ref _health, new DriverHealth(DriverState.Degraded, - Volatile.Read(ref _health).LastSuccessfulRead, - $"FOCAS status 0x{status:X8} reading {reference}")); + DegradeHealth(generation, $"FOCAS status 0x{status:X8} reading {reference}"); + linkDown = true; } else { - NoteRefusal(reference, def.Address, status, now); + NoteRefusal(reference, def.Address, status, now, generation); } } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { throw; } @@ -779,18 +807,48 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, // Timeout budget. Surface a recoverable comm error so the BadWaitingForInitialData // seed is overwritten and health degrades, instead of the read hanging forever. results[i] = new DataValueSnapshot(null, FocasStatusMapper.BadCommunicationError, null, now); - Volatile.Write(ref _health, new DriverHealth(DriverState.Degraded, - Volatile.Read(ref _health).LastSuccessfulRead, $"FOCAS read timed out for {reference}")); + DegradeHealth(generation, $"FOCAS read timed out for {reference}"); + linkDown = true; } catch (Exception ex) { results[i] = new DataValueSnapshot(null, FocasStatusMapper.BadCommunicationError, null, now); - Volatile.Write(ref _health, new DriverHealth(DriverState.Degraded, - Volatile.Read(ref _health).LastSuccessfulRead, ex.Message)); + DegradeHealth(generation, ex.Message); + linkDown = true; } } } + /// + /// Applies a data-path health transition — only while is still the live + /// lifecycle generation (#740 review minor 8): a read that started against the previous table must not + /// write its Degraded onto the next initialise's health. Taken under the lifecycle lock, so the check and + /// the write cannot straddle a publish; the lock is uncontended outside a lifecycle transition. + /// + /// The generation the caller's operation started under. + /// The transition; returns its input unchanged to leave health alone. + private void UpdateHealth(int generation, Func next) + { + lock (_lifecycleLock) + { + if (generation != Volatile.Read(ref _lifecycleGeneration)) return; + Volatile.Write(ref _health, next(Volatile.Read(ref _health))); + } + } + + /// A comms fault: Degraded with LastSuccessfulRead preserved; never downgrades Faulted. + private void DegradeHealth(int generation, string reason) => + UpdateHealth(generation, h => h.State == DriverState.Faulted + ? h + : new DriverHealth(DriverState.Degraded, h.LastSuccessfulRead, reason)); + + /// An answered read: Healthy with LastSuccessfulRead advanced. + private void CreditHealth(int generation, DateTime now) => + UpdateHealth(generation, _ => new DriverHealth(DriverState.Healthy, now, null)); + + /// The live lifecycle generation — captured by a data-path operation when it starts. + private int CurrentGeneration => Volatile.Read(ref _lifecycleGeneration); + /// /// RawPaths whose CNC refusal has already been logged — a refusal warns ONCE per tag, not once per poll, /// and re-arms when the tag next reads Good. Cleared at initialise. @@ -811,17 +869,14 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, /// state change — but the CNC DID answer, so LastSuccessfulRead advances (else a reachable control whose /// only tags are refused would look silent to the comms-loss watchdog). /// - private void NoteRefusal(string reference, string address, uint status, DateTime now) + private void NoteRefusal(string reference, string address, uint status, DateTime now, int generation) { - // CAS, not a plain write: devices are read concurrently, and a read-modify-write could otherwise put back - // the state another device's comms fault had just replaced. - while (true) - { - var current = Volatile.Read(ref _health); - if (current.LastSuccessfulRead is { } last && last >= now) break; - if (ReferenceEquals(Interlocked.CompareExchange(ref _health, current with { LastSuccessfulRead = now }, current), current)) - break; - } + // The link answered, so a value-path Degraded left by an earlier comms fault is over (review minor 2): a + // device that recovers and then only refuses must not stay Degraded forever. Under the lifecycle lock + // (UpdateHealth), so concurrent device reads cannot interleave a read-modify-write. + UpdateHealth(generation, h => h.State is DriverState.Healthy or DriverState.Degraded + ? new DriverHealth(DriverState.Healthy, h.LastSuccessfulRead is { } l && l > now ? l : now, null) + : h); if (_refusalWarned.TryAdd(reference, 0)) _logger.LogWarning( "FOCAS CNC refused {RawPath} ({Address}) with 0x{Status:X8}; the tag reads Bad until it is fixed (warned once). Driver={DriverInstanceId}", @@ -844,9 +899,9 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, /// /// Returns the served indices, or null when nothing was served. /// - private async Task?> CoalescePmcReadsAsync( + private async Task<(HashSet? Served, bool LinkDown)> CoalescePmcReadsAsync( DeviceState device, List indices, IReadOnlyList references, DataValueSnapshot[] results, - DateTime now, CancellationToken ct) + DateTime now, int generation, CancellationToken ct) { List? items = null; foreach (var i in indices) @@ -863,12 +918,13 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, /* malformed / unresolvable — the per-tag loop reports it with its usual status */ } } - if (items is null || items.Count < 2) return null; + if (items is null || items.Count < 2) return (null, false); var plans = FocasPmcCoalescer.Plan(items, _options.PmcCoalesce.MaxSpanBytes, _options.PmcCoalesce.MaxGapBytes); - if (!plans.Any(p => p.Members.Count >= 2 && !device.CoalesceRejected.ContainsKey(p.Key))) return null; + if (!plans.Any(p => p.Members.Count >= 2 && !device.CoalesceRejected.ContainsKey(p.Key))) return (null, false); HashSet? served = null; + var linkDown = false; // One connect attempt per device per batch: if it fails, every PMC tag of the device fails with // that status here instead of paying a connect attempt per plan and again per tag. IFocasClient client; @@ -878,13 +934,20 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, { FailMembers(items, FocasStatusMapper.BadCommunicationError, $"FOCAS connect failed reading PMC tags of {device.Options.HostAddress}"); - return served; + return (served, true); } foreach (var plan in plans) { if (plan.Members.Count < 2) continue; if (device.CoalesceRejected.ContainsKey(plan.Key)) continue; + if (linkDown) + { + // The link failed on an earlier range this batch: fail the rest without another timeout. + FailMembers(plan.Members, FocasStatusMapper.BadCommunicationError, + $"FOCAS transport failure on {device.Options.HostAddress} earlier in this batch"); + continue; + } var members = plan.Members; try @@ -903,7 +966,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, : new DataValueSnapshot(null, FocasStatusMapper.BadOutOfRange, now, now); (served ??= []).Add(m.Index); } - Volatile.Write(ref _health, new DriverHealth(DriverState.Healthy, now, null)); + CreditHealth(generation, now); } else { @@ -924,9 +987,10 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, // Transport failure / timeout: the link is suspect, so fail the members without per-tag retries. FailMembers(members, FocasStatusMapper.BadCommunicationError, $"FOCAS transport failure reading PMC range {plan.Key}"); + linkDown = true; } } - return served; + return (served, linkDown); void FailMembers(IEnumerable failed, uint status, string reason) { @@ -935,9 +999,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, results[m.Index] = new DataValueSnapshot(null, status, null, now); (served ??= []).Add(m.Index); } - Volatile.Write(ref _health, new DriverHealth(DriverState.Degraded, - Volatile.Read(ref _health).LastSuccessfulRead, - $"{reason} (0x{status:X8})")); + DegradeHealth(generation, $"{reason} (0x{status:X8})"); } } @@ -948,6 +1010,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, IReadOnlyList writes, CancellationToken cancellationToken) { ArgumentNullException.ThrowIfNull(writes); + var generation = CurrentGeneration; var results = new WriteResult[writes.Count]; for (var i = 0; i < writes.Count; i++) @@ -998,8 +1061,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, // Per-call timeout (not external cancellation) — the write stalled past the device // Timeout budget. Surface a recoverable comm error rather than aborting the batch. results[i] = new WriteResult(FocasStatusMapper.BadCommunicationError); - Volatile.Write(ref _health, new DriverHealth(DriverState.Degraded, - Volatile.Read(ref _health).LastSuccessfulRead, $"FOCAS write timed out for {w.FullReference}")); + DegradeHealth(generation, $"FOCAS write timed out for {w.FullReference}"); } catch (NotSupportedException nse) { @@ -1019,8 +1081,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, catch (Exception ex) { results[i] = new WriteResult(FocasStatusMapper.BadCommunicationError); - Volatile.Write(ref _health, new DriverHealth(DriverState.Degraded, - Volatile.Read(ref _health).LastSuccessfulRead, ex.Message)); + DegradeHealth(generation, ex.Message); } } @@ -1992,6 +2053,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, /// An answered fixed-tree pass of : clears its failure, advances LastSuccessfulRead. private void NoteFixedTreeAnswered(DeviceState state) { + if (state.IsRetired) return; _fixedTreeFailures.TryRemove(state.Options.HostAddress, out _); Interlocked.Exchange(ref _fixedTreeLastSuccessTicks, DateTime.UtcNow.Ticks); } @@ -2004,9 +2066,12 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, /// private void NoteFixedTreeFailed(DeviceState state, Exception error) { + if (state.IsRetired) return; // a released device's late tick says nothing about the live table if (FocasAlarmProjection.IsCncRefusal(error)) { - _fixedTreeFailures.TryRemove(state.Options.HostAddress, out _); + // The CNC answered — evidence of life (review minor 2): clear the failure AND credit the answer, so a + // control that only refuses after an outage does not look silent to the comms-loss watchdog. + NoteFixedTreeAnswered(state); return; } var since = state.LastFixedPollUtc ?? state.FixedTreeStartedUtc; diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasAlarmLayoutTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasAlarmLayoutTests.cs index bc1bc1b2..467e72c3 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasAlarmLayoutTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasAlarmLayoutTests.cs @@ -84,6 +84,24 @@ public sealed class FocasAlarmLayoutTests alarms[19].Message.ShouldBe("OVERTRAVEL +19"); } + /// + /// Review I2 edge: vendor records with type 0 and EMPTY messages also satisfy the legacy invariants (the + /// int32 read of type<<16 | axis 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. + /// + [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 == ""); + } + /// The unambiguous lengths are unchanged. [Fact] public void Unambiguous_lengths_are_unchanged() 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 9ca35cf7..23e0e4e3 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 @@ -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. /// - 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().ShouldAllBe(c => c.PeakInFlight == 1, + var clients = factory.Clients.Cast().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); + } + + /// A device that accepted TCP and then stopped answering: every read hangs until the call timeout. + 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); + } + } + + /// + /// #740 review I1 — Task.WhenAll 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. + /// + [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); } 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 b83898e8..fae11d58 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 @@ -24,6 +24,10 @@ public sealed class FocasFixedTreeHealthTests { public volatile bool Down; public volatile bool Refuse; + /// Transport failures served so far — what the tests wait on instead of wall-clock delays. + public int Failures; + /// Refusals served so far. + 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 GetSysInfoAsync(CancellationToken ct) => - Down ? throw new IOException("socket closed") : base.GetSysInfoAsync(ct); + public override Task GetSysInfoAsync(CancellationToken ct) + { + if (!Down) return base.GetSysInfoAsync(ct); + Interlocked.Increment(ref sw.Failures); + throw new IOException("socket closed"); + } public override Task 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; } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasReadHealthContractTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasReadHealthContractTests.cs index 88e7f206..923ec354 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasReadHealthContractTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasReadHealthContractTests.cs @@ -112,6 +112,78 @@ public sealed class FocasReadHealthContractTests await drv.ShutdownAsync(CancellationToken.None); } + /// + /// 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. + /// + [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); + } + + /// A read whose CNC answer is held until released, then fails at the transport. + 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"); + } + } + + /// + /// 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. + /// + [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(); + 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); + } + /// A write the driver refuses (NotSupportedException) is not a comms fault either. [Fact] public async Task Write_refusal_does_not_degrade()