diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs index 97754d94..60da49b5 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs @@ -423,7 +423,50 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover // ---- IReadable ---- + /// + /// How many devices one reads concurrently. Each device is still read serially (its + /// tags share one PCCC connection, and SLC/MicroLogix controllers accept few), so this bounds the number of + /// devices — not requests — in flight, which is what keeps a dead device from starving a healthy one. + /// + internal const int MaxConcurrentDevices = 8; + + /// + /// Consecutive comms faults within one device that mark it DOWN (opening its backoff window and failing the + /// rest of the pass fast). Two, not one: a single tag that times out (an odd address, a slow file) must not + /// blacken every sibling on a device that is answering. + /// + internal const int FaultsToMarkDeviceDown = 2; + + /// One resolved reference queued for its device's pass. + private sealed record ReadItem(int Index, string Reference, AbLegacyTagDefinition Def, AbLegacyAddress Parsed); + + /// + /// What one wire operation proved. = the device replied (a Good value OR a + /// deterministic device refusal such as "file not found") — the evidence that advances + /// LastSuccessfulRead; = a comms failure (timeout, dropped connection, garbage + /// reply); = failed fast inside the device's backoff window (Bad, Degraded, but no new + /// evidence for the backoff); = nothing reached the wire or it says nothing about + /// reachability (a local refusal, a bad index value). + /// + private enum OpVerdict { Neutral, Answered, Fault, Throttled } + + /// The health evidence one device pass produced (collected, then written once per read). + private struct DeviceReadOutcome + { + public bool AnyGood; + public bool AnyAnswered; + public string? FaultMessage; + } + /// + /// + /// Review 07 P7: references are bucketed by device and the devices are read concurrently (at most + /// ); within a device the tags are read in order. A device that produces + /// consecutive comms faults opens its + /// window and every remaining (and every later) read on it fails fast with BadCommunicationError, + /// touching no wire, until the window expires or the probe sees it answer. Health is written ONCE per call: + /// worst outcome wins, and LastSuccessfulRead advances when ANY device answered. + /// public async Task> ReadAsync( IReadOnlyList fullReferences, CancellationToken cancellationToken) { @@ -431,124 +474,255 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover var epoch = CurrentEpoch; var now = DateTime.UtcNow; var results = new DataValueSnapshot[fullReferences.Count]; + var devices = _devices; // one snapshot of the published table for the whole call + var buckets = new Dictionary>(ReferenceEqualityComparer.Instance); for (var i = 0; i < fullReferences.Count; i++) { var reference = fullReferences[i]; - if (!_resolver.TryResolve(reference, out var def)) + if (!_resolver.TryResolve(reference, out var def) || !devices.TryGetValue(def.DeviceHostAddress, out var device)) { results[i] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadNodeIdUnknown, null, now); continue; } - if (!_devices.TryGetValue(def.DeviceHostAddress, out var device)) + var parsed = AbLegacyAddress.TryParse(def.Address, device.Options.PlcFamily); + if (parsed is null) { + WarnMalformedOnce(device, def); results[i] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadNodeIdUnknown, null, now); continue; } + if (!buckets.TryGetValue(device, out var list)) buckets[device] = list = []; + list.Add(new ReadItem(i, reference, def, parsed)); + } + if (buckets.Count == 0) return results; - try + var passes = buckets.ToArray(); + var outcomes = new DeviceReadOutcome[passes.Length]; + using (var window = new SemaphoreSlim(MaxConcurrentDevices, MaxConcurrentDevices)) + { + var tasks = new Task[passes.Length]; + for (var d = 0; d < passes.Length; d++) + tasks[d] = RunPassAsync(d); + // WhenAll completes only once every pass has finished (it never short-circuits on a fault), so the + // window cannot be disposed under a pass still waiting on it. + await Task.WhenAll(tasks).ConfigureAwait(false); + + async Task RunPassAsync(int d) { - var preParsed = AbLegacyAddress.TryParse(def.Address, device.Options.PlcFamily); - if (preParsed is null) - { - WarnMalformedOnce(device, def); - results[i] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadNodeIdUnknown, null, now); - continue; - } - if (preParsed.IsIndirect) - { - // Indirect: read the index word(s) NOW (never cached - the index can change between - // polls), compose the concrete address, then run the normal path against it. - var res = await ResolveIndirectAsync(device, def, preParsed, cancellationToken).ConfigureAwait(false); - if (res.Def is null) - { - results[i] = new DataValueSnapshot(null, res.Status, null, now); - if (res.TransportFault) - SetHealthIfCurrent(epoch, new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, - $"transport fault resolving indirect address for {reference}")); - continue; - } - def = res.Def; - } - - var runtime = await EnsureTagRuntimeAsync(device, def, cancellationToken).ConfigureAwait(false); - - int status; - object? value; - // Serialise the Read → GetStatus → DecodeValue sequence against the shared - // runtime — the server read path and the poll loop both call ReadAsync against - // the same cached libplctag Tag handle, which is not concurrency-safe. - var opLock = device.GetRuntimeLock(def.Name); - await opLock.WaitAsync(cancellationToken).ConfigureAwait(false); + await window.WaitAsync(cancellationToken).ConfigureAwait(false); try { - await runtime.ReadAsync(cancellationToken).ConfigureAwait(false); - status = runtime.GetStatus(); - var parsed = AbLegacyAddress.TryParse(def.Address, device.Options.PlcFamily); - if (status != 0) - value = null; - else if (IsArrayTag(def)) - value = runtime.DecodeArray(def.DataType, EffectiveArrayLength(def)); - else - value = runtime.DecodeValue(def.DataType, parsed?.BitIndex); + outcomes[d] = await ReadDevicePassAsync(passes[d].Key, passes[d].Value, results, now, cancellationToken) + .ConfigureAwait(false); } finally { - opLock.Release(); + window.Release(); } - - if (status != 0) - { - // Evict the stale handle so the next call re-creates it (mirrors AbCip). - EvictRuntime(device, def.Name); - results[i] = new DataValueSnapshot(null, - AbLegacyStatusMapper.MapStatus(status), null, now); - SetHealthIfCurrent(epoch, new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, - $"transport status {status} reading {reference}")); - if (!device.FirstNonZeroStatusLogged) - { - device.FirstNonZeroStatusLogged = true; - _logger.LogWarning( - "AbLegacy non-zero transport status. Driver={DriverInstanceId} Device={DeviceHostAddress} Reference={Reference} Status={Status}", - _driverInstanceId, def.DeviceHostAddress, reference, status); - } - continue; - } - // Healthy read — re-arm the per-device first-failure log so a future non-zero - // status logs again rather than being suppressed by an old flag from a prior outage. - device.FirstNonZeroStatusLogged = false; - - results[i] = new DataValueSnapshot(value, AbLegacyStatusMapper.Good, now, now); - SetHealthIfCurrent(epoch, new DriverHealth(DriverState.Healthy, now, null)); - } - catch (OperationCanceledException) { throw; } - catch (AbTransportException ex) when (IsLocalRefusal(ex)) - { - // The runtime refused the operation before the wire (e.g. a native PLC-5 non-N read). - // Not a comms fault: BadNotSupported, keep the handle, leave health alone. - WarnRefusalOnce(device, def, "read", ex.Message); - results[i] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadNotSupported, null, now); - } - catch (NotSupportedException nse) - { - // A deterministic decode refusal (review 07 H16) — the same answer every poll. BadNotSupported, keep - // the handle (re-creating it only re-paid the init read), warn once, and NO health write. - WarnRefusalOnce(device, def, "read", nse.Message); - results[i] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadNotSupported, null, now); - } - catch (Exception ex) - { - // Transport exception — evict so the next read creates a fresh handle (mirrors AbCip). - EvictRuntime(device, def.Name); - results[i] = new DataValueSnapshot(null, - AbLegacyStatusMapper.BadCommunicationError, null, now); - SetHealthIfCurrent(epoch, new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message)); } } + ApplyReadHealth(epoch, outcomes, now); return results; } + /// + /// Writes the read's health once, after every device pass: Degraded if any pass hit a comms fault, else + /// Healthy if anything read Good; LastSuccessfulRead advances when any device answered (a Good read or a + /// deterministic device refusal — both prove the device is reachable). A read that reached no device's + /// wire, or only drew refusals, does not change the state. + /// + private void ApplyReadHealth(long epoch, DeviceReadOutcome[] outcomes, DateTime now) + { + var anyGood = false; + var anyAnswered = false; + string? fault = null; + foreach (var o in outcomes) + { + anyGood |= o.AnyGood; + anyAnswered |= o.AnyAnswered; + fault ??= o.FaultMessage; + } + if (fault is null && !anyGood && !anyAnswered) return; + + lock (_lifecycleLock) + { + if (epoch != _lifecycleEpoch) return; + var current = _health; + var lastRead = anyAnswered ? now : current.LastSuccessfulRead; + if (fault is not null) + _health = new DriverHealth(DriverState.Degraded, lastRead, fault); + else if (anyGood) + _health = new DriverHealth(DriverState.Healthy, lastRead, null); + else + _health = current with { LastSuccessfulRead = lastRead }; + } + } + + /// Reads one device's references in order, failing fast once the device is marked down. + private async Task ReadDevicePassAsync( + DeviceState device, List items, DataValueSnapshot[] results, DateTime now, CancellationToken ct) + { + var outcome = new DeviceReadOutcome(); + foreach (var item in items) + { + if (device.IsThrottled(DateTime.UtcNow)) + { + results[item.Index] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadCommunicationError, null, now); + outcome.FaultMessage ??= $"device {device.Options.HostAddress} is in its reconnect backoff window"; + continue; + } + + var (verdict, message) = await ReadOneAsync(device, item, results, now, ct).ConfigureAwait(false); + switch (verdict) + { + case OpVerdict.Answered: + outcome.AnyAnswered = true; + if (results[item.Index].StatusCode == AbLegacyStatusMapper.Good) outcome.AnyGood = true; + device.RecordAnswered(); + break; + case OpVerdict.Fault: + outcome.FaultMessage ??= message; + device.RecordCommsFault(DateTime.UtcNow); + break; + case OpVerdict.Throttled: + outcome.FaultMessage ??= message; + break; + } + } + return outcome; + } + + /// Reads one reference; writes its snapshot and reports what the operation proved. + private async Task<(OpVerdict Verdict, string? Message)> ReadOneAsync( + DeviceState device, ReadItem item, DataValueSnapshot[] results, DateTime now, CancellationToken ct) + { + var def = item.Def; + var reference = item.Reference; + try + { + if (item.Parsed.IsIndirect) + { + // Indirect: read the index word(s) NOW (never cached - the index can change between + // polls), compose the concrete address, then run the normal path against it. + var res = await ResolveIndirectAsync(device, def, item.Parsed, ct).ConfigureAwait(false); + if (res.Def is null) + { + results[item.Index] = new DataValueSnapshot(null, res.Status, null, now); + return res.TransportFault + ? (OpVerdict.Fault, $"transport fault resolving indirect address for {reference}") + : (OpVerdict.Neutral, null); + } + def = res.Def; + } + + var runtime = await EnsureTagRuntimeAsync(device, def, ct).ConfigureAwait(false); + + int status; + object? value; + // Serialise the Read → GetStatus → DecodeValue sequence against the shared + // runtime — the server read path and the poll loop both call ReadAsync against + // the same cached libplctag Tag handle, which is not concurrency-safe. + var opLock = device.GetRuntimeLock(def.Name); + await opLock.WaitAsync(ct).ConfigureAwait(false); + try + { + await runtime.ReadAsync(ct).ConfigureAwait(false); + status = runtime.GetStatus(); + var parsed = AbLegacyAddress.TryParse(def.Address, device.Options.PlcFamily); + if (status != 0) + value = null; + else if (IsArrayTag(def)) + value = runtime.DecodeArray(def.DataType, EffectiveArrayLength(def)); + else + value = runtime.DecodeValue(def.DataType, parsed?.BitIndex); + } + finally + { + opLock.Release(); + } + + if (status != 0) + { + results[item.Index] = new DataValueSnapshot(null, AbLegacyStatusMapper.MapStatus(status), null, now); + if (IsDeviceRefusal((AbStatus)status)) + { + // The device ANSWERED with a deterministic refusal (file/element missing, out of bounds…): per-tag + // Bad, the handle is kept (re-creating it would only re-pay the init read), warned once per tag, + // and it is evidence of reachability rather than a comms fault. + WarnRefusalOnce(device, def, "read", $"device status {(AbStatus)status}"); + return (OpVerdict.Answered, null); + } + // Evict the stale handle so the next call re-creates it (mirrors AbCip). + EvictRuntime(device, def.Name); + if (!device.FirstNonZeroStatusLogged) + { + device.FirstNonZeroStatusLogged = true; + _logger.LogWarning( + "AbLegacy non-zero transport status. Driver={DriverInstanceId} Device={DeviceHostAddress} Reference={Reference} Status={Status}", + _driverInstanceId, def.DeviceHostAddress, reference, status); + } + return (OpVerdict.Fault, $"transport status {status} reading {reference}"); + } + // Healthy read — re-arm the per-device first-failure log so a future non-zero + // status logs again rather than being suppressed by an old flag from a prior outage. + device.FirstNonZeroStatusLogged = false; + results[item.Index] = new DataValueSnapshot(value, AbLegacyStatusMapper.Good, now, now); + return (OpVerdict.Answered, null); + } + catch (OperationCanceledException) { throw; } + catch (ConnectionThrottledException tex) + { + // A cold connect refused inside the backoff window: no wire, but the tag has no value either. + results[item.Index] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadCommunicationError, null, now); + return (OpVerdict.Throttled, tex.Message); + } + catch (AbTransportException ex) when (IsLocalRefusal(ex)) + { + // The runtime refused the operation before the wire (e.g. a native PLC-5 non-N read). + // Not a comms fault: BadNotSupported, keep the handle, leave health alone. + WarnRefusalOnce(device, def, "read", ex.Message); + results[item.Index] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadNotSupported, null, now); + return (OpVerdict.Neutral, null); + } + catch (AbTransportException ex) when (IsDeviceRefusal(ex.Status)) + { + // The native runtime raises a device's refusal (e.g. STS 0x20 "file not found") as an exception. + WarnRefusalOnce(device, def, "read", ex.Message); + results[item.Index] = new DataValueSnapshot(null, AbLegacyStatusMapper.MapStatus(ex.Status), null, now); + return (OpVerdict.Answered, null); + } + catch (NotSupportedException nse) + { + // A deterministic decode refusal (review 07 H16) — the same answer every poll. BadNotSupported, keep + // the handle (re-creating it only re-paid the init read), warn once, and NO health write. + WarnRefusalOnce(device, def, "read", nse.Message); + results[item.Index] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadNotSupported, null, now); + return (OpVerdict.Neutral, null); + } + catch (Exception ex) + { + // Transport exception — evict so the next read creates a fresh handle (mirrors AbCip). + EvictRuntime(device, def.Name); + results[item.Index] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadCommunicationError, null, now); + return (OpVerdict.Fault, ex.Message); + } + } + + /// + /// True for a status that proves the DEVICE answered and refused this one operation deterministically — the + /// file or element does not exist, the access is not allowed, the data type or size is wrong. Such a status + /// is per-tag, keeps the handle, and is not a comms fault. Everything else non-zero (timeouts, dropped or + /// refused connections, malformed or partial replies, unknown codes) is treated as a comms fault. + /// is classified separately (). + /// + /// The transport status. + /// for a deterministic device refusal. + internal static bool IsDeviceRefusal(AbStatus status) => status is + AbStatus.NotFound or AbStatus.NotAllowed or AbStatus.OutOfBounds or AbStatus.TooLarge or AbStatus.TooSmall + or AbStatus.TypeMismatch or AbStatus.DeviceStateConflict; + // ---- IWritable ---- /// @@ -558,6 +732,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover ArgumentNullException.ThrowIfNull(writes); var epoch = CurrentEpoch; var results = new WriteResult[writes.Count]; + var devices = _devices; // one snapshot of the published table for the whole call for (var i = 0; i < writes.Count; i++) { @@ -572,7 +747,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover results[i] = new WriteResult(AbLegacyStatusMapper.BadNotWritable); continue; } - if (!_devices.TryGetValue(def.DeviceHostAddress, out var device)) + if (!devices.TryGetValue(def.DeviceHostAddress, out var device)) { results[i] = new WriteResult(AbLegacyStatusMapper.BadNodeIdUnknown); continue; @@ -619,14 +794,6 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover continue; } - // PCCC bit-within-word writes — RMW against a parallel parent-word runtime (strip the /N - // bit suffix). The per-parent-word lock serialises concurrent bit writers. Applies to every - // bit-addressable PCCC file: N-file (N7:0/3), B-file (B3:0/0), and the I/O image files - // (I:0/0, O:1/2); L-file bits RMW a 32-bit parent, the rest a 16-bit word. T/C/R sub-elements - // (T4:0.DN, even when authored Bit-typed) never reach this path: AbLegacyAddress.TryParse rejects a - // sub-element combined with a /N bit index, so their BitIndex is always null. NOTE: an Input-image (I) write is sent - // to the PLC like any other write — the device drives the input image from physical inputs and - // may reject it; we surface that real PCCC status rather than pre-rejecting at the driver. // Array writes (review 07 U10): convert and length-check BEFORE any runtime exists, then encode every // element. A bit / sub-element array has no flat encoding. Array? arrayValue = null; @@ -643,6 +810,14 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover continue; } } + // PCCC bit-within-word writes — RMW against a parallel parent-word runtime (strip the /N + // bit suffix). The per-parent-word lock serialises concurrent bit writers. Applies to every + // bit-addressable PCCC file: N-file (N7:0/3), B-file (B3:0/0), and the I/O image files + // (I:0/0, O:1/2); L-file bits RMW a 32-bit parent, the rest a 16-bit word. T/C/R sub-elements + // (T4:0.DN, even when authored Bit-typed) never reach this path: AbLegacyAddress.TryParse rejects a + // sub-element combined with a /N bit index, so their BitIndex is always null. NOTE: an Input-image (I) write is sent + // to the PLC like any other write — the device drives the input image from physical inputs and + // may reject it; we surface that real PCCC status rather than pre-rejecting at the driver. else if (def.DataType == AbLegacyDataType.Bit && parsed.BitIndex is int bit) { results[i] = new WriteResult( @@ -669,18 +844,38 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover opLock.Release(); } - if (status != 0) + if (status == 0) { - // Evict the stale handle so the next call re-creates it (mirrors AbCip). - EvictRuntime(device, def.Name); + device.RecordAnswered(); + results[i] = new WriteResult(AbLegacyStatusMapper.Good); + } + else if (IsDeviceRefusal((AbStatus)status)) + { + // The device answered and refused this write deterministically: keep the handle, warn once. + device.RecordAnswered(); + WarnRefusalOnce(device, def, "write", $"device status {(AbStatus)status}"); results[i] = new WriteResult(AbLegacyStatusMapper.MapStatus(status)); } else { - results[i] = new WriteResult(AbLegacyStatusMapper.Good); + // Evict the stale handle so the next call re-creates it (mirrors AbCip). + EvictRuntime(device, def.Name); + device.RecordCommsFault(DateTime.UtcNow); + results[i] = new WriteResult(AbLegacyStatusMapper.MapStatus(status)); } } catch (OperationCanceledException) { throw; } + catch (ConnectionThrottledException) + { + // The device is in its reconnect backoff window: fail fast, no wire, no new evidence. + results[i] = new WriteResult(AbLegacyStatusMapper.BadCommunicationError); + } + catch (AbTransportException ex) when (IsDeviceRefusal(ex.Status)) + { + device.RecordAnswered(); + WarnRefusalOnce(device, def, "write", ex.Message); + results[i] = new WriteResult(AbLegacyStatusMapper.MapStatus(ex.Status)); + } catch (NotSupportedException nse) { // Deterministic (review 07 H16): nothing reached the wire, so it says nothing about the device. @@ -706,6 +901,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover { // Transport exception — evict so the next write/read creates a fresh handle (mirrors AbCip). EvictRuntime(device, def.Name); + device.RecordCommsFault(DateTime.UtcNow); results[i] = new WriteResult(AbLegacyStatusMapper.BadCommunicationError); SetHealthIfCurrent(epoch, new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message)); } @@ -799,14 +995,21 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover state.ProbeInitialized = true; } await probeRuntime.ReadAsync(ct).ConfigureAwait(false); - success = probeRuntime.GetStatus() == 0; + var status = probeRuntime.GetStatus(); + success = status == 0; + // The probe's verdict feeds the data path's backoff (review 07 P7): an answer resets the window so a + // recovered device is read again at once; a comms fault counts toward marking it down. + if (success || IsDeviceRefusal((AbStatus)status)) state.RecordAnswered(); + else state.RecordCommsFault(DateTime.UtcNow); } catch (OperationCanceledException) when (ct.IsCancellationRequested) { break; } - catch + catch (Exception ex) { try { probeRuntime?.Dispose(); } catch { } probeRuntime = null; state.ProbeInitialized = false; + if (ex is AbTransportException ate && IsDeviceRefusal(ate.Status)) state.RecordAnswered(); + else if (!IsLocalRefusal(ex)) state.RecordCommsFault(DateTime.UtcNow); } TransitionDeviceState(state, success ? HostState.Running : HostState.Stopped); @@ -1043,10 +1246,16 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover await runtime.ReadAsync(ct).ConfigureAwait(false); } catch (OperationCanceledException) { throw; } + catch (ConnectionThrottledException) { throw; } // the caller answers it as a fail-fast catch (AbTransportException ex) when (IsLocalRefusal(ex)) { return (null, AbLegacyStatusMapper.BadNotSupported, false); } + catch (AbTransportException ex) when (IsDeviceRefusal(ex.Status)) + { + // The device answered: the index file/element does not exist. Per-tag, not a comms fault. + return (null, AbLegacyStatusMapper.MapStatus(ex.Status), false); + } catch { EvictParentRuntime(device, runtimeName); @@ -1054,6 +1263,8 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover } var status = runtime.GetStatus(); + if (status != 0 && IsDeviceRefusal((AbStatus)status)) + return (null, AbLegacyStatusMapper.MapStatus(status), false); if (status != 0) { EvictParentRuntime(device, runtimeName); @@ -1093,32 +1304,56 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover private async Task EnsureParentRuntimeAsync( AbLegacyDriver.DeviceState device, string parentName, CancellationToken ct) { - // Fast path: runtime already cached. - if (device.ParentRuntimes.TryGetValue(parentName, out var existing)) return existing; - - // Slow path: serialise creation per key so concurrent callers don't each create a - // runtime and one of them gets overwritten + leaked. Only one caller initialises; the - // others find the entry on the second TryGetValue inside the lock. - var creationLock = device.GetCreationLock($"parent:{parentName}"); - await creationLock.WaitAsync(ct).ConfigureAwait(false); - try - { - if (device.ParentRuntimes.TryGetValue(parentName, out existing)) return existing; - - var runtime = _tagFactory.Create(new AbLegacyTagCreateParams( + var timeout = _options.Timeout; + return await CreateSingleFlightAsync(device, device.ParentRuntimes, parentName, () => _tagFactory.Create( + new AbLegacyTagCreateParams( Gateway: device.ParsedAddress.Gateway, Port: device.ParsedAddress.Port, CipPath: device.EffectiveCipPath, LibplctagPlcAttribute: device.Profile.LibplctagPlcAttribute, TagName: parentName, - Timeout: _options.Timeout)); - await InitializeOrKeepRefusedAsync(runtime, () => device.Cache(device.ParentRuntimes, parentName, runtime), ct) - .ConfigureAwait(false); + Timeout: timeout)), ct).ConfigureAwait(false); + } + + /// + /// Thrown when a cold connect is refused because the device's window is open + /// — a fail-fast that costs no wire traffic. Reads and writes answer it with BadCommunicationError. + /// + internal sealed class ConnectionThrottledException(string message) : Exception(message); + + /// + /// Returns the cached runtime for , or creates and initialises one. The COLD path is + /// single-flight per DEVICE (, review 07 P7): concurrent callers for + /// different cold tags of one device connect one at a time, so a down device sees one connect attempt instead + /// of one per caller, and inside an open backoff window no attempt is made at all + /// (). The HOT path (cached handle) never touches the gate. The + /// caller records the outcome on the device's backoff (once — recording here as well would double-count). + /// + private static async Task CreateSingleFlightAsync( + DeviceState device, + System.Collections.Concurrent.ConcurrentDictionary table, + string key, + Func create, + CancellationToken ct) + { + if (table.TryGetValue(key, out var existing)) return existing; + + await device.ConnectGate.WaitAsync(ct).ConfigureAwait(false); + try + { + // Re-check under the gate: a sibling caller may have connected this very key while we waited. + if (table.TryGetValue(key, out existing)) return existing; + if (device.IsThrottled(DateTime.UtcNow)) + throw new ConnectionThrottledException( + $"AbLegacy connect for '{key}' on {device.Options.HostAddress} throttled: the device is in its reconnect backoff window."); + + var runtime = create(); + await InitializeOrKeepRefusedAsync(runtime, () => device.Cache(table, key, runtime), ct).ConfigureAwait(false); return runtime; } finally { - creationLock.Release(); + device.ConnectGate.Release(); } } @@ -1214,30 +1449,14 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover private async Task EnsureTagRuntimeAsync( DeviceState device, AbLegacyTagDefinition def, CancellationToken ct) { - // Fast path: runtime already cached. if (device.Runtimes.TryGetValue(def.Name, out var existing)) return existing; - // Slow path: serialise creation per tag name so concurrent callers for the same tag - // (server read path + poll loop) don't both create a runtime and one gets leaked. - var creationLock = device.GetCreationLock($"tag:{def.Name}"); - await creationLock.WaitAsync(ct).ConfigureAwait(false); - try - { - if (device.Runtimes.TryGetValue(def.Name, out existing)) return existing; - - var parsed = AbLegacyAddress.TryParse(def.Address, device.Options.PlcFamily) - ?? throw new InvalidOperationException( - $"AbLegacy tag '{def.Name}' has malformed Address '{def.Address}' for PLC family {device.Options.PlcFamily}."); - - var runtime = _tagFactory.Create(CreateParamsFor(device, def, parsed)); - await InitializeOrKeepRefusedAsync(runtime, () => device.Cache(device.Runtimes, def.Name, runtime), ct) - .ConfigureAwait(false); - return runtime; - } - finally - { - creationLock.Release(); - } + var parsed = AbLegacyAddress.TryParse(def.Address, device.Options.PlcFamily) + ?? throw new InvalidOperationException( + $"AbLegacy tag '{def.Name}' has malformed Address '{def.Address}' for PLC family {device.Options.PlcFamily}."); + var createParams = CreateParamsFor(device, def, parsed); + return await CreateSingleFlightAsync(device, device.Runtimes, def.Name, () => _tagFactory.Create(createParams), ct) + .ConfigureAwait(false); } /// @@ -1329,7 +1548,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover /// avoids the check-then-act race present on a plain Dictionary: two concurrent /// EnsureTagRuntimeAsync callers for the same key both miss the lookup on a /// plain dict and both create + store a runtime, leaking the loser. Access is guarded - /// by a per-key creation semaphore () so exactly one + /// by the per-device connect gate () so exactly one /// runtime is created per tag name. /// public System.Collections.Concurrent.ConcurrentDictionary Runtimes { get; } = @@ -1348,20 +1567,56 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover new(StringComparer.OrdinalIgnoreCase); /// - /// Per-key creation locks for and . - /// A caller holds this before the TryGetValue + Create + InitializeAsync + TryAdd - /// sequence so that a concurrent caller waits rather than creating a duplicate runtime - /// that would be leaked on . + /// Per-device connect gate (review 07 P7): every COLD runtime create + initialise on this device runs under + /// it, so connects are single-flight per device — a down device sees one attempt at a time, not one per + /// concurrent caller — and a concurrent caller for the same key finds the winner's runtime instead of + /// creating (and leaking) a duplicate. A cached runtime never touches it. Not disposed on retirement, for + /// the same reason as the per-key gates (an in-flight caller may still release it). /// - private readonly System.Collections.Concurrent.ConcurrentDictionary _creationLocks = new(StringComparer.OrdinalIgnoreCase); + public SemaphoreSlim ConnectGate { get; } = new(1, 1); /// - /// Gets or creates the creation lock for the specified key. + /// Per-device reconnect throttle: opened after consecutive comms faults, + /// 1 s doubling to 30 s, reset by any answer from the device (data path or probe). Not internally + /// synchronised — every touch is under . /// - /// The lock key. - /// The semaphore slim for the key. - public SemaphoreSlim GetCreationLock(string key) => - _creationLocks.GetOrAdd(key, _ => new SemaphoreSlim(1, 1)); + private readonly ConnectionBackoff _backoff = new(TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(30)); + + private readonly object _backoffLock = new(); + private int _faultStreak; + + /// True while the device's backoff window is open: operations fail fast without the wire. + /// The current UTC time. + /// when no attempt should be made now. + public bool IsThrottled(DateTime nowUtc) + { + lock (_backoffLock) return !_backoff.ShouldAttempt(nowUtc); + } + + /// The device answered (a Good value or a deterministic refusal): reset the streak and the window. + public void RecordAnswered() + { + lock (_backoffLock) + { + _faultStreak = 0; + _backoff.RecordSuccess(); + } + } + + /// + /// A comms fault: the th consecutive one marks the device down — opens + /// (or, after a failed retry, lengthens) the backoff window and restarts the streak. + /// + /// The current UTC time (the window is measured from here). + public void RecordCommsFault(DateTime nowUtc) + { + lock (_backoffLock) + { + if (++_faultStreak < FaultsToMarkDeviceDown) return; + _faultStreak = 0; + _backoff.RecordFailure(nowUtc); + } + } private readonly System.Collections.Concurrent.ConcurrentDictionary _rmwLocks = new(); @@ -1445,11 +1700,10 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover public void ForgetGates(string runtimeKey) { _runtimeLocks.TryRemove(runtimeKey, out _); - _creationLocks.TryRemove($"tag:{runtimeKey}", out _); } - /// The number of per-key gates currently held (runtime + creation + RMW). Tests only. - internal int GateCount => _runtimeLocks.Count + _creationLocks.Count + _rmwLocks.Count; + /// The number of per-key gates currently held (runtime + RMW). Tests only. + internal int GateCount => _runtimeLocks.Count + _rmwLocks.Count; private volatile bool _retired; diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.IntegrationTests/AbLegacyDeadDeviceIsolationTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.IntegrationTests/AbLegacyDeadDeviceIsolationTests.cs new file mode 100644 index 00000000..2e5be188 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.IntegrationTests/AbLegacyDeadDeviceIsolationTests.cs @@ -0,0 +1,65 @@ +using System.Diagnostics; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.IntegrationTests; + +/// +/// Review 07 P7 (High), wire level: one driver serving a live ab_server device AND a dead device (a +/// blackholed address, so every connect waits out the full timeout). Before the fix the dead device's 20 tags +/// cost 20 x Timeout per poll and the live device was read once per cycle; now the live device reads Good in +/// the same call, the dead device costs at most two timeouts, and the next poll does not touch it at all. +/// Read-only. +/// +[Collection(AbLegacyServerCollection.Name)] +[Trait("Category", "Integration")] +[Trait("Simulator", "ab_server-PCCC")] +public sealed class AbLegacyDeadDeviceIsolationTests(AbLegacyServerFixture sim) +{ + // TEST-NET-1 (RFC 5737): never routed, so a connect blackholes until the driver's timeout. + private const string DeadUri = "ab://192.0.2.1:44818/1,0"; + + /// The live device is served promptly while a blackholed sibling is throttled. + /// The simulator profile. + /// The transport stack the driver runs on. + [AbLegacyTheory] + [MemberData(nameof(TransportMatrix.Profiles), MemberType = typeof(TransportMatrix))] + public async Task Dead_device_does_not_stall_the_live_one(AbLegacyServerProfile profile, AbLegacyTransport transport) + { + if (sim.SkipReasonFor(profile) is { } skip) Assert.Skip(skip); + + var liveUri = $"ab://{sim.Host}:{sim.PortFor(profile)}/{sim.CipPath}"; + var timeout = TimeSpan.FromSeconds(2); + var dead = Enumerable.Range(0, 20) + .Select(i => new AbLegacyTagDefinition($"Dead{i}", DeadUri, $"N7:{i}", AbLegacyDataType.Int)) + .ToArray(); + await using var drv = new AbLegacyDriver(new AbLegacyDriverOptions + { + Devices = [new AbLegacyDeviceOptions(liveUri, profile.Family), new AbLegacyDeviceOptions(DeadUri, profile.Family)], + RawTags = AbLegacyRawTags.Entries( + [new AbLegacyTagDefinition("Live", liveUri, "N7:0", AbLegacyDataType.Int), .. dead]), + Timeout = timeout, + Probe = new AbLegacyProbeOptions { Enabled = false }, + Transport = transport, + }, driverInstanceId: $"ablegacy-p7-{profile.Family}-{transport}"); + var ct = TestContext.Current.CancellationToken; + await drv.InitializeAsync("{}", ct); + string[] refs = ["Live", .. dead.Select(d => d.Name)]; + + var sw = Stopwatch.StartNew(); + var first = await drv.ReadAsync(refs, ct); + sw.Stop(); + + first[0].StatusCode.ShouldBe(AbLegacyStatusMapper.Good, "the live device must be served in the same poll"); + first.Skip(1).ShouldAllBe(s => s.StatusCode != AbLegacyStatusMapper.Good); + sw.Elapsed.ShouldBeLessThan(timeout * 3, "a dead device may cost at most two timeouts per backoff window"); + + sw.Restart(); + var second = await drv.ReadAsync(refs, ct); + sw.Stop(); + second[0].StatusCode.ShouldBe(AbLegacyStatusMapper.Good); + sw.Elapsed.ShouldBeLessThan(timeout, "inside the backoff window the dead device is not dialled"); + drv.GetHealth().State.ShouldBe(DriverState.Degraded); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyDeviceIsolationTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyDeviceIsolationTests.cs new file mode 100644 index 00000000..a28b8b60 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyDeviceIsolationTests.cs @@ -0,0 +1,259 @@ +using System.Diagnostics; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport; + +namespace ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests; + +/// +/// Review 07 P7 (High): reads were strictly serial across every device with no per-device backoff, so one dead +/// device cost Timeout per tag per poll and starved every healthy device — long enough to trip the +/// comms-loss watchdog against a reachable one. Devices are now read in parallel (bounded), each device pass +/// stops paying for a dead device after two consecutive comms faults and opens a per-device +/// window, and cold connects are single-flight per device. +/// +[Trait("Category", "Unit")] +public sealed class AbLegacyDeviceIsolationTests +{ + private const string HostA = "ab://10.0.0.5/1,0"; + private const string HostB = "ab://10.0.0.6/1,0"; + + /// A tag on a dead device: each wire operation burns and then fails. + private sealed class DeadTag(AbLegacyTagCreateParams p, TimeSpan cost) : FakeAbLegacyTag(p) + { + public static int WireOps; + + public override async Task InitializeAsync(CancellationToken ct) + { + Interlocked.Increment(ref WireOps); + await Task.Delay(cost, ct); + throw new System.Net.Sockets.SocketException(10060); + } + + public override async Task ReadAsync(CancellationToken ct) + { + Interlocked.Increment(ref WireOps); + await Task.Delay(cost, ct); + throw new System.Net.Sockets.SocketException(10060); + } + } + + private static AbLegacyDriver NewDriver(FakeAbLegacyTagFactory factory, params AbLegacyTagDefinition[] tags) => + new(new AbLegacyDriverOptions + { + Devices = [new AbLegacyDeviceOptions(HostA), new AbLegacyDeviceOptions(HostB)], + RawTags = AbLegacyRawTags.Entries(tags), + Probe = new AbLegacyProbeOptions { Enabled = false }, + }, "drv-p7", factory); + + private static AbLegacyTagDefinition[] Tags(string prefix, string host, int n, string file = "N7") => + [.. Enumerable.Range(0, n).Select(i => new AbLegacyTagDefinition($"{prefix}{i}", host, $"{file}:{i * 3}", AbLegacyDataType.Int))]; + + [Fact] + public async Task Dead_device_costs_at_most_two_timeouts_then_fails_fast_and_never_stalls_the_healthy_one() + { + var cost = TimeSpan.FromMilliseconds(150); + var factory = new FakeAbLegacyTagFactory + { + Customise = p => p.Gateway == "10.0.0.6" ? new DeadTag(p, cost) : new FakeAbLegacyTag(p) { Value = 7 }, + }; + var a = Tags("A", HostA, 5); + var b = Tags("B", HostB, 20); + var drv = NewDriver(factory, [.. a, .. b]); + await drv.InitializeAsync("{}", CancellationToken.None); + var refs = a.Concat(b).Select(t => t.Name).ToList(); + var before = drv.GetHealth().LastSuccessfulRead; + Interlocked.Exchange(ref DeadTag.WireOps, 0); + + var sw = Stopwatch.StartNew(); + var first = await drv.ReadAsync(refs, CancellationToken.None); + sw.Stop(); + + // Serial-with-no-backoff would pay 20 x 150 ms = 3 s on device B alone. + sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(1.5)); + Volatile.Read(ref DeadTag.WireOps).ShouldBeLessThanOrEqualTo(2); + first.Take(5).ShouldAllBe(s => s.StatusCode == AbLegacyStatusMapper.Good); // positive control: A is served + first.Skip(5).ShouldAllBe(s => s.StatusCode == AbLegacyStatusMapper.BadCommunicationError); + var health = drv.GetHealth(); + health.State.ShouldBe(DriverState.Degraded); + health.LastSuccessfulRead.ShouldNotBeNull(); + health.LastSuccessfulRead!.Value.ShouldBeGreaterThan(before ?? DateTime.MinValue); // A's success advances it + + // Inside B's backoff window the next poll touches B's wire not at all. + var opsAfterFirst = Volatile.Read(ref DeadTag.WireOps); + sw.Restart(); + var second = await drv.ReadAsync(refs, CancellationToken.None); + sw.Stop(); + Volatile.Read(ref DeadTag.WireOps).ShouldBe(opsAfterFirst); + sw.Elapsed.ShouldBeLessThan(TimeSpan.FromMilliseconds(500)); + second.Take(5).ShouldAllBe(s => s.StatusCode == AbLegacyStatusMapper.Good); + await drv.ShutdownAsync(CancellationToken.None); + } + + [Fact] + public async Task Devices_are_read_in_parallel() + { + // Device A's read cannot complete until device B's read has STARTED: a serial walk deadlocks here. + var bStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var factory = new FakeAbLegacyTagFactory + { + Customise = p => p.Gateway == "10.0.0.5" + ? new GatedTag(p, wait: bStarted.Task) { Value = 1 } + : new GatedTag(p, signal: bStarted) { Value = 2 }, + }; + var drv = NewDriver(factory, + new AbLegacyTagDefinition("A", HostA, "N7:0", AbLegacyDataType.Int), + new AbLegacyTagDefinition("B", HostB, "N7:0", AbLegacyDataType.Int)); + await drv.InitializeAsync("{}", CancellationToken.None); + + var r = await drv.ReadAsync(["A", "B"], CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(5)); + + r[0].Value.ShouldBe(1); + r[1].Value.ShouldBe(2); + await drv.ShutdownAsync(CancellationToken.None); + } + + private sealed class GatedTag(AbLegacyTagCreateParams p, Task? wait = null, TaskCompletionSource? signal = null) + : FakeAbLegacyTag(p) + { + public override async Task ReadAsync(CancellationToken ct) + { + signal?.TrySetResult(); + if (wait is not null) await wait.WaitAsync(ct); + await base.ReadAsync(ct); + } + } + + [Fact] + public async Task One_timing_out_tag_does_not_blacken_the_rest_of_its_device() + { + var factory = new FakeAbLegacyTagFactory + { + Customise = p => p.TagName == "N7:0" + ? new FakeAbLegacyTag(p) { ThrowOnRead = true, Exception = new TimeoutException() } + : new FakeAbLegacyTag(p) { Value = 5 }, + }; + var drv = NewDriver(factory, Tags("A", HostA, 4)); + await drv.InitializeAsync("{}", CancellationToken.None); + + for (var poll = 0; poll < 3; poll++) + { + var r = await drv.ReadAsync(["A0", "A1", "A2", "A3"], CancellationToken.None); + r[0].StatusCode.ShouldBe(AbLegacyStatusMapper.BadCommunicationError); + r.Skip(1).ShouldAllBe(s => s.StatusCode == AbLegacyStatusMapper.Good, $"poll {poll}"); + } + drv.GetHealth().State.ShouldBe(DriverState.Degraded); + await drv.ShutdownAsync(CancellationToken.None); + } + + [Fact] + public async Task Device_refusal_is_per_tag_Bad_advances_LastSuccessfulRead_and_keeps_the_handle() + { + var logger = new AbLegacyTestLogger(); + var factory = new FakeAbLegacyTagFactory + { + Customise = p => new FakeAbLegacyTag(p) { Status = (int)AbStatus.NotFound }, + }; + var drv = new AbLegacyDriver(new AbLegacyDriverOptions + { + Devices = [new AbLegacyDeviceOptions(HostA)], + RawTags = AbLegacyRawTags.Entries(new AbLegacyTagDefinition("X", HostA, "N7:0", AbLegacyDataType.Int)), + Probe = new AbLegacyProbeOptions { Enabled = false }, + }, "drv-p7", factory, logger); + await drv.InitializeAsync("{}", CancellationToken.None); + var initHealth = drv.GetHealth(); + await Task.Delay(20); + + var r1 = await drv.ReadAsync(["X"], CancellationToken.None); + var r2 = await drv.ReadAsync(["X"], CancellationToken.None); + + r1[0].StatusCode.ShouldBe(AbLegacyStatusMapper.BadNodeIdUnknown); + r2[0].StatusCode.ShouldBe(AbLegacyStatusMapper.BadNodeIdUnknown); + var health = drv.GetHealth(); + health.State.ShouldBe(DriverState.Healthy); // the device answered: not a comms fault + health.LastSuccessfulRead!.Value.ShouldBeGreaterThan(initHealth.LastSuccessfulRead!.Value); + factory.Tags["N7:0"].InitializeCount.ShouldBe(1); + factory.Tags["N7:0"].ReadCount.ShouldBe(2); // positive control: both polls really asked the device + logger.Warnings.Count(w => w.Contains("Reference=X")).ShouldBe(1); + await drv.ShutdownAsync(CancellationToken.None); + } + + [Fact] + public async Task Cold_connects_on_one_device_are_single_flight() + { + var inFlight = 0; + var maxInFlight = 0; + var factory = new FakeAbLegacyTagFactory + { + Customise = p => new CountingInitTag(p, () => + { + var now = Interlocked.Increment(ref inFlight); + int seen; + while (now > (seen = Volatile.Read(ref maxInFlight)) && Interlocked.CompareExchange(ref maxInFlight, now, seen) != seen) { } + }, () => Interlocked.Decrement(ref inFlight)) { Value = 1 }, + }; + var tags = Tags("A", HostA, 6); + var drv = NewDriver(factory, tags); + await drv.InitializeAsync("{}", CancellationToken.None); + + // Three concurrent callers (poll loop, server read, write) each hitting different cold tags of one device. + await Task.WhenAll( + drv.ReadAsync(["A0", "A1"], CancellationToken.None), + drv.ReadAsync(["A2", "A3"], CancellationToken.None), + drv.WriteAsync([new WriteRequest("A4", 3), new WriteRequest("A5", 4)], CancellationToken.None)); + + maxInFlight.ShouldBe(1); + factory.Tags.Values.Count(t => t.InitializeCount == 1).ShouldBe(6); // positive control: all six connected + await drv.ShutdownAsync(CancellationToken.None); + } + + private sealed class CountingInitTag(AbLegacyTagCreateParams p, Action enter, Action exit) : FakeAbLegacyTag(p) + { + public override async Task InitializeAsync(CancellationToken ct) + { + enter(); + try + { + await Task.Delay(30, ct); + await base.InitializeAsync(ct); + } + finally { exit(); } + } + } + + [Fact] + public async Task A_recovered_device_is_read_again_once_its_window_expires() + { + var dead = true; + var creates = 0; + var factory = new FakeAbLegacyTagFactory + { + Customise = p => + { + Interlocked.Increment(ref creates); + return dead + ? new FakeAbLegacyTag(p) { ThrowOnInitialize = true, Exception = new System.Net.Sockets.SocketException(10061) } + : new FakeAbLegacyTag(p) { Value = 9 }; + }, + }; + var drv = NewDriver(factory, new AbLegacyTagDefinition("B", HostB, "N7:0", AbLegacyDataType.Int)); + await drv.InitializeAsync("{}", CancellationToken.None); + + // Two consecutive comms faults mark the device down and open its first (1 s) window. + (await drv.ReadAsync(["B"], CancellationToken.None))[0].StatusCode.ShouldBe(AbLegacyStatusMapper.BadCommunicationError); + (await drv.ReadAsync(["B"], CancellationToken.None))[0].StatusCode.ShouldBe(AbLegacyStatusMapper.BadCommunicationError); + var attempts = Volatile.Read(ref creates); + dead = false; + // Inside the window: fails fast without a connect attempt, although the device is back. + (await drv.ReadAsync(["B"], CancellationToken.None))[0].StatusCode.ShouldBe(AbLegacyStatusMapper.BadCommunicationError); + Volatile.Read(ref creates).ShouldBe(attempts); + await Task.Delay(TimeSpan.FromMilliseconds(1100)); + + var recovered = await drv.ReadAsync(["B"], CancellationToken.None); + recovered[0].StatusCode.ShouldBe(AbLegacyStatusMapper.Good); + recovered[0].Value.ShouldBe(9); + drv.GetHealth().State.ShouldBe(DriverState.Healthy); + await drv.ShutdownAsync(CancellationToken.None); + } +}