diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/CannedAgentClient.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/CannedAgentClient.cs
index 164457a8..1e9ec961 100644
--- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/CannedAgentClient.cs
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/CannedAgentClient.cs
@@ -178,6 +178,13 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
///
public TaskCompletionSource? CurrentGate { get; set; }
+ ///
+ /// When set, a /sample enumeration that is cancelled or whose client is disposed does NOT end until this
+ /// source completes — a pump that cannot unwind (a wedged socket read). The deterministic way to keep a retired
+ /// pump alive past the point a lifecycle restart retires it.
+ ///
+ public TaskCompletionSource? HoldSampleExit { get; set; }
+
///
/// Signalled the moment a /current request lands — before
/// parks it. This is the deterministic "the caller is now inside
@@ -360,6 +367,11 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable
{
scripted = await chunks.Reader.ReadAsync(lifetime.Token).ConfigureAwait(false);
}
+ catch (Exception ex) when (HoldSampleExit is { } hold && ex is OperationCanceledException or ChannelClosedException)
+ {
+ await hold.Task.ConfigureAwait(false);
+ throw;
+ }
catch (ChannelClosedException)
{
// Mirrors the production client: a stream that stops for any reason other than the
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectPumpGenerationTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectPumpGenerationTests.cs
index 476b8d53..2c093abb 100644
--- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectPumpGenerationTests.cs
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectPumpGenerationTests.cs
@@ -37,6 +37,7 @@ public sealed class MTConnectPumpGenerationTests
await driver.InitializeAsync("{}", Ct);
Task[] retired = [];
+ var holdStalePump = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var blockOnce = 0;
var blocking = new ManualResetEventSlim(false);
var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
@@ -68,7 +69,10 @@ public sealed class MTConnectPumpGenerationTests
(++spins).ShouldBeLessThan(500, "restart never cleared the pump slot");
}
- // A second subscriber arrives mid-restart: it starts a pump on the (still-live) old client.
+ // A second subscriber arrives mid-restart: it starts a pump on the (still-live) old client. That pump is
+ // held alive past the retire point (it cannot unwind until released), so the retention is observable
+ // deterministically rather than racing the pump's own exit.
+ clientA.HoldSampleExit = holdStalePump;
await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
var staleSessionPump = driver.SampleStreamTask.ShouldNotBeNull();
await clientA.WaitForSampleCallsAsync(2).WaitAsync(Watchdog, Ct);
@@ -82,8 +86,8 @@ public sealed class MTConnectPumpGenerationTests
// pump also stops by itself once it sees its session generation is gone, so it may already have
// finished before the restart got to retire it — either way it is not left running unjoined.
retired = driver.SupersededPumpTasks;
- (retired.Contains(staleSessionPump) || staleSessionPump.IsCompleted).ShouldBeTrue(
- "the stale-session pump must be retained for shutdown to join, or already be gone");
+ staleSessionPump.IsCompleted.ShouldBeFalse("positive control: the stale pump is still running here");
+ retired.ShouldContain(staleSessionPump, "a retired pump that is still running must be retained for shutdown to join");
// The restart installed a new client; the standing subscription must be streaming from IT.
await clientB.WaitForSampleCallsAsync(1).WaitAsync(Watchdog, Ct);
@@ -92,6 +96,7 @@ public sealed class MTConnectPumpGenerationTests
finally
{
blocking.Set();
+ holdStalePump.TrySetResult();
await driver.ShutdownAsync(CancellationToken.None);
}