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
new file mode 100644
index 00000000..a5443512
--- /dev/null
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/CipConnection.cs
@@ -0,0 +1,287 @@
+using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Eip;
+namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip;
+
+///
+/// One device's CIP channel: one EtherNet/IP session and, when , one
+/// class-3 connection. Opens lazily on first use, reconnects (through the )
+/// after the session faults, and keeps an idle connection alive.
+///
+///
+/// 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.
+/// 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
+/// Open), and the exchange by 2 × Timeout (waiting behind one in-flight exchange, then its own).
+/// Exceeding either surfaces as ().
+/// Connection identity. Every Forward Open attempt uses a fresh connection serial and T→O id,
+/// so a reopen is never mistaken by the target for a duplicate of a connection it has not yet expired.
+///
+public sealed class CipConnection : ICipChannel, IAsyncDisposable
+{
+ /// Forward Open frames sent by every connection in the process (fallback included). Gate diagnostics.
+ internal static int ForwardOpenCount;
+
+ /// Channel sends by every connection in the process (keep-alives included). Gate diagnostics.
+ internal static int RequestsSent;
+
+ /// 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];
+
+ /// Largest unconnected message a target is guaranteed to accept.
+ private const int UnconnectedMessageLimit = 504;
+
+ private sealed record Open(EncapsulationClient Client, ForwardOpenReply? Conn, ushort ConnectionSerial, int NegotiatedSize);
+
+ private readonly DeviceKey _key;
+ private readonly ConnectionOptions _o;
+ private readonly ForwardOpenThrottle _throttle;
+ private readonly TimeProvider _time;
+ private readonly Task _predecessorClosed;
+ private readonly byte[] _route;
+
+ // Never disposed: callers may still be queued on it when DisposeAsync runs, and Release on a
+ // disposed SemaphoreSlim throws. It owns no unmanaged handle unless AvailableWaitHandle is touched.
+ private readonly SemaphoreSlim _state = new(1, 1);
+ private readonly ITimer? _keepAlive;
+ private Open? _open; // written under _state; read lock-free via Volatile
+ private int _disposed; // checked under _state before any open; read lock-free elsewhere
+ private int _disposeStarted;
+ private long _lastSendTimestamp;
+ private int _keepAlivesSent;
+ private int _connectionSerial = Random.Shared.Next(1, ushort.MaxValue);
+
+ /// The device; is the route (ignored for Direct).
+ /// Routing, sizes, timeouts and keep-alive settings.
+ /// The device's process-lifetime Forward Open throttle.
+ /// 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.
+ public CipConnection(DeviceKey key, ConnectionOptions options, ForwardOpenThrottle throttle, TimeProvider time,
+ Task? predecessorClosed = null)
+ {
+ _key = key ?? throw new ArgumentNullException(nameof(key));
+ _o = options ?? throw new ArgumentNullException(nameof(options));
+ _throttle = throttle ?? throw new ArgumentNullException(nameof(throttle));
+ _time = time ?? throw new ArgumentNullException(nameof(time));
+ _predecessorClosed = predecessorClosed ?? Task.CompletedTask;
+ // Direct ignores the path entirely (B3), so it is not even parsed.
+ _route = options.Routing == CipRouting.Direct ? [] : CipPath.EncodeRoute(key.CipPath);
+ _lastSendTimestamp = time.GetTimestamp();
+ if (options.Routing == CipRouting.Connected)
+ _keepAlive = time.CreateTimer(static s => ((CipConnection)s!).OnKeepAliveTimer(), this,
+ options.KeepAliveInterval, Timeout.InfiniteTimeSpan);
+ }
+
+ /// The options this connection was built with.
+ internal ConnectionOptions Options => _o;
+
+ /// Test hook: keep-alives started (counted synchronously when the timer decides to send).
+ internal int KeepAlivesSent => Volatile.Read(ref _keepAlivesSent);
+
+ private TimeSpan OpenBudget => _o.MinForwardOpenInterval + 4 * _o.Timeout;
+ private TimeSpan ExchangeBudget => 2 * _o.Timeout;
+
+ public int MaxRequestBytes => _o.Routing switch
+ {
+ CipRouting.Connected => (Volatile.Read(ref _open)?.NegotiatedSize ?? Math.Min(_o.ConnectionSize, UnconnectedMessageLimit)) - 2,
+ CipRouting.UnconnectedSend => UnconnectedMessageLimit - (12 + _route.Length),
+ _ => UnconnectedMessageLimit,
+ };
+
+ public async Task> SendAsync(ReadOnlyMemory cipRequest, CancellationToken cancellationToken)
+ {
+ ThrowIfDisposed();
+ var open = await EnsureOpenAsync(cancellationToken).ConfigureAwait(false);
+ return await ExchangeAsync(open, cipRequest, cancellationToken).ConfigureAwait(false);
+ }
+
+ /// Returns the live (client, connection) pair, opening one if needed. The pair is returned
+ /// rather than re-read from fields afterwards, so a concurrent dispose can never hand the caller a
+ /// null (finding 16).
+ private async Task EnsureOpenAsync(CancellationToken ct)
+ {
+ if (Volatile.Read(ref _open) is { Client.IsConnected: true } live) return live;
+
+ using var deadline = new CancellationTokenSource(OpenBudget, _time);
+ using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, deadline.Token);
+ try
+ {
+ await _state.WaitAsync(linked.Token).ConfigureAwait(false);
+ try
+ {
+ ThrowIfDisposed();
+ var current = _open;
+ if (current is { Client.IsConnected: true }) return current;
+ if (current is not null)
+ {
+ Volatile.Write(ref _open, null);
+ await current.Client.DisposeAsync().ConfigureAwait(false); // faulted: closes without UnRegister
+ }
+ await _predecessorClosed.WaitAsync(linked.Token).ConfigureAwait(false);
+ var opened = await _throttle.RunAsync(OpenAsync, linked.Token).ConfigureAwait(false);
+ Volatile.Write(ref _open, opened);
+ return opened;
+ }
+ finally { _state.Release(); }
+ }
+ catch (OperationCanceledException) when (!ct.IsCancellationRequested && deadline.IsCancellationRequested)
+ {
+ throw new AbTransportException(AbStatus.Timeout,
+ $"Opening {_key.Host}:{_key.Port} did not complete within {OpenBudget.TotalMilliseconds:F0} ms.");
+ }
+ }
+
+ private async Task OpenAsync(CancellationToken ct)
+ {
+ ThrowIfDisposed(); // disposal may have begun while this attempt waited out the throttle
+ var client = await EncapsulationClient.ConnectAsync(_key.Host, _key.Port, _o.Timeout, ct).ConfigureAwait(false);
+ if (_o.Routing != CipRouting.Connected) return new Open(client, null, 0, UnconnectedMessageLimit);
+ try
+ {
+ var serial = NextConnectionSerial();
+ var toId = (uint)Random.Shared.NextInt64(1, uint.MaxValue);
+ var large = _o.ConnectionSize > 511;
+ var size = _o.ConnectionSize;
+ var reply = await ForwardOpenAsync(client, large, size, serial, toId, ct).ConfigureAwait(false);
+ if (large && ConnectionManagerServices.ShouldFallBackToRegularOpen(reply))
+ {
+ // Pre-v20 Logix / small-buffer targets: retry once, inside this same throttled attempt.
+ size = Math.Min(_o.ConnectionSize, UnconnectedMessageLimit);
+ reply = await ForwardOpenAsync(client, false, size, serial, toId, ct).ConfigureAwait(false);
+ }
+ return new Open(client, ConnectionManagerServices.ParseForwardOpenReply(reply), serial, size);
+ }
+ catch
+ {
+ await client.DisposeAsync().ConfigureAwait(false);
+ throw;
+ }
+ }
+
+ private async Task ForwardOpenAsync(EncapsulationClient client, bool large, int size, ushort serial,
+ uint toId, CancellationToken ct)
+ {
+ 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);
+ return CipReply.Parse(await client.SendRRDataAsync(frame, _o.Timeout, ct).ConfigureAwait(false));
+ }
+
+ private ushort NextConnectionSerial()
+ {
+ var s = (ushort)Interlocked.Increment(ref _connectionSerial);
+ return s == 0 ? (ushort)Interlocked.Increment(ref _connectionSerial) : s;
+ }
+
+ /// One request over an established pair, bounded by .
+ private async Task> ExchangeAsync(Open open, ReadOnlyMemory cip, CancellationToken ct)
+ {
+ Interlocked.Increment(ref RequestsSent);
+ Interlocked.Exchange(ref _lastSendTimestamp, _time.GetTimestamp());
+ using var deadline = new CancellationTokenSource(ExchangeBudget, _time);
+ using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, deadline.Token);
+ try
+ {
+ return _o.Routing switch
+ {
+ CipRouting.Connected => await open.Client.SendUnitDataAsync(open.Conn!.OTConnectionId, cip, _o.Timeout, linked.Token).ConfigureAwait(false),
+ 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),
+ };
+ }
+ catch (OperationCanceledException) when (!ct.IsCancellationRequested && deadline.IsCancellationRequested)
+ {
+ throw new AbTransportException(AbStatus.Timeout,
+ $"Request to {_key.Host}:{_key.Port} did not complete within {ExchangeBudget.TotalMilliseconds:F0} ms.");
+ }
+ }
+
+ /// One-shot timer, re-armed after every tick so the longest idle gap is one
+ /// (half the connection expiry), not two.
+ private void OnKeepAliveTimer()
+ {
+ if (Volatile.Read(ref _disposed) != 0) return;
+ var interval = _o.KeepAliveInterval;
+ var open = Volatile.Read(ref _open);
+ if (open is not { Client.IsConnected: true })
+ {
+ // Nothing to keep alive, and a keep-alive must never be what reopens a connection.
+ Rearm(interval);
+ return;
+ }
+ var idle = _time.GetElapsedTime(Interlocked.Read(ref _lastSendTimestamp));
+ if (idle < interval)
+ {
+ Rearm(interval - idle);
+ return;
+ }
+ Interlocked.Increment(ref _keepAlivesSent);
+ _ = KeepAliveAsync(open);
+ }
+
+ private async Task KeepAliveAsync(Open open)
+ {
+ try { await ExchangeAsync(open, KeepAliveRequest, CancellationToken.None).ConfigureAwait(false); }
+ catch (Exception)
+ {
+ // A timeout or I/O failure has already faulted the session; the next real send reconnects
+ // through the throttle. Nothing to report here: the keep-alive has no caller.
+ }
+ finally { Rearm(_o.KeepAliveInterval); }
+ }
+
+ private void Rearm(TimeSpan due)
+ {
+ if (Volatile.Read(ref _disposed) != 0) return;
+ try { _keepAlive?.Change(due, Timeout.InfiniteTimeSpan); }
+ catch (ObjectDisposedException) { /* disposed concurrently */ }
+ }
+
+ private void ThrowIfDisposed()
+ {
+ if (Volatile.Read(ref _disposed) != 0)
+ throw new AbTransportException(AbStatus.Abort, "Connection disposed.");
+ }
+
+ /// Stops the keep-alive, sends Forward Close (Connected only) and closes the session, whose
+ /// own dispose sends UnRegisterSession. Later sends throw .
+ public async ValueTask DisposeAsync()
+ {
+ if (Interlocked.Exchange(ref _disposeStarted, 1) != 0) return;
+
+ // Flag first: every caller still queued on _state sees it the moment it gets the lock and throws
+ // Abort without opening, so this waits behind at most the one open already in progress (bounded
+ // by OpenBudget), never behind a queue of them.
+ Volatile.Write(ref _disposed, 1);
+ await _state.WaitAsync().ConfigureAwait(false);
+ Open? open;
+ try
+ {
+ open = _open;
+ Volatile.Write(ref _open, null);
+ }
+ finally { _state.Release(); }
+
+ if (_keepAlive is not null) await _keepAlive.DisposeAsync().ConfigureAwait(false);
+ if (open is null) return;
+ if (open.Conn is not null && open.Client.IsConnected)
+ {
+ try
+ {
+ var budget = TimeSpan.FromSeconds(1);
+ using var cts = new CancellationTokenSource(budget);
+ await open.Client.SendRRDataAsync(
+ ConnectionManagerServices.ForwardClose(open.ConnectionSerial, _o.VendorId, _o.OriginatorSerial, _route),
+ budget, cts.Token).ConfigureAwait(false);
+ }
+ catch (Exception ex) when (ex is AbTransportException or OperationCanceledException)
+ {
+ // The target expires the connection after RPI × multiplier anyway.
+ }
+ }
+ await open.Client.DisposeAsync().ConfigureAwait(false);
+ }
+}
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ConnectionOptions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ConnectionOptions.cs
new file mode 100644
index 00000000..a8f95831
--- /dev/null
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ConnectionOptions.cs
@@ -0,0 +1,100 @@
+namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip;
+
+/// How CIP requests reach the target's message router (review finding B3; the captures decide
+/// each family's mode).
+public enum CipRouting
+{
+ /// Forward Open a class-3 connection, then SendUnitData. Logix and Micro800 (empty route).
+ Connected,
+
+ /// SendRRData wrapping each request in Unconnected Send (0x52) carrying the route.
+ UnconnectedSend,
+
+ /// SendRRData carrying the request bare; any configured CIP path is ignored (PCCC families).
+ Direct,
+}
+
+/// Per-device connection settings. Validated on assignment so an unusable value fails at the
+/// configuration seam rather than on the wire.
+public sealed record ConnectionOptions
+{
+ /// Press safety floor: no setting may let Forward Opens to one device start closer together.
+ public static readonly TimeSpan MinimumForwardOpenInterval = TimeSpan.FromSeconds(1);
+
+ /// Originator serial shared by every connection this process opens.
+ private static readonly uint ProcessOriginatorSerial = (uint)Random.Shared.Next(1, int.MaxValue);
+
+ private readonly TimeSpan _timeout = TimeSpan.FromSeconds(5);
+ private readonly int _connectionSize = 4002;
+ private readonly TimeSpan _minForwardOpenInterval = MinimumForwardOpenInterval;
+ private readonly uint _rpiMicros = 2_000_000;
+ private readonly byte _timeoutMultiplier = 1;
+ private readonly TimeSpan? _keepAliveInterval;
+
+ public CipRouting Routing { get; init; } = CipRouting.Connected;
+
+ /// Pack queued requests into Multiple Service Packets (the scheduler's switch).
+ public bool AllowPacking { get; init; }
+
+ /// Bound on each single exchange (TCP connect, RegisterSession, Forward Open, request).
+ /// Must be finite and positive (ruling R9-2): an infinite timeout would let an abandoned exchange
+ /// hold the session's single-flight gate forever.
+ public TimeSpan Timeout
+ {
+ get => _timeout;
+ init => _timeout = value > TimeSpan.Zero && value != System.Threading.Timeout.InfiniteTimeSpan
+ ? value
+ : throw new ArgumentOutOfRangeException(nameof(Timeout), value, "Timeout must be finite and positive.");
+ }
+
+ /// Requested connection size in bytes. Above 511 the open is a Large Forward Open (0x5B).
+ public int ConnectionSize
+ {
+ get => _connectionSize;
+ init => _connectionSize = value is >= 1 and <= ushort.MaxValue
+ ? value
+ : throw new ArgumentOutOfRangeException(nameof(ConnectionSize), value, "Connection size must be 1..65535.");
+ }
+
+ /// Minimum time between Forward Open attempt starts for one device (≥ 1 s, press safety).
+ public TimeSpan MinForwardOpenInterval
+ {
+ get => _minForwardOpenInterval;
+ init => _minForwardOpenInterval = value >= MinimumForwardOpenInterval && value != System.Threading.Timeout.InfiniteTimeSpan
+ ? value
+ : throw new ArgumentOutOfRangeException(nameof(MinForwardOpenInterval), value, "Must be finite and at least 1 s (press safety).");
+ }
+
+ /// Requested packet interval for both directions, in microseconds.
+ public uint RpiMicros
+ {
+ get => _rpiMicros;
+ init => _rpiMicros = value > 0 ? value : throw new ArgumentOutOfRangeException(nameof(RpiMicros), value, "RPI must be positive.");
+ }
+
+ /// Encoded connection timeout multiplier: the connection expires after RPI × (4 << value).
+ /// The default 1 means ×8, so 2 s × 8 = 16 s.
+ public byte TimeoutMultiplier
+ {
+ get => _timeoutMultiplier;
+ init => _timeoutMultiplier = value <= 7 ? value : throw new ArgumentOutOfRangeException(nameof(TimeoutMultiplier), value, "Multiplier must be 0..7.");
+ }
+
+ /// How long an idle connected session waits before the target would drop it.
+ public TimeSpan ConnectionExpiry => TimeSpan.FromTicks(RpiMicros * 10L * (4L << TimeoutMultiplier));
+
+ /// Idle time after which a keep-alive is sent (Connected only). Defaults to half the
+ /// connection expiry: RPI × 8 / 2 = 8 s.
+ public TimeSpan KeepAliveInterval
+ {
+ get => _keepAliveInterval ?? ConnectionExpiry / 2;
+ init => _keepAliveInterval = value > TimeSpan.Zero && value != System.Threading.Timeout.InfiniteTimeSpan
+ ? value
+ : throw new ArgumentOutOfRangeException(nameof(KeepAliveInterval), value, "Keep-alive interval must be finite and positive.");
+ }
+
+ public ushort VendorId { get; init; } = 0x1337;
+
+ /// Random per process; together with the vendor id and connection serial it names a connection.
+ public uint OriginatorSerial { get; init; } = ProcessOriginatorSerial;
+}
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/DeviceKey.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/DeviceKey.cs
new file mode 100644
index 00000000..f0498555
--- /dev/null
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/DeviceKey.cs
@@ -0,0 +1,22 @@
+namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip;
+
+///
+/// Identity of one physical CIP connection: exactly one exists per key
+/// (press safety). and compare case-insensitively;
+/// and compare exactly.
+///
+public sealed record DeviceKey(string Host, int Port, string CipPath, string PlcAttribute)
+{
+ public bool Equals(DeviceKey? other) =>
+ other is not null
+ && string.Equals(Host, other.Host, StringComparison.OrdinalIgnoreCase)
+ && Port == other.Port
+ && string.Equals(CipPath, other.CipPath, StringComparison.Ordinal)
+ && string.Equals(PlcAttribute, other.PlcAttribute, StringComparison.OrdinalIgnoreCase);
+
+ public override int GetHashCode() => HashCode.Combine(
+ StringComparer.OrdinalIgnoreCase.GetHashCode(Host),
+ Port,
+ StringComparer.Ordinal.GetHashCode(CipPath),
+ StringComparer.OrdinalIgnoreCase.GetHashCode(PlcAttribute));
+}
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
new file mode 100644
index 00000000..f25b24a5
--- /dev/null
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ForwardOpenThrottle.cs
@@ -0,0 +1,55 @@
+namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip;
+
+/// Press safety: at most one Forward Open attempt in flight per device and a minimum
+/// interval between attempt starts. A misconfigured client once drove ~15 Forward Opens/s at a
+/// production press (ignitionoee vite-press-oee-bind.md, 2026-09-23); this is what prevents it.
+/// One instance per , kept by for the
+/// process lifetime so a reconnect after the last lease was released is still limited. Its semaphore is
+/// therefore never disposed. The interval is measured from the previous attempt's start, whatever
+/// that attempt's outcome.
+public sealed class ForwardOpenThrottle
+{
+ private readonly SemaphoreSlim _inFlight = new(1, 1);
+ private readonly TimeSpan _minInterval;
+ private readonly TimeProvider _time;
+ private DateTimeOffset? _lastStart; // only touched while holding _inFlight
+
+ public ForwardOpenThrottle(TimeSpan minInterval, TimeProvider time)
+ {
+ if (minInterval < TimeSpan.Zero || minInterval == Timeout.InfiniteTimeSpan)
+ throw new ArgumentOutOfRangeException(nameof(minInterval), minInterval, "Interval must be finite and non-negative.");
+ _minInterval = minInterval;
+ _time = time ?? throw new ArgumentNullException(nameof(time));
+ }
+
+ /// Run one open attempt once the throttle allows it. Throws
+ /// () when
+ /// fires before the attempt could start; the attempt itself sees too.
+ public async Task RunAsync(Func> open, CancellationToken ct)
+ {
+ ArgumentNullException.ThrowIfNull(open);
+ try { await _inFlight.WaitAsync(ct).ConfigureAwait(false); }
+ catch (OperationCanceledException) when (ct.IsCancellationRequested)
+ { throw Throttled("another attempt is in flight"); }
+
+ try
+ {
+ if (_lastStart is { } last)
+ {
+ var wait = last + _minInterval - _time.GetUtcNow();
+ if (wait > TimeSpan.Zero)
+ {
+ try { await Task.Delay(wait, _time, ct).ConfigureAwait(false); }
+ catch (OperationCanceledException) when (ct.IsCancellationRequested)
+ { throw Throttled("the minimum interval had not elapsed"); }
+ }
+ }
+ _lastStart = _time.GetUtcNow();
+ return await open(ct).ConfigureAwait(false);
+ }
+ finally { _inFlight.Release(); }
+ }
+
+ private static AbTransportException Throttled(string why) =>
+ new(AbStatus.BadConnection, $"Forward Open throttled; caller deadline reached before {why}.");
+}
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
new file mode 100644
index 00000000..b254f80c
--- /dev/null
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/DeviceSessionPool.cs
@@ -0,0 +1,129 @@
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip;
+namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport;
+
+///
+/// Process-wide owner of device connections: exactly one and one
+/// per (review B1), handed out as ref-counted
+/// s. The last lease's dispose closes the connection; a later acquire builds
+/// fresh instances, which do not open until the previous connection has finished closing.
+///
+/// Forward Open throttles are kept per key for the process lifetime and never pruned, so a
+/// reconnect after the last release is still rate-limited against the previous attempt.
+public sealed class DeviceSessionPool
+{
+ /// The process-wide pool the drivers use. Tests build their own.
+ public static DeviceSessionPool Shared { get; } = new();
+
+ internal sealed class Entry(CipConnection connection, RequestScheduler scheduler, ConnectionOptions options)
+ {
+ public CipConnection Connection { get; } = connection;
+ public RequestScheduler Scheduler { get; } = scheduler;
+ public ConnectionOptions Options { get; } = options;
+ public int RefCount; // under the pool lock
+ public bool MismatchLogged; // under the pool lock
+ }
+
+ private readonly object _lock = new();
+ private readonly Dictionary _entries = new();
+ private readonly Dictionary _throttles = new();
+ private readonly Dictionary _closing = new();
+ private readonly TimeProvider _time;
+ private readonly ILogger _logger;
+
+ public DeviceSessionPool(TimeProvider? time = null, ILogger? logger = null)
+ {
+ _time = time ?? TimeProvider.System;
+ _logger = logger ?? NullLogger.Instance;
+ }
+
+ /// Test hook: keys with a live entry.
+ internal int EntryCount { get { lock (_lock) return _entries.Count; } }
+
+ /// Lease the shared connection for , creating it on first use. Performs
+ /// no I/O: the connection opens on its first request. When the key already has an entry, that entry's
+ /// options win; a differing is logged once per entry.
+ public ConnectionLease Acquire(DeviceKey key, ConnectionOptions options)
+ {
+ ArgumentNullException.ThrowIfNull(key);
+ ArgumentNullException.ThrowIfNull(options);
+ lock (_lock)
+ {
+ if (!_entries.TryGetValue(key, out var entry))
+ {
+ if (!_throttles.TryGetValue(key, out var throttle))
+ _throttles[key] = throttle = new ForwardOpenThrottle(options.MinForwardOpenInterval, _time);
+ var predecessor = _closing.GetValueOrDefault(key);
+ var connection = new CipConnection(key, options, throttle, _time, predecessor);
+ entry = new Entry(connection, new RequestScheduler(connection, options.AllowPacking), options);
+ _entries[key] = entry;
+ }
+ else if (!entry.Options.Equals(options) && !entry.MismatchLogged)
+ {
+ entry.MismatchLogged = true;
+ _logger.LogWarning(
+ "Device {Host}:{Port} path '{CipPath}' is already connected with different connection options; " +
+ "the existing options ({Existing}) stay in force and the requested ({Requested}) are ignored.",
+ key.Host, key.Port, key.CipPath, entry.Options, options);
+ }
+ entry.RefCount++;
+ return new ConnectionLease(this, key, entry);
+ }
+ }
+
+ internal async ValueTask ReleaseAsync(DeviceKey key, Entry entry)
+ {
+ TaskCompletionSource closed;
+ lock (_lock)
+ {
+ if (--entry.RefCount > 0) return;
+ _entries.Remove(key);
+ closed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ _closing[key] = closed.Task;
+ }
+
+ // Outside the lock: scheduler first (fails its waiters, stops sending), then Forward Close +
+ // UnRegisterSession via the connection.
+ try
+ {
+ await entry.Scheduler.DisposeAsync().ConfigureAwait(false);
+ await entry.Connection.DisposeAsync().ConfigureAwait(false);
+ }
+ finally
+ {
+ lock (_lock)
+ {
+ if (_closing.TryGetValue(key, out var t) && t == closed.Task) _closing.Remove(key);
+ }
+ closed.TrySetResult(); // never faults: a successor only needs to know the close is over
+ }
+ }
+}
+
+/// A reference on a device's shared connection. Dispose exactly once (further disposes are no-ops).
+public sealed class ConnectionLease : IAsyncDisposable
+{
+ private readonly DeviceSessionPool _pool;
+ private readonly DeviceKey _key;
+ private readonly DeviceSessionPool.Entry _entry;
+ private int _disposed;
+
+ internal ConnectionLease(DeviceSessionPool pool, DeviceKey key, DeviceSessionPool.Entry entry)
+ {
+ _pool = pool;
+ _key = key;
+ _entry = entry;
+ }
+
+ public DeviceKey Key => _key;
+
+ /// The device's channel. Prefer : it serializes and packs requests.
+ public ICipChannel Channel => _entry.Connection;
+
+ /// The device's single request scheduler, shared by every lease on the key.
+ public RequestScheduler Scheduler => _entry.Scheduler;
+
+ public ValueTask DisposeAsync() =>
+ Interlocked.Exchange(ref _disposed, 1) == 0 ? _pool.ReleaseAsync(_key, _entry) : ValueTask.CompletedTask;
+}
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.csproj b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.csproj
index 2f48e07a..c15cdd13 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.csproj
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.csproj
@@ -15,5 +15,8 @@
+
+
+
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
new file mode 100644
index 00000000..84c25f02
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/CipConnectionTests.cs
@@ -0,0 +1,385 @@
+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.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.
+[CollectionDefinition(Name, DisableParallelization = true)]
+public sealed class CipConnectionStatics
+{
+ public const string Name = "CipConnection statics";
+}
+
+/// against . Wire I/O is real loopback;
+/// the throttle, deadlines and keep-alive run on .
+[Collection(CipConnectionStatics.Name)]
+public sealed class CipConnectionTests
+{
+ private static readonly TimeSpan Guard = TimeSpan.FromSeconds(10);
+ private static readonly byte[] ReadTag = [0x4C, 0x02, 0x91, 0x03, 0x54, 0x61, 0x67, 0x00, 0x01, 0x00];
+ private static CancellationToken TestCt => TestContext.Current.CancellationToken;
+
+ private static CipConnection Create(FakeCipTarget target, ConnectionOptions o, ManualTimeProvider time, string path = "1,0") =>
+ new(new DeviceKey("127.0.0.1", target.Port, path, "ControlLogix"), o,
+ new ForwardOpenThrottle(o.MinForwardOpenInterval, time), time);
+
+ private static ConnectionOptions Options(CipRouting routing = CipRouting.Connected, int size = 4002, double timeoutSeconds = 2) =>
+ new() { Routing = routing, ConnectionSize = size, Timeout = TimeSpan.FromSeconds(timeoutSeconds) };
+
+ private static ushort RegularOTParams(byte[] fo) => BinaryPrimitives.ReadUInt16LittleEndian(fo.AsSpan(32));
+
+ // ---- options ---------------------------------------------------------------------------------
+
+ [Fact]
+ public void Defaults_match_the_press_safety_values()
+ {
+ var o = new ConnectionOptions();
+ o.MinForwardOpenInterval.ShouldBe(TimeSpan.FromSeconds(1));
+ o.RpiMicros.ShouldBe(2_000_000u);
+ o.TimeoutMultiplier.ShouldBe((byte)1);
+ o.ConnectionExpiry.ShouldBe(TimeSpan.FromSeconds(16));
+ o.KeepAliveInterval.ShouldBe(TimeSpan.FromSeconds(8));
+ o.VendorId.ShouldBe((ushort)0x1337);
+ new ConnectionOptions().OriginatorSerial.ShouldBe(o.OriginatorSerial); // per process, not per instance
+ }
+
+ public static TheoryData UnusableTimeouts => new() { Timeout.InfiniteTimeSpan, TimeSpan.Zero, TimeSpan.FromSeconds(-1) };
+
+ [Theory]
+ [MemberData(nameof(UnusableTimeouts))]
+ public void Options_reject_a_non_finite_or_non_positive_timeout(TimeSpan timeout) =>
+ Should.Throw(() => new ConnectionOptions { Timeout = timeout });
+
+ [Fact]
+ public void Options_reject_a_forward_open_interval_below_one_second() =>
+ Should.Throw(() => new ConnectionOptions { MinForwardOpenInterval = TimeSpan.FromMilliseconds(999) });
+
+ // ---- Large Forward Open fallback -------------------------------------------------------------
+
+ [Fact]
+ public async Task Large_open_service_not_supported_falls_back_to_regular_open_once()
+ {
+ await using var target = new FakeCipTarget();
+ 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);
+
+ await conn.SendAsync(ReadTag, TestCt);
+
+ var fos = target.ForwardOpenRequests;
+ fos.Count.ShouldBe(2);
+ fos[0][0].ShouldBe((byte)0x5B);
+ 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);
+ target.ConnectedRequests.ShouldBe(1);
+ }
+
+ [Fact]
+ public async Task Large_open_bad_size_falls_back_to_regular_open_once()
+ {
+ await using var target = new FakeCipTarget();
+ target.ForwardOpenReply = fo => fo[0] == 0x5B ? FakeCipTarget.Refuse(0x01, 0x0109)(fo) : FakeCipTarget.Accept(fo);
+ var time = new ManualTimeProvider();
+ await using var conn = Create(target, Options(), time);
+
+ await conn.SendAsync(ReadTag, TestCt);
+
+ target.ForwardOpenRequests.Select(f => f[0]).ShouldBe(new byte[] { 0x5B, 0x54 });
+ RegularOTParams(target.ForwardOpenRequests[1]).ShouldBe((ushort)(0x4200 | 504));
+ }
+
+ [Fact]
+ public async Task A_refused_fallback_is_not_retried_again()
+ {
+ await using var target = new FakeCipTarget { ForwardOpenReply = FakeCipTarget.Refuse(0x08) };
+ var time = new ManualTimeProvider();
+ await using var conn = Create(target, Options(), time);
+
+ await Should.ThrowAsync(() => conn.SendAsync(ReadTag, TestCt));
+
+ target.ForwardOpens.ShouldBe(2); // Large, then exactly one regular
+ }
+
+ [Fact]
+ public async Task Connection_in_use_does_not_fall_back()
+ {
+ await using var target = new FakeCipTarget { ForwardOpenReply = FakeCipTarget.Refuse(0x01, 0x0100) };
+ var time = new ManualTimeProvider();
+ await using var conn = Create(target, Options(), time);
+
+ var ex = await Should.ThrowAsync(() => conn.SendAsync(ReadTag, TestCt));
+
+ ex.Message.ShouldContain("Forward Open refused");
+ target.ForwardOpens.ShouldBe(1);
+ target.ConnectedRequests.ShouldBe(0);
+ }
+
+ [Fact]
+ public async Task A_small_connection_size_uses_a_regular_open()
+ {
+ await using var target = new FakeCipTarget();
+ var time = new ManualTimeProvider();
+ await using var conn = Create(target, Options(size: 504), time);
+
+ await conn.SendAsync(ReadTag, TestCt);
+
+ target.ForwardOpenRequests.Single()[0].ShouldBe((byte)0x54);
+ conn.MaxRequestBytes.ShouldBe(502);
+ }
+
+ // ---- routing (B3) ----------------------------------------------------------------------------
+
+ [Fact]
+ public async Task Direct_routing_sends_bare_cip_even_with_a_path()
+ {
+ await using var target = new FakeCipTarget();
+ var time = new ManualTimeProvider();
+ await using var conn = Create(target, Options(CipRouting.Direct), time, path: "1,0");
+
+ var reply = await conn.SendAsync(ReadTag, TestCt);
+
+ CipReply.Parse(reply).GeneralStatus.ShouldBe((byte)0);
+ target.DirectCips.Single().ShouldBe(ReadTag);
+ target.ForwardOpens.ShouldBe(0);
+ target.UnconnectedSends.ShouldBe(0);
+ target.ConnectedRequests.ShouldBe(0);
+ conn.MaxRequestBytes.ShouldBe(504);
+ }
+
+ [Fact]
+ public async Task UnconnectedSend_routing_wraps_in_0x52()
+ {
+ await using var target = new FakeCipTarget();
+ var time = new ManualTimeProvider();
+ await using var conn = Create(target, Options(CipRouting.UnconnectedSend), time, path: "1,0");
+
+ await conn.SendAsync(ReadTag, TestCt);
+
+ target.UnconnectedEmbedded.Single().ShouldBe(ReadTag);
+ var raw = FakeEipServer.RequestCip(target.Server.Received.Last(f => FakeEipServer.Command(f) == 0x006F));
+ raw.ShouldBe(ConnectionManagerServices.UnconnectedSend(ReadTag, new byte[] { 0x01, 0x00 }));
+ target.ForwardOpens.ShouldBe(0);
+ target.DirectRequests.ShouldBe(0);
+ conn.MaxRequestBytes.ShouldBe(504 - (12 + 2));
+ }
+
+ [Fact]
+ public async Task Connected_routing_never_sends_bare()
+ {
+ await using var target = new FakeCipTarget();
+ var time = new ManualTimeProvider();
+ await using var conn = Create(target, Options(), time);
+
+ for (var i = 0; i < 3; i++) await conn.SendAsync(ReadTag, TestCt);
+
+ target.ConnectedCips.Count.ShouldBe(3);
+ target.ConnectedCips.ShouldAllBe(c => c.SequenceEqual(ReadTag));
+ target.ForwardOpens.ShouldBe(1);
+ target.DirectRequests.ShouldBe(0);
+ target.UnconnectedSends.ShouldBe(0);
+ conn.MaxRequestBytes.ShouldBe(4000);
+ }
+
+ [Fact]
+ public async Task Connected_routing_with_an_empty_route_opens_straight_to_the_message_router()
+ {
+ // R9-1: Micro800 is Connected with an empty route; libplctag's Forward Open path is just 20 02 24 01.
+ await using var target = new FakeCipTarget();
+ var time = new ManualTimeProvider();
+ await using var conn = Create(target, Options(size: 504), time, path: "");
+
+ await conn.SendAsync(ReadTag, TestCt);
+
+ var fo = target.ForwardOpenRequests.Single();
+ fo[^6..].ShouldBe(new byte[] { 0xA3, 0x02, 0x20, 0x02, 0x24, 0x01 });
+ target.ConnectedRequests.ShouldBe(1);
+ }
+
+ // ---- reconnect -------------------------------------------------------------------------------
+
+ [Fact]
+ public async Task Caller_cancellation_does_not_reconnect()
+ {
+ await using var target = new FakeCipTarget();
+ var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ var held = 0;
+ target.ConnectedHandler = async cip =>
+ {
+ if (Interlocked.Increment(ref held) == 1) await release.Task;
+ return [(byte)(cip[0] | 0x80), 0, 0, 0];
+ };
+ var time = new ManualTimeProvider();
+ await using var conn = Create(target, Options(), time);
+ var before = Volatile.Read(ref CipConnection.ForwardOpenCount);
+
+ using var cts = CancellationTokenSource.CreateLinkedTokenSource(TestCt);
+ var pending = conn.SendAsync(ReadTag, cts.Token);
+ await FakeCipTarget.Until(() => target.ConnectedRequests == 1, Guard, "the connected request");
+ await cts.CancelAsync();
+ await Should.ThrowAsync(() => pending);
+ release.SetResult();
+
+ await conn.SendAsync(ReadTag, TestCt);
+
+ (Volatile.Read(ref CipConnection.ForwardOpenCount) - before).ShouldBe(1);
+ target.ForwardOpens.ShouldBe(1);
+ target.Server.Connections.ShouldBe(1);
+ target.ConnectedRequests.ShouldBe(2);
+ }
+
+ [Fact]
+ public async Task Timeout_reconnects_through_the_throttle()
+ {
+ await using var target = new FakeCipTarget();
+ var answer = 1;
+ target.ConnectedHandler = cip => Task.FromResult(
+ Volatile.Read(ref answer) == 1 ? [(byte)(cip[0] | 0x80), 0, 0, 0] : null);
+ var time = new ManualTimeProvider();
+ await using var conn = Create(target, Options(timeoutSeconds: 0.3), time);
+
+ await conn.SendAsync(ReadTag, TestCt); // first Forward Open, at fake t0
+ Volatile.Write(ref answer, 0);
+ var timedOut = await Should.ThrowAsync(() => conn.SendAsync(ReadTag, TestCt));
+ timedOut.Status.ShouldBe(AbStatus.Timeout); // the failing call keeps its own error
+ target.ForwardOpens.ShouldBe(1);
+
+ Volatile.Write(ref answer, 1);
+ var reconnect = conn.SendAsync(ReadTag, TestCt);
+ await time.WaitForTimerDueAtAsync(time.Start + TimeSpan.FromSeconds(1), Guard); // parked in the throttle
+ target.ForwardOpens.ShouldBe(1);
+ reconnect.IsCompleted.ShouldBeFalse();
+
+ time.Advance(TimeSpan.FromSeconds(1));
+ await reconnect.WaitAsync(Guard, TestCt);
+
+ target.ForwardOpens.ShouldBe(2);
+ target.Server.Connections.ShouldBe(2);
+ // A fresh connection serial per attempt, so the target never sees a duplicate of the stale connection.
+ var serials = target.ForwardOpenRequests.Select(f => BinaryPrimitives.ReadUInt16LittleEndian(f.AsSpan(16))).ToArray();
+ serials[1].ShouldNotBe(serials[0]);
+ }
+
+ [Fact]
+ public async Task A_stalled_open_is_bounded_by_the_deadline()
+ {
+ // R9-3: no caller token at all, and the target never answers the Forward Open.
+ await using var target = new FakeCipTarget();
+ var time = new ManualTimeProvider();
+ await using var conn = Create(target, Options(timeoutSeconds: 30), time);
+ target.Server.Stall = true;
+
+ var send = conn.SendAsync(ReadTag, CancellationToken.None);
+ await FakeCipTarget.Until(() => target.ForwardOpens == 0 && target.Server.Received.Count >= 2, Guard, "the Forward Open frame");
+ var openBudget = TimeSpan.FromSeconds(1) + 4 * TimeSpan.FromSeconds(30);
+ await time.WaitForTimerDueAtAsync(time.Start + openBudget, Guard);
+ time.Advance(openBudget);
+
+ var ex = await Should.ThrowAsync(() => send.WaitAsync(Guard, TestCt));
+ ex.Status.ShouldBe(AbStatus.Timeout);
+ }
+
+ // ---- keep-alive ------------------------------------------------------------------------------
+
+ [Fact]
+ public async Task Idle_connection_sends_one_keepalive_per_interval()
+ {
+ 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);
+ await conn.SendAsync(ReadTag, TestCt);
+
+ time.Advance(TimeSpan.FromSeconds(8));
+ conn.KeepAlivesSent.ShouldBe(1);
+ await FakeCipTarget.Until(() => target.ConnectedRequests == 2, Guard, "the keep-alive");
+ target.ConnectedCips[1].ShouldBe(CipConnection.KeepAliveRequest);
+ await time.WaitForTimerDueAtAsync(time.Start + TimeSpan.FromSeconds(16), Guard); // re-armed
+
+ time.Advance(TimeSpan.FromSeconds(7));
+ conn.KeepAlivesSent.ShouldBe(1); // exactly one per idle interval
+
+ 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
+ target.ForwardOpens.ShouldBe(1);
+ }
+
+ [Fact]
+ public async Task Traffic_suppresses_keepalive()
+ {
+ await using var target = new FakeCipTarget();
+ var time = new ManualTimeProvider();
+ await using var conn = Create(target, Options(), time);
+ await conn.SendAsync(ReadTag, TestCt);
+
+ for (var i = 0; i < 4; i++)
+ {
+ time.Advance(TimeSpan.FromSeconds(5));
+ await conn.SendAsync(ReadTag, TestCt);
+ }
+
+ conn.KeepAlivesSent.ShouldBe(0);
+ target.ConnectedCips.ShouldAllBe(c => c.SequenceEqual(ReadTag));
+ target.ConnectedRequests.ShouldBe(5);
+ }
+
+ [Fact]
+ public async Task No_keepalive_is_sent_when_the_connection_was_never_opened()
+ {
+ await using var target = new FakeCipTarget();
+ var time = new ManualTimeProvider();
+ await using var conn = Create(target, Options(), time);
+
+ time.Advance(TimeSpan.FromSeconds(30));
+
+ conn.KeepAlivesSent.ShouldBe(0);
+ target.ForwardOpens.ShouldBe(0);
+ target.Server.Connections.ShouldBe(0);
+ }
+
+ // ---- disposal --------------------------------------------------------------------------------
+
+ [Fact]
+ public async Task Send_after_dispose_throws_Abort()
+ {
+ await using var target = new FakeCipTarget();
+ var time = new ManualTimeProvider();
+ var conn = Create(target, Options(), time);
+ await conn.SendAsync(ReadTag, TestCt);
+
+ await conn.DisposeAsync();
+
+ var ex = await Should.ThrowAsync(() => conn.SendAsync(ReadTag, TestCt));
+ ex.Status.ShouldBe(AbStatus.Abort);
+ target.ForwardCloses.ShouldBe(1);
+ await conn.DisposeAsync(); // idempotent
+ }
+
+ [Fact]
+ public async Task Callers_queued_behind_an_open_get_Abort_when_disposed()
+ {
+ await using var target = new FakeCipTarget { ForwardOpenReply = FakeCipTarget.Refuse(0x01, 0x0113) };
+ var time = new ManualTimeProvider();
+ var conn = Create(target, Options(), time);
+ await Should.ThrowAsync(() => conn.SendAsync(ReadTag, TestCt)); // attempt 1, t0
+
+ var parked = conn.SendAsync(ReadTag, TestCt); // waits in the throttle
+ await time.WaitForTimerDueAtAsync(time.Start + TimeSpan.FromSeconds(1), Guard);
+ var queued = Enumerable.Range(0, 5).Select(_ => conn.SendAsync(ReadTag, TestCt)).ToArray();
+ var dispose = conn.DisposeAsync().AsTask();
+ time.Advance(TimeSpan.FromSeconds(1));
+
+ (await Should.ThrowAsync(() => parked.WaitAsync(Guard, TestCt))).Status.ShouldBe(AbStatus.Abort);
+ foreach (var q in queued)
+ (await Should.ThrowAsync(() => q.WaitAsync(Guard, TestCt))).Status.ShouldBe(AbStatus.Abort);
+ await dispose.WaitAsync(Guard, TestCt);
+ target.ForwardOpens.ShouldBe(1); // nothing is opened once disposal has begun
+ }
+}
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
new file mode 100644
index 00000000..bf2124de
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/ForwardOpenThrottleTests.cs
@@ -0,0 +1,156 @@
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip;
+using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Fakes;
+namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Cip;
+
+/// Press safety: on fake time, and the storm it exists to stop.
+[Collection(CipConnectionStatics.Name)] // the storm test moves CipConnection's static counters
+public sealed class ForwardOpenThrottleTests
+{
+ private static readonly TimeSpan Second = TimeSpan.FromSeconds(1);
+ private static readonly TimeSpan Guard = TimeSpan.FromSeconds(10);
+ private static CancellationToken TestCt => TestContext.Current.CancellationToken;
+
+ [Fact]
+ public async Task Second_attempt_inside_interval_waits()
+ {
+ var time = new ManualTimeProvider();
+ var throttle = new ForwardOpenThrottle(Second, time);
+ await throttle.RunAsync(_ => Task.FromResult(1), TestCt);
+ time.Advance(TimeSpan.FromMilliseconds(400));
+
+ var started = 0;
+ var second = throttle.RunAsync(_ => { Interlocked.Increment(ref started); return Task.FromResult(2); }, TestCt);
+ await time.WaitForTimerDueAtAsync(time.Start + Second, Guard); // waits for the remainder, measured from attempt 1's start
+ time.Advance(TimeSpan.FromMilliseconds(599));
+ Volatile.Read(ref started).ShouldBe(0);
+ second.IsCompleted.ShouldBeFalse();
+
+ time.Advance(TimeSpan.FromMilliseconds(1));
+ (await second.WaitAsync(Guard, TestCt)).ShouldBe(2);
+ Volatile.Read(ref started).ShouldBe(1);
+ }
+
+ [Fact]
+ public async Task Attempt_after_interval_proceeds_immediately()
+ {
+ var time = new ManualTimeProvider();
+ var throttle = new ForwardOpenThrottle(Second, time);
+ await throttle.RunAsync(_ => Task.FromResult(1), TestCt);
+ time.Advance(Second);
+
+ var started = false;
+ var second = throttle.RunAsync(_ => { started = true; return Task.FromResult(2); }, TestCt);
+
+ started.ShouldBeTrue(); // ran synchronously: no wait was needed
+ second.IsCompletedSuccessfully.ShouldBeTrue();
+ time.ArmedTimers.ShouldBe(0);
+ }
+
+ [Fact]
+ public async Task Only_one_attempt_in_flight()
+ {
+ var time = new ManualTimeProvider();
+ var throttle = new ForwardOpenThrottle(Second, time);
+ var inFlight = 0;
+ var maxInFlight = 0;
+ var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+
+ async Task Attempt(CancellationToken _)
+ {
+ var now = Interlocked.Increment(ref inFlight);
+ InterlockedMax(ref maxInFlight, now);
+ await release.Task;
+ Interlocked.Decrement(ref inFlight);
+ return now;
+ }
+
+ 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);
+ Volatile.Read(ref inFlight).ShouldBe(1);
+ others.ShouldAllBe(t => !t.IsCompleted);
+
+ release.SetResult();
+ await first.WaitAsync(Guard, TestCt);
+ // Attempt 2 starts at once (attempt 1 started 5 s ago); each later one then waits out the
+ // interval from its predecessor's start.
+ await others[0].WaitAsync(Guard, TestCt);
+ for (var i = 1; i < others.Length; i++)
+ {
+ await time.WaitForTimerDueAtAsync(time.GetUtcNow() + Second, Guard);
+ time.Advance(Second);
+ await FakeCipTarget.Until(() => others.Count(t => t.IsCompleted) == i + 1, Guard, $"attempt {i + 2}");
+ }
+ Volatile.Read(ref maxInFlight).ShouldBe(1);
+ }
+
+ [Fact]
+ public async Task Caller_deadline_while_waiting_throws_throttled()
+ {
+ var time = new ManualTimeProvider();
+ var throttle = new ForwardOpenThrottle(Second, time);
+ await throttle.RunAsync(_ => Task.FromResult(1), TestCt);
+ using var cts = new CancellationTokenSource();
+ var invoked = false;
+
+ var second = throttle.RunAsync(_ => { invoked = true; return Task.FromResult(2); }, cts.Token);
+ await time.WaitForTimerDueAtAsync(time.Start + Second, Guard);
+ await cts.CancelAsync();
+
+ var ex = await Should.ThrowAsync(() => second.WaitAsync(Guard, TestCt));
+ ex.Status.ShouldBe(AbStatus.BadConnection);
+ ex.Message.ShouldContain("Forward Open throttled");
+ invoked.ShouldBeFalse();
+ }
+
+ [Fact]
+ public async Task Refusing_target_sees_at_most_one_forward_open_per_interval()
+ {
+ // The storm: 50 callers hammering a device that refuses every Forward Open, for 3 s of fake time.
+ // Attempt starts are >= 1 s apart, so t = 0, 1, 2, 3 is the most there can be.
+ 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);
+ using var stop = CancellationTokenSource.CreateLinkedTokenSource(TestCt);
+ var failures = 0;
+
+ var callers = Enumerable.Range(0, 50).Select(_ => Task.Run(async () =>
+ {
+ while (!stop.IsCancellationRequested)
+ {
+ try { await conn.SendAsync(new byte[] { 0x4C, 0x00 }, stop.Token); }
+ catch (AbTransportException) { Interlocked.Increment(ref failures); }
+ catch (OperationCanceledException) { /* stopping */ }
+ }
+ }, TestCt)).ToArray();
+
+ for (var step = 0; step < 30; step++)
+ {
+ 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
+
+ await stop.CancelAsync();
+ await Task.WhenAll(callers).WaitAsync(Guard, TestCt);
+ await conn.DisposeAsync();
+
+ time.GetUtcNow().ShouldBe(time.Start + TimeSpan.FromSeconds(3));
+ target.ForwardOpens.ShouldBeLessThanOrEqualTo(4);
+ target.ForwardOpens.ShouldBeGreaterThanOrEqualTo(1);
+ Volatile.Read(ref failures).ShouldBeGreaterThanOrEqualTo(target.ForwardOpens);
+ target.ConnectedRequests.ShouldBe(0);
+ }
+
+ private static void InterlockedMax(ref int location, int value)
+ {
+ int current;
+ while (value > (current = Volatile.Read(ref location)) &&
+ Interlocked.CompareExchange(ref location, value, current) != current) { }
+ }
+}
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
new file mode 100644
index 00000000..065ed32c
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/DeviceSessionPoolTests.cs
@@ -0,0 +1,173 @@
+using Microsoft.Extensions.Logging;
+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.Cip;
+using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Fakes;
+namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests;
+
+[Collection(CipConnectionStatics.Name)]
+public sealed class DeviceSessionPoolTests
+{
+ private static readonly TimeSpan Guard = TimeSpan.FromSeconds(10);
+ private static readonly byte[] ReadTag = [0x4C, 0x02, 0x91, 0x03, 0x54, 0x61, 0x67, 0x00, 0x01, 0x00];
+ private static CancellationToken TestCt => TestContext.Current.CancellationToken;
+
+ private static ConnectionOptions Options(int size = 4002) => new() { ConnectionSize = size, Timeout = TimeSpan.FromSeconds(2) };
+
+ [Fact]
+ public void Device_keys_compare_host_and_plc_attribute_case_insensitively()
+ {
+ var a = new DeviceKey("Press-07.plant.local", 44818, "1,0", "ControlLogix");
+ var b = new DeviceKey("PRESS-07.PLANT.LOCAL", 44818, "1,0", "controllogix");
+ a.ShouldBe(b);
+ a.GetHashCode().ShouldBe(b.GetHashCode());
+ a.ShouldNotBe(a with { CipPath = "1,1" });
+ a.ShouldNotBe(a with { Port = 2222 });
+ }
+
+ [Fact]
+ public async Task Leases_for_one_key_share_one_scheduler()
+ {
+ var pool = new DeviceSessionPool(new ManualTimeProvider());
+ var key = new DeviceKey("plc1", 44818, "1,0", "ControlLogix");
+
+ await using var a = pool.Acquire(key, Options());
+ await using var b = pool.Acquire(key with { Host = "PLC1" }, Options());
+ await using var other = pool.Acquire(key with { CipPath = "1,2" }, Options());
+
+ b.Scheduler.ShouldBeSameAs(a.Scheduler);
+ b.Channel.ShouldBeSameAs(a.Channel);
+ other.Scheduler.ShouldNotBeSameAs(a.Scheduler);
+ other.Channel.ShouldNotBeSameAs(a.Channel);
+ pool.EntryCount.ShouldBe(2);
+ }
+
+ [Fact]
+ public async Task A_second_acquire_with_different_options_reuses_the_first_and_logs_once()
+ {
+ var logger = new CapturingLogger();
+ var pool = new DeviceSessionPool(new ManualTimeProvider(), logger);
+ var key = new DeviceKey("plc1", 44818, "1,0", "ControlLogix");
+ var first = Options(4002);
+ var second = Options(504) with { AllowPacking = true };
+
+ await using var a = pool.Acquire(key, first);
+ await using var b = pool.Acquire(key, second);
+ await using var c = pool.Acquire(key, second);
+ await using var d = pool.Acquire(key, first);
+
+ ((CipConnection)b.Channel).Options.ShouldBe(first);
+ ((CipConnection)c.Channel).Options.ShouldBe(first);
+ logger.Warnings.ShouldBe(1);
+ }
+
+ [Fact]
+ public async Task Last_lease_dispose_sends_forward_close_then_unregister()
+ {
+ await using var target = new FakeCipTarget();
+ var pool = new DeviceSessionPool(new ManualTimeProvider());
+ var key = new DeviceKey("127.0.0.1", target.Port, "1,0", "ControlLogix");
+ var a = pool.Acquire(key, Options());
+ var b = pool.Acquire(key, Options());
+ (await a.Scheduler.SendAsync(ReadTag, TestCt)).GeneralStatus.ShouldBe((byte)0);
+
+ await a.DisposeAsync();
+ await a.DisposeAsync(); // idempotent: must not release b's reference
+ target.ForwardCloses.ShouldBe(0);
+ (await b.Scheduler.SendAsync(ReadTag, TestCt)).GeneralStatus.ShouldBe((byte)0);
+
+ await b.DisposeAsync();
+ await target.Server.WaitForFrameAsync(f => FakeEipServer.Command(f) == EncapCommand.UnRegisterSession, Guard);
+
+ target.ForwardCloses.ShouldBe(1);
+ var frames = target.Server.Received.ToList();
+ var close = frames.FindIndex(f => FakeEipServer.Command(f) == EncapCommand.SendRRData
+ && FakeCipTarget.IsForwardClose(FakeEipServer.RequestCip(f)));
+ var unregister = frames.FindIndex(f => FakeEipServer.Command(f) == EncapCommand.UnRegisterSession);
+ close.ShouldBeGreaterThanOrEqualTo(0);
+ close.ShouldBeLessThan(unregister);
+ // The Forward Close names the connection that was opened.
+ target.ForwardCloseRequests.Single().AsSpan(8, 8).ToArray()
+ .ShouldBe(target.ForwardOpenRequests.Single().AsSpan(16, 8).ToArray());
+ pool.EntryCount.ShouldBe(0);
+ await Should.ThrowAsync(() => b.Scheduler.SendAsync(ReadTag, TestCt));
+ }
+
+ [Fact]
+ public async Task A_later_acquire_builds_new_instances()
+ {
+ var pool = new DeviceSessionPool(new ManualTimeProvider());
+ var key = new DeviceKey("plc1", 44818, "1,0", "ControlLogix");
+ var first = pool.Acquire(key, Options());
+ var scheduler = first.Scheduler;
+ var channel = first.Channel;
+ await first.DisposeAsync();
+
+ await using var second = pool.Acquire(key, Options());
+
+ second.Scheduler.ShouldNotBeSameAs(scheduler);
+ second.Channel.ShouldNotBeSameAs(channel);
+ }
+
+ [Fact]
+ public async Task A_reconnect_after_the_last_release_is_still_throttled()
+ {
+ await using var target = new FakeCipTarget();
+ var time = new ManualTimeProvider();
+ var pool = new DeviceSessionPool(time);
+ var key = new DeviceKey("127.0.0.1", target.Port, "1,0", "ControlLogix");
+
+ await using (var first = pool.Acquire(key, Options()))
+ await first.Scheduler.SendAsync(ReadTag, TestCt); // Forward Open at fake t0
+
+ await using var second = pool.Acquire(key, Options());
+ var send = second.Scheduler.SendAsync(ReadTag, TestCt);
+ await time.WaitForTimerDueAtAsync(time.Start + TimeSpan.FromSeconds(1), Guard);
+ target.ForwardOpens.ShouldBe(1);
+
+ time.Advance(TimeSpan.FromSeconds(1));
+ await send.WaitAsync(Guard, TestCt);
+ target.ForwardOpens.ShouldBe(2);
+ }
+
+ [Fact]
+ public async Task Concurrent_acquire_and_release_is_safe()
+ {
+ var pool = new DeviceSessionPool(new ManualTimeProvider());
+ var keys = Enumerable.Range(0, 3).Select(i => new DeviceKey($"plc{i}", 44818, "1,0", "ControlLogix")).ToArray();
+ var schedulers = new System.Collections.Concurrent.ConcurrentDictionary<(int, RequestScheduler), byte>();
+
+ var tasks = Enumerable.Range(0, 100).Select(i => Task.Run(async () =>
+ {
+ var lease = pool.Acquire(keys[i % 3], Options());
+ schedulers.TryAdd((i % 3, lease.Scheduler), 0);
+ await Task.Yield();
+ await lease.DisposeAsync();
+ }, TestCt)).ToArray();
+ await Task.WhenAll(tasks).WaitAsync(Guard, TestCt);
+
+ pool.EntryCount.ShouldBe(0);
+ // Every lease got a scheduler for its own key only, never another key's.
+ schedulers.Keys.GroupBy(k => k.Item2).ShouldAllBe(g => g.Select(k => k.Item1).Distinct().Count() == 1);
+ }
+
+ [Fact]
+ public void Shared_is_one_process_wide_instance() =>
+ DeviceSessionPool.Shared.ShouldBeSameAs(DeviceSessionPool.Shared);
+
+ private sealed class CapturingLogger : ILogger
+ {
+ private int _warnings;
+ public int Warnings => Volatile.Read(ref _warnings);
+ public IDisposable? BeginScope(TState state) where TState : notnull => null;
+ public bool IsEnabled(LogLevel logLevel) => true;
+
+ public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception,
+ Func formatter)
+ {
+ if (logLevel == LogLevel.Warning) Interlocked.Increment(ref _warnings);
+ }
+ }
+}
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeCipTarget.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeCipTarget.cs
new file mode 100644
index 00000000..0ce9f4eb
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeCipTarget.cs
@@ -0,0 +1,123 @@
+using System.Buffers.Binary;
+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.Wire;
+namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Fakes;
+
+///
+/// A CIP target on top of : answers Forward Open / Forward Close /
+/// Unconnected Send / Direct (bare SendRRData) / connected SendUnitData requests and counts each kind.
+///
+internal sealed class FakeCipTarget : IAsyncDisposable
+{
+ public const uint OTConnectionId = 0xC0FFEE01;
+
+ private readonly List _forwardOpens = [];
+ private readonly List _forwardCloses = [];
+ private readonly List _unconnectedEmbedded = [];
+ private readonly List _direct = [];
+ private readonly List _connected = [];
+
+ public FakeCipTarget()
+ {
+ Server.HandlerAsync = HandleAsync;
+ }
+
+ public FakeEipServer Server { get; } = new();
+
+ public int Port => Server.Port;
+
+ /// Forward Open request CIP → reply CIP. Defaults to .
+ public Func ForwardOpenReply { get; set; } = Accept;
+
+ /// Reply to an ordinary service request (Direct, embedded in Unconnected Send, or connected).
+ /// Defaults to success with two data bytes (a Vendor ID, for the keep-alive).
+ public Func ServiceReply { get; set; } = req => [(byte)(req[0] | 0x80), 0, 0, 0, 0x01, 0x00];
+
+ /// Connected requests only: may delay, or return null to never answer. Defaults to .
+ public Func>? ConnectedHandler { get; set; }
+
+ public int ForwardOpens { get { lock (_forwardOpens) return _forwardOpens.Count; } }
+ public int ForwardCloses { get { lock (_forwardCloses) return _forwardCloses.Count; } }
+ public int UnconnectedSends { get { lock (_unconnectedEmbedded) return _unconnectedEmbedded.Count; } }
+ public int DirectRequests { get { lock (_direct) return _direct.Count; } }
+ public int ConnectedRequests { get { lock (_connected) return _connected.Count; } }
+
+ public IReadOnlyList ForwardOpenRequests { get { lock (_forwardOpens) return [.. _forwardOpens]; } }
+ public IReadOnlyList ForwardCloseRequests { get { lock (_forwardCloses) return [.. _forwardCloses]; } }
+ public IReadOnlyList UnconnectedEmbedded { get { lock (_unconnectedEmbedded) return [.. _unconnectedEmbedded]; } }
+ public IReadOnlyList DirectCips { get { lock (_direct) return [.. _direct]; } }
+ public IReadOnlyList ConnectedCips { get { lock (_connected) return [.. _connected]; } }
+
+ /// A successful (Large) Forward Open reply echoing the request's ids and serials.
+ public static byte[] Accept(byte[] fo) => new CipWriter()
+ .U8((byte)(fo[0] | 0x80)).U8(0).U8(0).U8(0)
+ .U32(OTConnectionId)
+ .U32(BinaryPrimitives.ReadUInt32LittleEndian(fo.AsSpan(12)))
+ .U16(BinaryPrimitives.ReadUInt16LittleEndian(fo.AsSpan(16)))
+ .U16(BinaryPrimitives.ReadUInt16LittleEndian(fo.AsSpan(18)))
+ .U32(BinaryPrimitives.ReadUInt32LittleEndian(fo.AsSpan(20)))
+ .U32(2_000_000).U32(2_000_000).U8(0).U8(0)
+ .ToArray();
+
+ /// A refusal with the given general and extended status.
+ public static Func Refuse(byte general, params ushort[] extended) => fo =>
+ {
+ var w = new CipWriter().U8((byte)(fo[0] | 0x80)).U8(0).U8(general).U8((byte)extended.Length);
+ foreach (var e in extended) w.U16(e);
+ return w.ToArray();
+ };
+
+ public static bool IsForwardOpen(byte[] cip) => cip[0] is 0x54 or 0x5B && AddressesConnectionManager(cip);
+
+ public static bool IsForwardClose(byte[] cip) => cip[0] == 0x4E && AddressesConnectionManager(cip);
+
+ private static bool AddressesConnectionManager(byte[] cip) =>
+ cip.Length >= 6 && cip.AsSpan(2, 4).SequenceEqual(CipPath.ConnectionManager);
+
+ private async Task HandleAsync(byte[] frame)
+ {
+ var cip = FakeEipServer.RequestCip(frame);
+ if (FakeEipServer.Command(frame) == EncapCommand.SendUnitData)
+ {
+ lock (_connected) _connected.Add(cip);
+ var reply = ConnectedHandler is null ? ServiceReply(cip) : await ConnectedHandler(cip);
+ return reply is null ? null : FakeEipServer.UnitReply(frame, reply);
+ }
+
+ if (IsForwardOpen(cip))
+ {
+ lock (_forwardOpens) _forwardOpens.Add(cip);
+ return FakeEipServer.RRReply(frame, ForwardOpenReply(cip));
+ }
+ if (IsForwardClose(cip))
+ {
+ lock (_forwardCloses) _forwardCloses.Add(cip);
+ var w = new CipWriter().U8(0xCE).U8(0).U8(0).U8(0).Bytes(cip.AsSpan(8, 8)).U8(0).U8(0);
+ return FakeEipServer.RRReply(frame, w.ToArray());
+ }
+ if (cip[0] == 0x52 && AddressesConnectionManager(cip))
+ {
+ var len = BinaryPrimitives.ReadUInt16LittleEndian(cip.AsSpan(8));
+ var embedded = cip.AsSpan(10, len).ToArray();
+ lock (_unconnectedEmbedded) _unconnectedEmbedded.Add(embedded);
+ return FakeEipServer.RRReply(frame, ServiceReply(embedded));
+ }
+
+ lock (_direct) _direct.Add(cip);
+ return FakeEipServer.RRReply(frame, ServiceReply(cip));
+ }
+
+ /// Polls (real time) until holds.
+ public static async Task Until(Func condition, TimeSpan timeout, string what)
+ {
+ var giveUp = DateTime.UtcNow + timeout;
+ while (!condition())
+ {
+ if (DateTime.UtcNow > giveUp) throw new TimeoutException($"Timed out waiting for {what}.");
+ await Task.Delay(5);
+ }
+ }
+
+ public ValueTask DisposeAsync() => Server.DisposeAsync();
+}
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/ManualTimeProvider.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/ManualTimeProvider.cs
new file mode 100644
index 00000000..49b5ee77
--- /dev/null
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/ManualTimeProvider.cs
@@ -0,0 +1,102 @@
+namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Fakes;
+
+///
+/// Deterministic : time moves only on , which fires every
+/// timer falling due on the way, in due order, synchronously on the calling thread.
+/// (Microsoft.Extensions.TimeProvider.Testing is not pinned in this repo and no package may be added.)
+///
+internal sealed class ManualTimeProvider : TimeProvider
+{
+ private readonly object _lock = new();
+ private readonly List _timers = [];
+ private DateTimeOffset _now;
+
+ public ManualTimeProvider() => Start = _now = new DateTimeOffset(2026, 10, 7, 0, 0, 0, TimeSpan.Zero);
+
+ public DateTimeOffset Start { get; }
+
+ public override DateTimeOffset GetUtcNow() { lock (_lock) return _now; }
+
+ public override long GetTimestamp() => GetUtcNow().UtcTicks;
+
+ public override long TimestampFrequency => TimeSpan.TicksPerSecond;
+
+ public override ITimer CreateTimer(TimerCallback callback, object? state, TimeSpan dueTime, TimeSpan period)
+ {
+ var t = new ManualTimer(this, callback, state);
+ lock (_lock) _timers.Add(t);
+ t.Change(dueTime, period);
+ return t;
+ }
+
+ /// Armed timers (not disposed, with a due time).
+ public int ArmedTimers { get { lock (_lock) return _timers.Count(t => t.Due is not null); } }
+
+ public bool HasTimerDueAt(DateTimeOffset at) { lock (_lock) return _timers.Any(t => t.Due == at); }
+
+ /// Waits (in real time) until some code under test has armed a timer due at ,
+ /// i.e. is genuinely parked on the fake clock rather than merely not yet scheduled.
+ public async Task WaitForTimerDueAtAsync(DateTimeOffset at, TimeSpan timeout)
+ {
+ var giveUp = DateTime.UtcNow + timeout;
+ while (!HasTimerDueAt(at))
+ {
+ if (DateTime.UtcNow > giveUp) throw new TimeoutException($"No timer armed for {at:O} (now {GetUtcNow():O}).");
+ await Task.Delay(5);
+ }
+ }
+
+ public void Advance(TimeSpan by)
+ {
+ DateTimeOffset target;
+ lock (_lock) target = _now + by;
+ while (true)
+ {
+ ManualTimer? next;
+ lock (_lock)
+ {
+ next = _timers.Where(t => t.Due <= target).MinBy(t => t.Due);
+ if (next is null) { _now = target; return; }
+ _now = next.Due!.Value;
+ next.Due = next.Period is { } p ? _now + p : null;
+ }
+ next.Fire();
+ }
+ }
+
+ private sealed class ManualTimer(ManualTimeProvider owner, TimerCallback callback, object? state) : ITimer
+ {
+ public DateTimeOffset? Due; // guarded by owner._lock
+ public TimeSpan? Period; // guarded by owner._lock
+ private bool _disposed;
+
+ public void Fire() => callback(state);
+
+ public bool Change(TimeSpan dueTime, TimeSpan period)
+ {
+ lock (owner._lock)
+ {
+ if (_disposed) return false;
+ Due = dueTime == Timeout.InfiniteTimeSpan ? null : owner._now + dueTime;
+ Period = period == Timeout.InfiniteTimeSpan || period == TimeSpan.Zero ? null : period;
+ return true;
+ }
+ }
+
+ public void Dispose()
+ {
+ lock (owner._lock)
+ {
+ _disposed = true;
+ Due = null;
+ owner._timers.Remove(this);
+ }
+ }
+
+ public ValueTask DisposeAsync()
+ {
+ Dispose();
+ return ValueTask.CompletedTask;
+ }
+ }
+}