diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDecodeException.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDecodeException.cs new file mode 100644 index 00000000..06d9a639 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDecodeException.cs @@ -0,0 +1,17 @@ +namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus; + +/// +/// The device answered with well-framed data the tag's codec cannot represent (a non-BCD nibble in a +/// BCD register, for example). The device is reachable, so this is a per-tag +/// BadDecodingError with a once-per-tag Warning — never a comms fault, never a health write. +/// +public sealed class ModbusDecodeException : Exception +{ + /// Initializes a new instance of the class. + /// A human-readable description. + /// The codec's own exception. + public ModbusDecodeException(string message, Exception? inner = null) + : base(message, inner) + { + } +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs index fc85ddde..648e4c99 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs @@ -43,7 +43,33 @@ public sealed class ModbusDriver // v3 authored-tag table, keyed by RawPath (the tag identity + driver wire reference). RawPath is // case-sensitive ordinal. Built in InitializeAsync from _options.RawTags via the pure mapper. - private readonly Dictionary _tagsByRawPath = new(StringComparer.Ordinal); + // BUILD-AND-SWAP: a fresh dictionary is built off to the side and published by reference + // (Volatile.Write, under _stateLock with the generation check); it is never mutated or .Clear()ed + // in place, because retained poll loops read it concurrently across a re-initialize. + private Dictionary _tagsByRawPath = new(StringComparer.Ordinal); + + /// + /// Lifecycle generation. Every InitializeAsync / ShutdownAsync bumps it under + /// ; every health write and every table/transport publish carries the + /// generation it was computed under and is DROPPED once the generation has moved on — so a + /// superseded (abandoned) initialize can never publish over a newer one, a read that straddled a + /// re-initialize cannot write health about the new transport, and a shutdown's final health + /// write cannot clobber a newer init. + /// + private int _generation; + + /// Guards , the transport/table publish and the health fields. + private readonly object _stateLock = new(); + + // Health is composed from two independent sides so a good write clears only a WRITE-side + // Degraded (and a good read only a read-side one). LastSuccessfulRead advances on any evidence the + // device answered — a good read, a good write, a refusal, or a probe answer. + private string? _readFault; + private string? _writeFault; + + // Once-per-tag Warning for deterministic device refusals (exception 01/02/03, undecodable data). + // Replaced (never cleared in place) on every table publish. + private ConcurrentDictionary _refusalWarned = new(StringComparer.Ordinal); // Resolves a read/write/subscribe RawPath reference to its tag definition via a single hit on the // authored _tagsByRawPath table (v3: the reference is always a RawPath; a miss is a miss). @@ -87,6 +113,11 @@ public sealed class ModbusDriver private IModbusTransport? _transport; private CancellationTokenSource? _probeCts; private CancellationTokenSource? _reprobeCts; + private Task? _probeTask; + private Task? _reprobeTask; + + /// Bounded join on the probe / re-probe loops at teardown — they honour their token, so this is a backstop. + private static readonly TimeSpan BackgroundLoopJoinTimeout = TimeSpan.FromSeconds(2); /// /// Guards / / @@ -105,7 +136,7 @@ public sealed class ModbusDriver /// (a connect that never returns) must not turn shutdown or dispose into a hang — past the /// timeout they tear down anyway, which is the lesser evil of the two. /// - private static readonly TimeSpan LifecycleGateTimeout = TimeSpan.FromSeconds(10); + internal TimeSpan LifecycleGateTimeout { get; set; } = TimeSpan.FromSeconds(10); private DriverHealth _health = new(DriverState.Unknown, null, null); @@ -152,7 +183,7 @@ public sealed class ModbusDriver _driverInstanceId = driverInstanceId; _logger = logger ?? NullLogger.Instance; _resolver = new EquipmentTagRefResolver( - r => _tagsByRawPath.TryGetValue(r, out var t) ? t : null); + r => Volatile.Read(ref _tagsByRawPath).TryGetValue(r, out var t) ? t : null); _transportFactory = transportFactory ?? (o => ModbusTransportFactory.Create(o)); _poll = new PollGroupEngine( @@ -208,9 +239,151 @@ public sealed class ModbusDriver internal void HandlePollError(Exception ex) { _logger.LogWarning(ex, "Modbus poll reader failed. Driver={DriverInstanceId}", _driverInstanceId); - var current = ReadHealth(); - if (current.State != DriverState.Faulted) - WriteHealth(new DriverHealth(DriverState.Degraded, current.LastSuccessfulRead, ex.Message)); + lock (_stateLock) + { + // Lifecycle-owned states are left alone: Faulted is a stronger, config-level signal, and an + // Initializing/Unknown driver has no transport for a poll to have failed against — the + // failure is the re-initialize window itself, which the lifecycle reports on its own. + var current = _health; + if (current.State is DriverState.Faulted or DriverState.Initializing or DriverState.Unknown) return; + _readFault = ex.Message; + PublishComposedHealthLocked(current.LastSuccessfulRead); + } + } + + // ---- health composition ---- + + /// Which half of the driver a batch outcome came from. + private enum HealthSide { Read, Write } + + /// + /// Per-batch accounting: did the device answer anything, did a comms fault happen, and which units + /// (or the whole endpoint) must the rest of this batch fail fast against. + /// + private sealed class BatchOutcome + { + public bool Answered; + public string? CommsFault; + public bool EndpointDown; + private HashSet? _failedUnits; + + public void FailUnit(byte unit) => (_failedUnits ??= []).Add(unit); + + public bool IsFailedFast(byte unit) => EndpointDown || (_failedUnits?.Contains(unit) ?? false); + } + + /// + /// Applies one batch's outcome to the driver health, under , and only when + /// is still current. A batch that touched no device (every reference + /// unknown, every write refused before the wire) changes nothing. + /// + private void ReportOutcome(int generation, HealthSide side, BatchOutcome batch, DateTime now) + { + if (!batch.Answered && batch.CommsFault is null) return; + lock (_stateLock) + { + if (generation != _generation) return; + var current = _health; + if (current.State is DriverState.Faulted or DriverState.Initializing or DriverState.Unknown) return; + if (side == HealthSide.Read) _readFault = batch.CommsFault; + else _writeFault = batch.CommsFault; + PublishComposedHealthLocked(batch.Answered ? now : current.LastSuccessfulRead); + } + } + + /// Writes Healthy/Degraded from the two fault sides. Caller holds . + private void PublishComposedHealthLocked(DateTime? lastSuccessfulRead) + { + var fault = _readFault ?? _writeFault; + WriteHealth(fault is null + ? new DriverHealth(DriverState.Healthy, lastSuccessfulRead, null) + : new DriverHealth(DriverState.Degraded, lastSuccessfulRead, fault)); + } + + /// + /// A probe answer proves the device is reachable: advance LastSuccessfulRead without + /// touching either fault side (it says nothing about the tags). + /// + private void NoteDeviceAnswered(int generation) + { + lock (_stateLock) + { + if (generation != _generation) return; + var current = _health; + if (current.State is not (DriverState.Healthy or DriverState.Degraded)) return; + WriteHealth(current with { LastSuccessfulRead = DateTime.UtcNow }); + } + } + + /// Lifecycle health write, dropped when has been superseded. + private void WriteLifecycleHealth(int generation, DriverHealth health) + { + lock (_stateLock) + { + if (generation != _generation) return; + _readFault = null; + _writeFault = null; + WriteHealth(health); + } + } + + /// + /// Classifies one failed read/write against the health contract and returns the per-tag status: + /// + /// Exception 01/02/03 — a deterministic REFUSAL: the device answered; per-tag Bad, a + /// once-per-tag Warning, no Degraded (it used to latch Degraded and, if every tag refused, + /// trip the 90 s watchdog against a reachable device). + /// Exception 04/05/06 — the device answered but is not serving: Degraded, answered. + /// Exception 0A/0B (and unknown codes) — the gateway answered for a target that did not: + /// Degraded, not answered; 0B fails the rest of that unit's batch fast. + /// Undecodable data — the device answered with a value we cannot represent: per-tag + /// BadDecodingError, once-per-tag Warning, no Degraded. + /// No socket / throttled reconnect — the endpoint is down: Degraded, the rest of the + /// batch fails fast. + /// Any other transport failure (timeout, desync, EOF) — Degraded; the rest of that + /// unit's batch fails fast instead of serialising N timeouts. + /// + /// + private uint ClassifyFailure(Exception ex, byte unit, string tagRef, BatchOutcome batch, bool isWrite) + { + switch (ex) + { + case ModbusException mex when IsRefusalCode(mex.ExceptionCode): + batch.Answered = true; + WarnRefusalOnce(tagRef, isWrite, $"exception 0x{mex.ExceptionCode:X2} (fc 0x{mex.FunctionCode:X2})"); + return MapModbusExceptionToStatus(mex.ExceptionCode); + case ModbusException mex when mex.ExceptionCode is 0x04 or 0x05 or 0x06: + batch.Answered = true; + batch.CommsFault ??= mex.Message; + return MapModbusExceptionToStatus(mex.ExceptionCode); + case ModbusException mex: + batch.CommsFault ??= mex.Message; + if (mex.ExceptionCode == 0x0B) batch.FailUnit(unit); + return MapModbusExceptionToStatus(mex.ExceptionCode); + case ModbusDecodeException dex: + batch.Answered = true; + WarnRefusalOnce(tagRef, isWrite, dex.Message); + return StatusBadDecodingError; + case ModbusTransportUnavailableException or ObjectDisposedException or System.Net.Sockets.SocketException: + batch.CommsFault ??= ex.Message; + batch.EndpointDown = true; + return StatusBadCommunicationError; + default: + batch.CommsFault ??= ex.Message; + batch.FailUnit(unit); + return StatusBadCommunicationError; + } + } + + /// Exception 01 Illegal Function / 02 Illegal Data Address / 03 Illegal Data Value — the device refused the request. + internal static bool IsRefusalCode(byte exceptionCode) => exceptionCode is 0x01 or 0x02 or 0x03; + + private void WarnRefusalOnce(string tagRef, bool isWrite, string detail) + { + if (!Volatile.Read(ref _refusalWarned).TryAdd((isWrite ? "w:" : "r:") + tagRef, 0)) return; + _logger.LogWarning( + "Modbus device refused {Operation} of tag {RawPath}: {Detail}. The tag reads/writes Bad; driver health is unaffected (the device answered). Driver={DriverInstanceId}", + isWrite ? "a write" : "a read", tagRef, detail, _driverInstanceId); } // @@ -310,11 +483,26 @@ public sealed class ModbusDriver /// The gate-held body of . private async Task InitializeCoreAsync(string driverConfigJson, CancellationToken cancellationToken) { - WriteHealth(new DriverHealth(DriverState.Initializing, null, null)); + // Claim a new generation and take the previous transport out of service atomically: any read or + // write that captured the old generation can no longer write health, and nothing can capture the + // old transport from here on. + int generation; + IModbusTransport? previous; + lock (_stateLock) + { + generation = ++_generation; + previous = _transport; + _transport = null; + _readFault = null; + _writeFault = null; + WriteHealth(new DriverHealth(DriverState.Initializing, null, null)); + } + + IModbusTransport? transport = null; try { // Release whatever a PREVIOUS initialize published before building the replacements. All - // three fields assigned below (_transport, _probeCts, _reprobeCts) were simply overwritten, + // fields assigned below (_transport, _probeCts, _reprobeCts) were simply overwritten, // so a second initialize on a live driver stranded a connected transport plus two running // background loops that no later dispose could reach. The IDriver contract lets a // caller re-initialise a live driver without ShutdownAsync, and DriverInstanceActor's retry tick @@ -323,20 +511,9 @@ public sealed class ModbusDriver // every ACTIVE SUBSCRIPTION — and nothing re-subscribes them afterwards, because a // mid-life re-initialize is not a shutdown the OPC UA layer knows to re-drive. A caller // that genuinely wants the subscriptions gone calls ShutdownAsync. A first initialize - // sees three nulls, so this block is a no-op on the initial call. - try { _probeCts?.Cancel(); } catch { } - _probeCts?.Dispose(); - _probeCts = null; - - try { _reprobeCts?.Cancel(); } catch { } - _reprobeCts?.Dispose(); - _reprobeCts = null; - - if (_transport is not null) - { - await _transport.DisposeAsync().ConfigureAwait(false); - _transport = null; - } + // sees nulls, so this block is a no-op on the initial call. + await StopBackgroundLoopsAsync().ConfigureAwait(false); + if (previous is not null) await previous.DisposeAsync().ConfigureAwait(false); // re-parse the supplied DriverConfig so a config change delivered through the IDriver // contract is honoured. Without this the driver keeps serving the options it was CONSTRUCTED @@ -345,17 +522,19 @@ public sealed class ModbusDriver if (HasConfigBody(driverConfigJson)) _options = ModbusDriverFactoryExtensions.ParseOptions(_driverInstanceId, driverConfigJson); - _transport = _transportFactory(_options); - await _transport.ConnectAsync(cancellationToken).ConfigureAwait(false); - // Build the RawPath → definition table from the authored raw tags. Each entry's TagConfig - // blob is mapped by the pure factory; the entry's WriteIdempotent flag is threaded onto the - // def (it lives on the RawTagEntry, not inside the blob). A mapper miss is skipped (logged), - // never thrown — a bad tag must not fail the whole driver init. + transport = _transportFactory(_options); + await transport.ConnectAsync(cancellationToken).ConfigureAwait(false); + + // Build the RawPath → definition table OFF TO THE SIDE from the authored raw tags. Each + // entry's TagConfig blob is mapped by the pure factory; the entry's WriteIdempotent flag is + // threaded onto the def (it lives on the RawTagEntry, not inside the blob). A mapper miss is + // skipped (logged), never thrown — a bad tag must not fail the whole driver init. + var table = new Dictionary(StringComparer.Ordinal); _skippedTags.BeginRebuild(_options.RawTags.Count); foreach (var entry in _options.RawTags) { if (ModbusTagDefinitionFactory.FromTagConfig(entry.TagConfig, entry.RawPath, out var def)) - _tagsByRawPath[entry.RawPath] = def with { WriteIdempotent = entry.WriteIdempotent }; + table[entry.RawPath] = def with { WriteIdempotent = entry.WriteIdempotent }; else { _skippedTags.Skip(entry.RawPath, "TagConfig did not map to a definition"); @@ -365,30 +544,76 @@ public sealed class ModbusDriver } } _skippedTags.Commit(); - WriteHealth(new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null)); - // PR 23: kick off the probe loop once the transport is up. Initial state stays - // Unknown until the first probe tick succeeds — avoids broadcasting a premature - // Running transition before any register round-trip has happened. - if (_options.Probe.Enabled) + // Publish transport + table + Healthy as ONE generation-checked step. A superseded attempt + // (a shutdown/initialize ran while this one was abandoned mid-connect) publishes nothing and + // releases its own transport. + lock (_stateLock) { - _probeCts = new CancellationTokenSource(); - _ = Task.Run(() => ProbeLoopAsync(_probeCts.Token), _probeCts.Token); - } + if (generation != _generation) + throw new OperationCanceledException( + "Modbus initialize was superseded by a later lifecycle operation; its transport was released"); + _transport = transport; + Volatile.Write(ref _tagsByRawPath, table); + Volatile.Write(ref _refusalWarned, new ConcurrentDictionary(StringComparer.Ordinal)); + WriteHealth(new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null)); - if (_options.AutoProhibitReprobeInterval is not null) - { - _reprobeCts = new CancellationTokenSource(); - _ = Task.Run(() => ReprobeLoopAsync(_reprobeCts.Token), _reprobeCts.Token); + // PR 23: kick off the probe loop once the transport is up. Initial state stays + // Unknown until the first probe tick succeeds — avoids broadcasting a premature + // Running transition before any register round-trip has happened. Started inside the + // generation-checked publish so a teardown that bypassed the lifecycle gate can never + // run between the publish and the loop start and strand a loop. + if (_options.Probe.Enabled) + { + var cts = new CancellationTokenSource(); + var probeTransport = transport; + _probeCts = cts; + _probeTask = Task.Run(() => ProbeLoopAsync(generation, probeTransport, cts.Token), cts.Token); + } + + if (_options.AutoProhibitReprobeInterval is not null) + { + var cts = new CancellationTokenSource(); + _reprobeCts = cts; + _reprobeTask = Task.Run(() => ReprobeLoopAsync(cts.Token), cts.Token); + } } + transport = null; // published — owned by the driver now } catch (Exception ex) { - WriteHealth(new DriverHealth(DriverState.Faulted, null, ex.Message)); + // A failed attempt releases the transport it built — it is never left half-published. + if (transport is not null) + { + try { await transport.DisposeAsync().ConfigureAwait(false); } catch { } + } + WriteLifecycleHealth(generation, new DriverHealth(DriverState.Faulted, null, ex.Message)); throw; } } + /// Cancels, bounded-joins and disposes the probe and re-probe loops. Idempotent. + private async Task StopBackgroundLoopsAsync() + { + var probeCts = _probeCts; + var reprobeCts = _reprobeCts; + var tasks = new[] { _probeTask, _reprobeTask }.Where(t => t is not null).Cast().ToArray(); + _probeCts = null; + _reprobeCts = null; + _probeTask = null; + _reprobeTask = null; + + try { probeCts?.Cancel(); } catch { } + try { reprobeCts?.Cancel(); } catch { } + if (tasks.Length > 0) + { + try { await Task.WhenAll(tasks).WaitAsync(BackgroundLoopJoinTimeout).ConfigureAwait(false); } + catch { /* a loop that ignores its token is abandoned, never awaited forever */ } + } + probeCts?.Dispose(); + reprobeCts?.Dispose(); + } + /// public async Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) { @@ -435,9 +660,17 @@ public sealed class ModbusDriver /// . private async Task ShutdownCoreAsync() { - var lastRead = ReadHealth().LastSuccessfulRead; + int generation; + DateTime? lastRead; + lock (_stateLock) + { + // A shutdown is a lifecycle operation too: it supersedes any in-flight (abandoned) initialize, + // and its own final write below is dropped if a NEWER initialize started meanwhile. + generation = ++_generation; + lastRead = _health.LastSuccessfulRead; + } await TeardownAsync().ConfigureAwait(false); - WriteHealth(new DriverHealth(DriverState.Unknown, lastRead, null)); + WriteLifecycleHealth(generation, new DriverHealth(DriverState.Unknown, lastRead, null)); } /// @@ -489,57 +722,83 @@ public sealed class ModbusDriver public async Task> ReadAsync( IReadOnlyList fullReferences, CancellationToken cancellationToken) { - var transport = RequireTransport(); + var (generation, transport, table) = CaptureLifecycle(); var now = DateTime.UtcNow; var results = new DataValueSnapshot[fullReferences.Count]; + var batch = new BatchOutcome(); var coalesced = _options.MaxReadGap > 0 - ? await ReadCoalescedAsync(transport, fullReferences, results, now, cancellationToken).ConfigureAwait(false) + ? await ReadCoalescedAsync(transport, table, fullReferences, results, now, batch, cancellationToken).ConfigureAwait(false) : new HashSet(); for (var i = 0; i < fullReferences.Count; i++) { if (coalesced.Contains(i)) continue; - if (!_resolver.TryResolve(fullReferences[i], out var tag)) + if (!table.TryGetValue(fullReferences[i], out var tag)) { results[i] = new DataValueSnapshot(null, StatusBadNodeIdUnknown, null, now); continue; } + var unit = ResolveUnitId(tag); + if (batch.IsFailedFast(unit)) + { + // Fail fast: this unit (or the whole endpoint) already failed at the transport this batch; + // paying the same timeout once per remaining tag would stall the poll N-fold. + results[i] = new DataValueSnapshot(null, StatusBadCommunicationError, null, now); + continue; + } try { var value = await ReadOneAsync(transport, tag, cancellationToken).ConfigureAwait(false); results[i] = new DataValueSnapshot(value, 0u, now, now); - WriteHealth(new DriverHealth(DriverState.Healthy, now, null)); + batch.Answered = true; // Invalidate the WriteOnChangeOnly cache when the read returns a different value // — typically an HMI-side or PLC-internal change. Without this, a setpoint // tweaked at the panel could be silently re-suppressed when our client tried // to restore it. - if (_options.WriteOnChangeOnly) - { - lock (_lastWrittenLock) - { - if (_lastWrittenByRef.TryGetValue(fullReferences[i], out var prev) && !Equals(prev, value)) - _lastWrittenByRef.Remove(fullReferences[i]); - } - } + InvalidateWriteCacheIfDiverged(fullReferences[i], value); } - catch (ModbusException mex) + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { - results[i] = new DataValueSnapshot(null, MapModbusExceptionToStatus(mex.ExceptionCode), null, now); - WriteHealth(new DriverHealth(DriverState.Degraded, ReadHealth().LastSuccessfulRead, mex.Message)); + throw; } catch (Exception ex) { - // Non-Modbus-layer failure: socket dropped, timeout, malformed response. Surface - // as communication error so callers can distinguish it from tag-level faults. - results[i] = new DataValueSnapshot(null, StatusBadCommunicationError, null, now); - WriteHealth(new DriverHealth(DriverState.Degraded, ReadHealth().LastSuccessfulRead, ex.Message)); + results[i] = new DataValueSnapshot(null, ClassifyFailure(ex, unit, fullReferences[i], batch, isWrite: false), null, now); } } + + // ONE health write per batch, worst-wins over the batch, generation-checked. It used to be + // last-writer-wins per tag, so health flapped with tag order. + ReportOutcome(generation, HealthSide.Read, batch, now); return results; } + /// + /// Captures the generation, transport and tag table a read/write batch runs against, atomically + /// with respect to a lifecycle publish. + /// + private (int Generation, IModbusTransport Transport, Dictionary Table) CaptureLifecycle() + { + lock (_stateLock) + { + var transport = _transport ?? throw new InvalidOperationException("ModbusDriver not initialized"); + return (_generation, transport, _tagsByRawPath); + } + } + + /// Runs a codec decode, normalising its failure to so the + /// classifier can tell undecodable device data from a transport fault. + private static object Decode(Func decode, ModbusTagDefinition tag) + { + try { return decode(); } + catch (Exception ex) when (ex is InvalidDataException or OverflowException or ArgumentException or InvalidCastException) + { + throw new ModbusDecodeException($"value of {tag.Name} ({tag.DataType}) could not be decoded: {ex.Message}", ex); + } + } + private async Task ReadOneAsync(IModbusTransport transport, ModbusTagDefinition tag, CancellationToken ct) { var arrayCount = tag.ArrayCount ?? 1; @@ -577,9 +836,9 @@ public sealed class ModbusDriver : await ReadRegisterBlockChunkedAsync(transport, unitId, fc, tag.Address, totalRegs, cap, ct).ConfigureAwait(false); if (!tag.ArrayCount.HasValue) - return DecodeRegister(data, tag); + return Decode(() => DecodeRegister(data, tag), tag); - return DecodeRegisterArray(data, tag, elementRegs, arrayCount); + return Decode(() => DecodeRegisterArray(data, tag, elementRegs, arrayCount), tag); } default: throw new InvalidOperationException($"Unknown region {tag.Region}"); @@ -913,9 +1172,11 @@ public sealed class ModbusDriver /// private async Task> ReadCoalescedAsync( IModbusTransport transport, + Dictionary table, IReadOnlyList fullReferences, DataValueSnapshot[] results, DateTime timestamp, + BatchOutcome batch, CancellationToken ct) { // Eligible: known tag, register region (HoldingRegisters or InputRegisters), scalar @@ -923,7 +1184,7 @@ public sealed class ModbusDriver var eligible = new List<(int Index, string Ref, ModbusTagDefinition Tag)>(); for (var i = 0; i < fullReferences.Count; i++) { - if (!_resolver.TryResolve(fullReferences[i], out var tag)) continue; + if (!table.TryGetValue(fullReferences[i], out var tag)) continue; if (tag.CoalesceProhibited) continue; if (tag.ArrayCount.HasValue) continue; if (tag.Region is not (ModbusRegion.HoldingRegisters or ModbusRegion.InputRegisters)) continue; @@ -983,38 +1244,66 @@ public sealed class ModbusDriver } var qty = (ushort)(block.End - block.Start + 1); - try + if (batch.IsFailedFast(group.Key.Unit)) { - var data = await ReadRegisterBlockAsync(transport, group.Key.Unit, fc, block.Start, qty, ct).ConfigureAwait(false); - foreach (var (idx, tag) in block.Members) - { - var sliceOffsetBytes = (tag.Address - block.Start) * 2; - var sliceLenBytes = RegisterCount(tag) * 2; - var value = DecodeRegister(data.AsSpan(sliceOffsetBytes, sliceLenBytes), tag); - results[idx] = new DataValueSnapshot(value, 0u, timestamp, timestamp); - handled.Add(idx); - InvalidateWriteCacheIfDiverged(fullReferences[idx], value); - } - WriteHealth(new DriverHealth(DriverState.Healthy, timestamp, null)); - } - catch (ModbusException mex) - { - RecordAutoProhibition(group.Key.Unit, group.Key.Region, block.Start, block.End); - WriteHealth(new DriverHealth(DriverState.Degraded, ReadHealth().LastSuccessfulRead, mex.Message)); - } - catch (Exception ex) - { - // Communication failures (timeout, socket drop) aren't a structural reason - // to prohibit the range — the same coalesced read might succeed once the - // transport recovers. Mark members Bad for this scan but don't auto-prohibit - // and don't deflect to per-tag fallback (which would just hit the same dead - // socket). foreach (var (idx, _) in block.Members) { results[idx] = new DataValueSnapshot(null, StatusBadCommunicationError, null, timestamp); handled.Add(idx); } - WriteHealth(new DriverHealth(DriverState.Degraded, ReadHealth().LastSuccessfulRead, ex.Message)); + continue; + } + byte[] data; + try + { + data = await ReadRegisterBlockAsync(transport, group.Key.Unit, fc, block.Start, qty, ct).ConfigureAwait(false); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + throw; + } + catch (ModbusException mex) when (IsRefusalCode(mex.ExceptionCode)) + { + // The device refused the merged range (a protected hole inside it): record the range + // as auto-prohibited and leave the members UNhandled so the per-tag fallback reads each + // surviving address individually. The device answered, so health is not degraded. + batch.Answered = true; + RecordAutoProhibition(group.Key.Unit, group.Key.Region, block.Start, block.End); + continue; + } + catch (Exception ex) + { + // Communication failures (timeout, socket drop) and non-refusal exception codes aren't + // a structural reason to prohibit the range — the same coalesced read might succeed + // once the transport recovers. Mark members Bad for this scan but don't auto-prohibit + // and don't deflect to per-tag fallback (which would just hit the same dead socket). + var status = ClassifyFailure(ex, group.Key.Unit, fullReferences[block.Members[0].Index], batch, isWrite: false); + foreach (var (idx, _) in block.Members) + { + results[idx] = new DataValueSnapshot(null, status, null, timestamp); + handled.Add(idx); + } + continue; + } + + batch.Answered = true; + foreach (var (idx, tag) in block.Members) + { + var sliceOffsetBytes = (tag.Address - block.Start) * 2; + var sliceLenBytes = RegisterCount(tag) * 2; + handled.Add(idx); + try + { + var value = Decode(() => DecodeRegister(data.AsSpan(sliceOffsetBytes, sliceLenBytes), tag), tag); + results[idx] = new DataValueSnapshot(value, 0u, timestamp, timestamp); + InvalidateWriteCacheIfDiverged(fullReferences[idx], value); + } + catch (ModbusDecodeException dex) + { + // One undecodable member never fails its neighbours. + results[idx] = new DataValueSnapshot(null, + ClassifyFailure(dex, group.Key.Unit, fullReferences[idx], batch, isWrite: false), null, timestamp); + } } } } @@ -1124,12 +1413,13 @@ public sealed class ModbusDriver public async Task> WriteAsync( IReadOnlyList writes, CancellationToken cancellationToken) { - var transport = RequireTransport(); + var (generation, transport, table) = CaptureLifecycle(); var results = new WriteResult[writes.Count]; + var batch = new BatchOutcome(); for (var i = 0; i < writes.Count; i++) { var w = writes[i]; - if (!_resolver.TryResolve(w.FullReference, out var tag)) + if (!table.TryGetValue(w.FullReference, out var tag)) { results[i] = new WriteResult(StatusBadNodeIdUnknown); continue; @@ -1144,30 +1434,67 @@ public sealed class ModbusDriver results[i] = new WriteResult(0u); continue; } + var unit = ResolveUnitId(tag); + if (batch.IsFailedFast(unit)) + { + results[i] = new WriteResult(StatusBadCommunicationError); + continue; + } try { await WriteOneAsync(transport, tag, w.Value, cancellationToken).ConfigureAwait(false); results[i] = new WriteResult(0u); + batch.Answered = true; if (_options.WriteOnChangeOnly) lock (_lastWrittenLock) _lastWrittenByRef[w.FullReference] = w.Value; } - catch (ModbusException mex) + catch (Exception ex) { - results[i] = new WriteResult(MapModbusExceptionToStatus(mex.ExceptionCode)); - } - catch (InvalidDataException) - { - results[i] = new WriteResult(StatusBadCommunicationError); - } - catch (Exception) - { - results[i] = new WriteResult(StatusBadInternalError); + results[i] = new WriteResult(MapWriteFailure(ex, unit, w.FullReference, batch, cancellationToken)); } } + + // Write-side health: a comms failure degrades it (LastSuccessfulRead preserved); a good write or a + // device refusal proves the device answered and clears a write-side Degraded. Generation-checked. + ReportOutcome(generation, HealthSide.Write, batch, DateTime.UtcNow); return results; } + /// + /// Write-status mapping (review 07 H7). Failures raised BEFORE the wire — a value that does not fit + /// (BadOutOfRange), cannot convert (BadTypeMismatch) or is not supported + /// (BadNotSupported) — never touch health. Device answers go through + /// (refusal / device fault / comms fault). Transport failures — + /// desync, timeout, socket, EOF — are BadCommunicationError, no longer + /// BadInternalError; only a genuinely unexpected exception still is. + /// + private uint MapWriteFailure(Exception ex, byte unit, string tagRef, BatchOutcome batch, CancellationToken ct) + { + switch (ex) + { + case OperationCanceledException when ct.IsCancellationRequested: + // The caller's write deadline fired (e.g. waiting on the single-flight gate). The device + // said nothing, so health is untouched. + return StatusBadTimeout; + case ModbusWriteRefusedException refused: + return refused.StatusCode; + case OverflowException or ArgumentOutOfRangeException: + return StatusBadOutOfRange; + case FormatException or InvalidCastException or ArgumentException: + return StatusBadTypeMismatch; + case NotSupportedException: + return StatusBadNotSupported; + case ModbusException or ModbusDecodeException or IOException or InvalidDataException + or System.Net.Sockets.SocketException or ObjectDisposedException or TimeoutException + or OperationCanceledException: + return ClassifyFailure(ex, unit, tagRef, batch, isWrite: true); + default: + _logger.LogWarning(ex, "Modbus write to {RawPath} failed unexpectedly. Driver={DriverInstanceId}", tagRef, _driverInstanceId); + return StatusBadInternalError; + } + } + private bool IsRedundantWrite(string tagRef, object? value) { lock (_lastWrittenLock) @@ -1256,7 +1583,7 @@ public sealed class ModbusDriver var qty = (ushort)(bytes.Length / 2); var writeCap = _options.MaxRegistersPerWrite == 0 ? (ushort)123 : _options.MaxRegistersPerWrite; if (qty > writeCap) - throw new InvalidOperationException( + throw new ModbusWriteRefusedException(StatusBadOutOfRange, $"Write of {qty} registers to {tag.Name} exceeds MaxRegistersPerWrite={writeCap}. " + $"Split the tag (e.g. shorter StringLength or smaller ArrayCount) — partial FC16 chunks would lose atomicity."); var pdu = new byte[6 + 1 + bytes.Length]; @@ -1270,7 +1597,7 @@ public sealed class ModbusDriver return; } default: - throw new InvalidOperationException($"Writes not supported for region {tag.Region}"); + throw new ModbusWriteRefusedException(StatusBadNotWritable, $"Writes not supported for region {tag.Region}"); } } @@ -1281,9 +1608,12 @@ public sealed class ModbusDriver private static byte[] EncodeRegisterArray(object? value, ModbusTagDefinition tag) { var count = tag.ArrayCount!.Value; - if (value is not System.Collections.IList list || list.Count != count) - throw new InvalidOperationException( + if (value is not System.Collections.IList list) + throw new ModbusWriteRefusedException(StatusBadTypeMismatch, $"Array write to {tag.Name} expects an IList of length {count}; got {value?.GetType().Name ?? "null"}"); + if (list.Count != count) + throw new ModbusWriteRefusedException(StatusBadOutOfRange, + $"Array write to {tag.Name} expects {count} elements; got {list.Count}"); var elementBytes = ElementByteCount(tag); var result = new byte[count * elementBytes]; @@ -1308,7 +1638,10 @@ public sealed class ModbusDriver for (var i = 0; i < expectedCount; i++) arr[i] = Convert.ToBoolean(list[i]); return arr; } - throw new InvalidOperationException( + if (value is System.Collections.IList wrong) + throw new ModbusWriteRefusedException(StatusBadOutOfRange, + $"Coil-array write to {tagName} expects {expectedCount} elements; got {wrong.Count}"); + throw new ModbusWriteRefusedException(StatusBadTypeMismatch, $"Coil-array write to {tagName} expects a bool[] (or convertible IList) of length {expectedCount}; got {value?.GetType().Name ?? "null"}"); } @@ -1392,12 +1725,12 @@ public sealed class ModbusDriver /// public string HostName => $"{_options.Host}:{_options.Port}"; - private async Task ProbeLoopAsync(CancellationToken ct) + private async Task ProbeLoopAsync(int generation, IModbusTransport transport, CancellationToken ct) { - var transport = _transport; // captured reference; disposal tears the loop down via ct + // The transport is captured at start; disposal tears the loop down via ct. while (!ct.IsCancellationRequested) { - var success = false; + var answered = false; try { using var probeCts = CancellationTokenSource.CreateLinkedTokenSource(ct); @@ -1405,19 +1738,28 @@ public sealed class ModbusDriver var pdu = new byte[] { 0x03, (byte)(_options.Probe.ProbeAddress >> 8), (byte)(_options.Probe.ProbeAddress & 0xFF), 0x00, 0x01 }; - _ = await transport!.SendAsync(_options.UnitId, pdu, probeCts.Token).ConfigureAwait(false); - success = true; + _ = await transport.SendAsync(_options.UnitId, pdu, probeCts.Token).ConfigureAwait(false); + answered = true; } catch (OperationCanceledException) when (ct.IsCancellationRequested) { return; } + catch (ModbusException mex) when (mex.ExceptionCode is not (0x0A or 0x0B)) + { + // An exception PDU from the device itself (e.g. a locked probe register) proves the host + // is up — it answered. Only a gateway reporting its target unreachable (0A/0B) is Stopped. + answered = true; + } catch { - // transport / timeout / exception PDU — treated as Stopped below + // transport / timeout / gateway-target failure — treated as Stopped below } - TransitionTo(success ? HostState.Running : HostState.Stopped); + // A probe that finished after teardown cancelled the loop must not raise a stale transition. + if (ct.IsCancellationRequested) return; + if (answered) NoteDeviceAnswered(generation); + TransitionTo(answered ? HostState.Running : HostState.Stopped); try { await Task.Delay(_options.Probe.Interval, ct).ConfigureAwait(false); } catch (OperationCanceledException) { return; } @@ -1743,8 +2085,6 @@ public sealed class ModbusDriver return result; } - private IModbusTransport RequireTransport() => - _transport ?? throw new InvalidOperationException("ModbusDriver not initialized"); private const uint StatusBadInternalError = 0x80020000u; private const uint StatusBadNodeIdUnknown = 0x80340000u; @@ -1753,6 +2093,9 @@ public sealed class ModbusDriver private const uint StatusBadNotSupported = 0x803D0000u; private const uint StatusBadDeviceFailure = 0x808B0000u; private const uint StatusBadCommunicationError = 0x80050000u; + private const uint StatusBadDecodingError = 0x80070000u; + private const uint StatusBadTimeout = 0x800A0000u; + private const uint StatusBadTypeMismatch = 0x80740000u; /// /// Map a server-returned Modbus exception code to the most informative OPC UA @@ -1789,6 +2132,10 @@ public sealed class ModbusDriver /// A task that represents the asynchronous operation. public async ValueTask DisposeAsync() { + // Dispose supersedes any in-flight (abandoned) initialize up front, even when it then has to + // bypass the gate: that initialize will find its generation moved on and release its own + // transport instead of publishing it after this teardown. + lock (_stateLock) _generation++; if (await _lifecycleGate.WaitAsync(LifecycleGateTimeout).ConfigureAwait(false)) { try @@ -1816,25 +2163,27 @@ public sealed class ModbusDriver /// private async Task TeardownAsync() { - try { _probeCts?.Cancel(); } catch { } - _probeCts?.Dispose(); - _probeCts = null; + await StopBackgroundLoopsAsync().ConfigureAwait(false); - try { _reprobeCts?.Cancel(); } catch { } - _reprobeCts?.Dispose(); - _reprobeCts = null; + // Drain the poll loops BEFORE dropping the tables and caches they read, so no in-flight cycle can + // re-seed a cache after it was cleared. + await _poll.DisposeAsync().ConfigureAwait(false); + // After the engine's loops are drained, no onChange can re-populate the per-cycle buffers. + _cycleBatchItems.Clear(); - _tagsByRawPath.Clear(); + IModbusTransport? transport; + lock (_stateLock) + { + transport = _transport; + _transport = null; + // Swap, never .Clear() in place: a reader that captured the old table keeps a consistent view. + Volatile.Write(ref _tagsByRawPath, new Dictionary(StringComparer.Ordinal)); + } _resolver.Clear(); // no-op in v3 (the resolver reads the live table by closure); kept for symmetry _lastPublishedByRef.Clear(); lock (_lastWrittenLock) _lastWrittenByRef.Clear(); lock (_autoProhibitedLock) _autoProhibited.Clear(); - await _poll.DisposeAsync().ConfigureAwait(false); - // After the engine's loops are drained, no onChange can re-populate the per-cycle buffers. - _cycleBatchItems.Clear(); - - if (_transport is not null) await _transport.DisposeAsync().ConfigureAwait(false); - _transport = null; + if (transport is not null) await transport.DisposeAsync().ConfigureAwait(false); } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusRtuOverTcpTransport.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusRtuOverTcpTransport.cs index 69ccf915..7dcc5dff 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusRtuOverTcpTransport.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusRtuOverTcpTransport.cs @@ -46,7 +46,7 @@ public sealed class ModbusRtuOverTcpTransport : IModbusTransport private readonly Stream? _testStream; private readonly TimeSpan _timeout; private readonly SemaphoreSlim _gate = new(1, 1); - private bool _disposed; + private int _disposed; /// Initializes a new instance of the class. /// The host address or hostname of the Modbus gateway. @@ -90,7 +90,7 @@ public sealed class ModbusRtuOverTcpTransport : IModbusTransport /// public async Task SendAsync(byte unitId, byte[] pdu, CancellationToken ct) { - if (_disposed) throw new ObjectDisposedException(nameof(ModbusRtuOverTcpTransport)); + if (Volatile.Read(ref _disposed) != 0) throw new ObjectDisposedException(nameof(ModbusRtuOverTcpTransport)); if (_life is null && _testStream is null) throw new InvalidOperationException("Transport not connected"); // Single-flight: RTU carries no transaction id, so at most one transaction may be on the bus @@ -187,9 +187,8 @@ public sealed class ModbusRtuOverTcpTransport : IModbusTransport /// A task that represents the asynchronous operation. public async ValueTask DisposeAsync() { - if (_disposed) return; - _disposed = true; + if (Interlocked.Exchange(ref _disposed, 1) != 0) return; if (_life is not null) await _life.DisposeAsync().ConfigureAwait(false); - _gate.Dispose(); + // _gate is deliberately NOT disposed: a SendAsync still in flight at teardown releases it in its finally. } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusSocketLifecycle.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusSocketLifecycle.cs index 7a032cdd..34fcbdd3 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusSocketLifecycle.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusSocketLifecycle.cs @@ -76,6 +76,7 @@ public sealed class ModbusSocketLifecycle : IAsyncDisposable { // A previous socket (failed, torn down or idle-reaped) is released FIRST: overwriting _client // used to strand the old TcpClient for the finalizer — up to nine per backoff loop. + ThrowIfDisposed(); await TearDownAsync().ConfigureAwait(false); // DNS resolution runs under the same per-op deadline as the TCP connect: an unresolvable or @@ -98,6 +99,7 @@ public sealed class ModbusSocketLifecycle : IAsyncDisposable client = new TcpClient(target.AddressFamily); EnableKeepAlive(client, _keepAlive); await client.ConnectAsync(target, _port, cts.Token).ConfigureAwait(false); + ThrowIfDisposed(); // disposed while dialling — the finally releases the fresh client _stream = client.GetStream(); _client = client; client = null; // ownership transferred @@ -258,11 +260,16 @@ public sealed class ModbusSocketLifecycle : IAsyncDisposable /// A task that represents the asynchronous operation. public async ValueTask DisposeAsync() { - try - { - if (_stream is not null) await _stream.DisposeAsync().ConfigureAwait(false); - } - catch { /* best-effort */ } - _client?.Dispose(); + // Latch first: a SendAsync still in flight (the driver tearing down under it) must not reconnect + // a fresh socket after this point — ConnectAsync checks the latch both before and after dialling. + Volatile.Write(ref _disposed, 1); + await TearDownAsync().ConfigureAwait(false); + } + + private int _disposed; + + private void ThrowIfDisposed() + { + if (Volatile.Read(ref _disposed) != 0) throw new ObjectDisposedException(nameof(ModbusSocketLifecycle)); } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusTcpTransport.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusTcpTransport.cs index 26534610..f21ce678 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusTcpTransport.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusTcpTransport.cs @@ -36,7 +36,7 @@ public sealed class ModbusTcpTransport : IModbusTransport private readonly TimeSpan _timeout; private readonly SemaphoreSlim _gate = new(1, 1); private ushort _nextTx; - private bool _disposed; + private int _disposed; /// Initializes a new instance of the class. /// The host address or hostname of the Modbus server. @@ -65,7 +65,7 @@ public sealed class ModbusTcpTransport : IModbusTransport /// public async Task SendAsync(byte unitId, byte[] pdu, CancellationToken ct) { - if (_disposed) throw new ObjectDisposedException(nameof(ModbusTcpTransport)); + if (Volatile.Read(ref _disposed) != 0) throw new ObjectDisposedException(nameof(ModbusTcpTransport)); await _gate.WaitAsync(ct).ConfigureAwait(false); try @@ -228,9 +228,8 @@ public sealed class ModbusTcpTransport : IModbusTransport /// A task that represents the asynchronous operation. public async ValueTask DisposeAsync() { - if (_disposed) return; - _disposed = true; + if (Interlocked.Exchange(ref _disposed, 1) != 0) return; await _life.DisposeAsync().ConfigureAwait(false); - _gate.Dispose(); + // _gate is deliberately NOT disposed: a SendAsync still in flight at teardown releases it in its finally. } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusWriteRefusedException.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusWriteRefusedException.cs new file mode 100644 index 00000000..866990ec --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusWriteRefusedException.cs @@ -0,0 +1,21 @@ +namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus; + +/// +/// The driver refused a write BEFORE anything reached the wire (wrong array length, a payload over +/// MaxRegistersPerWrite, an unsupported shape). Carries the OPC UA status the write returns; +/// it never touches driver health — nothing was sent. +/// +public sealed class ModbusWriteRefusedException : Exception +{ + /// Initializes a new instance of the class. + /// The OPC UA status the write returns. + /// A human-readable description. + public ModbusWriteRefusedException(uint statusCode, string message) + : base(message) + { + StatusCode = statusCode; + } + + /// Gets the OPC UA status the write returns. + public uint StatusCode { get; } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusHealthContractTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusHealthContractTests.cs new file mode 100644 index 00000000..50282ec3 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusHealthContractTests.cs @@ -0,0 +1,341 @@ +using System.Text.Json; +using Microsoft.Extensions.Logging; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests; + +/// +/// The driver health contract (review 07 H4 + H7, and the lifecycle/fail-fast rules every driver lane +/// must meet): a device REFUSAL is per-tag Bad with no Degraded and proves the device answered; a comms +/// fault degrades and preserves LastSuccessfulRead; health is one write per batch; a comms timeout on +/// one unit fails the rest of that unit's batch fast; writes map transport failures to +/// BadCommunicationError and value problems to BadOutOfRange / BadTypeMismatch. +/// +[Trait("Category", "Unit")] +public sealed class ModbusHealthContractTests +{ + private const uint Good = 0u; + private const uint BadOutOfRange = 0x803C0000u; + private const uint BadDeviceFailure = 0x808B0000u; + private const uint BadCommunicationError = 0x80050000u; + private const uint BadTypeMismatch = 0x80740000u; + private const uint BadNodeIdUnknown = 0x80340000u; + + private static ModbusDriverOptions Opts(params ModbusTagDefinition[] tags) => new() + { + Host = "f", RawTags = ModbusRawTags.Entries(tags), Probe = new ModbusProbeOptions { Enabled = false }, + }; + + private static ModbusTagDefinition Hr(string name, ushort addr, byte? unit = null, ModbusDataType type = ModbusDataType.Int16) => + new(name, ModbusRegion.HoldingRegisters, addr, type, UnitId: unit); + + private static Task Refuse(byte code, byte[] pdu) => + Task.FromException(new ModbusException(pdu[0], code, $"exception {code}")); + + private static ushort Addr(byte[] pdu) => (ushort)((pdu[1] << 8) | pdu[2]); + + private static async Task<(ModbusDriver Driver, ScriptedTransport Transport)> StartAsync( + ModbusDriverOptions opts, ILogger? logger = null) + { + var t = new ScriptedTransport(); + var d = new ModbusDriver(opts, "m1", _ => t, logger); + await d.InitializeAsync("{}", TestContext.Current.CancellationToken); + return (d, t); + } + + // ---------------- H4: refusals ---------------- + + [Fact] + public async Task All_tags_refused_keeps_Healthy_and_advances_LastSuccessfulRead() + { + var (drv, t) = await StartAsync(Opts(Hr("A", 10), Hr("B", 11))); + await using var _ = drv; + t.Handler = (_, pdu, _) => Refuse(0x02, pdu); + var before = drv.GetHealth().LastSuccessfulRead; + + var r = await drv.ReadAsync(["A", "B"], TestContext.Current.CancellationToken); + + r.ShouldAllBe(s => s.StatusCode == BadOutOfRange); + var h = drv.GetHealth(); + h.State.ShouldBe(DriverState.Healthy, "a refusal means the device answered — no Degraded, no watchdog fodder"); + h.LastSuccessfulRead.ShouldNotBeNull(); + h.LastSuccessfulRead!.Value.ShouldBeGreaterThanOrEqualTo(before!.Value); + } + + [Fact] + public async Task A_refusal_after_a_comms_fault_restores_Healthy() + { + var (drv, t) = await StartAsync(Opts(Hr("A", 10))); + await using var _ = drv; + t.Handler = (_, _, _) => Task.FromException(new IOException("socket dropped")); + await drv.ReadAsync(["A"], TestContext.Current.CancellationToken); + var degraded = drv.GetHealth(); + degraded.State.ShouldBe(DriverState.Degraded); + + t.Handler = (_, pdu, _) => Refuse(0x02, pdu); + await drv.ReadAsync(["A"], TestContext.Current.CancellationToken); + var h = drv.GetHealth(); + h.State.ShouldBe(DriverState.Healthy); + h.LastSuccessfulRead!.Value.ShouldBeGreaterThanOrEqualTo(degraded.LastSuccessfulRead!.Value); + } + + [Theory] + [InlineData(0x04)] + [InlineData(0x0B)] + public async Task Non_refusal_exception_codes_still_degrade(byte code) + { + // Positive control for the refusal tests: a server-failure / gateway code is NOT a refusal. + var (drv, t) = await StartAsync(Opts(Hr("A", 10))); + await using var _ = drv; + t.Handler = (_, pdu, _) => Refuse(code, pdu); + await drv.ReadAsync(["A"], TestContext.Current.CancellationToken); + drv.GetHealth().State.ShouldBe(DriverState.Degraded); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task Mixed_refused_and_good_batch_is_Healthy_regardless_of_tag_order(bool refusedFirst) + { + var (drv, t) = await StartAsync(Opts(Hr("Refused", 10), Hr("Fine", 50))); + await using var _ = drv; + t.Handler = (_, pdu, _) => Addr(pdu) == 10 ? Refuse(0x02, pdu) : null; + string[] refs = refusedFirst ? ["Refused", "Fine"] : ["Fine", "Refused"]; + var r = await drv.ReadAsync(refs, TestContext.Current.CancellationToken); + r.Count(s => s.StatusCode == Good).ShouldBe(1); + drv.GetHealth().State.ShouldBe(DriverState.Healthy); + } + + [Fact] + public async Task A_refused_tag_warns_once_not_once_per_poll() + { + var logger = new ListLogger(); + var (drv, t) = await StartAsync(Opts(Hr("A", 10)), logger); + await using var _ = drv; + t.Handler = (_, pdu, _) => Refuse(0x02, pdu); + for (var i = 0; i < 3; i++) await drv.ReadAsync(["A"], TestContext.Current.CancellationToken); + logger.Warnings.Count(m => m.Contains("refused") && m.Contains("A")).ShouldBe(1); + } + + // ---------------- fail fast per unit / endpoint ---------------- + + [Fact] + public async Task A_comms_timeout_on_one_unit_fails_the_rest_of_that_units_batch_fast() + { + var (drv, t) = await StartAsync(Opts(Hr("U2a", 10, 2), Hr("U2b", 11, 2), Hr("U2c", 12, 2), Hr("U1", 13, 1))); + await using var _ = drv; + t.Bank.HoldingRegisters[13] = 42; + t.Handler = (unit, _, _) => unit == 2 + ? Task.FromException(new ModbusTransportDesyncException( + ModbusTransportDesyncException.DesyncReason.Timeout, "timeout")) + : null; + + var r = await drv.ReadAsync(["U2a", "U2b", "U2c", "U1"], TestContext.Current.CancellationToken); + + t.SendsTo(2).ShouldBe(1, "one timeout per unit per batch, not one per tag"); + r.Take(3).ShouldAllBe(s => s.StatusCode == BadCommunicationError); + // Positive control: the healthy unit in the same batch is still read. + r[3].StatusCode.ShouldBe(Good); + r[3].Value.ShouldBe((short)42); + drv.GetHealth().State.ShouldBe(DriverState.Degraded); + } + + [Fact] + public async Task An_unreachable_endpoint_fails_the_whole_batch_fast() + { + var (drv, t) = await StartAsync(Opts(Hr("A", 10, 1), Hr("B", 11, 2), Hr("C", 12, 3))); + await using var _ = drv; + t.Handler = (_, _, _) => Task.FromException(new ModbusTransportUnavailableException("throttled")); + var r = await drv.ReadAsync(["A", "B", "C"], TestContext.Current.CancellationToken); + t.TotalSends.ShouldBe(1); + r.ShouldAllBe(s => s.StatusCode == BadCommunicationError); + } + + // ---------------- H7: write status mapping + write-side health ---------------- + + [Fact] + public async Task Write_transport_failure_is_BadCommunicationError_and_degrades_the_write_side() + { + var (drv, t) = await StartAsync(Opts(Hr("A", 10))); + await using var _ = drv; + t.Handler = (_, pdu, _) => pdu[0] == 0x06 ? Task.FromException(new IOException("dropped")) : null; + + var w = await drv.WriteAsync([new WriteRequest("A", (short)1)], TestContext.Current.CancellationToken); + w[0].StatusCode.ShouldBe(BadCommunicationError); + drv.GetHealth().State.ShouldBe(DriverState.Degraded); + + // A good READ does not clear a write-side fault... + (await drv.ReadAsync(["A"], TestContext.Current.CancellationToken))[0].StatusCode.ShouldBe(Good); + drv.GetHealth().State.ShouldBe(DriverState.Degraded); + + // ...a good WRITE does. + t.Handler = null; + (await drv.WriteAsync([new WriteRequest("A", (short)2)], TestContext.Current.CancellationToken))[0].StatusCode.ShouldBe(Good); + drv.GetHealth().State.ShouldBe(DriverState.Healthy); + } + + [Fact] + public async Task A_write_refusal_counts_as_answered_and_clears_a_write_side_fault() + { + var (drv, t) = await StartAsync(Opts(Hr("A", 10))); + await using var _ = drv; + t.Handler = (_, pdu, _) => pdu[0] == 0x06 ? Task.FromException(new IOException("dropped")) : null; + await drv.WriteAsync([new WriteRequest("A", (short)1)], TestContext.Current.CancellationToken); + drv.GetHealth().State.ShouldBe(DriverState.Degraded); + + t.Handler = (_, pdu, _) => pdu[0] == 0x06 ? Refuse(0x03, pdu) : null; + var w = await drv.WriteAsync([new WriteRequest("A", (short)1)], TestContext.Current.CancellationToken); + w[0].StatusCode.ShouldBe(BadOutOfRange); + drv.GetHealth().State.ShouldBe(DriverState.Healthy); + } + + [Fact] + public async Task Write_value_problems_map_to_OutOfRange_and_TypeMismatch_without_touching_health() + { + var arr = new ModbusTagDefinition("Arr", ModbusRegion.HoldingRegisters, 20, ModbusDataType.Int16, ArrayCount: 3); + var (drv, t) = await StartAsync(Opts(Hr("Bcd", 10, type: ModbusDataType.Bcd16), Hr("I16", 11), arr)); + await using var _ = drv; + + var w = await drv.WriteAsync([ + new WriteRequest("Bcd", 12345), // > 4 BCD digits + new WriteRequest("I16", 70000), // > Int16 + new WriteRequest("I16", "not a number"), + new WriteRequest("Arr", new short[] { 1, 2 }), // length mismatch + new WriteRequest("Arr", new int[] { 1, 2, 3 }), // CLR list of another element type converts (ScadaBridge shape) + ], TestContext.Current.CancellationToken); + + w[0].StatusCode.ShouldBe(BadOutOfRange); + w[1].StatusCode.ShouldBe(BadOutOfRange); + w[2].StatusCode.ShouldBe(BadTypeMismatch); + w[3].StatusCode.ShouldBe(BadOutOfRange); + w[4].StatusCode.ShouldBe(Good); + t.Bank.HoldingRegisters[22].ShouldBe((ushort)3); + drv.GetHealth().State.ShouldBe(DriverState.Healthy); + } + + // ---------------- lifecycle generation ---------------- + + [Fact] + public async Task A_superseded_initialize_never_publishes_over_the_shutdown_that_overtook_it() + { + var t = new ScriptedTransport { ConnectGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously) }; + var drv = new ModbusDriver(Opts(Hr("A", 10)), "m1", _ => t) { LifecycleGateTimeout = TimeSpan.FromSeconds(1) }; + var init = drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + await t.ConnectEntered.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + + // The init is wedged in its connect; shutdown waits out the gate timeout and tears down anyway. + await drv.ShutdownAsync(TestContext.Current.CancellationToken); + drv.GetHealth().State.ShouldBe(DriverState.Unknown); + + t.ConnectGate.SetResult(); // the abandoned connect finally completes + await Should.ThrowAsync(() => init); + + drv.GetHealth().State.ShouldBe(DriverState.Unknown, "the superseded init must not publish Healthy"); + t.DisposeCount.ShouldBe(1, "the superseded init releases the transport it built"); + await Should.ThrowAsync(() => drv.ReadAsync(["A"], TestContext.Current.CancellationToken)); + } + + [Fact] + public async Task A_read_that_straddles_a_reinitialize_does_not_write_health_about_the_new_transport() + { + var first = new ScriptedTransport(); + var second = new ScriptedTransport(); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + first.Handler = (_, _, _) => { entered.TrySetResult(); return release.Task; }; + var calls = 0; + var drv = new ModbusDriver(Opts(Hr("A", 10)), "m1", _ => Interlocked.Increment(ref calls) == 1 ? first : second); + await using var _ = drv; + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + + var read = drv.ReadAsync(["A"], TestContext.Current.CancellationToken); + await entered.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); // re-init onto `second` + drv.GetHealth().State.ShouldBe(DriverState.Healthy); + + release.SetException(new IOException("old socket died")); // the straddling read now fails + (await read)[0].StatusCode.ShouldBe(BadCommunicationError); + drv.GetHealth().State.ShouldBe(DriverState.Healthy, "a read on the superseded transport must not degrade the new one"); + + // Positive control: a failure on the CURRENT transport still degrades. + second.Handler = (_, _, _) => Task.FromException(new IOException("new socket died")); + await drv.ReadAsync(["A"], TestContext.Current.CancellationToken); + drv.GetHealth().State.ShouldBe(DriverState.Degraded); + } + + [Fact] + public async Task Reinitialize_with_a_new_tag_set_drops_tags_the_new_config_no_longer_has() + { + var t = new ScriptedTransport(); + var drv = new ModbusDriver(Opts(Hr("Old", 10)), "m1", _ => t); + await using var _ = drv; + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + (await drv.ReadAsync(["Old"], TestContext.Current.CancellationToken))[0].StatusCode.ShouldBe(Good); + + var config = JsonSerializer.Serialize(new + { + host = "f", + probe = new { enabled = false }, + rawTags = ModbusRawTags.Entries([Hr("New", 11)]), + }); + await drv.InitializeAsync(config, TestContext.Current.CancellationToken); + + var r = await drv.ReadAsync(["Old", "New"], TestContext.Current.CancellationToken); + r[0].StatusCode.ShouldBe(BadNodeIdUnknown, "a re-init publishes a fresh table, it does not merge into the old one"); + r[1].StatusCode.ShouldBe(Good); + } + + // ---------------- complete teardown ---------------- + + [Fact] + public async Task Shutdown_joins_the_probe_loop_so_no_host_transition_fires_after_it_returns() + { + var t = new ScriptedTransport(); + var probeBlocked = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var probes = 0; + t.Handler = async (_, pdu, ct) => + { + if (Interlocked.Increment(ref probes) == 1) return await t.Bank.SendAsync(1, pdu, ct); // first probe: Running + probeBlocked.TrySetResult(); + try { await Task.Delay(Timeout.Infinite, ct); } + catch (OperationCanceledException) { } + throw new ObjectDisposedException("transport"); // a teardown-time failure, NOT an OCE + }; + var opts = new ModbusDriverOptions + { + Host = "f", RawTags = ModbusRawTags.Entries([Hr("A", 10)]), + Probe = new ModbusProbeOptions { Enabled = true, Interval = TimeSpan.FromMilliseconds(50), Timeout = TimeSpan.FromSeconds(30) }, + }; + var drv = new ModbusDriver(opts, "m1", _ => t); + var running = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var afterShutdown = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var shutdownReturned = false; + drv.OnHostStatusChanged += (_, e) => + { + if (e.NewState == HostState.Running) running.TrySetResult(); + if (Volatile.Read(ref shutdownReturned)) afterShutdown.TrySetResult(); + }; + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + await running.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); // positive control + await probeBlocked.Task.WaitAsync(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken); + + await drv.ShutdownAsync(TestContext.Current.CancellationToken); + Volatile.Write(ref shutdownReturned, true); + + await Should.ThrowAsync( + () => afterShutdown.Task.WaitAsync(TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken)); + } + + private sealed class ListLogger : ILogger + { + public System.Collections.Concurrent.ConcurrentQueue Messages { get; } = new(); + public IEnumerable Warnings => Messages.Where(m => m.StartsWith("Warning:", StringComparison.Ordinal)); + public IDisposable? BeginScope(TState state) where TState : notnull => null; + public bool IsEnabled(LogLevel logLevel) => true; + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) + => Messages.Enqueue($"{logLevel}: {formatter(state, exception)}"); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusLoggerInjectionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusLoggerInjectionTests.cs index f4caef3f..867b20ee 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusLoggerInjectionTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusLoggerInjectionTests.cs @@ -92,7 +92,7 @@ public sealed class ModbusLoggerInjectionTests // Scan 1 — coalesced read fails. Expect exactly one warning. await drv.ReadAsync(["T100", "T102", "T104"], CancellationToken.None); - var warnings = logger.Entries.Where(e => e.Level == LogLevel.Warning).ToList(); + var warnings = logger.Entries.Where(e => e.Level == LogLevel.Warning && e.Message.Contains("auto-prohibited")).ToList(); warnings.Count.ShouldBe(1); warnings[0].Message.ShouldContain("drv-logged"); warnings[0].Message.ShouldContain("Start=100"); @@ -102,7 +102,13 @@ public sealed class ModbusLoggerInjectionTests // the prohibition and skips the merge; even if it didn't, the de-dupe in // RecordAutoProhibition would suppress). await drv.ReadAsync(["T100", "T102", "T104"], CancellationToken.None); - logger.Entries.Count(e => e.Level == LogLevel.Warning).ShouldBe(1, "re-fire of same range stays silent"); + logger.Entries.Count(e => e.Level == LogLevel.Warning && e.Message.Contains("auto-prohibited")) + .ShouldBe(1, "re-fire of same range stays silent"); + // The protected register's per-tag fallback is a device refusal: one Warning for that tag across + // both scans (once-per-tag), and none for its readable neighbours. + var refusals = logger.Entries.Where(e => e.Level == LogLevel.Warning && e.Message.Contains("refused")).ToList(); + refusals.Count.ShouldBe(1); + refusals[0].Message.ShouldContain("T102"); await drv.ShutdownAsync(CancellationToken.None); } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ScriptedTransport.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ScriptedTransport.cs new file mode 100644 index 00000000..e865bc34 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ScriptedTransport.cs @@ -0,0 +1,51 @@ +using System.Collections.Concurrent; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests; + +/// +/// In-memory transport whose every send is answered by a test-supplied handler; falls back to the +/// register/coil bank fake () when the handler returns null. +/// Counts sends per unit and dispose calls; the connect can be blocked on a gate. +/// +internal sealed class ScriptedTransport : IModbusTransport +{ + private readonly ConcurrentDictionary _sendsByUnit = new(); + private int _disposeCount; + + /// The bank answering un-scripted sends. + public ModbusDriverTests.FakeTransport Bank { get; } = new(); + + /// Per-send handler; return null to fall through to . + public Func?>? Handler { get; set; } + + /// When set, waits for it before completing. + public TaskCompletionSource? ConnectGate { get; set; } + + /// Signalled when is entered. + public TaskCompletionSource ConnectEntered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public int DisposeCount => Volatile.Read(ref _disposeCount); + + public int SendsTo(byte unit) => _sendsByUnit.TryGetValue(unit, out var n) ? n : 0; + + public int TotalSends => _sendsByUnit.Values.Sum(); + + public async Task ConnectAsync(CancellationToken ct) + { + ConnectEntered.TrySetResult(); + if (ConnectGate is { } gate) await gate.Task.WaitAsync(ct); + } + + public Task SendAsync(byte unitId, byte[] pdu, CancellationToken ct) + { + _sendsByUnit.AddOrUpdate(unitId, 1, (_, n) => n + 1); + var scripted = Handler?.Invoke(unitId, pdu, ct); + return scripted ?? Bank.SendAsync(unitId, pdu, ct); + } + + public ValueTask DisposeAsync() + { + Interlocked.Increment(ref _disposeCount); + return ValueTask.CompletedTask; + } +}