diff --git a/docs/drivers/AbCip.md b/docs/drivers/AbCip.md index af3f0497..2a8696b9 100644 --- a/docs/drivers/AbCip.md +++ b/docs/drivers/AbCip.md @@ -172,6 +172,16 @@ The driver has two interchangeable wire transports, selected by one `DriverConfi two types into one value. - **Forward Open throttle:** at most one attempt in flight, at least 1 s between starts, bounded by the open budget. +- **A target-closed CIP connection is recycled at once (#566).** When a connected reply says the controller + no longer holds the class-3 connection while TCP stays up — general status `0x01` with extended `0x0107` + (target connection not found) or `0x0203` (connection timed out), e.g. after a controller restart or a + connection timeout — the session is closed (UnRegisterSession; no Forward Close, there is nothing to close) + and the next request reopens through the throttle. The request that got the reply fails + `BadCommunicationError` (transport `BadConnection`); a keep-alive that gets it recycles the connection + before any request fails. Every other status — tag-level errors, other `0x01` extended codes — is left to + the request. ⚠️ RESIDUAL: which of the two codes a real Logix controller returns for a forgotten + connection (and whether it answers at all rather than dropping TCP) is unobserved; the check is correct + whichever it does — a silent target still faults the session through the exchange timeout. - `ConnectionSize` / `Timeout` / `AllowPacking` are honoured per pooled connection, and the **first** driver/device to open a given device key wins — including a Test Connect probe on a fused `admin,driver` node (its connection carries the probe timeout). A later opener with different options is ignored and diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.LibplctagShared/LibplctagOrphanRelease.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.LibplctagShared/LibplctagOrphanRelease.cs index 5ac1f8d1..e484bf50 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.LibplctagShared/LibplctagOrphanRelease.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.LibplctagShared/LibplctagOrphanRelease.cs @@ -47,10 +47,7 @@ internal static class LibplctagOrphanRelease private static int _released; /// True when the pinned wrapper's private fields were found, so a failed tag can be destroyed. - internal static bool CanReadHandles => HandleField is not null && !ForceRootingForTest; - - /// Test seam (review r1 M1): take the rooting fallback as if the wrapper's fields had been renamed. - internal static bool ForceRootingForTest { get; set; } + internal static bool CanReadHandles => HandleField is not null; /// Orphaned native tags destroyed so far (process-wide; diagnostics and tests). internal static int Released => Volatile.Read(ref _released); @@ -117,12 +114,19 @@ internal static class LibplctagOrphanRelease /// /// The status the native tag held just before it was destroyed (what GetStatus() would have /// reported), or null when nothing was destroyed. - internal static Status? ReleaseIfOrphaned(Tag tag) + internal static Status? ReleaseIfOrphaned(Tag tag) => ReleaseIfOrphaned(tag, CanReadHandles); + + /// + /// with the handle-readability decision passed in, so a test can take the + /// rooting fallback (review r1 M1) without a process-wide switch (Gitea #581). Production always passes + /// . + /// + internal static Status? ReleaseIfOrphaned(Tag tag, bool canReadHandles) { try { if (tag.IsInitialized) return null; - if (!CanReadHandles) + if (!canReadHandles) { Rooted.Add(tag); return null; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/CipConnection.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/CipConnection.cs index 5004d731..eb0fa39b 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/CipConnection.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/CipConnection.cs @@ -10,7 +10,9 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip; /// /// Reconnect. Only a faulted session (timeout, I/O error, malformed frame) is reopened, and /// only by the next send; the failing call throws its own exception. Caller cancellation never -/// faults the session (B2), so it never causes a reconnect. +/// faults the session (B2), so it never causes a reconnect. A connected reply saying the target no longer +/// holds the class-3 connection (general 0x01, extended 0x0107 / 0x0203 — the session itself still up) retires +/// the pair at once, whether a request or a keep-alive got it, so the next send reopens it (Gitea #566). /// Bounded. Every is bounded even if the caller passes no token /// (ruling R9-3): opening (including the throttle wait and any wait behind another opener) is bounded by /// MinForwardOpenInterval + 4 × Timeout (connect, RegisterSession, Forward Open, fallback Forward @@ -21,70 +23,6 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip; /// public sealed class CipConnection : ICipChannel, IAsyncDisposable { - /// Forward Open frames sent by every connection in the process (fallback included). Gate diagnostics. - internal static int ForwardOpenCount; - - /// Forward Close requests put on the wire by every connection in the process. Gate diagnostics. - internal static int ForwardClosesSent; - - /// Forward Close requests that got a reply within their budget. Gate diagnostics. - internal static int ForwardClosesAcked; - - /// Channel sends by every connection in the process (keep-alives included). Gate diagnostics. - internal static int RequestsSent; - - // Round-trip statistics of completed exchanges, process-wide (gate diagnostics: T21 latency attribution). - // Two Stopwatch reads and four interlocked ops per request; reset by the harness between timed passes. - private static long _exchangeCount; - private static long _exchangeTotalTicks; - private static long _exchangeMinTicks = long.MaxValue; - private static long _exchangeMaxTicks; - - /// Round-trip statistics of the exchanges completed since the last . - internal readonly record struct ExchangeStats(long Count, double MinMs, double MaxMs, double MeanMs); - - /// Clears the process-wide round-trip statistics. Gate diagnostics. - internal static void ResetExchangeStats() - { - Interlocked.Exchange(ref _exchangeCount, 0); - Interlocked.Exchange(ref _exchangeTotalTicks, 0); - Interlocked.Exchange(ref _exchangeMinTicks, long.MaxValue); - Interlocked.Exchange(ref _exchangeMaxTicks, 0); - } - - /// Snapshot of the process-wide round-trip statistics. Gate diagnostics. - internal static ExchangeStats ReadExchangeStats() - { - var count = Interlocked.Read(ref _exchangeCount); - if (count == 0) return new ExchangeStats(0, 0, 0, 0); - static double Ms(long ticks) => ticks * 1000.0 / Stopwatch.Frequency; - return new ExchangeStats(count, Ms(Interlocked.Read(ref _exchangeMinTicks)), Ms(Interlocked.Read(ref _exchangeMaxTicks)), - Ms(Interlocked.Read(ref _exchangeTotalTicks)) / count); - } - - private static void RecordExchange(long ticks) - { - Interlocked.Increment(ref _exchangeCount); - Interlocked.Add(ref _exchangeTotalTicks, ticks); - long seen; - while (ticks < (seen = Interlocked.Read(ref _exchangeMinTicks)) - && Interlocked.CompareExchange(ref _exchangeMinTicks, ticks, seen) != seen) { } - while (ticks > (seen = Interlocked.Read(ref _exchangeMaxTicks)) - && Interlocked.CompareExchange(ref _exchangeMaxTicks, ticks, seen) != seen) { } - } - - /// Open attempts (TCP connect + RegisterSession + Forward Open, a fallback open included) started by - /// every connection in the process — counted before any I/O, so a refused attempt counts too. Gate diagnostics. - internal static int OpenAttempts; - - /// - /// Process-wide cap on ; -1 (the default) is unlimited. An attempt beyond - /// the cap fails with before any I/O — no TCP connect, no Forward Open. Only the - /// read-only differential harness (Task 18/21) sets it, so a production press that refuses or drops the - /// connection sees exactly one open attempt instead of one per remaining operation. - /// - internal static int OpenAttemptBudget = -1; - /// Get Attribute Single, Identity (class 1) instance 1 attribute 1 (Vendor ID). internal static readonly byte[] KeepAliveRequest = [0x0E, 0x03, 0x20, 0x01, 0x24, 0x01, 0x30, 0x01]; @@ -98,6 +36,7 @@ public sealed class CipConnection : ICipChannel, IAsyncDisposable private readonly ForwardOpenThrottle _throttle; private readonly TimeProvider _time; private readonly Task _predecessorClosed; + private readonly ConnectionDiagnostics _diagnostics; private readonly byte[] _route; private readonly object _sync = new(); @@ -119,9 +58,11 @@ public sealed class CipConnection : ICipChannel, IAsyncDisposable /// Clock for the throttle, deadlines and keep-alive. /// Completes when the previous connection for the same key has /// finished closing, so two connections to one device never overlap. + /// The owning pool's counters and open-attempt limits; a private set when null. internal CipConnection(DeviceKey key, ConnectionOptions options, ForwardOpenThrottle throttle, TimeProvider time, - Task? predecessorClosed = null) + Task? predecessorClosed = null, ConnectionDiagnostics? diagnostics = null) { + _diagnostics = diagnostics ?? new ConnectionDiagnostics(); _key = key ?? throw new ArgumentNullException(nameof(key)); _o = options ?? throw new ArgumentNullException(nameof(options)); _throttle = throttle ?? throw new ArgumentNullException(nameof(throttle)); @@ -141,6 +82,12 @@ public sealed class CipConnection : ICipChannel, IAsyncDisposable /// Test hook: keep-alives started (counted synchronously when the timer decides to send). internal int KeepAlivesSent => Volatile.Read(ref _keepAlivesSent); + /// This connection's counters (its pool's, when it has one). + internal ConnectionDiagnostics Diagnostics => _diagnostics; + + /// Test hook: true while a shared open is in progress. + internal bool OpenInProgress { get { lock (_sync) return _pendingOpen is not null; } } + private TimeSpan OpenBudget => _o.MinForwardOpenInterval + 4 * _o.Timeout; private TimeSpan ExchangeBudget => 2 * _o.Timeout; @@ -210,6 +157,7 @@ public sealed class CipConnection : ICipChannel, IAsyncDisposable Volatile.Write(ref _open, null); } if (stale is not null) await stale.Client.DisposeAsync().ConfigureAwait(false); // faulted: no UnRegister + if (!_predecessorClosed.IsCompleted) _diagnostics.CountPredecessorWait(); await _predecessorClosed.WaitAsync(beforeAttempt.Token).ConfigureAwait(false); opened = await _throttle.RunAsync(_ => OpenAsync(deadline.Token), beforeAttempt.Token).ConfigureAwait(false); } @@ -235,11 +183,7 @@ public sealed class CipConnection : ICipChannel, IAsyncDisposable private async Task OpenAsync(CancellationToken ct) { ThrowIfDisposed(); // disposal may have begun while this attempt waited out the throttle - var attempt = Interlocked.Increment(ref OpenAttempts); - var budget = Volatile.Read(ref OpenAttemptBudget); - if (budget >= 0 && attempt > budget) - throw new AbTransportException(AbStatus.Abort, - $"Open attempt {attempt} to {_key.Host}:{_key.Port} refused before any I/O: the process-wide open-attempt budget of {budget} is spent."); + if (_diagnostics.AdmitOpenAttempt(_key) is { } refused) throw new AbTransportException(AbStatus.Abort, refused); var client = await EncapsulationClient.ConnectAsync(_key.Host, _key.Port, _o.Timeout, ct).ConfigureAwait(false); if (Volatile.Read(ref _disposed) != 0) { @@ -276,7 +220,7 @@ public sealed class CipConnection : ICipChannel, IAsyncDisposable var p = new ForwardOpenParams(large, toId, serial, _o.VendorId, _o.OriginatorSerial, _o.TimeoutMultiplier, _o.RpiMicros, size, _route); var frame = ConnectionManagerServices.ForwardOpen(p); - Interlocked.Increment(ref ForwardOpenCount); + _diagnostics.CountForwardOpen(); return CipReply.Parse(await client.SendRRDataAsync(frame, _o.Timeout, ct).ConfigureAwait(false)); } @@ -289,7 +233,7 @@ public sealed class CipConnection : ICipChannel, IAsyncDisposable /// One request over an established pair, bounded by . private async Task> ExchangeAsync(Open open, ReadOnlyMemory cip, CancellationToken ct) { - Interlocked.Increment(ref RequestsSent); + _diagnostics.CountRequest(); Interlocked.Exchange(ref _lastSendTimestamp, _time.GetTimestamp()); using var deadline = new CancellationTokenSource(ExchangeBudget, _time); using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, deadline.Token); @@ -302,7 +246,14 @@ public sealed class CipConnection : ICipChannel, IAsyncDisposable CipRouting.UnconnectedSend => await open.Client.SendRRDataAsync(ConnectionManagerServices.UnconnectedSend(cip.Span, _route), _o.Timeout, linked.Token).ConfigureAwait(false), _ => await open.Client.SendRRDataAsync(cip, _o.Timeout, linked.Token).ConfigureAwait(false), }; - RecordExchange(Stopwatch.GetTimestamp() - started); + _diagnostics.RecordExchange(Stopwatch.GetTimestamp() - started); + if (_o.Routing == CipRouting.Connected && ConnectionManagerServices.IsConnectionGone(reply.Span) is { } extended) + { + await RetireClosedByTargetAsync(open).ConfigureAwait(false); + throw new AbTransportException(AbStatus.BadConnection, + $"{_key.Host}:{_key.Port} no longer holds the CIP connection (general status 0x01, extended 0x{extended:X4}); " + + "the session was closed and the next request reopens it."); + } return reply; } catch (OperationCanceledException) when (!ct.IsCancellationRequested && deadline.IsCancellationRequested) @@ -312,6 +263,24 @@ public sealed class CipConnection : ICipChannel, IAsyncDisposable } } + /// + /// The target answered over a session that is still up, but says the class-3 connection itself is gone (it + /// timed the connection out, or was restarted / re-downloaded and forgot it). Every later connected request + /// would get the same answer while stays true, so nothing else + /// would ever reopen it (Gitea #566): retire the pair now, so the next send reopens through the throttle. + /// No Forward Close — the target has just said there is nothing to close. + /// + private async Task RetireClosedByTargetAsync(Open open) + { + lock (_sync) + { + if (!ReferenceEquals(_open, open)) return; // a concurrent exchange (keep-alive) already retired it + Volatile.Write(ref _open, null); + } + _diagnostics.CountClosedByTarget(); + await open.Client.DisposeAsync().ConfigureAwait(false); // UnRegisterSession, then close the socket + } + /// One-shot timer, re-armed after every tick so the longest idle gap is one /// (half the connection expiry), not two. private void OnKeepAliveTimer() @@ -396,11 +365,11 @@ public sealed class CipConnection : ICipChannel, IAsyncDisposable { var budget = TimeSpan.FromSeconds(1); using var cts = new CancellationTokenSource(budget, _time); - Interlocked.Increment(ref ForwardClosesSent); + _diagnostics.CountForwardCloseSent(); await open.Client.SendRRDataAsync( ConnectionManagerServices.ForwardClose(open.ConnectionSerial, _o.VendorId, _o.OriginatorSerial, _route), budget, cts.Token).ConfigureAwait(false); - Interlocked.Increment(ref ForwardClosesAcked); + _diagnostics.CountForwardCloseAcked(); } catch (Exception ex) when (ex is AbTransportException or OperationCanceledException) { diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ConnectionDiagnostics.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ConnectionDiagnostics.cs new file mode 100644 index 00000000..f89cf3f3 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ConnectionDiagnostics.cs @@ -0,0 +1,163 @@ +using System.Diagnostics; +namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip; + +/// +/// Counters and the open-attempt limit for the s one +/// owns. Gate and harness diagnostics: nothing in production reads them, and a limit is only ever set by the +/// read-only differential harness. +/// +/// +/// Instance-scoped (Gitea #581): every pool — and every connection a test builds on its own — has its own, so +/// no process-wide mutable state ships in the transport, and two tests can never move each other's counts. The +/// open-attempt limit is scoped to one (Gitea #566), so limiting the press under test +/// can never refuse an open to any other device in the process. +/// +internal sealed class ConnectionDiagnostics +{ + private readonly object _lock = new(); + private readonly Dictionary _limits = new(); + + private int _forwardOpens; + private int _forwardClosesSent; + private int _forwardClosesAcked; + private int _requestsSent; + private int _openAttempts; + private int _predecessorWaits; + private int _connectionsClosedByTarget; + + // Round-trip statistics of completed exchanges (T21 latency attribution). Two Stopwatch reads and four + // interlocked ops per request; reset by the harness between timed passes. + private long _exchangeCount; + private long _exchangeTotalTicks; + private long _exchangeMinTicks = long.MaxValue; + private long _exchangeMaxTicks; + + /// Forward Open frames sent (fallback included). + public int ForwardOpens => Volatile.Read(ref _forwardOpens); + + /// Forward Close requests put on the wire. + public int ForwardClosesSent => Volatile.Read(ref _forwardClosesSent); + + /// Forward Close requests that got a reply within their budget. + public int ForwardClosesAcked => Volatile.Read(ref _forwardClosesAcked); + + /// Channel sends (keep-alives included). + public int RequestsSent => Volatile.Read(ref _requestsSent); + + /// Open attempts (TCP connect + RegisterSession + Forward Open, a fallback open included), counted + /// before any I/O, so an attempt refused by a limit counts too. + public int OpenAttempts => Volatile.Read(ref _openAttempts); + + /// Opens that found an earlier connection for the same device still closing and parked behind it. + public int PredecessorWaits => Volatile.Read(ref _predecessorWaits); + + /// Connections retired because the target reported the CIP connection itself gone (TCP still up). + public int ConnectionsClosedByTarget => Volatile.Read(ref _connectionsClosedByTarget); + + /// Zeroes the Forward Open / Forward Close counters (the harness, before its native phase). + public void ResetForwardCounts() + { + Interlocked.Exchange(ref _forwardOpens, 0); + Interlocked.Exchange(ref _forwardClosesSent, 0); + Interlocked.Exchange(ref _forwardClosesAcked, 0); + } + + internal void CountForwardOpen() => Interlocked.Increment(ref _forwardOpens); + internal void CountForwardCloseSent() => Interlocked.Increment(ref _forwardClosesSent); + internal void CountForwardCloseAcked() => Interlocked.Increment(ref _forwardClosesAcked); + internal void CountRequest() => Interlocked.Increment(ref _requestsSent); + internal void CountPredecessorWait() => Interlocked.Increment(ref _predecessorWaits); + internal void CountClosedByTarget() => Interlocked.Increment(ref _connectionsClosedByTarget); + + /// Round-trip statistics of the exchanges completed since the last . + internal readonly record struct ExchangeStats(long Count, double MinMs, double MaxMs, double MeanMs); + + /// Clears the round-trip statistics. + public void ResetExchangeStats() + { + Interlocked.Exchange(ref _exchangeCount, 0); + Interlocked.Exchange(ref _exchangeTotalTicks, 0); + Interlocked.Exchange(ref _exchangeMinTicks, long.MaxValue); + Interlocked.Exchange(ref _exchangeMaxTicks, 0); + } + + /// Snapshot of the round-trip statistics. + public ExchangeStats ReadExchangeStats() + { + var count = Interlocked.Read(ref _exchangeCount); + if (count == 0) return new ExchangeStats(0, 0, 0, 0); + static double Ms(long ticks) => ticks * 1000.0 / Stopwatch.Frequency; + return new ExchangeStats(count, Ms(Interlocked.Read(ref _exchangeMinTicks)), Ms(Interlocked.Read(ref _exchangeMaxTicks)), + Ms(Interlocked.Read(ref _exchangeTotalTicks)) / count); + } + + internal void RecordExchange(long ticks) + { + Interlocked.Increment(ref _exchangeCount); + Interlocked.Add(ref _exchangeTotalTicks, ticks); + long seen; + while (ticks < (seen = Interlocked.Read(ref _exchangeMinTicks)) + && Interlocked.CompareExchange(ref _exchangeMinTicks, ticks, seen) != seen) { } + while (ticks > (seen = Interlocked.Read(ref _exchangeMaxTicks)) + && Interlocked.CompareExchange(ref _exchangeMaxTicks, ticks, seen) != seen) { } + } + + /// + /// Caps the open attempts to at until the returned scope is + /// disposed. An attempt beyond the cap fails with before any I/O — no TCP connect, + /// no Forward Open — so a press that refuses or drops the connection sees exactly the budgeted attempts instead + /// of one per remaining operation. Only the read-only differential harness (Task 18/21) sets one; one limit per + /// key at a time. + /// + public OpenAttemptLimit LimitOpenAttempts(DeviceKey key, int attempts) + { + ArgumentNullException.ThrowIfNull(key); + ArgumentOutOfRangeException.ThrowIfNegative(attempts); + var limit = new OpenAttemptLimit(this, key, attempts); + lock (_lock) + { + if (!_limits.TryAdd(key, limit)) + throw new InvalidOperationException($"An open-attempt limit for {key.Host}:{key.Port} is already in force."); + } + return limit; + } + + /// True while a limit is in force for . + public bool HasOpenAttemptLimit(DeviceKey key) { lock (_lock) return _limits.ContainsKey(key); } + + /// Counts one open attempt to ; null when it may proceed, else why it may not. + internal string? AdmitOpenAttempt(DeviceKey key) + { + var attempt = Interlocked.Increment(ref _openAttempts); + OpenAttemptLimit? limit; + lock (_lock) _limits.TryGetValue(key, out limit); + return limit is null || limit.TryTake() + ? null + : $"Open attempt {attempt} to {key.Host}:{key.Port} refused before any I/O: the open-attempt budget of {limit.Budget} for this device is spent."; + } + + private void Remove(OpenAttemptLimit limit) + { + lock (_lock) + { + if (_limits.TryGetValue(limit.Key, out var current) && current == limit) _limits.Remove(limit.Key); + } + } + + /// An open-attempt cap on one device; disposing it lifts the cap. + internal sealed class OpenAttemptLimit(ConnectionDiagnostics owner, DeviceKey key, int budget) : IDisposable + { + private int _attempts; + + public DeviceKey Key { get; } = key; + + public int Budget { get; } = budget; + + /// Attempts counted against this limit, the refused ones included. + public int Attempts => Volatile.Read(ref _attempts); + + internal bool TryTake() => Interlocked.Increment(ref _attempts) <= Budget; + + public void Dispose() => owner.Remove(this); + } +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ConnectionManagerServices.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ConnectionManagerServices.cs index 58235c4c..96dcd63c 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ConnectionManagerServices.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ConnectionManagerServices.cs @@ -83,6 +83,26 @@ public static class ConnectionManagerServices r.GeneralStatus == 0x08 || (r.GeneralStatus == 0x01 && r.ExtendedStatus.Length > 0 && r.ExtendedStatus[0] == 0x0109); + /// Connection Manager extended status: target connection not found. + public const ushort TargetConnectionNotFound = 0x0107; + + /// Connection Manager extended status: the connection timed out. + public const ushort ConnectionTimedOut = 0x0203; + + /// + /// For a reply received over a class-3 connection: the extended status when the target says the connection + /// itself no longer exists — general 0x01 (Connection Failure) with extended 0x0107 (target connection not + /// found) or 0x0203 (connection timed out) — else null. Any other status, a tag-level one included, describes + /// the request and leaves the connection alone. + /// + public static ushort? IsConnectionGone(ReadOnlySpan reply) + { + // Reply header: service | reserved | general status | extended-status word count | words... + if (reply.Length < 6 || reply[2] != 0x01 || reply[3] == 0) return null; + var extended = (ushort)(reply[4] | (reply[5] << 8)); + return extended is TargetConnectionNotFound or ConnectionTimedOut ? extended : null; + } + public static byte[] ForwardClose(ushort connectionSerial, ushort vendorId, uint originatorSerial, ReadOnlySpan route) { var connPath = new CipWriter().Bytes(route).Bytes(CipPath.MessageRouter).ToArray(); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ForwardOpenThrottle.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ForwardOpenThrottle.cs index b315aa06..c1179ae5 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ForwardOpenThrottle.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ForwardOpenThrottle.cs @@ -13,6 +13,8 @@ public sealed class ForwardOpenThrottle private readonly TimeSpan _minInterval; private readonly TimeProvider _time; private long? _lastStartTimestamp; // monotonic (TimeProvider.GetTimestamp); only touched while holding _inFlight + private int _queued; + private Task? _intervalWait; // the pending interval wait (test hook only) /// Internal: the only legitimate owner is , which keeps exactly /// one per device. A second throttle for the same device would halve the guaranteed interval. @@ -24,6 +26,13 @@ public sealed class ForwardOpenThrottle _time = time ?? throw new ArgumentNullException(nameof(time)); } + /// Test hook: callers waiting behind an in-flight attempt. + internal int Queued => Volatile.Read(ref _queued); + + /// Test hook: true while the next attempt is parked on the minimum interval and that wait has not + /// yet elapsed — no attempt is then in flight, and none can start until the clock moves. + internal bool IsWaitingOutInterval => Volatile.Read(ref _intervalWait) is { IsCompleted: false }; + /// Run one open attempt once the throttle allows it. bounds only the wait /// BEFORE the attempt (queued behind an in-flight attempt, or the interval): when it fires there, this /// throws (, "Forward Open @@ -33,9 +42,11 @@ public sealed class ForwardOpenThrottle public async Task RunAsync(Func> open, CancellationToken ct) { ArgumentNullException.ThrowIfNull(open); + Interlocked.Increment(ref _queued); try { await _inFlight.WaitAsync(ct).ConfigureAwait(false); } catch (OperationCanceledException) when (ct.IsCancellationRequested) { throw Throttled("another attempt is in flight"); } + finally { Interlocked.Decrement(ref _queued); } try { @@ -45,9 +56,12 @@ public sealed class ForwardOpenThrottle var wait = _minInterval - _time.GetElapsedTime(last); if (wait > TimeSpan.Zero) { - try { await Task.Delay(wait, _time, ct).ConfigureAwait(false); } + var delay = Task.Delay(wait, _time, ct); + Volatile.Write(ref _intervalWait, delay); + try { await delay.ConfigureAwait(false); } catch (OperationCanceledException) when (ct.IsCancellationRequested) { throw Throttled("the minimum interval had not elapsed"); } + finally { Volatile.Write(ref _intervalWait, null); } } } _lastStartTimestamp = _time.GetTimestamp(); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/DeviceSessionPool.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/DeviceSessionPool.cs index 1572caa3..9d137d13 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/DeviceSessionPool.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/DeviceSessionPool.cs @@ -52,6 +52,9 @@ public sealed class DeviceSessionPool /// Wiring-test hook: the logger this pool currently writes to (final review I2). internal ILogger Logger => _logger; + /// Counters and open-attempt limits for this pool's connections (gate and harness diagnostics). + internal ConnectionDiagnostics Diagnostics { get; } = new(); + /// Test hook: keys with a live entry. internal int EntryCount { get { lock (_lock) return _entries.Count; } } @@ -69,7 +72,7 @@ public sealed class DeviceSessionPool if (!_throttles.TryGetValue(key, out var throttle)) _throttles[key] = throttle = new ForwardOpenThrottle(options.MinForwardOpenInterval, _time); var predecessor = _closing.GetValueOrDefault(key) ?? Task.CompletedTask; - var connection = new CipConnection(key, options, throttle, _time, predecessor); + var connection = new CipConnection(key, options, throttle, _time, predecessor, Diagnostics); entry = new Entry(connection, new RequestScheduler(connection, options.AllowPacking), options, predecessor); _entries[key] = entry; } @@ -109,28 +112,36 @@ public sealed class DeviceSessionPool } }, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); - // Outside the lock: scheduler first (fails its waiters, stops sending), then Forward Close + - // UnRegisterSession via the connection. The connection is disposed even if the scheduler's - // dispose throws: otherwise its keep-alive would hold the press connection open forever. + // Outside the lock. try { - try { await entry.Scheduler.DisposeAsync().ConfigureAwait(false); } - catch (Exception ex) - { - _logger.LogWarning(ex, "Disposing the request scheduler for {Host}:{Port} failed; closing the connection anyway.", - key.Host, key.Port); - } - try { await entry.Connection.DisposeAsync().ConfigureAwait(false); } - catch (Exception ex) - { - _logger.LogWarning(ex, "Closing the connection to {Host}:{Port} failed.", key.Host, key.Port); - } + await CloseAsync(key, entry.Scheduler, entry.Connection, _logger).ConfigureAwait(false); } finally { closed.TrySetResult(); // never faults: a successor only needs to know the close is over } } + + /// + /// Scheduler first (fails its waiters, stops sending), then Forward Close + UnRegisterSession via the + /// connection. The connection is disposed even if the scheduler's dispose throws: otherwise its keep-alive + /// would hold the press connection open forever. Never throws; each failure is logged. + /// + internal static async Task CloseAsync(DeviceKey key, IAsyncDisposable scheduler, IAsyncDisposable connection, ILogger logger) + { + try { await scheduler.DisposeAsync().ConfigureAwait(false); } + catch (Exception ex) + { + logger.LogWarning(ex, "Disposing the request scheduler for {Host}:{Port} failed; closing the connection anyway.", + key.Host, key.Port); + } + try { await connection.DisposeAsync().ConfigureAwait(false); } + catch (Exception ex) + { + logger.LogWarning(ex, "Closing the connection to {Host}:{Port} failed.", key.Host, key.Port); + } + } } /// A reference on a device's shared connection. Dispose exactly once (further disposes are no-ops). diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/CipConnectionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/CipConnectionTests.cs index e76e9c98..bffdec3f 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/CipConnectionTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/CipConnectionTests.cs @@ -2,11 +2,14 @@ using System.Buffers.Binary; using Shouldly; using Xunit; using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip; +using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Eip; using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Fakes; namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Cip; -/// Serializes every test class that asserts on 's process-wide -/// diagnostic counters, so no parallel test can move them mid-assertion. +/// Serializes the loopback-socket test classes. Their diagnostic counters are per connection / per +/// pool since Gitea #581 (no longer process-wide statics), but still +/// reconfigures 's logger, and the timing-sensitive storm test is kept +/// off a thread pool its neighbours are saturating. [CollectionDefinition(Name, DisableParallelization = true)] public sealed class CipConnectionStatics { @@ -66,7 +69,7 @@ public sealed class CipConnectionTests target.ForwardOpenReply = fo => fo[0] == 0x5B ? FakeCipTarget.Refuse(0x08)(fo) : FakeCipTarget.Accept(fo); var time = new ManualTimeProvider(); await using var conn = Create(target, Options(), time); - var before = Volatile.Read(ref CipConnection.ForwardOpenCount); + var before = conn.Diagnostics.ForwardOpens; await conn.SendAsync(ReadTag, TestCt); @@ -76,7 +79,7 @@ public sealed class CipConnectionTests fos[1][0].ShouldBe((byte)0x54); RegularOTParams(fos[1]).ShouldBe((ushort)(0x4200 | 504)); conn.MaxRequestBytes.ShouldBe(502); - (Volatile.Read(ref CipConnection.ForwardOpenCount) - before).ShouldBe(2); + (conn.Diagnostics.ForwardOpens - before).ShouldBe(2); target.ConnectedRequests.ShouldBe(1); } @@ -109,14 +112,12 @@ public sealed class CipConnectionTests [Fact] public async Task A_spent_open_attempt_budget_refuses_the_reconnect_before_any_io() { - // Task 18 review r1 (C1): the differential harness caps the process at one open attempt, so a press + // Task 18 review r1 (C1): the differential harness caps the press at one open attempt, so a press // that refuses the Forward Open sees one TCP connect and one Forward Open, never one per remaining tag. await using var target = new FakeCipTarget { ForwardOpenReply = FakeCipTarget.Refuse(0x01, 0x0100) }; var time = new ManualTimeProvider(); await using var conn = Create(target, Options(size: 504), time); - var before = Volatile.Read(ref CipConnection.OpenAttempts); - Volatile.Write(ref CipConnection.OpenAttemptBudget, before + 1); - try + using (conn.Diagnostics.LimitOpenAttempts(Key(target), 1)) { await Should.ThrowAsync(() => conn.SendAsync(ReadTag, TestCt)); var retry = conn.SendAsync(ReadTag, TestCt); @@ -128,13 +129,43 @@ public sealed class CipConnectionTests refused.Message.ShouldContain("budget"); target.Server.Connections.ShouldBe(1); target.ForwardOpens.ShouldBe(1); - } - finally - { - Volatile.Write(ref CipConnection.OpenAttemptBudget, -1); + conn.Diagnostics.OpenAttempts.ShouldBe(2); // the refused one is counted too } } + [Fact] + public async Task An_open_attempt_limit_is_scoped_to_one_device_and_lifted_by_its_scope() + { + // Gitea #566 / #581: the limit is per DeviceKey on one pool's diagnostics, not a process-wide static, so + // limiting the press under test can never refuse an open to any other device. + await using var limited = new FakeCipTarget(); + await using var other = new FakeCipTarget(); + var time = new ManualTimeProvider(); + var pool = new DeviceSessionPool(time); + var limitedKey = Key(limited); + + using (pool.Diagnostics.LimitOpenAttempts(limitedKey, 0)) + { + await using var otherLease = pool.Acquire(Key(other), Options()); + (await otherLease.Scheduler.SendAsync(ReadTag, TestCt)).GeneralStatus.ShouldBe((byte)0); + + await using var limitedLease = pool.Acquire(limitedKey, Options()); + var refused = await Should.ThrowAsync(() => limitedLease.Scheduler.SendAsync(ReadTag, TestCt)); + refused.Status.ShouldBe(AbStatus.Abort); + refused.Message.ShouldContain("budget"); + Should.Throw(() => pool.Diagnostics.LimitOpenAttempts(limitedKey, 5)); // one at a time + } + + pool.Diagnostics.HasOpenAttemptLimit(limitedKey).ShouldBeFalse(); + time.Advance(TimeSpan.FromSeconds(5)); // the refused attempt still started the device's throttle interval + await using var lifted = pool.Acquire(limitedKey, Options()); + (await lifted.Scheduler.SendAsync(ReadTag, TestCt).WaitAsync(Guard, TestCt)).GeneralStatus.ShouldBe((byte)0); + limited.ForwardOpens.ShouldBe(1); // the refused attempt never reached the wire; the lifted one did + other.ForwardOpens.ShouldBe(1); + } + + private static DeviceKey Key(FakeCipTarget target) => new("127.0.0.1", target.Port, "1,0", "ControlLogix"); + [Fact] public async Task Connection_in_use_does_not_fall_back() { @@ -245,7 +276,7 @@ public sealed class CipConnectionTests }; var time = new ManualTimeProvider(); await using var conn = Create(target, Options(), time); - var before = Volatile.Read(ref CipConnection.ForwardOpenCount); + var before = conn.Diagnostics.ForwardOpens; using var cts = CancellationTokenSource.CreateLinkedTokenSource(TestCt); var pending = conn.SendAsync(ReadTag, cts.Token); @@ -256,7 +287,7 @@ public sealed class CipConnectionTests await conn.SendAsync(ReadTag, TestCt); - (Volatile.Read(ref CipConnection.ForwardOpenCount) - before).ShouldBe(1); + (conn.Diagnostics.ForwardOpens - before).ShouldBe(1); target.ForwardOpens.ShouldBe(1); target.Server.Connections.ShouldBe(1); target.ConnectedRequests.ShouldBe(2); @@ -321,7 +352,7 @@ public sealed class CipConnectionTests await using var target = new FakeCipTarget(); var time = new ManualTimeProvider(); await using var conn = Create(target, Options(), time); - var before = Volatile.Read(ref CipConnection.RequestsSent); + var before = conn.Diagnostics.RequestsSent; await conn.SendAsync(ReadTag, TestCt); time.Advance(TimeSpan.FromSeconds(8)); @@ -336,7 +367,7 @@ public sealed class CipConnectionTests time.Advance(TimeSpan.FromSeconds(1)); conn.KeepAlivesSent.ShouldBe(2); await FakeCipTarget.Until(() => target.ConnectedRequests == 3, Guard, "the second keep-alive"); - (Volatile.Read(ref CipConnection.RequestsSent) - before).ShouldBe(3); // keep-alives count + (conn.Diagnostics.RequestsSent - before).ShouldBe(3); // keep-alives count target.ForwardOpens.ShouldBe(1); } @@ -368,6 +399,9 @@ public sealed class CipConnectionTests time.Advance(TimeSpan.FromSeconds(30)); + // Positive control (#583): the keep-alive timer really ticked (at 8, 16 and 24 s) and re-armed itself, so + // the zero below is the timer declining to send, not a timer that never ran. + time.HasTimerDueAt(time.Start + TimeSpan.FromSeconds(32)).ShouldBeTrue(); conn.KeepAlivesSent.ShouldBe(0); target.ForwardOpens.ShouldBe(0); target.Server.Connections.ShouldBe(0); @@ -411,4 +445,159 @@ public sealed class CipConnectionTests await dispose.WaitAsync(Guard, TestCt); target.ForwardOpens.ShouldBe(1); // nothing is opened once disposal has begun } + + [Fact] + public async Task Disposing_during_the_session_registration_never_puts_a_Forward_Open_on_the_wire() + { + // Gitea #566: disposal that begins while the TCP connect / RegisterSession is on the wire lets that step + // finish (it is bounded), then releases the session instead of sending a Forward Open nobody would close. + await using var target = new FakeCipTarget(); + var registerHeld = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + target.Server.RegisterSessionGate = () => registerHeld.Task; + var time = new ManualTimeProvider(); + var conn = Create(target, Options(), time); + + var send = conn.SendAsync(ReadTag, TestCt); + await target.Server.WaitForFrameAsync(f => FakeEipServer.Command(f) == EncapCommand.RegisterSession, Guard); + var dispose = conn.DisposeAsync().AsTask(); + conn.OpenInProgress.ShouldBeTrue(); // disposal is waiting out the attempt on the wire... + dispose.IsCompleted.ShouldBeFalse(); // ...rather than abandoning it + registerHeld.SetResult(); + + (await Should.ThrowAsync(() => send.WaitAsync(Guard, TestCt))).Status.ShouldBe(AbStatus.Abort); + await dispose.WaitAsync(Guard, TestCt); + // The session is released (positive control); the server records frames in arrival order, so a Forward + // Open sent before the UnRegisterSession would already be counted. + await target.Server.WaitForFrameAsync(f => FakeEipServer.Command(f) == EncapCommand.UnRegisterSession, Guard); + target.ForwardOpens.ShouldBe(0); + target.ForwardCloses.ShouldBe(0); + conn.Diagnostics.OpenAttempts.ShouldBe(1); + } + + private const ushort AbandonedOpenRefusal = 0x7A11; // an extended status no other test uses + + [Fact] + public async Task An_open_every_waiter_abandoned_leaves_no_unobserved_task_exception() + { + // Gitea #566: the shared open fails AFTER its only waiter gave up (its own token). Nothing awaits the + // open's task then; without the observer it would surface as UnobservedTaskException at finalization. + var unobserved = await UnobservedTaskProbe.CollectAsync(AbandonAFailingOpenAsync, Guard); + + unobserved.ShouldNotContain(e => e.Message.Contains($"0x{AbandonedOpenRefusal:X4}")); + } + + // Kept out of the test body so no local in the test's own state machine roots the abandoned open. + [System.Runtime.CompilerServices.MethodImpl(System.Runtime.CompilerServices.MethodImplOptions.NoInlining)] + private static async Task AbandonAFailingOpenAsync() + { + await using var target = new FakeCipTarget { ForwardOpenReply = FakeCipTarget.Refuse(0x01, AbandonedOpenRefusal) }; + var openHeld = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + target.ForwardOpenGate = () => openHeld.Task; + var time = new ManualTimeProvider(); + await using var conn = Create(target, Options(size: 504), time); + using (var cts = new CancellationTokenSource()) + { + var waiter = conn.SendAsync(ReadTag, cts.Token); + await FakeCipTarget.Until(() => target.ForwardOpens == 1, Guard, "the Forward Open"); + await cts.CancelAsync(); + await Should.ThrowAsync(() => waiter); + } + openHeld.SetResult(); // the target refuses, after every waiter has gone + await FakeCipTarget.Until(() => !conn.OpenInProgress, Guard, "the abandoned open to fail"); + } + + // ---- CIP-level close (Gitea #566) ------------------------------------------------------------ + + private static byte[] Ok(byte[] cip) => [(byte)(cip[0] | 0x80), 0, 0, 0]; + + private static byte[] Status(byte[] cip, byte general, ushort extended) => + [(byte)(cip[0] | 0x80), 0, general, 1, (byte)extended, (byte)(extended >> 8)]; + + [Theory] + [InlineData(ConnectionManagerServices.TargetConnectionNotFound)] + [InlineData(ConnectionManagerServices.ConnectionTimedOut)] + public async Task A_reply_saying_the_CIP_connection_is_gone_retires_it_and_the_next_send_reopens(ushort extended) + { + // The target dropped the class-3 connection but the TCP session is still up: every later connected request + // would get the same answer with the session looking healthy, so the connection is recycled at once. + await using var target = new FakeCipTarget(); + var gone = 0; + target.ConnectedHandler = cip => Task.FromResult( + Interlocked.Exchange(ref gone, 0) == 1 ? Status(cip, 0x01, extended) : Ok(cip)); + var time = new ManualTimeProvider(); + await using var conn = Create(target, Options(), time); + await conn.SendAsync(ReadTag, TestCt); + + Volatile.Write(ref gone, 1); + var ex = await Should.ThrowAsync(() => conn.SendAsync(ReadTag, TestCt)); + + ex.Status.ShouldBe(AbStatus.BadConnection); + ex.Message.ShouldContain($"0x{extended:X4}"); + conn.Diagnostics.ConnectionsClosedByTarget.ShouldBe(1); + await target.Server.WaitForFrameAsync(f => FakeEipServer.Command(f) == EncapCommand.UnRegisterSession, Guard); + await FakeCipTarget.Until(() => target.Server.OpenConnections == 0, Guard, "the session to close"); + target.ForwardCloses.ShouldBe(0); // the target just said there is nothing to close + + var reopen = conn.SendAsync(ReadTag, TestCt); + await time.WaitForTimerDueAtAsync(time.Start + TimeSpan.FromSeconds(1), Guard); // through the throttle + time.Advance(TimeSpan.FromSeconds(1)); + CipReply.Parse(await reopen.WaitAsync(Guard, TestCt)).GeneralStatus.ShouldBe((byte)0); + target.ForwardOpens.ShouldBe(2); + target.Server.Connections.ShouldBe(2); + } + + [Fact] + public async Task A_keepalive_told_the_connection_is_gone_recycles_it_before_any_request_fails() + { + await using var target = new FakeCipTarget(); + target.ConnectedHandler = cip => Task.FromResult( + cip.SequenceEqual(CipConnection.KeepAliveRequest) ? Status(cip, 0x01, 0x0107) : Ok(cip)); + var time = new ManualTimeProvider(); + await using var conn = Create(target, Options(), time); + await conn.SendAsync(ReadTag, TestCt); + + time.Advance(TimeSpan.FromSeconds(8)); // idle: the keep-alive goes out and is told the connection is gone + await FakeCipTarget.Until(() => conn.Diagnostics.ConnectionsClosedByTarget == 1, Guard, "the retirement"); + await FakeCipTarget.Until(() => target.Server.OpenConnections == 0, Guard, "the session to close"); + + // The next request reopens (the throttle interval is long past) instead of failing on a dead connection. + CipReply.Parse(await conn.SendAsync(ReadTag, TestCt).WaitAsync(Guard, TestCt)).GeneralStatus.ShouldBe((byte)0); + target.ForwardOpens.ShouldBe(2); + } + + [Fact] + public async Task Other_error_statuses_describe_the_request_and_leave_the_connection_alone() + { + // Negative control for the recycle: a tag-level error and a connection-failure status that is not about + // this connection come back as ordinary replies, and the next request rides the same connection. + await using var target = new FakeCipTarget(); + var replies = new Queue>( + [ + cip => Status(cip, 0x04, 0x0000), // path segment error: the tag + cip => Status(cip, 0x01, 0x0204), // unconnected request timed out: not this connection + Ok, + ]); + target.ConnectedHandler = cip => Task.FromResult(replies.Dequeue()(cip)); + var time = new ManualTimeProvider(); + await using var conn = Create(target, Options(), time); + + CipReply.Parse(await conn.SendAsync(ReadTag, TestCt)).GeneralStatus.ShouldBe((byte)0x04); + CipReply.Parse(await conn.SendAsync(ReadTag, TestCt)).GeneralStatus.ShouldBe((byte)0x01); + CipReply.Parse(await conn.SendAsync(ReadTag, TestCt)).GeneralStatus.ShouldBe((byte)0); // same connection + + conn.Diagnostics.ConnectionsClosedByTarget.ShouldBe(0); + target.ForwardOpens.ShouldBe(1); + target.Server.Connections.ShouldBe(1); + } + + [Fact] + public void Only_connection_failure_0107_and_0203_mean_the_connection_is_gone() + { + ConnectionManagerServices.IsConnectionGone([0xCC, 0, 0x01, 1, 0x07, 0x01]).ShouldBe((ushort)0x0107); + ConnectionManagerServices.IsConnectionGone([0xCC, 0, 0x01, 1, 0x03, 0x02]).ShouldBe((ushort)0x0203); + ConnectionManagerServices.IsConnectionGone([0xCC, 0, 0x01, 1, 0x04, 0x02]).ShouldBeNull(); // 0x0204 + ConnectionManagerServices.IsConnectionGone([0xCC, 0, 0x01, 0]).ShouldBeNull(); // no extended word + ConnectionManagerServices.IsConnectionGone([0xCC, 0, 0x05, 1, 0x07, 0x01]).ShouldBeNull(); // not 0x01 + ConnectionManagerServices.IsConnectionGone([0xCC, 0, 0, 0]).ShouldBeNull(); + } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/ForwardOpenThrottleTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/ForwardOpenThrottleTests.cs index 6a4d3d74..77797cae 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/ForwardOpenThrottleTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/ForwardOpenThrottleTests.cs @@ -69,7 +69,8 @@ public sealed class ForwardOpenThrottleTests var first = throttle.RunAsync(Attempt, TestCt); time.Advance(TimeSpan.FromSeconds(5)); // the interval is no obstacle; only the in-flight one is var others = Enumerable.Range(0, 4).Select(_ => throttle.RunAsync(Attempt, TestCt)).ToArray(); - await Task.Delay(50, TestCt); + // Positive control (#583): all four have reached the throttle and are parked behind the in-flight attempt. + await FakeCipTarget.Until(() => throttle.Queued == 4, Guard, "four callers queued behind the first"); Volatile.Read(ref inFlight).ShouldBe(1); others.ShouldAllBe(t => !t.IsCompleted); @@ -131,8 +132,8 @@ public sealed class ForwardOpenThrottleTests await using var target = new FakeCipTarget { ForwardOpenReply = FakeCipTarget.Refuse(0x01, 0x0113) }; var time = new ManualTimeProvider(); var options = new ConnectionOptions { ConnectionSize = 504, Timeout = TimeSpan.FromSeconds(2) }; - var conn = new CipConnection(new DeviceKey("127.0.0.1", target.Port, "1,0", "ControlLogix"), options, - new ForwardOpenThrottle(options.MinForwardOpenInterval, time), time); + var throttle = new ForwardOpenThrottle(options.MinForwardOpenInterval, time); + var conn = new CipConnection(new DeviceKey("127.0.0.1", target.Port, "1,0", "ControlLogix"), options, throttle, time); using var stop = CancellationTokenSource.CreateLinkedTokenSource(TestCt); var failures = 0; @@ -151,7 +152,10 @@ public sealed class ForwardOpenThrottleTests await Task.Delay(10, TestCt); time.Advance(TimeSpan.FromMilliseconds(100)); } - await Task.Delay(100, TestCt); // let any attempt released by the last step reach the wire + // Positive control (#583), instead of a fixed settle: wait until the next attempt is parked on the interval. + // Then no attempt is in flight and none can start until the clock moves, so every attempt the last step + // released has already reached the wire and been counted. + await FakeCipTarget.Until(() => throttle.IsWaitingOutInterval, Guard, "the next attempt to park on the interval"); await stop.CancelAsync(); await Task.WhenAll(callers).WaitAsync(Guard, TestCt); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/DeviceSessionPoolTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/DeviceSessionPoolTests.cs index 618ccb4c..bfbc3295 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/DeviceSessionPoolTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/DeviceSessionPoolTests.cs @@ -203,7 +203,9 @@ public sealed class DeviceSessionPoolTests await using var c = pool.Acquire(key, Options()); var cSend = c.Scheduler.SendAsync(ReadTag, TestCt); - await Task.Delay(300, TestCt); + // Positive control (#583): C's open has reached the predecessor wait and parked there, so the absence of + // its Forward Open below is the wait holding it, not C merely not having got that far yet. + await FakeCipTarget.Until(() => pool.Diagnostics.PredecessorWaits == 1, Guard, "C to park behind A's close"); target.ForwardOpens.ShouldBe(1); // C is held back behind A cSend.IsCompleted.ShouldBeFalse(); @@ -292,7 +294,7 @@ public sealed class DeviceSessionPoolTests time.Advance(TimeSpan.FromSeconds(5)); // the throttle is no obstacle await using var b = pool.Acquire(key, Options()); var bSend = b.Scheduler.SendAsync(ReadTag, TestCt); - await Task.Delay(300, TestCt); + await FakeCipTarget.Until(() => pool.Diagnostics.PredecessorWaits == 1, Guard, "B to park behind A's close"); target.ForwardOpens.ShouldBe(1); // held behind A's close bSend.IsCompleted.ShouldBeFalse(); @@ -304,6 +306,72 @@ public sealed class DeviceSessionPoolTests pool.EntryCount.ShouldBe(1); } + /// + /// Gitea #566: releasing a lease while the device's shared open is still on the wire does not wait for that + /// open (up to its whole open budget) — (#579) returns at + /// once, and the teardown it hands back still Forward Closes what the open produced once it lands. + /// + [Fact] + public async Task Release_in_background_does_not_wait_behind_an_in_flight_shared_open() + { + await using var target = new FakeCipTarget(); + var openHeld = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + target.ForwardOpenGate = () => openHeld.Task; + var pool = new DeviceSessionPool(new ManualTimeProvider()); + var key = new DeviceKey("127.0.0.1", target.Port, "1,0", "ControlLogix"); + + var lease = pool.Acquire(key, Options()); + var send = lease.Scheduler.SendAsync(ReadTag, TestCt); + await FakeCipTarget.Until(() => target.ForwardOpens == 1, Guard, "the Forward Open"); + ((CipConnection)lease.Channel).OpenInProgress.ShouldBeTrue(); + + var teardown = lease.ReleaseInBackground(); // returns synchronously: the open is still held + + pool.EntryCount.ShouldBe(0); + teardown.IsCompleted.ShouldBeFalse("the teardown waits for the open on the wire; the release call did not"); + (await Should.ThrowAsync(() => send.WaitAsync(Guard, TestCt))).Status.ShouldBe(AbStatus.BadConnection); + + openHeld.SetResult(); + await teardown.WaitAsync(Guard, TestCt); + target.ForwardCloses.ShouldBe(1); // what the late-accepted open created is still closed + } + + /// Gitea #566: the close path disposes the connection even when the scheduler's dispose throws — + /// otherwise the connection's keep-alive would hold the press connection open forever. + [Fact] + public async Task Close_disposes_the_connection_even_when_the_scheduler_dispose_throws() + { + var logger = new CapturingLogger(); + var scheduler = new Disposable(() => throw new InvalidOperationException("scheduler dispose failed")); + var connection = new Disposable(() => { }); + var key = new DeviceKey("plc1", 44818, "1,0", "ControlLogix"); + + await DeviceSessionPool.CloseAsync(key, scheduler, connection, logger); + + scheduler.Disposed.ShouldBe(1); + connection.Disposed.ShouldBe(1); + logger.Warnings.ShouldBe(1); + + // And a throwing connection dispose is logged, never propagated to the release. + var failing = new Disposable(() => throw new InvalidOperationException("close failed")); + await DeviceSessionPool.CloseAsync(key, new Disposable(() => { }), failing, logger); + failing.Disposed.ShouldBe(1); + logger.Warnings.ShouldBe(2); + } + + private sealed class Disposable(Action onDispose) : IAsyncDisposable + { + private int _disposed; + public int Disposed => Volatile.Read(ref _disposed); + + public ValueTask DisposeAsync() + { + Interlocked.Increment(ref _disposed); + onDispose(); + return ValueTask.CompletedTask; + } + } + [Fact] public void Shared_is_one_process_wide_instance() => DeviceSessionPool.Shared.ShouldBeSameAs(DeviceSessionPool.Shared); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Eip/EncapsulationClientTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Eip/EncapsulationClientTests.cs index 56bcda11..e70fb159 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Eip/EncapsulationClientTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Eip/EncapsulationClientTests.cs @@ -271,28 +271,10 @@ public sealed class EncapsulationClientTests [Fact] public async Task An_abandoned_exchange_that_times_out_leaves_no_unobserved_task_exception() { - var unobserved = new List(); - EventHandler onUnobserved = (_, e) => - { - if (e.Exception.Flatten().InnerExceptions.Any(x => x.Message.Contains("within 317 ms"))) - lock (unobserved) unobserved.Add(e.Exception); - }; - TaskScheduler.UnobservedTaskException += onUnobserved; - try - { - await AbandonAnExchangeThatTimesOut(); - for (var i = 0; i < 5; i++) - { - GC.Collect(); - GC.WaitForPendingFinalizers(); - await Task.Delay(20, TestCt); - } - lock (unobserved) unobserved.ShouldBeEmpty(); - } - finally - { - TaskScheduler.UnobservedTaskException -= onUnobserved; - } + // Gitea #583: collected against a positive control (a deliberately unobserved fault that MUST be reported), + // not after a fixed number of GC cycles. + var unobserved = await UnobservedTaskProbe.CollectAsync(AbandonAnExchangeThatTimesOut, Guard); + unobserved.ShouldNotContain(x => x.Message.Contains("within 317 ms")); } // Kept out of the test body so no local in the test's own state machine roots the abandoned task. diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeEipServer.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeEipServer.cs index b6866fcd..b8ff62cb 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeEipServer.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeEipServer.cs @@ -61,6 +61,9 @@ internal sealed class FakeEipServer : IAsyncDisposable /// Synchronous convenience setter for . public Func Handler { set => HandlerAsync = req => Task.FromResult(value(req)); } + /// Awaited before a RegisterSession is answered (the request is already recorded). + public Func? RegisterSessionGate { get; set; } + /// Never answer non-session requests. public bool Stall { get; set; } @@ -160,6 +163,7 @@ internal sealed class FakeEipServer : IAsyncDisposable switch (Command(frame)) { case EncapCommand.RegisterSession: + if (RegisterSessionGate is { } gate) await gate(); reply = (byte[])frame.Clone(); BinaryPrimitives.WriteUInt32LittleEndian(reply.AsSpan(4), Session); break; diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/UnobservedTaskProbe.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/UnobservedTaskProbe.cs new file mode 100644 index 00000000..1313e9d0 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/UnobservedTaskProbe.cs @@ -0,0 +1,54 @@ +using System.Collections.Concurrent; +using System.Runtime.CompilerServices; +namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Fakes; + +/// +/// Collects s raised after some code abandons work, with a +/// positive control (Gitea #583): after the abandon, a task is deliberately faulted and dropped, and garbage is +/// collected until THAT task's exception has been reported. Only then is the collection returned — so "nothing +/// of ours was reported" means the finalizer pass that would have reported it really ran, not merely that a +/// fixed number of GC cycles went by before the runtime got round to it. +/// +internal static class UnobservedTaskProbe +{ + public static async Task> CollectAsync(Func abandon, TimeSpan timeout) + { + var marker = $"unobserved-task-probe-control-{Guid.NewGuid():N}"; + var seen = new ConcurrentQueue(); + var controlReported = 0; + EventHandler handler = (_, e) => + { + foreach (var ex in e.Exception.Flatten().InnerExceptions) + { + if (ex.Message == marker) Volatile.Write(ref controlReported, 1); + else seen.Enqueue(ex); + } + }; + TaskScheduler.UnobservedTaskException += handler; + try + { + await abandon(); + AbandonControl(marker); + var giveUp = DateTime.UtcNow + timeout; + while (Volatile.Read(ref controlReported) == 0) + { + if (DateTime.UtcNow > giveUp) + throw new TimeoutException("The positive control's unobserved exception was never reported."); + GC.Collect(); + GC.WaitForPendingFinalizers(); + await Task.Delay(10); + } + // One more full pass, so anything that became garbage in the same cycle as the control is reported too. + GC.Collect(); + GC.WaitForPendingFinalizers(); + } + finally + { + TaskScheduler.UnobservedTaskException -= handler; + } + return [.. seen]; + } + + [MethodImpl(MethodImplOptions.NoInlining)] + private static void AbandonControl(string marker) => _ = Task.FromException(new InvalidOperationException(marker)); +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialRun.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialRun.cs index ee9e9b7c..4450b63a 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialRun.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialRun.cs @@ -92,9 +92,9 @@ internal sealed class DifferentialAbortedException(string message, Exception? in /// Stop on first failure (review r1 C1). The first connection-level failure on either side — /// connect / RegisterSession / Forward Open refused, timeout, session lost — aborts the whole run with the /// operation and status; nothing is retried. The native side is additionally capped at ONE open attempt -/// process-wide (), so a reconnect cannot reach the wire even +/// to the press under test (), so a reconnect cannot reach the wire even /// from inside an enumerator that swallows a per-program failure, and the run aborts the moment -/// passes 1. +/// passes 1. /// internal static class DifferentialRun { @@ -222,9 +222,9 @@ internal static class DifferentialRun /// /// Runs with every native operation sharing ONE connection, held open by an anchor - /// lease: the process-wide Forward Open counter is reset, the open-attempt budget is set to exactly one - /// more attempt, and receives a check to call after each operation that aborts the - /// moment a second Forward Open reaches the wire. The anchor is released and the budget restored in + /// lease: the shared pool's Forward Open counters are reset, the press is limited to exactly one + /// open attempt, and receives a check to call after each operation that aborts the + /// moment a second Forward Open reaches the wire. The anchor is released and the limit lifted in /// finally. Shared by the differential and the latency comparison. /// /// The body's result, the Forward Opens counted between the reset and the anchor release, and the @@ -233,43 +233,38 @@ internal static class DifferentialRun string phase, DifferentialTarget target, Func, Task> body) { var anchorParams = target.Params("@tags"); + var key = NativeAbCipDevice.Key(anchorParams); + var diagnostics = DeviceSessionPool.Shared.Diagnostics; void CheckOpens(string operation) { - var opens = Volatile.Read(ref CipConnection.ForwardOpenCount); + var opens = diagnostics.ForwardOpens; if (opens > 1) throw new DifferentialAbortedException( $"Differential ABORTED in the {phase} phase after {operation}: {opens} Forward Open frames on the wire, " + "expected exactly one (a reconnect, or a Large→regular fallback). Nothing was retried."); } - Interlocked.Exchange(ref CipConnection.ForwardOpenCount, 0); - Interlocked.Exchange(ref CipConnection.ForwardClosesSent, 0); - Interlocked.Exchange(ref CipConnection.ForwardClosesAcked, 0); - Volatile.Write(ref CipConnection.OpenAttemptBudget, Volatile.Read(ref CipConnection.OpenAttempts) + 1); + diagnostics.ResetForwardCounts(); T result; int forwardOpens; NativeCloses forwardCloses; - try + // Exactly one open attempt to the press under test; lifted when the scope ends, whatever happens. + using (diagnostics.LimitOpenAttempts(key, 1)) { // Acquire does no I/O; the key and options carry no tag-specific field, so every lease shares this entry. - var anchor = DeviceSessionPool.Shared.Acquire(NativeAbCipDevice.Key(anchorParams), NativeAbCipDevice.Options(anchorParams)); + var anchor = DeviceSessionPool.Shared.Acquire(key, NativeAbCipDevice.Options(anchorParams)); try { result = await body(CheckOpens).ConfigureAwait(false); } finally { - forwardOpens = Volatile.Read(ref CipConnection.ForwardOpenCount); + forwardOpens = diagnostics.ForwardOpens; await anchor.DisposeAsync().ConfigureAwait(false); - forwardCloses = new NativeCloses( - Volatile.Read(ref CipConnection.ForwardClosesSent), Volatile.Read(ref CipConnection.ForwardClosesAcked)); + forwardCloses = new NativeCloses(diagnostics.ForwardClosesSent, diagnostics.ForwardClosesAcked); } } - finally - { - Volatile.Write(ref CipConnection.OpenAttemptBudget, -1); - } return (result, forwardOpens, forwardCloses); } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialSafetyTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialSafetyTests.cs index d7842605..173c88a7 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialSafetyTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/DifferentialSafetyTests.cs @@ -30,12 +30,28 @@ public sealed class DifferentialSafetyTests () => DifferentialRun.RunNativePhaseAsync(target, selection, TestContext.Current.CancellationToken)); ex.Message.ShouldContain("native phase at browse"); - // Long enough for a throttled (≥ 1 s) reconnect to have reached the listener, had one been allowed. - await Task.Delay(TimeSpan.FromSeconds(2.5), TestContext.Current.CancellationToken); - endpoint.Connections.ShouldBe(1); - Volatile.Read(ref CipConnection.OpenAttemptBudget).ShouldBe(-1, "the budget is restored after the phase"); + DeviceSessionPool.Shared.Diagnostics.HasOpenAttemptLimit(Key(target)).ShouldBeFalse("the limit is lifted after the phase"); + await ControlOpenAsync(target); + endpoint.Connections.ShouldBe(2, "the phase's one connect, then the control's — no reconnect from the phase"); } + /// + /// Positive control (Gitea #583) for "no reconnect reached the listener": one more open to the same device, + /// made after the phase, once its limit is lifted. It goes through the same per-device Forward Open throttle, + /// whose waiters are served in arrival order, so by the time it has reached the listener (it is counted on + /// accept, before the hang-up that fails it) any reconnect the phase had queued would already have got there + /// — and, the limit being gone, nothing would have refused it. + /// + private static async Task ControlOpenAsync(DifferentialTarget target) + { + // A native tag initialisation (an existence read), like the harness's own: same pool, same device key. + using var runtime = AbCipTransportFactories.For(AbCipTransport.Native).Tags + .Create(target.Params("A", 1, false, AbCipDataType.DInt)); + await Should.ThrowAsync(() => runtime.InitializeAsync(TestContext.Current.CancellationToken)); + } + + private static DeviceKey Key(DifferentialTarget target) => NativeAbCipDevice.Key(target.Params("@tags")); + [Fact] public async Task A_failing_endpoint_aborts_the_libplctag_phase_before_any_tag_is_read() { @@ -146,13 +162,13 @@ public sealed class DifferentialSafetyTests public async Task The_open_attempt_budget_alone_stops_a_second_open_even_when_nothing_aborts_the_phase() { // Unlike the abort test above, the body here SWALLOWS every failure and keeps going, so the harness's - // abort-on-first-failure can never end the phase: only CipConnection.OpenAttemptBudget stands between + // abort-on-first-failure can never end the phase: only the per-device open-attempt limit stands between // the second request and the wire. await using var endpoint = HangUpListener.Start(); var target = Target(endpoint.Port); var statuses = new List(); string? secondMessage = null; - var attemptsBefore = Volatile.Read(ref CipConnection.OpenAttempts); + var attemptsBefore = DeviceSessionPool.Shared.Diagnostics.OpenAttempts; var tags = AbCipTransportFactories.For(AbCipTransport.Native).Tags; await DifferentialRun.WithNativeAnchorAsync("native", target, async _ => @@ -178,11 +194,12 @@ public sealed class DifferentialSafetyTests statuses.Count.ShouldBe(2, "both requests failed; neither aborted the phase"); statuses[1].ShouldBe(AbStatus.Abort); secondMessage.ShouldNotBeNull().ShouldContain("budget"); - Volatile.Read(ref CipConnection.OpenAttempts).ShouldBe(attemptsBefore + 2, "the second attempt was counted, then refused"); - // Long enough for a throttled reconnect to have reached the listener had the budget let one through. - await Task.Delay(TimeSpan.FromSeconds(1.5), TestContext.Current.CancellationToken); - endpoint.Connections.ShouldBe(1, "exactly the one budgeted open attempt reached the wire"); - Volatile.Read(ref CipConnection.OpenAttemptBudget).ShouldBe(-1, "the budget is restored afterwards"); + // The pool is process-wide, so another test's opens could move this counter; at least these two are in it. + (DeviceSessionPool.Shared.Diagnostics.OpenAttempts - attemptsBefore).ShouldBeGreaterThanOrEqualTo(2, + "the second attempt was counted, then refused"); + DeviceSessionPool.Shared.Diagnostics.HasOpenAttemptLimit(Key(target)).ShouldBeFalse("the limit is lifted afterwards"); + await ControlOpenAsync(target); + endpoint.Connections.ShouldBe(2, "exactly the one budgeted open attempt reached the wire, then the control's"); } [Fact] diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/PollLatencyComparisonTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/PollLatencyComparisonTests.cs index 283cd63a..68a828b1 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/PollLatencyComparisonTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/PollLatencyComparisonTests.cs @@ -126,7 +126,7 @@ public sealed class PollLatencyComparisonTests(DifferentialRunFixture fixture) $"{passes.Sum(p => p.TagFailures)} |"; } - private static string Rtt(ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip.CipConnection.ExchangeStats? s) => + private static string Rtt(ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip.ConnectionDiagnostics.ExchangeStats? s) => s is not { Count: > 0 } x ? "-" : $"{Ms(x.MinMs)}/{Ms(x.MeanMs)}/{Ms(x.MaxMs)}"; private static string Ms(double ms) => ms.ToString("0.0", CultureInfo.InvariantCulture); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/PollLatencyRun.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/PollLatencyRun.cs index c774cb60..7e339d49 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/PollLatencyRun.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/PollLatencyRun.cs @@ -5,13 +5,13 @@ using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip; namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests.Differential; /// One timed burst: every tag read once, concurrently. -/// Native only: delta over the pass (keep-alives included). +/// Native only: delta over the pass (keep-alives included). /// Reads that ended with a tag-level status (a connection-level one aborts instead). /// Collections that ran during the pass, per generation (T21 review r1 M8: a client-side stall shows here). /// Native only: round trip of each CIP exchange in the pass, from (a slow /// controller shows here; a slow pass with fast exchanges points at the client). internal sealed record LatencyPass( - int Index, TimeSpan Wall, int? Requests, int TagFailures, GcDelta Gc = default, CipConnection.ExchangeStats? Rtt = null); + int Index, TimeSpan Wall, int? Requests, int TagFailures, GcDelta Gc = default, ConnectionDiagnostics.ExchangeStats? Rtt = null); /// GC collections per generation over a window (gen0 counts include gen1/gen2 collections, as the runtime reports). internal readonly record struct GcDelta(int Gen0, int Gen1, int Gen2) @@ -137,7 +137,7 @@ internal static class PollLatencyRun var runtimes = new List<(SelectedTag Tag, IAbCipTagRuntime Runtime)>(); try { - var before = Volatile.Read(ref CipConnection.RequestsSent); + var before = DeviceSessionPool.Shared.Diagnostics.RequestsSent; foreach (var tag in tags) { var rt = await CreateAsync("latency native", target, nativeFactory, tag, ct).ConfigureAwait(false); @@ -145,7 +145,7 @@ internal static class PollLatencyRun if (rt is null) { problems.Add($"native could not read '{tag.Name}', which libplctag read"); continue; } runtimes.Add((tag, rt)); } - var setup = Volatile.Read(ref CipConnection.RequestsSent) - before; + var setup = DeviceSessionPool.Shared.Diagnostics.RequestsSent - before; var passes = await RunPassesAsync("latency native", target, runtimes, countRequests: true, checkOpens, ct) .ConfigureAwait(false); return (passes, setup); @@ -214,15 +214,15 @@ internal static class PollLatencyRun { if (p > 1 && target.LatencyPassGapMs > 0) await Task.Delay(target.LatencyPassGapMs, ct).ConfigureAwait(false); - var requestsBefore = Volatile.Read(ref CipConnection.RequestsSent); - CipConnection.ResetExchangeStats(); + var requestsBefore = DeviceSessionPool.Shared.Diagnostics.RequestsSent; + DeviceSessionPool.Shared.Diagnostics.ResetExchangeStats(); var gcBefore = GcDelta.Now(); var clock = Stopwatch.StartNew(); var outcomes = await Task.WhenAll(runtimes.Select(r => ReadOnceAsync(r.Runtime, ct))).ConfigureAwait(false); clock.Stop(); var gc = GcDelta.Now() - gcBefore; - var rtt = countRequests ? CipConnection.ReadExchangeStats() : (CipConnection.ExchangeStats?)null; - var requests = Volatile.Read(ref CipConnection.RequestsSent) - requestsBefore; + var rtt = countRequests ? DeviceSessionPool.Shared.Diagnostics.ReadExchangeStats() : (ConnectionDiagnostics.ExchangeStats?)null; + var requests = DeviceSessionPool.Shared.Diagnostics.RequestsSent - requestsBefore; var failures = 0; for (var i = 0; i < outcomes.Length; i++) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipAlarmProjectionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipAlarmProjectionTests.cs index d7c253cc..8542a977 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipAlarmProjectionTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipAlarmProjectionTests.cs @@ -48,26 +48,33 @@ public sealed class AbCipAlarmProjectionTests AbCipAlarmProjection.MapSeverity(900).ShouldBe(AlarmSeverity.Critical); } + private static AbCipDriverOptions Options(bool enabled) => new() + { + Devices = [new AbCipDeviceOptions(Device)], + RawTags = AbCipRawTags.From(AlmdTag("HighTemp")), + EnableAlarmProjection = enabled, + AlarmPollInterval = TimeSpan.FromMilliseconds(20), + // The ALMD projection here drives the parent-UDT runtime via offset-keyed values, + // so it needs the declaration-only whole-UDT grouping fast path (Driver.AbCip-003). + EnableDeclarationOnlyUdtGrouping = true, + }; + /// Verifies that disabled alarm projection returns a valid handle but does not poll. [Fact] public async Task FeatureFlag_Off_SubscribeAlarms_Returns_Handle_But_Never_Polls() { var factory = new FakeAbCipTagFactory(); - var opts = new AbCipDriverOptions - { - Devices = [new AbCipDeviceOptions(Device)], - RawTags = AbCipRawTags.From(AlmdTag("HighTemp")), - EnableAlarmProjection = false, // explicit; also the default - }; - var drv = new AbCipDriver(opts, "drv-1", factory); + var drv = new AbCipDriver(Options(enabled: false), "drv-1", factory); await drv.InitializeAsync("{}", CancellationToken.None); var handle = await drv.SubscribeAlarmsAsync(["HighTemp"], CancellationToken.None); handle.ShouldNotBeNull(); handle.DiagnosticId.ShouldContain("abcip-alarm-sub-"); - // Wait a touch — if polling were active, a fake member-read would be triggered. - await Task.Delay(100); + // Positive control (#583): the same options with the flag ON poll three times — so a loop this + // subscription had started would have read (and so created) the ALMD runtime by now. + await PolledThreeTimesWithTheFlagOnAsync(); + factory.Tags.ShouldNotContainKey("HighTemp"); factory.Tags.ShouldNotContainKey("HighTemp.InFaulted"); factory.Tags.ShouldNotContainKey("HighTemp.Severity"); @@ -80,17 +87,7 @@ public sealed class AbCipAlarmProjectionTests public async Task FeatureFlag_On_Subscribe_Starts_Polling_And_Fires_Raise_On_0_to_1() { var factory = new FakeAbCipTagFactory(); - var opts = new AbCipDriverOptions - { - Devices = [new AbCipDeviceOptions(Device)], - RawTags = AbCipRawTags.From(AlmdTag("HighTemp")), - EnableAlarmProjection = true, - AlarmPollInterval = TimeSpan.FromMilliseconds(20), - // The ALMD projection here drives the parent-UDT runtime via offset-keyed values, - // so it needs the declaration-only whole-UDT grouping fast path (Driver.AbCip-003). - EnableDeclarationOnlyUdtGrouping = true, - }; - var drv = new AbCipDriver(opts, "drv-1", factory); + var drv = new AbCipDriver(Options(enabled: true), "drv-1", factory); await drv.InitializeAsync("{}", CancellationToken.None); var events = new List(); @@ -99,21 +96,16 @@ public sealed class AbCipAlarmProjectionTests var handle = await drv.SubscribeAlarmsAsync(["HighTemp"], CancellationToken.None); // The ALMD UDT is declared so whole-UDT grouping kicks in; the parent HighTemp runtime - // gets created + polled. Set InFaulted offset-value to 0 first (clear), wait a tick, - // then flip to 1 (fault) + wait for the raise event. + // gets created + polled. Set InFaulted offset-value to 0 first (clear), let polls seed the + // "last-seen false" state, then flip to 1 (fault) + wait for the raise event. await WaitForTagCreation(factory, "HighTemp"); factory.Tags["HighTemp"].ValuesByOffset[0] = 0; // InFaulted=false at offset 0 factory.Tags["HighTemp"].ValuesByOffset[8] = 500; // Severity at offset 8 (after InFaulted+Acked) - await Task.Delay(80); // let a tick seed the "last-seen false" state + await WaitForMoreReadsAsync(factory, "HighTemp", 2); factory.Tags["HighTemp"].ValuesByOffset[0] = 1; // flip to faulted - await Task.Delay(200); // allow several polls to be safe - - lock (events) - { - events.ShouldContain(e => e.SourceNodeId == "HighTemp" && e.AlarmType == "ALMD" - && e.Message.Contains("raised")); - } + await WaitForEventAsync(events, e => e.SourceNodeId == "HighTemp" && e.AlarmType == "ALMD" + && e.Message.Contains("raised")); await drv.UnsubscribeAlarmsAsync(handle, CancellationToken.None); await drv.ShutdownAsync(CancellationToken.None); @@ -135,15 +127,7 @@ public sealed class AbCipAlarmProjectionTests public async Task Resubscribe_Collapses_To_Single_Loop_No_Duplicate_Raise() { var factory = new FakeAbCipTagFactory(); - var opts = new AbCipDriverOptions - { - Devices = [new AbCipDeviceOptions(Device)], - RawTags = AbCipRawTags.From(AlmdTag("HighTemp")), - EnableAlarmProjection = true, - AlarmPollInterval = TimeSpan.FromMilliseconds(20), - EnableDeclarationOnlyUdtGrouping = true, - }; - var drv = new AbCipDriver(opts, "drv-1", factory); + var drv = new AbCipDriver(Options(enabled: true), "drv-1", factory); await drv.InitializeAsync("{}", CancellationToken.None); var raises = new List(); @@ -158,17 +142,18 @@ public sealed class AbCipAlarmProjectionTests await WaitForTagCreation(factory, "HighTemp"); factory.Tags["HighTemp"].ValuesByOffset[0] = 0; // InFaulted=false at offset 0 factory.Tags["HighTemp"].ValuesByOffset[8] = 500; // Severity at offset 8 - await Task.Delay(80); // let sub1's loop seed its "last-seen false" baseline + await WaitForMoreReadsAsync(factory, "HighTemp", 2); // sub1's loop has seeded its baseline // Second subscribe = the actor re-subscribing across a reconnect. Its loop reads the same - // shared HighTemp runtime; give it time to seed its own "last-seen false" baseline too. + // shared HighTemp runtime; let it poll (and so seed its own "last-seen false" baseline) too. var sub2 = await drv.SubscribeAlarmsAsync(["HighTemp"], CancellationToken.None); - await Task.Delay(80); + await WaitForMoreReadsAsync(factory, "HighTemp", 3); factory.Tags["HighTemp"].ValuesByOffset[0] = 1; // one 0->1 transition - // Wait past several 20ms poll intervals so any leaked second loop has ample time to also - // fire its duplicate raise before we assert. - await Task.Delay(250); + await WaitForEventAsync(raises, _ => true); + // Processed-count wait (#583), not a fixed delay: six more reads of the shared runtime after the raise, + // so a leaked second loop — which reads it too — has had polls in which to fire its duplicate. + await WaitForMoreReadsAsync(factory, "HighTemp", 6); await drv.UnsubscribeAlarmsAsync(sub1, CancellationToken.None); await drv.UnsubscribeAlarmsAsync(sub2, CancellationToken.None); @@ -186,17 +171,7 @@ public sealed class AbCipAlarmProjectionTests public async Task Clear_Event_Fires_On_1_to_0_Transition() { var factory = new FakeAbCipTagFactory(); - var opts = new AbCipDriverOptions - { - Devices = [new AbCipDeviceOptions(Device)], - RawTags = AbCipRawTags.From(AlmdTag("HighTemp")), - EnableAlarmProjection = true, - AlarmPollInterval = TimeSpan.FromMilliseconds(20), - // The ALMD projection here drives the parent-UDT runtime via offset-keyed values, - // so it needs the declaration-only whole-UDT grouping fast path (Driver.AbCip-003). - EnableDeclarationOnlyUdtGrouping = true, - }; - var drv = new AbCipDriver(opts, "drv-1", factory); + var drv = new AbCipDriver(Options(enabled: true), "drv-1", factory); await drv.InitializeAsync("{}", CancellationToken.None); var events = new List(); @@ -207,16 +182,10 @@ public sealed class AbCipAlarmProjectionTests factory.Tags["HighTemp"].ValuesByOffset[0] = 1; factory.Tags["HighTemp"].ValuesByOffset[8] = 500; - await Task.Delay(80); // observe raise + await WaitForEventAsync(events, e => e.Message.Contains("raised")); factory.Tags["HighTemp"].ValuesByOffset[0] = 0; - await Task.Delay(200); - - lock (events) - { - events.ShouldContain(e => e.Message.Contains("raised")); - events.ShouldContain(e => e.Message.Contains("cleared")); - } + await WaitForEventAsync(events, e => e.Message.Contains("cleared")); await drv.UnsubscribeAlarmsAsync(handle, CancellationToken.None); await drv.ShutdownAsync(CancellationToken.None); @@ -227,33 +196,59 @@ public sealed class AbCipAlarmProjectionTests public async Task Unsubscribe_Stops_The_Poll_Loop() { var factory = new FakeAbCipTagFactory(); - var opts = new AbCipDriverOptions - { - Devices = [new AbCipDeviceOptions(Device)], - RawTags = AbCipRawTags.From(AlmdTag("HighTemp")), - EnableAlarmProjection = true, - AlarmPollInterval = TimeSpan.FromMilliseconds(20), - // The ALMD projection here drives the parent-UDT runtime via offset-keyed values, - // so it needs the declaration-only whole-UDT grouping fast path (Driver.AbCip-003). - EnableDeclarationOnlyUdtGrouping = true, - }; - var drv = new AbCipDriver(opts, "drv-1", factory); + var drv = new AbCipDriver(Options(enabled: true), "drv-1", factory); await drv.InitializeAsync("{}", CancellationToken.None); var handle = await drv.SubscribeAlarmsAsync(["HighTemp"], CancellationToken.None); await WaitForTagCreation(factory, "HighTemp"); - var preUnsubReadCount = factory.Tags["HighTemp"].ReadCount; + await WaitForMoreReadsAsync(factory, "HighTemp", 1); // the loop is running + // Unsubscribe awaits the loop's exit, so no read can land after it returns. await drv.UnsubscribeAlarmsAsync(handle, CancellationToken.None); - await Task.Delay(100); // well past several poll intervals if the loop were still alive + var postUnsubReadCount = factory.Tags["HighTemp"].ReadCount; - var postDelayReadCount = factory.Tags["HighTemp"].ReadCount; - // Allow at most one straggler read between the unsubscribe-cancel + the loop exit. - (postDelayReadCount - preUnsubReadCount).ShouldBeLessThanOrEqualTo(1); + // Positive control (#583): an identical, still-subscribed driver polls three more times on the same + // interval — so the unsubscribed loop, had it survived, would have read again by now. + await PolledThreeTimesWithTheFlagOnAsync(); + factory.Tags["HighTemp"].ReadCount.ShouldBe(postUnsubReadCount); await drv.ShutdownAsync(CancellationToken.None); } + /// A control driver with the projection ON polls its ALMD runtime three times. + private static async Task PolledThreeTimesWithTheFlagOnAsync() + { + var factory = new FakeAbCipTagFactory(); + var control = new AbCipDriver(Options(enabled: true), "drv-control", factory); + await control.InitializeAsync("{}", CancellationToken.None); + var handle = await control.SubscribeAlarmsAsync(["HighTemp"], CancellationToken.None); + await WaitForTagCreation(factory, "HighTemp"); + await WaitForMoreReadsAsync(factory, "HighTemp", 3); + await control.UnsubscribeAlarmsAsync(handle, CancellationToken.None); + await control.ShutdownAsync(CancellationToken.None); + } + + /// Waits for further reads of the fake runtime (a processed-count wait). + private static async Task WaitForMoreReadsAsync(FakeAbCipTagFactory factory, string tagName, int more) + { + var tag = factory.Tags[tagName]; + var target = tag.ReadCount + more; + var deadline = DateTime.UtcNow.AddSeconds(5); + while (tag.ReadCount < target && DateTime.UtcNow < deadline) await Task.Delay(5); + tag.ReadCount.ShouldBeGreaterThanOrEqualTo(target, $"'{tagName}' should have been read {more} more times"); + } + + private static async Task WaitForEventAsync(List events, Func match) + { + var deadline = DateTime.UtcNow.AddSeconds(5); + while (DateTime.UtcNow < deadline) + { + lock (events) if (events.Any(match)) return; + await Task.Delay(5); + } + throw new TimeoutException("The expected alarm event was never raised."); + } + private static async Task WaitForTagCreation(FakeAbCipTagFactory factory, string tagName) { var deadline = DateTime.UtcNow.AddSeconds(2); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipDriverCodeReviewRegressionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipDriverCodeReviewRegressionTests.cs index f1fabd0f..fed49d46 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipDriverCodeReviewRegressionTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipDriverCodeReviewRegressionTests.cs @@ -152,8 +152,11 @@ public sealed class AbCipDriverCodeReviewRegressionTests }, "drv-1", factory); await drv.InitializeAsync("{}", CancellationToken.None); - // Give the probe loop a moment to actually start spinning. - await Task.Delay(60); + // Wait until the probe loop is actually spinning (#583: a processed count, not a fixed moment). + var deadline = DateTime.UtcNow.AddSeconds(5); + while (!(factory.Tags.TryGetValue("ProbeTag", out var probe) && probe.ReadCount >= 2) && DateTime.UtcNow < deadline) + await Task.Delay(5, TestContext.Current.CancellationToken); + factory.Tags["ProbeTag"].ReadCount.ShouldBeGreaterThanOrEqualTo(2); // Must complete cleanly — no ObjectDisposedException from a loop racing a disposed CTS. await Should.NotThrowAsync(() => drv.ShutdownAsync(CancellationToken.None)); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipHostProbeTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipHostProbeTests.cs index bbb25483..c77b0e8e 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipHostProbeTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipHostProbeTests.cs @@ -100,11 +100,36 @@ public sealed class AbCipHostProbeTests drv.OnHostStatusChanged += (_, e) => transitions.Enqueue(e); await drv.InitializeAsync("{}", CancellationToken.None); - await Task.Delay(300); + // #583: structural, not a fixed wait — no probe loop was ever started, so no transition can follow. + drv.GetDeviceState("ab://10.0.0.5/1,0")!.ProbeTask.ShouldBeNull(); + factory.Tags.ShouldBeEmpty(); transitions.ShouldBeEmpty(); drv.GetHostStatuses().Single().State.ShouldBe(HostState.Unknown); await drv.ShutdownAsync(CancellationToken.None); + await ProbeLoopIsObservableAsync(); + } + + /// + /// Positive control for the "no probe loop" assertions above: the same observation (the device's + /// ProbeTask) on a driver whose probe IS configured is non-null, and that loop does produce a + /// transition — so a null ProbeTask really means "no loop", not "a loop the hook cannot see". + /// + private static async Task ProbeLoopIsObservableAsync() + { + var factory = new FakeAbCipTagFactory { Customise = p => new FakeAbCipTag(p) { Status = 0 } }; + var transitions = new ConcurrentQueue(); + var control = new AbCipDriver(new AbCipDriverOptions + { + Devices = [new AbCipDeviceOptions("ab://10.0.0.5/1,0")], + Probe = new AbCipProbeOptions { Enabled = true, ProbeTagPath = "@raw_cpu_type", Interval = TimeSpan.FromMilliseconds(100) }, + }, "drv-control", factory); + control.OnHostStatusChanged += (_, e) => transitions.Enqueue(e); + await control.InitializeAsync("{}", CancellationToken.None); + (control.GetDeviceState("ab://10.0.0.5/1,0")!.ProbeTask is not null).ShouldBeTrue(); + await WaitForAsync(() => !transitions.IsEmpty, TimeSpan.FromSeconds(2)); + transitions.ShouldNotBeEmpty(); + await control.ShutdownAsync(CancellationToken.None); } /// Verifies that the probe is skipped when ProbeTagPath is null. @@ -118,10 +143,11 @@ public sealed class AbCipHostProbeTests }, "drv-1"); await drv.InitializeAsync("{}", CancellationToken.None); - await Task.Delay(200); + drv.GetDeviceState("ab://10.0.0.5/1,0")!.ProbeTask.ShouldBeNull(); // #583: no loop, so no transition drv.GetHostStatuses().Single().State.ShouldBe(HostState.Unknown); await drv.ShutdownAsync(CancellationToken.None); + await ProbeLoopIsObservableAsync(); } /// Verifies that the probe loops across multiple devices independently. diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipLoggingTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipLoggingTests.cs index 1704614b..515179b2 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipLoggingTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipLoggingTests.cs @@ -65,8 +65,12 @@ public sealed class AbCipLoggingTests await drv.InitializeAsync("{}", CancellationToken.None); - // Give the probe loop a couple of ticks to log. - await Task.Delay(200); + // Wait (bounded) for the entry itself rather than a fixed couple of ticks (#583). + var deadline = DateTime.UtcNow.AddSeconds(5); + while (!logger.Entries.Any(e => e.Message.Contains("probe", StringComparison.OrdinalIgnoreCase) + || (e.Exception?.Message.Contains("simulated probe init failure") ?? false)) + && DateTime.UtcNow < deadline) + await Task.Delay(5, TestContext.Current.CancellationToken); await drv.ShutdownAsync(CancellationToken.None); @@ -186,8 +190,10 @@ public sealed class AbCipLoggingTests /// Test logger that captures all log entries. internal sealed class CapturingLogger : ILogger { - /// Gets the captured log entries. - public List<(LogLevel Level, string Message, Exception? Exception)> Entries { get; } = new(); + private readonly List<(LogLevel Level, string Message, Exception? Exception)> _entries = new(); + + /// Gets a snapshot of the captured log entries (the probe loop logs from another thread). + public List<(LogLevel Level, string Message, Exception? Exception)> Entries { get { lock (_entries) return [.. _entries]; } } /// Begins a scope (stub implementation). /// The type of the scope state. @@ -206,7 +212,7 @@ public sealed class AbCipLoggingTests public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) { - Entries.Add((logLevel, formatter(state, exception), exception)); + lock (_entries) _entries.Add((logLevel, formatter(state, exception), exception)); } private sealed class NullScope : IDisposable diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipSubscriptionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipSubscriptionTests.cs index c4625681..a839e8bb 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipSubscriptionTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipSubscriptionTests.cs @@ -67,7 +67,8 @@ public sealed class AbCipSubscriptionTests drv.OnDataChange += (_, e) => events.Enqueue(e); var handle = await drv.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(100), CancellationToken.None); - await Task.Delay(500); + // Processed-count wait (#583): five polls have read the unchanged value, not "500 ms went by". + await WaitForReadsAsync(factory, "Speed", 5); await drv.UnsubscribeAsync(handle, CancellationToken.None); events.Count.ShouldBe(1); @@ -101,22 +102,30 @@ public sealed class AbCipSubscriptionTests public async Task Unsubscribe_halts_polling() { var (drv, factory) = NewDriver( - new AbCipTagDefinition("Speed", "ab://10.0.0.5/1,0", "Speed", AbCipDataType.DInt)); + new AbCipTagDefinition("Speed", "ab://10.0.0.5/1,0", "Speed", AbCipDataType.DInt), + new AbCipTagDefinition("Temp", "ab://10.0.0.5/1,0", "Temp", AbCipDataType.DInt)); await drv.InitializeAsync("{}", CancellationToken.None); var tagRef = new FakeAbCipTag(new AbCipTagCreateParams("10.0.0.5", 44818, "1,0", "controllogix", "Speed", TimeSpan.FromSeconds(2))) { Value = 1 }; - factory.Customise = _ => tagRef; + factory.Customise = p => p.TagName == "Speed" ? tagRef : new FakeAbCipTag(p) { Value = 1 }; var events = new ConcurrentQueue(); drv.OnDataChange += (_, e) => events.Enqueue(e); var handle = await drv.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(100), CancellationToken.None); - await WaitForAsync(() => events.Count >= 1, TimeSpan.FromSeconds(1)); + var controlHandle = await drv.SubscribeAsync(["Temp"], TimeSpan.FromMilliseconds(100), CancellationToken.None); + await WaitForAsync(() => events.Any(e => e.FullReference == "Speed"), TimeSpan.FromSeconds(1)); await drv.UnsubscribeAsync(handle, CancellationToken.None); - var afterUnsub = events.Count; + var afterUnsub = events.Count(e => e.FullReference == "Speed"); + var readsAfterUnsub = tagRef.ReadCount; tagRef.Value = 999; - await Task.Delay(400); - events.Count.ShouldBe(afterUnsub); + // Positive control (#583): the still-subscribed sibling, on the same interval, polls three more times — + // so Speed's loop, had it survived the unsubscribe, has had three chances to read the change. + await WaitForReadsAsync(factory, "Temp", factory.Tags["Temp"].ReadCount + 3); + + events.Count(e => e.FullReference == "Speed").ShouldBe(afterUnsub); + tagRef.ReadCount.ShouldBe(readsAfterUnsub); + await drv.UnsubscribeAsync(controlHandle, CancellationToken.None); } /// Verifies that polling intervals below 100ms are floored to the minimum. @@ -132,12 +141,17 @@ public sealed class AbCipSubscriptionTests drv.OnDataChange += (_, e) => events.Enqueue(e); var handle = await drv.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(5), CancellationToken.None); - await Task.Delay(300); + // Processed-count wait (#583): three polls, then measure the spacing they actually ran at. + var started = System.Diagnostics.Stopwatch.StartNew(); + await WaitForReadsAsync(factory, "Speed", 3); + var elapsed = started.Elapsed; await drv.UnsubscribeAsync(handle, CancellationToken.None); - // Value is stable → only the initial-data push fires; the 100 ms floor keeps polls sparse enough - // that no extra event is produced against a stable value. + // Value is stable → only the initial-data push fires. events.Count.ShouldBe(1); + // The floor: the initial read is immediate, so the two later ones took at least two 100 ms intervals. + // At the requested 5 ms they would have taken ~10 ms. + elapsed.ShouldBeGreaterThanOrEqualTo(TimeSpan.FromMilliseconds(180)); } /// Verifies that ShutdownAsync cancels all active subscriptions. @@ -154,11 +168,25 @@ public sealed class AbCipSubscriptionTests _ = await drv.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(100), CancellationToken.None); await WaitForAsync(() => events.Count >= 1, TimeSpan.FromSeconds(1)); + var speed = factory.Tags["Speed"]; await drv.ShutdownAsync(CancellationToken.None); var afterShutdown = events.Count; - await Task.Delay(300); + var readsAfterShutdown = speed.ReadCount; + speed.Value = 2; // a change any surviving loop would raise + + // Positive control (#583): a second, independent driver polling on the same interval reads three more + // times — so the shut-down driver's loop, had it survived, has had three chances to raise the change. + var (control, controlFactory) = NewDriver( + new AbCipTagDefinition("Speed", "ab://10.0.0.5/1,0", "Speed", AbCipDataType.DInt)); + await control.InitializeAsync("{}", CancellationToken.None); + controlFactory.Customise = p => new FakeAbCipTag(p) { Value = 1 }; + _ = await control.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(100), CancellationToken.None); + await WaitForReadsAsync(controlFactory, "Speed", 3); + await control.ShutdownAsync(CancellationToken.None); + events.Count.ShouldBe(afterShutdown); + speed.ReadCount.ShouldBe(readsAfterShutdown); } /// Verifies that subscriptions on UDT members use the synthesized full reference. @@ -195,4 +223,12 @@ public sealed class AbCipSubscriptionTests while (!condition() && DateTime.UtcNow < deadline) await Task.Delay(20); } + + /// Waits until the fake runtime for has served at least + /// reads (a processed-count wait), failing — never passing vacuously — if it does not get there. + private static async Task WaitForReadsAsync(FakeAbCipTagFactory factory, string tag, int reads) + { + await WaitForAsync(() => factory.Tags.TryGetValue(tag, out var t) && t.ReadCount >= reads, TimeSpan.FromSeconds(5)); + factory.Tags[tag].ReadCount.ShouldBeGreaterThanOrEqualTo(reads, $"'{tag}' should have been polled {reads} times"); + } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/FakeAbCipTag.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/FakeAbCipTag.cs index 3d8be6b3..740e0985 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/FakeAbCipTag.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/FakeAbCipTag.cs @@ -25,7 +25,8 @@ internal class FakeAbCipTag : IAbCipTagRuntime /// Gets the count of invocations. public int InitializeCount { get; private set; } /// Gets the count of invocations. - public int ReadCount { get; private set; } + public int ReadCount => Volatile.Read(ref _readCount); + private int _readCount; // interlocked: poll loops read it from another thread (#583 processed-count waits) /// Gets the count of invocations. public int WriteCount { get; private set; } /// Gets a value indicating whether the tag has been disposed. @@ -48,7 +49,7 @@ internal class FakeAbCipTag : IAbCipTagRuntime /// The cancellation token for the operation. public virtual Task ReadAsync(CancellationToken cancellationToken) { - ReadCount++; + Interlocked.Increment(ref _readCount); if (ThrowOnRead) throw Exception ?? new InvalidOperationException("fake read failure"); return Task.CompletedTask; } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/LibplctagOrphanReleaseTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/LibplctagOrphanReleaseTests.cs index 228a602c..4adc8de5 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/LibplctagOrphanReleaseTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/LibplctagOrphanReleaseTests.cs @@ -18,7 +18,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests; [CollectionDefinition(Name, DisableParallelization = true)] public sealed class LibplctagOrphanReleaseCollection { - /// The release counters and the rooting seam are process-wide: nothing else may run alongside. + /// The release counters are process-wide: nothing else may run alongside. public const string Name = "libplctag orphan release"; } @@ -90,23 +90,29 @@ public sealed class LibplctagOrphanReleaseTests [Fact] public async Task Without_a_readable_handle_the_failed_tag_is_rooted_instead() { + // Gitea #581: the "unreadable handle" decision is passed in, not flipped through a process-wide switch. var rooted = LibplctagOrphanRelease.RootedCount; var released = LibplctagOrphanRelease.Released; - var rt = new LibplctagTagRuntime(Refused()); - LibplctagOrphanRelease.ForceRootingForTest = true; + var tag = LibplctagTagRuntime.BuildTag(Refused()); try { - await Should.ThrowAsync(() => rt.InitializeAsync(CancellationToken.None)); + await Should.ThrowAsync(() => tag.InitializeAsync(CancellationToken.None)); + var handle = LibplctagOrphanRelease.HandleOf(tag); + handle.ShouldBeGreaterThan(0, "plc_tag_create_ex returned a handle before the failure"); + + LibplctagOrphanRelease.ReleaseIfOrphaned(tag, canReadHandles: false).ShouldBeNull(); + LibplctagOrphanRelease.RootedCount.ShouldBe(rooted + 1); LibplctagOrphanRelease.Released.ShouldBe(released); + NativeTagExists(handle).ShouldBeTrue("the rooting fallback destroys nothing"); + LibplctagOrphanRelease.HandleOf(tag).ShouldBe(handle); } finally { - LibplctagOrphanRelease.ForceRootingForTest = false; // Clean up the native tag this test deliberately leaked (its callback was unregistered by the timeout). - var handle = rt.NativeHandle; + var handle = LibplctagOrphanRelease.HandleOf(tag); if (handle > 0) plctag.plc_tag_destroy(handle); - rt.Dispose(); + tag.Dispose(); } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/NativeAbCipLeaseReleaseTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/NativeAbCipLeaseReleaseTests.cs index 1203432a..6fe1a56f 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/NativeAbCipLeaseReleaseTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/NativeAbCipLeaseReleaseTests.cs @@ -72,7 +72,9 @@ public sealed class NativeAbCipLeaseReleaseTests // the first one's Forward Close is in flight (review finding 22) — the pool holds it behind the close. var second = (NativeAbCipTagRuntime)factory.Create(Params(target.Port)); var init = second.InitializeAsync(Ct); - await Task.Delay(300, Ct); + // Positive control (#583): the second open has reached the pool's predecessor wait and parked there, so + // the missing Forward Open below is that wait holding it — not the open simply not having got that far. + await FakeCipTarget.Until(() => pool.Diagnostics.PredecessorWaits == 1, Guard, "the re-create to park behind the close"); target.ForwardOpens.ShouldBe(1); init.IsCompleted.ShouldBeFalse(); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyCapabilityTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyCapabilityTests.cs index 99977356..3d4da6fb 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyCapabilityTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyCapabilityTests.cs @@ -71,11 +71,17 @@ public sealed class AbLegacyCapabilityTests { var tagRef = new FakeAbLegacyTag( new AbLegacyTagCreateParams("10.0.0.5", 44818, "1,0", "slc500", "N7:0", TimeSpan.FromSeconds(2))) { Value = 1 }; - var factory = new FakeAbLegacyTagFactory { Customise = _ => tagRef }; + var control = new FakeAbLegacyTag( + new AbLegacyTagCreateParams("10.0.0.5", 44818, "1,0", "slc500", "N7:1", TimeSpan.FromSeconds(2))) { Value = 1 }; + var factory = new FakeAbLegacyTagFactory { Customise = p => p.TagName == "N7:1" ? control : tagRef }; var drv = new AbLegacyDriver(new AbLegacyDriverOptions { Devices = [new AbLegacyDeviceOptions("ab://10.0.0.5/1,0")], - RawTags = AbLegacyRawTags.Entries([new AbLegacyTagDefinition("X", "ab://10.0.0.5/1,0", "N7:0", AbLegacyDataType.Int)]), + RawTags = AbLegacyRawTags.Entries( + [ + new AbLegacyTagDefinition("X", "ab://10.0.0.5/1,0", "N7:0", AbLegacyDataType.Int), + new AbLegacyTagDefinition("Y", "ab://10.0.0.5/1,0", "N7:1", AbLegacyDataType.Int), + ]), Probe = new AbLegacyProbeOptions { Enabled = false }, }, "drv-1", factory); await drv.InitializeAsync("{}", CancellationToken.None); @@ -84,13 +90,22 @@ public sealed class AbLegacyCapabilityTests drv.OnDataChange += (_, e) => events.Enqueue(e); var handle = await drv.SubscribeAsync(["X"], TimeSpan.FromMilliseconds(100), CancellationToken.None); - await WaitForAsync(() => events.Count >= 1, TimeSpan.FromSeconds(1)); + var controlHandle = await drv.SubscribeAsync(["Y"], TimeSpan.FromMilliseconds(100), CancellationToken.None); + await WaitForAsync(() => events.Count(e => e.FullReference == "X") >= 1, TimeSpan.FromSeconds(1)); await drv.UnsubscribeAsync(handle, CancellationToken.None); - var afterUnsub = events.Count; + var afterUnsub = events.Count(e => e.FullReference == "X"); + var readsAfterUnsub = tagRef.ReadCount; tagRef.Value = 999; - await Task.Delay(300, TestContext.Current.CancellationToken); - events.Count.ShouldBe(afterUnsub); + // Positive control (#583): the still-subscribed sibling, on the same interval, polls three more times — + // so X's loop, had it survived, has had three chances to read the change and raise it. + var controlReads = control.ReadCount; + await WaitForAsync(() => control.ReadCount >= controlReads + 3, TimeSpan.FromSeconds(3)); + control.ReadCount.ShouldBeGreaterThanOrEqualTo(controlReads + 3); + + events.Count(e => e.FullReference == "X").ShouldBe(afterUnsub); + tagRef.ReadCount.ShouldBe(readsAfterUnsub); + await drv.UnsubscribeAsync(controlHandle, CancellationToken.None); } // ---- IHostConnectivityProbe ---- @@ -171,10 +186,30 @@ public sealed class AbLegacyCapabilityTests Probe = new AbLegacyProbeOptions { Enabled = true, ProbeAddress = null }, }, "drv-1"); await drv.InitializeAsync("{}", CancellationToken.None); - await Task.Delay(200, TestContext.Current.CancellationToken); + // #583: structural, not a fixed wait — no probe loop was ever started, so no transition can follow. + drv.GetDeviceState("ab://10.0.0.5/1,0")!.ProbeCts.ShouldBeNull(); drv.GetHostStatuses().Single().State.ShouldBe(HostState.Unknown); await drv.ShutdownAsync(CancellationToken.None); + + // Positive control: the same observation on a driver whose probe IS configured sees the loop, and that loop + // produces a transition — so a null ProbeCts really means "no loop". + var transitions = new ConcurrentQueue(); + var control = new AbLegacyDriver(new AbLegacyDriverOptions + { + Devices = [new AbLegacyDeviceOptions("ab://10.0.0.5/1,0")], + Probe = new AbLegacyProbeOptions + { + Enabled = true, Interval = TimeSpan.FromMilliseconds(100), + Timeout = TimeSpan.FromMilliseconds(50), ProbeAddress = "S:0", + }, + }, "drv-control", new FakeAbLegacyTagFactory()); + control.OnHostStatusChanged += (_, e) => transitions.Enqueue(e); + await control.InitializeAsync("{}", CancellationToken.None); + control.GetDeviceState("ab://10.0.0.5/1,0")!.ProbeCts.ShouldNotBeNull(); + await WaitForAsync(() => !transitions.IsEmpty, TimeSpan.FromSeconds(2)); + transitions.ShouldNotBeEmpty(); + await control.ShutdownAsync(CancellationToken.None); } // ---- IPerCallHostResolver ---- diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/FakeAbLegacyTag.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/FakeAbLegacyTag.cs index bd177379..ba0a4618 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/FakeAbLegacyTag.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/FakeAbLegacyTag.cs @@ -43,7 +43,8 @@ internal class FakeAbLegacyTag : IAbLegacyTagRuntime public int InitializeCount { get; private set; } /// Gets the count of read calls. - public int ReadCount { get; private set; } + public int ReadCount => Volatile.Read(ref _readCount); + private int _readCount; // interlocked: poll loops read it from another thread (#583 processed-count waits) /// Gets the count of write calls. public int WriteCount { get; private set; } @@ -70,7 +71,7 @@ internal class FakeAbLegacyTag : IAbLegacyTagRuntime /// A task representing the asynchronous operation. public virtual Task ReadAsync(CancellationToken ct) { - ReadCount++; + Interlocked.Increment(ref _readCount); if (ThrowOnRead) throw Exception ?? new InvalidOperationException(); return Task.CompletedTask; }