fix(ab-transport): harden Task 9 pool and connection (close chaining, uncancellable shared open, monotonic throttle, canonical CIP path, internal ctors, shared-pool logger)

2026-10-07 16:45:26 -04:00
parent b82fffd269
commit a251954bf7
8 changed files with 280 additions and 60 deletions
@@ -41,12 +41,14 @@ public sealed class CipConnection : ICipChannel, IAsyncDisposable
    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 object _sync = new();
+    // Cancels only the waits BEFORE an attempt (predecessor close, throttle interval), never an attempt
+    // already on the wire. Never disposed: it has no timer, and late registrations must stay legal.
+    private readonly CancellationTokenSource _disposing = new();
    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 Task<Open>? _pendingOpen;   // the one shared open in progress; under _sync
+    private Open? _open;                // written under _sync (Volatile); read lock-free via Volatile
+    private int _disposed;              // set under _sync; read lock-free via Volatile
    private int _disposeStarted;
    private long _lastSendTimestamp;
    private int _keepAlivesSent;
@@ -58,7 +60,7 @@ public sealed class CipConnection : ICipChannel, IAsyncDisposable
    /// <param name="time">Clock for the throttle, deadlines and keep-alive.</param>
    /// <param name="predecessorClosed">Completes when the previous connection for the same key has
    /// finished closing, so two connections to one device never overlap.</param>
-    public CipConnection(DeviceKey key, ConnectionOptions options, ForwardOpenThrottle throttle, TimeProvider time,
+    internal CipConnection(DeviceKey key, ConnectionOptions options, ForwardOpenThrottle throttle, TimeProvider time,
        Task? predecessorClosed = null)
    {
        _key = key ?? throw new ArgumentNullException(nameof(key));
@@ -100,37 +102,68 @@ public sealed class CipConnection : ICipChannel, IAsyncDisposable
    /// <summary>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).</summary>
-    private async Task<Open> EnsureOpenAsync(CancellationToken ct)
+    /// <remarks>All callers share ONE open, which runs under its own deadline only. A caller's token (or the
+    /// scheduler's stop token) ends that caller's <i>wait</i>, never the open itself: cancelling a Forward
+    /// Open mid-flight could leave a connection the target accepted but nobody ever closes. A completed
+    /// open is stored even if disposal began meanwhile, so <see cref="DisposeAsync"/> Forward Closes it.</remarks>
+    private Task<Open> EnsureOpenAsync(CancellationToken ct)
    {
-        if (Volatile.Read(ref _open) is { Client.IsConnected: true } live) return live;
+        if (Volatile.Read(ref _open) is { Client.IsConnected: true } live) return Task.FromResult(live);

+        Task<Open> pending;
+        TaskCompletionSource<Open>? started = null;
+        lock (_sync)
+        {
+            ThrowIfDisposed();
+            if (_open is { Client.IsConnected: true } current) return Task.FromResult(current);
+            if (_pendingOpen is null)
+            {
+                started = new TaskCompletionSource<Open>(TaskCreationOptions.RunContinuationsAsynchronously);
+                _pendingOpen = started.Task;
+            }
+            pending = _pendingOpen;
+        }
+        if (started is not null) _ = RunOpenAsync(started);
+        return pending.WaitAsync(ct);
+    }
+
+    /// <summary>The shared open. Bounded by <see cref="OpenBudget"/> (ruling R9-3); its outcome, success or
+    /// failure, goes to every caller waiting on it.</summary>
+    private async Task RunOpenAsync(TaskCompletionSource<Open> result)
+    {
        using var deadline = new CancellationTokenSource(OpenBudget, _time);
-        using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, deadline.Token);
+        using var beforeAttempt = CancellationTokenSource.CreateLinkedTokenSource(deadline.Token, _disposing.Token);
+        Open? opened = null;
+        Exception? error = null;
        try
        {
-            await _state.WaitAsync(linked.Token).ConfigureAwait(false);
-            try
+            Open? stale;
+            lock (_sync)
            {
-                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;
+                stale = _open;
+                Volatile.Write(ref _open, null);
            }
-            finally { _state.Release(); }
+            if (stale is not null) await stale.Client.DisposeAsync().ConfigureAwait(false);   // faulted: no UnRegister
+            await _predecessorClosed.WaitAsync(beforeAttempt.Token).ConfigureAwait(false);
+            opened = await _throttle.RunAsync(_ => OpenAsync(deadline.Token), beforeAttempt.Token).ConfigureAwait(false);
        }
-        catch (OperationCanceledException) when (!ct.IsCancellationRequested && deadline.IsCancellationRequested)
+        catch (Exception ex)
        {
-            throw new AbTransportException(AbStatus.Timeout,
-                $"Opening {_key.Host}:{_key.Port} did not complete within {OpenBudget.TotalMilliseconds:F0} ms.");
+            error = Volatile.Read(ref _disposed) != 0
+                ? new AbTransportException(AbStatus.Abort, "Connection disposed.", ex)
+                : ex is OperationCanceledException && deadline.IsCancellationRequested
+                    ? new AbTransportException(AbStatus.Timeout,
+                        $"Opening {_key.Host}:{_key.Port} did not complete within {OpenBudget.TotalMilliseconds:F0} ms.", ex)
+                    : ex;
        }
+
+        lock (_sync)
+        {
+            if (opened is not null) Volatile.Write(ref _open, opened);
+            _pendingOpen = null;
+        }
+        if (opened is not null) result.SetResult(opened);
+        else result.SetException(error!);
    }

    private async Task<Open> OpenAsync(CancellationToken ct)
@@ -252,18 +285,28 @@ public sealed class CipConnection : ICipChannel, IAsyncDisposable
    {
        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);
+        // Under the lock, so no caller can start a new open after this point. Then cancel the shared
+        // open's pre-attempt waits; an attempt already on the wire is allowed to finish (bounded by
+        // OpenBudget) so that whatever it opened is Forward Closed below.
+        Task<Open>? pending;
+        lock (_sync)
+        {
+            Volatile.Write(ref _disposed, 1);
+            pending = _pendingOpen;
+        }
+        await _disposing.CancelAsync().ConfigureAwait(false);
+        if (pending is not null)
+        {
+            try { await pending.ConfigureAwait(false); }
+            catch (Exception) { /* its callers have the error; nothing was opened */ }
+        }
+
        Open? open;
-        try
+        lock (_sync)
        {
            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;
@@ -272,7 +315,7 @@ public sealed class CipConnection : ICipChannel, IAsyncDisposable
            try
            {
                var budget = TimeSpan.FromSeconds(1);
-                using var cts = new CancellationTokenSource(budget);
+                using var cts = new CancellationTokenSource(budget, _time);
                await open.Client.SendRRDataAsync(
                    ConnectionManagerServices.ForwardClose(open.ConnectionSerial, _o.VendorId, _o.OriginatorSerial, _route),
                    budget, cts.Token).ConfigureAwait(false);
@@ -1,12 +1,25 @@
 namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip;

 /// <summary>
-/// Identity of one physical CIP connection: exactly one <see cref="CipConnection"/> exists per key
-/// (press safety). <see cref="Host"/> and <see cref="PlcAttribute"/> compare case-insensitively;
-/// <see cref="Port"/> and <see cref="CipPath"/> compare exactly.
+/// Identity of one physical CIP connection: exactly one <see cref="CipConnection"/> (and one Forward Open
+/// throttle) exists per key (press safety). <see cref="Host"/> and <see cref="PlcAttribute"/> compare
+/// case-insensitively; <see cref="Port"/> compares exactly; <see cref="CipPath"/> is canonicalised (all
+/// whitespace removed, so <c>"1,0"</c> and <c>" 1, 0 "</c> are the same device) and then compares exactly.
 /// </summary>
 public sealed record DeviceKey(string Host, int Port, string CipPath, string PlcAttribute)
 {
+    private readonly string _cipPath = Canonicalise(CipPath);
+
+    /// <summary>The route, whitespace-free. Canonicalised on construction and on <c>with</c>.</summary>
+    public string CipPath
+    {
+        get => _cipPath;
+        init => _cipPath = Canonicalise(value);
+    }
+
+    private static string Canonicalise(string? path) =>
+        path is null ? "" : string.Concat(path.Where(c => !char.IsWhiteSpace(c)));
+
    public bool Equals(DeviceKey? other) =>
        other is not null
        && string.Equals(Host, other.Host, StringComparison.OrdinalIgnoreCase)
@@ -12,9 +12,11 @@ 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
+    private long? _lastStartTimestamp;    // monotonic (TimeProvider.GetTimestamp); only touched while holding _inFlight

-    public ForwardOpenThrottle(TimeSpan minInterval, TimeProvider time)
+    /// <remarks>Internal: the only legitimate owner is <see cref="DeviceSessionPool"/>, which keeps exactly
+    /// one per device. A second throttle for the same device would halve the guaranteed interval.</remarks>
+    internal 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.");
@@ -34,9 +36,10 @@ public sealed class ForwardOpenThrottle

        try
        {
-            if (_lastStart is { } last)
+            if (_lastStartTimestamp is { } last)
            {
-                var wait = last + _minInterval - _time.GetUtcNow();
+                // Monotonic: a wall-clock step (NTP, DST, an operator) can never shorten the interval.
+                var wait = _minInterval - _time.GetElapsedTime(last);
                if (wait > TimeSpan.Zero)
                {
                    try { await Task.Delay(wait, _time, ct).ConfigureAwait(false); }
@@ -44,7 +47,7 @@ public sealed class ForwardOpenThrottle
                    { throw Throttled("the minimum interval had not elapsed"); }
                }
            }
-            _lastStart = _time.GetUtcNow();
+            _lastStartTimestamp = _time.GetTimestamp();
            return await open(ct).ConfigureAwait(false);
        }
        finally { _inFlight.Release(); }
@@ -16,9 +16,12 @@ public sealed class DeviceSessionPool
    /// <summary>The process-wide pool the drivers use. Tests build their own.</summary>
    public static DeviceSessionPool Shared { get; } = new();

-    internal sealed class Entry(CipConnection connection, RequestScheduler scheduler, ConnectionOptions options)
+    internal sealed class Entry(CipConnection connection, RequestScheduler scheduler, ConnectionOptions options,
+        Task predecessorClosed)
    {
        public CipConnection Connection { get; } = connection;
+        /// <summary>Completes when every earlier connection for this key has finished closing.</summary>
+        public Task PredecessorClosed { get; } = predecessorClosed;
        public RequestScheduler Scheduler { get; } = scheduler;
        public ConnectionOptions Options { get; } = options;
        public int RefCount;            // under the pool lock
@@ -30,7 +33,7 @@ public sealed class DeviceSessionPool
    private readonly Dictionary<DeviceKey, ForwardOpenThrottle> _throttles = new();
    private readonly Dictionary<DeviceKey, Task> _closing = new();
    private readonly TimeProvider _time;
-    private readonly ILogger _logger;
+    private volatile ILogger _logger;

    public DeviceSessionPool(TimeProvider? time = null, ILogger? logger = null)
    {
@@ -38,6 +41,14 @@ public sealed class DeviceSessionPool
        _logger = logger ?? NullLogger.Instance;
    }

+    /// <summary>Give <see cref="Shared"/> a logger (it is built before any host exists), so its
+    /// options-mismatch and close-failure warnings reach the service log. Call once at startup.</summary>
+    public static void ConfigureSharedLogger(ILoggerFactory loggerFactory)
+    {
+        ArgumentNullException.ThrowIfNull(loggerFactory);
+        Shared._logger = loggerFactory.CreateLogger<DeviceSessionPool>();
+    }
+
    /// <summary>Test hook: keys with a live entry.</summary>
    internal int EntryCount { get { lock (_lock) return _entries.Count; } }

@@ -54,9 +65,9 @@ public sealed class DeviceSessionPool
            {
                if (!_throttles.TryGetValue(key, out var throttle))
                    _throttles[key] = throttle = new ForwardOpenThrottle(options.MinForwardOpenInterval, _time);
-                var predecessor = _closing.GetValueOrDefault(key);
+                var predecessor = _closing.GetValueOrDefault(key) ?? Task.CompletedTask;
                var connection = new CipConnection(key, options, throttle, _time, predecessor);
-                entry = new Entry(connection, new RequestScheduler(connection, options.AllowPacking), options);
+                entry = new Entry(connection, new RequestScheduler(connection, options.AllowPacking), options, predecessor);
                _entries[key] = entry;
            }
            else if (!entry.Options.Equals(options) && !entry.MismatchLogged)
@@ -75,27 +86,45 @@ public sealed class DeviceSessionPool
    internal async ValueTask ReleaseAsync(DeviceKey key, Entry entry)
    {
        TaskCompletionSource closed;
+        Task chain;
        lock (_lock)
        {
            if (--entry.RefCount > 0) return;
            _entries.Remove(key);
            closed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
-            _closing[key] = closed.Task;
+            // The successor must wait for THIS close and for every earlier one still running. An entry
+            // that never opened closes instantly; without the chain it would hide a slow predecessor
+            // (A closing, B acquired+released unopened, C opens while A's Forward Close is still pending).
+            chain = Task.WhenAll(entry.PredecessorClosed, closed.Task);
+            _closing[key] = chain;
        }
-
-        // 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
+        _ = chain.ContinueWith(_ =>
        {
            lock (_lock)
            {
-                if (_closing.TryGetValue(key, out var t) && t == closed.Task) _closing.Remove(key);
+                if (_closing.TryGetValue(key, out var t) && t == chain) _closing.Remove(key);
            }
+        }, 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.
+        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);
+            }
+        }
+        finally
+        {
            closed.TrySetResult();   // never faults: a successor only needs to know the close is over
        }
    }
@@ -106,6 +106,23 @@ public sealed class ForwardOpenThrottleTests
        invoked.ShouldBeFalse();
    }

+    [Fact]
+    public async Task A_wall_clock_step_does_not_shorten_the_interval()
+    {
+        var time = new ManualTimeProvider();
+        var throttle = new ForwardOpenThrottle(Second, time);
+        await throttle.RunAsync(_ => Task.FromResult(1), TestCt);
+
+        time.WallClockOffset = TimeSpan.FromHours(1);   // NTP / DST / operator steps the clock forward
+        var started = 0;
+        var second = throttle.RunAsync(_ => { Interlocked.Increment(ref started); return Task.FromResult(2); }, TestCt);
+
+        await time.WaitForTimerDueAtAsync(time.Start + Second, Guard);   // still parked for the full interval
+        Volatile.Read(ref started).ShouldBe(0);
+        time.Advance(Second);
+        (await second.WaitAsync(Guard, TestCt)).ShouldBe(2);
+    }
+
    [Fact]
    public async Task Refusing_target_sees_at_most_one_forward_open_per_interval()
    {
@@ -153,6 +153,109 @@ public sealed class DeviceSessionPoolTests
        schedulers.Keys.GroupBy(k => k.Item2).ShouldAllBe(g => g.Select(k => k.Item1).Distinct().Count() == 1);
    }

+    [Fact]
+    public async Task Device_keys_ignore_whitespace_in_the_cip_path()
+    {
+        var a = new DeviceKey("plc1", 44818, "1,0", "ControlLogix");
+        var b = new DeviceKey("plc1", 44818, " 1, 0 ", "ControlLogix");
+        b.CipPath.ShouldBe("1,0");
+        b.ShouldBe(a);
+        b.GetHashCode().ShouldBe(a.GetHashCode());
+        (a with { CipPath = "1, 2" }).CipPath.ShouldBe("1,2");
+
+        var pool = new DeviceSessionPool(new ManualTimeProvider());
+        await using var la = pool.Acquire(a, Options());
+        await using var lb = pool.Acquire(b, Options());
+        lb.Scheduler.ShouldBeSameAs(la.Scheduler);   // one connection, one throttle
+        pool.EntryCount.ShouldBe(1);
+    }
+
+    [Fact]
+    public async Task A_new_connection_waits_for_every_earlier_close_not_just_the_latest()
+    {
+        // A is closing (its Forward Close is held); B is acquired and released without ever opening, so
+        // its own close is instant; C must still not open until A's close has finished.
+        await using var target = new FakeCipTarget();
+        var closeHeld = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        target.ForwardCloseGate = () => closeHeld.Task;
+        var time = new ManualTimeProvider();
+        var pool = new DeviceSessionPool(time);
+        var key = new DeviceKey("127.0.0.1", target.Port, "1,0", "ControlLogix");
+
+        var a = pool.Acquire(key, Options());
+        await a.Scheduler.SendAsync(ReadTag, TestCt);
+        var aClosing = a.DisposeAsync().AsTask();
+        await FakeCipTarget.Until(() => target.ForwardCloses == 1, Guard, "A's Forward Close");
+
+        await pool.Acquire(key, Options()).DisposeAsync();            // B: never opened
+        time.Advance(TimeSpan.FromSeconds(5));                        // the throttle is no obstacle for C
+
+        await using var c = pool.Acquire(key, Options());
+        var cSend = c.Scheduler.SendAsync(ReadTag, TestCt);
+        await Task.Delay(300, TestCt);
+        target.ForwardOpens.ShouldBe(1);                              // C is held back behind A
+        cSend.IsCompleted.ShouldBeFalse();
+
+        closeHeld.SetResult();
+        await aClosing.WaitAsync(Guard, TestCt);
+        (await cSend.WaitAsync(Guard, TestCt)).GeneralStatus.ShouldBe((byte)0);
+        target.ForwardOpens.ShouldBe(2);
+    }
+
+    [Fact]
+    public async Task Disposing_during_an_open_still_forward_closes_what_the_target_accepted()
+    {
+        // The scheduler's stop token (and any caller's) ends only the WAIT; the Forward Open on the wire
+        // completes, and the connection it created is closed.
+        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");
+        var disposing = lease.DisposeAsync().AsTask();
+
+        var failed = await Should.ThrowAsync<AbTransportException>(() => send.WaitAsync(Guard, TestCt));
+        failed.Status.ShouldBe(AbStatus.BadConnection);               // the scheduler's waiter is released at once
+        target.ForwardCloses.ShouldBe(0);
+
+        openHeld.SetResult();                                         // the target accepts, late
+        await disposing.WaitAsync(Guard, TestCt);
+
+        target.ForwardCloses.ShouldBe(1);
+        target.ForwardCloseRequests.Single().AsSpan(8, 8).ToArray()
+            .ShouldBe(target.ForwardOpenRequests.Single().AsSpan(16, 8).ToArray());
+        target.ConnectedRequests.ShouldBe(0);
+    }
+
+    [Fact]
+    public async Task The_shared_pool_logger_can_be_configured()
+    {
+        var logger = new CapturingLogger();
+        DeviceSessionPool.ConfigureSharedLogger(new SingleLoggerFactory(logger));
+        try
+        {
+            var key = new DeviceKey($"shared-logger-{Guid.NewGuid():N}", 44818, "1,0", "ControlLogix");
+            await using var a = DeviceSessionPool.Shared.Acquire(key, Options(4002));
+            await using var b = DeviceSessionPool.Shared.Acquire(key, Options(504));
+            logger.Warnings.ShouldBe(1);
+        }
+        finally
+        {
+            DeviceSessionPool.ConfigureSharedLogger(Microsoft.Extensions.Logging.Abstractions.NullLoggerFactory.Instance);
+        }
+    }
+
+    private sealed class SingleLoggerFactory(ILogger logger) : ILoggerFactory
+    {
+        public ILogger CreateLogger(string categoryName) => logger;
+        public void AddProvider(ILoggerProvider provider) { }
+        public void Dispose() { }
+    }
+
    [Fact]
    public void Shared_is_one_process_wide_instance() =>
        DeviceSessionPool.Shared.ShouldBeSameAs(DeviceSessionPool.Shared);
@@ -37,6 +37,12 @@ internal sealed class FakeCipTarget : IAsyncDisposable
    /// <summary>Connected requests only: may delay, or return null to never answer. Defaults to <see cref="ServiceReply"/>.</summary>
    public Func<byte[], Task<byte[]?>>? ConnectedHandler { get; set; }

+    /// <summary>Awaited before a Forward Open is answered (the request is already counted).</summary>
+    public Func<Task>? ForwardOpenGate { get; set; }
+
+    /// <summary>Awaited before a Forward Close is answered (the request is already counted).</summary>
+    public Func<Task>? ForwardCloseGate { 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; } }
@@ -88,11 +94,13 @@ internal sealed class FakeCipTarget : IAsyncDisposable
        if (IsForwardOpen(cip))
        {
            lock (_forwardOpens) _forwardOpens.Add(cip);
+            if (ForwardOpenGate is { } openGate) await openGate();
            return FakeEipServer.RRReply(frame, ForwardOpenReply(cip));
        }
        if (IsForwardClose(cip))
        {
            lock (_forwardCloses) _forwardCloses.Add(cip);
+            if (ForwardCloseGate is { } closeGate) await closeGate();
            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());
        }
@@ -15,9 +15,13 @@ internal sealed class ManualTimeProvider : TimeProvider

    public DateTimeOffset Start { get; }

-    public override DateTimeOffset GetUtcNow() { lock (_lock) return _now; }
+    /// <summary>Steps the wall clock (<see cref="GetUtcNow"/>) without moving monotonic time
+    /// (<see cref="GetTimestamp"/>) or any timer, as an NTP or operator clock change does.</summary>
+    public TimeSpan WallClockOffset { get; set; }

-    public override long GetTimestamp() => GetUtcNow().UtcTicks;
+    public override DateTimeOffset GetUtcNow() { lock (_lock) return _now + WallClockOffset; }
+
+    public override long GetTimestamp() { lock (_lock) return _now.UtcTicks; }

    public override long TimestampFrequency => TimeSpan.TicksPerSecond;