fix(ab-transport): per-pool diagnostics + per-device open-attempt limit, CIP-level close recycle, CipConnection/pool test gaps (#566, #581)

- Process-wide CipConnection statics (ForwardOpenCount, Forward Close counts, RequestsSent, exchange stats, OpenAttempts, OpenAttemptBudget) move to an instance ConnectionDiagnostics owned by each DeviceSessionPool (and by a stand-alone CipConnection). The open-attempt budget becomes a disposable per-DeviceKey limit (LimitOpenAttempts); the differential harness scopes one to the press under test. - LibplctagOrphanRelease.ForceRootingForTest is gone: ReleaseIfOrphaned(tag, canReadHandles) overload. - A connected reply with general 0x01 / extended 0x0107 or 0x0203 (target no longer holds the class-3 connection, TCP still up) retires the session at once; the next send reopens through the throttle. - New tests: dispose during RegisterSession, abandoned-open observer (falsified), scheduler-dispose-throws guard (DeviceSessionPool.CloseAsync), release-in-background with a shared open in flight, per-device limit scoping, CIP-close recycle (request + keep-alive + negative control). Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-07 23:21:02 -04:00
parent 501f6ea4e5
commit 3edff3304b
18 changed files with 695 additions and 188 deletions
@@ -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;

-/// <summary>Serializes every test class that asserts on <see cref="CipConnection"/>'s process-wide
-/// diagnostic counters, so no parallel test can move them mid-assertion.</summary>
+/// <summary>Serializes the loopback-socket test classes. Their diagnostic counters are per connection / per
+/// pool since Gitea #581 (no longer process-wide statics), but <see cref="DeviceSessionPoolTests"/> still
+/// reconfigures <see cref="DeviceSessionPool.Shared"/>'s logger, and the timing-sensitive storm test is kept
+/// off a thread pool its neighbours are saturating.</summary>
 [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<AbTransportException>(() => 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<AbTransportException>(() => limitedLease.Scheduler.SendAsync(ReadTag, TestCt));
+            refused.Status.ShouldBe(AbStatus.Abort);
+            refused.Message.ShouldContain("budget");
+            Should.Throw<InvalidOperationException>(() => 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);
    }

@@ -411,4 +442,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<AbTransportException>(() => 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<OperationCanceledException>(() => 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<byte[]?>(
+            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<AbTransportException>(() => 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<byte[]?>(
+            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<Func<byte[], byte[]>>(
+        [
+            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<byte[]?>(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();
+    }
 }
@@ -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);
@@ -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);
    }

+    /// <summary>
+    ///     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) — <see cref="ConnectionLease.ReleaseInBackground"/> (#579) returns at
+    ///     once, and the teardown it hands back still Forward Closes what the open produced once it lands.
+    /// </summary>
+    [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<AbTransportException>(() => 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
+    }
+
+    /// <summary>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.</summary>
+    [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);
@@ -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<Exception>();
-        EventHandler<UnobservedTaskExceptionEventArgs> 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.
@@ -61,6 +61,9 @@ internal sealed class FakeEipServer : IAsyncDisposable
    /// <summary>Synchronous convenience setter for <see cref="HandlerAsync"/>.</summary>
    public Func<byte[], byte[]?> Handler { set => HandlerAsync = req => Task.FromResult(value(req)); }

+    /// <summary>Awaited before a RegisterSession is answered (the request is already recorded).</summary>
+    public Func<Task>? RegisterSessionGate { get; set; }
+
    /// <summary>Never answer non-session requests.</summary>
    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;
@@ -0,0 +1,54 @@
+using System.Collections.Concurrent;
+using System.Runtime.CompilerServices;
+namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Fakes;
+
+/// <summary>
+///     Collects <see cref="TaskScheduler.UnobservedTaskException"/>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.
+/// </summary>
+internal static class UnobservedTaskProbe
+{
+    public static async Task<IReadOnlyList<Exception>> CollectAsync(Func<Task> abandon, TimeSpan timeout)
+    {
+        var marker = $"unobserved-task-probe-control-{Guid.NewGuid():N}";
+        var seen = new ConcurrentQueue<Exception>();
+        var controlReported = 0;
+        EventHandler<UnobservedTaskExceptionEventArgs> 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));
+}
@@ -92,9 +92,9 @@ internal sealed class DifferentialAbortedException(string message, Exception? in
 ///     <para><b>Stop on first failure (review r1 C1).</b> 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 (<see cref="CipConnection.OpenAttemptBudget"/>), so a reconnect cannot reach the wire even
+///     to the press under test (<see cref="ConnectionDiagnostics.LimitOpenAttempts"/>), so a reconnect cannot reach the wire even
 ///     from inside an enumerator that swallows a per-program failure, and the run aborts the moment
-///     <see cref="CipConnection.ForwardOpenCount"/> passes 1.</para>
+///     <see cref="ConnectionDiagnostics.ForwardOpens"/> passes 1.</para>
 /// </summary>
 internal static class DifferentialRun
 {
@@ -222,9 +222,9 @@ internal static class DifferentialRun

    /// <summary>
    ///     Runs <paramref name="body"/> 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 <paramref name="body"/> 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 <paramref name="body"/> 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
    ///     <c>finally</c>. Shared by the differential and the latency comparison.
    /// </summary>
    /// <returns>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<Action<string>, Task<T>> 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);
    }

@@ -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");
    }

+    /// <summary>
+    ///     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.
+    /// </summary>
+    private static async Task ControlOpenAsync(DifferentialTarget target)
+    {
+        var p = target.Params("@tags");
+        await using var lease = DeviceSessionPool.Shared.Acquire(NativeAbCipDevice.Key(p), NativeAbCipDevice.Options(p));
+        await Should.ThrowAsync<AbTransportException>(
+            () => lease.Scheduler.SendAsync([0x4C, 0x00], 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<AbStatus>();
        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]
@@ -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);
@@ -5,13 +5,13 @@ using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip;
 namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests.Differential;

 /// <summary>One timed burst: every tag read once, concurrently.</summary>
-/// <param name="Requests">Native only: <see cref="CipConnection.RequestsSent"/> delta over the pass (keep-alives included).</param>
+/// <param name="Requests">Native only: <see cref="ConnectionDiagnostics.RequestsSent"/> delta over the pass (keep-alives included).</param>
 /// <param name="TagFailures">Reads that ended with a tag-level status (a connection-level one aborts instead).</param>
 /// <param name="Gc">Collections that ran during the pass, per generation (T21 review r1 M8: a client-side stall shows here).</param>
 /// <param name="Rtt">Native only: round trip of each CIP exchange in the pass, from <see cref="CipConnection"/> (a slow
 ///     controller shows here; a slow pass with fast exchanges points at the client).</param>
 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);

 /// <summary>GC collections per generation over a window (gen0 counts include gen1/gen2 collections, as the runtime reports).</summary>
 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++)
@@ -18,7 +18,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests;
 [CollectionDefinition(Name, DisableParallelization = true)]
 public sealed class LibplctagOrphanReleaseCollection
 {
-    /// <summary>The release counters and the rooting seam are process-wide: nothing else may run alongside.</summary>
+    /// <summary>The release counters are process-wide: nothing else may run alongside.</summary>
    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<AbTransportException>(() => rt.InitializeAsync(CancellationToken.None));
+            await Should.ThrowAsync<Exception>(() => 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();
        }
    }