From 66d099c7201122f1ce31af3d48230b84c38f77c1 Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Fri, 31 Jul 2026 13:28:34 -0600 Subject: [PATCH 1/5] feat(device): cancellable async connect/disconnect and IAsyncDisposable (closes #341) Adds ConnectAsync/DisconnectAsync to the device surface, implements IAsyncDisposable on DaqifiDevice, and threads a CancellationToken through IStreamTransport.ConnectAsync so an in-flight dial can be abandoned. The new transport overloads ship as default interface implementations that forward to the existing uncancellable ones, so third-party IStreamTransport / IDevice implementations keep compiling unchanged. The synchronous Connect, Disconnect and Dispose remain and behave exactly as before. Co-Authored-By: Claude Opus 5 --- README.md | 10 +- docs/DEVICE_INTERFACES.md | 50 +- .../Transport/ConnectRetryExecutorTests.cs | 101 +++- .../StreamTransportCancellationTests.cs | 149 ++++++ .../Device/DaqifiDeviceAsyncLifecycleTests.cs | 443 ++++++++++++++++++ .../Transport/ConnectRetryExecutor.cs | 31 +- .../Transport/IStreamTransport.cs | 28 ++ .../Transport/SerialStreamTransport.cs | 29 +- .../Transport/TcpStreamTransport.cs | 41 +- src/Daqifi.Core/Device/DaqifiDevice.cs | 432 +++++++++++++---- src/Daqifi.Core/Device/DaqifiDeviceFactory.cs | 9 +- src/Daqifi.Core/Device/IDevice.cs | 58 ++- 12 files changed, 1249 insertions(+), 132 deletions(-) create mode 100644 src/Daqifi.Core.Tests/Communication/Transport/StreamTransportCancellationTests.cs create mode 100644 src/Daqifi.Core.Tests/Device/DaqifiDeviceAsyncLifecycleTests.cs diff --git a/README.md b/README.md index 52bcff27..de3623fe 100644 --- a/README.md +++ b/README.md @@ -37,7 +37,7 @@ using Daqifi.Core.Channel; // Connect — transport and device initialization handled for you. The factory returns the base // DaqifiDevice type, but the constructed instance is always a DaqifiStreamingDevice. -using var device = (DaqifiStreamingDevice)await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); +await using var device = (DaqifiStreamingDevice)await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); // Subscribe to decoded, per-channel samples var ai0 = device.GetChannelsSnapshot().First(c => c.Type == ChannelType.Analog && c.ChannelNumber == 0); @@ -97,7 +97,7 @@ Pick whichever transport fits your setup — each snippet is a standalone, copy- **TCP with a resilient retry preset** (5 retries, longer timeouts): ```csharp -using var device = await DaqifiDeviceFactory.ConnectTcpAsync( +await using var device = await DaqifiDeviceFactory.ConnectTcpAsync( "192.168.1.100", 9760, DeviceConnectionOptions.Resilient); ``` @@ -106,7 +106,7 @@ using var device = await DaqifiDeviceFactory.ConnectTcpAsync( ```csharp // Replace with your OS-specific port: // Windows: "COM3" • macOS: "/dev/cu.usbmodem1" • Linux: "/dev/ttyACM0" -using var device = await DaqifiDeviceFactory.ConnectSerialAsync("COM3"); +await using var device = await DaqifiDeviceFactory.ConnectSerialAsync("COM3"); ``` **From a discovered device:** @@ -114,7 +114,7 @@ using var device = await DaqifiDeviceFactory.ConnectSerialAsync("COM3"); ```csharp using var finder = new WiFiDeviceFinder(); var devices = await finder.DiscoverAsync(TimeSpan.FromSeconds(5)); -using var device = await DaqifiDeviceFactory.ConnectFromDeviceInfoAsync(devices.First()); +await using var device = await DaqifiDeviceFactory.ConnectFromDeviceInfoAsync(devices.First()); ``` ### Custom retry options @@ -132,7 +132,7 @@ var options = new DeviceConnectionOptions }, InitializeDevice = true }; -using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760, options); +await using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760, options); ``` ### Device discovery diff --git a/docs/DEVICE_INTERFACES.md b/docs/DEVICE_INTERFACES.md index 4c234f61..35348093 100644 --- a/docs/DEVICE_INTERFACES.md +++ b/docs/DEVICE_INTERFACES.md @@ -15,7 +15,7 @@ using Daqifi.Core.Device; using Daqifi.Core.Communication.Producers; // Connect to a device (handles transport, connection, and initialization) -using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); +await using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); // Subscribe to incoming data device.MessageReceived += (sender, e) => @@ -141,7 +141,7 @@ foreach (var deviceInfo in devices) // Connect to the first discovered device if (devices.Any()) { - using var device = await DaqifiDeviceFactory.ConnectFromDeviceInfoAsync(devices.First()); + await using var device = await DaqifiDeviceFactory.ConnectFromDeviceInfoAsync(devices.First()); // Device is ready to use } ``` @@ -334,25 +334,47 @@ For cases where you need more control over the connection process: using Daqifi.Core.Device; using Daqifi.Core.Communication.Transport; -// Create and connect transport manually +// Create and connect transport manually. Every step takes a CancellationToken, so a user who +// cancels stops the attempt where it stands instead of waiting out the retries. var transport = new TcpStreamTransport("192.168.1.100", 9760); -await transport.ConnectAsync(new ConnectionRetryOptions { MaxAttempts = 3 }); +await transport.ConnectAsync(new ConnectionRetryOptions { MaxAttempts = 3 }, cancellationToken); // Create device with transport -using var device = new DaqifiDevice("My Device", transport); -device.Connect(); -await device.InitializeAsync(); +await using var device = new DaqifiDevice("My Device", transport); +await device.ConnectAsync(cancellationToken); +await device.InitializeAsync(cancellationToken: cancellationToken); // Now ready to send commands device.Send(ScpiMessageProducer.GetDeviceInfo); ``` +### Connecting and disconnecting without blocking + +`ConnectAsync`, `DisconnectAsync` and `DisposeAsync` are the non-blocking forms of `Connect`, +`Disconnect` and `Dispose`. Prefer them on a UI thread: the synchronous disconnect waits (up to ten +seconds) for any command exchange still in flight, which on a UI thread is a visible freeze. + +```csharp +// Disposal never blocks the caller — this is the recommended pattern. +await using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); +``` + +A cancelled `ConnectAsync` throws `OperationCanceledException` and leaves nothing half-open — if the +transport had already come up when the cancel landed, it is closed again. + +`DisconnectAsync` treats its token differently, and deliberately: cancelling it skips the wait for an +in-flight command exchange and proceeds straight to teardown. It never aborts the disconnect and +never throws `OperationCanceledException`, because a teardown abandoned part-way would leave the +device in an indeterminate state. On return the device is always disconnected. + +The synchronous `Connect`, `Disconnect` and `Dispose` remain, unchanged, for existing callers. + ### Error Handling ```csharp try { - using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); + await using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); device.Send(ScpiMessageProducer.StartStreaming(100)); } catch (ArgumentOutOfRangeException ex) when (ex.ParamName == "port") @@ -419,7 +441,7 @@ Notes: ### Connection Status Monitoring ```csharp -using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); +await using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); device.StatusChanged += (sender, args) => { @@ -495,7 +517,7 @@ reader and writer loops report every read/write outcome to a transport that does After initialization, device metadata is populated: ```csharp -using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); +await using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); // Access device information Console.WriteLine($"Part Number: {device.Metadata.PartNumber}"); @@ -530,7 +552,7 @@ using Daqifi.Core.Device; // DaqifiDeviceFactory methods return the base DaqifiDevice type, but the constructed instance is // always a DaqifiStreamingDevice — cast (or pattern-match with `is`) to reach its streaming API. -using var device = (DaqifiStreamingDevice)await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); +await using var device = (DaqifiStreamingDevice)await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); var ai0 = device.GetChannelsSnapshot().First(c => c.Type == ChannelType.Analog && c.ChannelNumber == 0); ai0.SampleReceived += (sender, e) => @@ -562,7 +584,7 @@ cancellation and backpressure instead of hand-building an event/queue bridge. Ea the decoded `IDataSample` with the `IChannel` that produced it. ```csharp -using var device = (DaqifiStreamingDevice)await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); +await using var device = (DaqifiStreamingDevice)await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); device.EnableChannels(device.GetChannelsSnapshot().Where(c => c.Type == ChannelType.Analog)); device.StreamingFrequency = 100; // Hz @@ -585,7 +607,7 @@ call `StopStreaming()` for that. This is additive: `SampleReceived` and `Message #### Raw protobuf frames ```csharp -using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); +await using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); var sampleCount = 0; device.MessageReceived += (sender, e) => @@ -697,7 +719,7 @@ to reach these members — the same way as `INetworkConfigurable` below. ```csharp using Daqifi.Core.Device.Diagnostics; -using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); +await using var device = await DaqifiDeviceFactory.ConnectTcpAsync("192.168.1.100", 9760); if (device is IDeviceDiagnostics diagnostics) { diff --git a/src/Daqifi.Core.Tests/Communication/Transport/ConnectRetryExecutorTests.cs b/src/Daqifi.Core.Tests/Communication/Transport/ConnectRetryExecutorTests.cs index 624f1349..d98917bf 100644 --- a/src/Daqifi.Core.Tests/Communication/Transport/ConnectRetryExecutorTests.cs +++ b/src/Daqifi.Core.Tests/Communication/Transport/ConnectRetryExecutorTests.cs @@ -30,7 +30,7 @@ public async Task ExecuteAsync_SucceedsFirstAttempt_ReportsConnectedOnce() await ConnectRetryExecutor.ExecuteAsync( FastRetry(3), - connectAttempt: _ => { attempts++; return Task.CompletedTask; }, + connectAttempt: (_, _) => { attempts++; return Task.CompletedTask; }, onAttemptFailed: () => failedCleanups++, onStatusChanged: (c, e) => statuses.Add((c, e))); @@ -49,7 +49,7 @@ public async Task ExecuteAsync_ReceivesResolvedOptions_InConnectAttempt() await ConnectRetryExecutor.ExecuteAsync( options, - connectAttempt: o => { seen = o; return Task.CompletedTask; }, + connectAttempt: (o, _) => { seen = o; return Task.CompletedTask; }, onAttemptFailed: () => { }, onStatusChanged: (_, _) => { }); @@ -66,7 +66,7 @@ public async Task ExecuteAsync_NullOptions_UsesSingleNoRetryAttempt() var thrown = await Assert.ThrowsAsync(() => ConnectRetryExecutor.ExecuteAsync( retryOptions: null, - connectAttempt: _ => { attempts++; throw boom; }, + connectAttempt: (_, _) => { attempts++; throw boom; }, onAttemptFailed: () => failedCleanups++, onStatusChanged: (_, _) => { })); @@ -84,7 +84,7 @@ public async Task ExecuteAsync_RetriesUntilSuccess_CleansUpEachFailedAttempt() await ConnectRetryExecutor.ExecuteAsync( FastRetry(3), - connectAttempt: _ => + connectAttempt: (_, _) => { attempts++; if (attempts < 3) @@ -120,7 +120,7 @@ public async Task ExecuteAsync_AllAttemptsFail_ThrowsLastExceptionAndReportsItUn var thrown = await Assert.ThrowsAsync(() => ConnectRetryExecutor.ExecuteAsync( FastRetry(3), - connectAttempt: _ => + connectAttempt: (_, _) => { attempts++; throw attempts < 3 ? new InvalidOperationException($"fail {attempts}") : lastBoom; @@ -155,7 +155,7 @@ public async Task ExecuteAsync_RetryDisabledWithMultipleMaxAttempts_TriesOnce() await Assert.ThrowsAsync(() => ConnectRetryExecutor.ExecuteAsync( options, - connectAttempt: _ => { attempts++; throw new InvalidOperationException("boom"); }, + connectAttempt: (_, _) => { attempts++; throw new InvalidOperationException("boom"); }, onAttemptFailed: () => { }, onStatusChanged: (_, _) => { })); @@ -179,7 +179,7 @@ public async Task ExecuteAsync_AppliesBackoffDelayBetweenAttempts() await ConnectRetryExecutor.ExecuteAsync( options, - connectAttempt: _ => + connectAttempt: (_, _) => { attempts++; if (attempts == 1) @@ -198,4 +198,91 @@ await ConnectRetryExecutor.ExecuteAsync( Assert.True(sw.ElapsedMilliseconds >= 80, $"Expected a backoff delay before retry, but only {sw.ElapsedMilliseconds}ms elapsed."); } + + [Fact] + public async Task ExecuteAsync_TokenAlreadyCanceled_NeverAttempts() + { + var attempts = 0; + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + + await Assert.ThrowsAnyAsync(() => + ConnectRetryExecutor.ExecuteAsync( + FastRetry(3), + connectAttempt: (_, _) => { attempts++; return Task.CompletedTask; }, + onAttemptFailed: () => { }, + onStatusChanged: (_, _) => { }, + cancellationToken: cts.Token)); + + Assert.Equal(0, attempts); + } + + [Fact] + public async Task ExecuteAsync_CanceledAttempt_IsNotRetried() + { + // A cancelled attempt is the caller walking away, not a transient failure — the loop must + // stop rather than burn through the remaining attempts. This is what lets an auto-reconnect + // loop (issue #379) be torn down promptly. + var attempts = 0; + var failedCleanups = 0; + var statuses = new List<(bool Connected, Exception? Error)>(); + using var cts = new CancellationTokenSource(); + + await Assert.ThrowsAnyAsync(() => + ConnectRetryExecutor.ExecuteAsync( + FastRetry(5), + connectAttempt: (_, token) => + { + attempts++; + cts.Cancel(); + token.ThrowIfCancellationRequested(); + return Task.CompletedTask; + }, + onAttemptFailed: () => failedCleanups++, + onStatusChanged: (c, e) => statuses.Add((c, e)), + cancellationToken: cts.Token)); + + Assert.Equal(1, attempts); + // The half-built handle is still cleaned up, and the transport is reported disconnected. + Assert.Equal(1, failedCleanups); + var status = Assert.Single(statuses); + Assert.False(status.Connected); + Assert.IsAssignableFrom(status.Error); + } + + [Fact] + public async Task ExecuteAsync_CanceledDuringBackoffDelay_StopsWaiting() + { + var options = new ConnectionRetryOptions + { + Enabled = true, + MaxAttempts = 3, + InitialDelay = TimeSpan.FromSeconds(30), + MaxDelay = TimeSpan.FromSeconds(30), + BackoffMultiplier = 1.0 + }; + + var attempts = 0; + using var cts = new CancellationTokenSource(); + var sw = Stopwatch.StartNew(); + + var run = ConnectRetryExecutor.ExecuteAsync( + options, + connectAttempt: (_, _) => + { + attempts++; + throw new InvalidOperationException("boom"); + }, + onAttemptFailed: () => { }, + onStatusChanged: (_, _) => { }, + cancellationToken: cts.Token); + + cts.CancelAfter(TimeSpan.FromMilliseconds(100)); + await Assert.ThrowsAnyAsync(() => run); + sw.Stop(); + + Assert.Equal(1, attempts); + Assert.True(sw.Elapsed < TimeSpan.FromSeconds(10), + $"The 30s backoff delay was not cancellable ({sw.ElapsedMilliseconds}ms elapsed)."); + } } diff --git a/src/Daqifi.Core.Tests/Communication/Transport/StreamTransportCancellationTests.cs b/src/Daqifi.Core.Tests/Communication/Transport/StreamTransportCancellationTests.cs new file mode 100644 index 00000000..f25051a9 --- /dev/null +++ b/src/Daqifi.Core.Tests/Communication/Transport/StreamTransportCancellationTests.cs @@ -0,0 +1,149 @@ +using Daqifi.Core.Communication.Transport; +using System.Diagnostics; +using System.Net; + +namespace Daqifi.Core.Tests.Communication.Transport; + +/// +/// Pins the cancellation contract added to +/// in issue #341: an in-flight dial can be abandoned, a caller-driven cancel is never disguised as +/// a connect timeout, and a transport written against the pre-#341 interface still works. +/// +public class StreamTransportCancellationTests +{ + [Fact] + public async Task TcpConnectAsync_CanceledMidAttempt_ThrowsOperationCanceledPromptly() + { + // A never-completing connect task plus a 30s timeout: without the token being honored this + // would sit here for the full half-minute. + using var transport = new TcpStreamTransport(IPAddress.Parse("192.0.2.1"), 9760); + transport.ConnectTaskFactory = _ => Task.Delay(Timeout.Infinite); + var options = new ConnectionRetryOptions + { + Enabled = false, + MaxAttempts = 1, + ConnectionTimeout = TimeSpan.FromSeconds(30) + }; + + using var cts = new CancellationTokenSource(); + var connect = transport.ConnectAsync(options, cts.Token); + + var sw = Stopwatch.StartNew(); + cts.CancelAfter(TimeSpan.FromMilliseconds(100)); + + var thrown = await Assert.ThrowsAnyAsync(() => connect); + sw.Stop(); + + Assert.IsNotType(thrown); + Assert.True(sw.Elapsed < TimeSpan.FromSeconds(10), + $"Cancellation took {sw.ElapsedMilliseconds}ms; the connect timeout was not short-circuited."); + Assert.False(transport.IsConnected); + } + + [Fact] + public async Task TcpConnectAsync_TokenAlreadyCanceled_NeverStartsAnAttempt() + { + using var transport = new TcpStreamTransport(IPAddress.Parse("192.0.2.1"), 9760); + var attempts = 0; + transport.ConnectTaskFactory = _ => + { + attempts++; + return Task.Delay(Timeout.Infinite); + }; + + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + + await Assert.ThrowsAnyAsync(() => transport.ConnectAsync(cts.Token)); + Assert.Equal(0, attempts); + } + + [Fact] + public async Task TcpConnectAsync_TimeoutWithALiveToken_StillSurfacesAsTimeoutException() + { + // Regression guard for the linked cancellation source: now that the caller's token shares a + // source with the timeout, the two must still be told apart. A timeout is a device problem + // the caller must see as such (daqifi-desktop#517), not a TaskCanceledException. + using var transport = new TcpStreamTransport(IPAddress.Parse("192.0.2.1"), 9760); + transport.ConnectTaskFactory = _ => Task.Delay(Timeout.Infinite); + var options = new ConnectionRetryOptions + { + Enabled = false, + MaxAttempts = 1, + ConnectionTimeout = TimeSpan.FromMilliseconds(250) + }; + + using var cts = new CancellationTokenSource(); + + var ex = await Assert.ThrowsAsync(() => transport.ConnectAsync(options, cts.Token)); + Assert.Contains("192.0.2.1:9760", ex.Message); + Assert.IsAssignableFrom(ex.InnerException); + } + + [Fact] + public async Task SerialConnectAsync_TokenAlreadyCanceled_ThrowsWithoutOpeningThePort() + { + // A port name nothing can open: if the token were ignored the attempt would fail with an + // IO/argument exception from Open() instead of the cancellation the caller asked for. + using var transport = new SerialStreamTransport("/dev/null-daqifi-does-not-exist"); + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + + await Assert.ThrowsAnyAsync(() => transport.ConnectAsync(cts.Token)); + Assert.False(transport.IsConnected); + } + + [Fact] + public async Task LegacyTransport_WrittenBeforeTheTokenExisted_StillConnectsThroughTheNewOverload() + { + // Source-compatibility guard: this transport implements only the pre-#341 members. It must + // keep compiling, and the cancellable overload must fall back to the uncancellable one. + IStreamTransport transport = new LegacyStreamTransport(); + + await transport.ConnectAsync(CancellationToken.None); + + Assert.True(transport.IsConnected); + transport.Dispose(); + } + + /// + /// An implementation frozen at the pre-#341 shape — it does not + /// override either cancellable overload, so it exercises their default implementations. + /// + private sealed class LegacyStreamTransport : IStreamTransport + { + private readonly MemoryStream _stream = new(); + private bool _isConnected; + + public Stream Stream => _stream; + public bool IsConnected => _isConnected; + public string ConnectionInfo => "Legacy"; + + public event EventHandler? StatusChanged; + + public Task ConnectAsync() => ConnectAsync(null); + + public Task ConnectAsync(ConnectionRetryOptions? retryOptions) + { + _isConnected = true; + StatusChanged?.Invoke(this, new TransportStatusEventArgs(true, ConnectionInfo)); + return Task.CompletedTask; + } + + public Task DisconnectAsync() + { + _isConnected = false; + return Task.CompletedTask; + } + + public void Connect() => ConnectAsync().GetAwaiter().GetResult(); + + public void Disconnect() => DisconnectAsync().GetAwaiter().GetResult(); + + public void Dispose() + { + _isConnected = false; + _stream.Dispose(); + } + } +} diff --git a/src/Daqifi.Core.Tests/Device/DaqifiDeviceAsyncLifecycleTests.cs b/src/Daqifi.Core.Tests/Device/DaqifiDeviceAsyncLifecycleTests.cs new file mode 100644 index 00000000..e7ed7037 --- /dev/null +++ b/src/Daqifi.Core.Tests/Device/DaqifiDeviceAsyncLifecycleTests.cs @@ -0,0 +1,443 @@ +using Daqifi.Core.Communication.Producers; +using Daqifi.Core.Communication.Transport; +using Daqifi.Core.Device; +using System.Diagnostics; + +namespace Daqifi.Core.Tests.Device; + +/// +/// Covers the cancellable async connect/disconnect surface and +/// added in issue #341, together with the guarantee that the pre-existing synchronous entry +/// points still behave exactly as they did. +/// +public class DaqifiDeviceAsyncLifecycleTests +{ + [Fact] + public async Task ConnectAsync_Succeeds_ReportsConnectedAndStartsSending() + { + using var transport = new GatedMockTransport(); + transport.OpenConnectGate(); + await using var device = new DaqifiDevice("Mock Device", transport); + + var statusChanges = new List(); + device.StatusChanged += (_, args) => statusChanges.Add(args.Status); + + await device.ConnectAsync(); + + Assert.True(device.IsConnected); + Assert.Equal(ConnectionStatus.Connected, device.Status); + Assert.Equal(DeviceState.Connected, device.State); + Assert.Equal(new[] { ConnectionStatus.Connecting, ConnectionStatus.Connected }, statusChanges); + + // The producer really is running over the transport's stream, not just flagged as such. + device.Send(ScpiMessageProducer.GetDeviceInfo); + Assert.True(await transport.WaitForWrittenTextAsync("SYSTem:SYSInfoPB?")); + } + + [Fact] + public async Task ConnectAsync_TokenAlreadyCanceled_NeverTouchesTheTransport() + { + using var transport = new GatedMockTransport(); + using var device = new DaqifiDevice("Mock Device", transport); + using var cts = new CancellationTokenSource(); + cts.Cancel(); + + await Assert.ThrowsAnyAsync(() => device.ConnectAsync(cts.Token)); + + Assert.Equal(0, transport.ConnectAttempts); + Assert.False(device.IsConnected); + // The device never even claimed to be connecting, so no consumer sees a spurious transition. + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + } + + [Fact] + public async Task ConnectAsync_CanceledMidAttempt_AbandonsItAndReportsDisconnected() + { + // The transport blocks inside ConnectAsync until its gate opens — the stand-in for a real + // dial that has not answered yet. Cancelling must break that wait rather than wait it out. + using var transport = new GatedMockTransport(); + using var device = new DaqifiDevice("Mock Device", transport); + using var cts = new CancellationTokenSource(); + + var statusChanges = new List(); + device.StatusChanged += (_, args) => statusChanges.Add(args.Status); + + var connect = device.ConnectAsync(cts.Token); + Assert.True(await transport.WaitForConnectEnteredAsync()); + Assert.False(connect.IsCompleted); + + await cts.CancelAsync(); + + await Assert.ThrowsAnyAsync(() => connect); + Assert.False(device.IsConnected); + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + Assert.Equal(DeviceState.Disconnected, device.State); + Assert.Equal(new[] { ConnectionStatus.Connecting, ConnectionStatus.Disconnected }, statusChanges); + } + + [Fact] + public async Task ConnectAsync_CanceledAfterTheTransportOpened_ClosesItAgain() + { + // A cancel that lands in the window between "transport up" and "pumps started" must not + // leak a live connection owned by a device that reports itself disconnected. + using var transport = new GatedMockTransport(); + transport.OpenConnectGate(); + using var cts = new CancellationTokenSource(); + transport.OnConnected = () => cts.Cancel(); + + using var device = new DaqifiDevice("Mock Device", transport); + + await Assert.ThrowsAnyAsync(() => device.ConnectAsync(cts.Token)); + + Assert.Equal(1, transport.ConnectAttempts); + Assert.Equal(1, transport.DisconnectCount); + Assert.False(transport.IsConnected); + Assert.False(device.IsConnected); + } + + [Fact] + public async Task ConnectAsync_TransportThrows_ReportsDisconnectedAndRethrows() + { + using var transport = new GatedMockTransport { ConnectFailure = new InvalidOperationException("no route") }; + using var device = new DaqifiDevice("Mock Device", transport); + + var thrown = await Assert.ThrowsAsync(() => device.ConnectAsync()); + + Assert.Equal("no route", thrown.Message); + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + Assert.Equal(DeviceState.Disconnected, device.State); + } + + [Fact] + public async Task DisconnectAsync_TearsDownTheTransportAndAllowsReconnect() + { + using var transport = new GatedMockTransport(); + transport.OpenConnectGate(); + using var device = new DaqifiDevice("Mock Device", transport); + + await device.ConnectAsync(); + await device.DisconnectAsync(); + + Assert.Equal(1, transport.DisconnectCount); + Assert.False(transport.IsConnected); + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + Assert.Equal(DeviceState.Disconnected, device.State); + + // Still reusable afterwards — teardown must not have poisoned anything. + await device.ConnectAsync(); + Assert.True(device.IsConnected); + } + + [Fact] + public async Task DisconnectAsync_NeverReportsLost() + { + using var transport = new GatedMockTransport(); + transport.OpenConnectGate(); + using var device = new DaqifiDevice("Mock Device", transport); + await device.ConnectAsync(); + + var statusChanges = new List(); + device.StatusChanged += (_, args) => statusChanges.Add(args.Status); + + await device.DisconnectAsync(); + + Assert.DoesNotContain(ConnectionStatus.Lost, statusChanges); + Assert.Contains(ConnectionStatus.Disconnected, statusChanges); + } + + [Fact] + public async Task DisconnectAsync_WhileATextExchangeHoldsTheLock_CancelingSkipsTheWait() + { + // The acceptance criterion behind #341: the sync Disconnect() can sit on the text-exchange + // lock for its full budget, which on a UI thread is a multi-second freeze. Cancelling the + // async one gives up that courtesy wait immediately and tears down anyway. + using var transport = new GatedMockTransport(); + transport.OpenConnectGate(); + using var device = new TextExchangeProbeDevice("Mock Device", transport); + await device.ConnectAsync(); + + var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var heldExchange = device.HoldTextExchangeAsync(entered, responseTimeoutMs: 3000); + Assert.Same(entered.Task, await Task.WhenAny(entered.Task, Task.Delay(TimeSpan.FromSeconds(10)))); + + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + + var sw = Stopwatch.StartNew(); + var disconnect = device.DisconnectAsync(cts.Token); + var finishedFirst = await Task.WhenAny(disconnect, heldExchange); + sw.Stop(); + + Assert.Same(disconnect, finishedFirst); + await disconnect; + Assert.True(sw.Elapsed < TimeSpan.FromSeconds(2), + $"DisconnectAsync waited {sw.ElapsedMilliseconds}ms despite a cancelled token."); + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + + // Let the abandoned exchange unwind; it is expected to fail now that the device is gone. + try + { + await heldExchange; + } + catch (Exception) + { + // Losing the race with teardown is the documented outcome, not a test failure. + } + } + + [Fact] + public async Task AwaitUsing_DisconnectsAndDisposesTheTransport() + { + var transport = new GatedMockTransport(); + transport.OpenConnectGate(); + + await using (var device = new DaqifiDevice("Mock Device", transport)) + { + await device.ConnectAsync(); + Assert.True(device.IsConnected); + } + + Assert.True(transport.IsDisposed); + Assert.Equal(1, transport.DisconnectCount); + } + + [Fact] + public async Task DisposeAsync_IsIdempotentAndInterchangeableWithDispose() + { + var transport = new GatedMockTransport(); + transport.OpenConnectGate(); + var device = new DaqifiDevice("Mock Device", transport); + await device.ConnectAsync(); + + await device.DisposeAsync(); + await device.DisposeAsync(); + device.Dispose(); + + // Exactly one teardown, no matter how many times (or which way) it is disposed. + Assert.Equal(1, transport.DisconnectCount); + Assert.True(transport.IsDisposed); + } + + [Fact] + public async Task DisposeAsync_AfterSyncDispose_IsANoOp() + { + var transport = new GatedMockTransport(); + transport.OpenConnectGate(); + var device = new DaqifiDevice("Mock Device", transport); + await device.ConnectAsync(); + + device.Dispose(); + await device.DisposeAsync(); + + Assert.Equal(1, transport.DisconnectCount); + } + + // ---- Synchronous parity: the pre-#341 entry points must behave identically ---- + + [Fact] + public void Connect_Disconnect_StillDriveTheTransportSynchronously() + { + using var transport = new GatedMockTransport(); + transport.OpenConnectGate(); + using var device = new DaqifiDevice("Mock Device", transport); + + var statusChanges = new List(); + device.StatusChanged += (_, args) => statusChanges.Add(args.Status); + + device.Connect(); + + Assert.True(device.IsConnected); + Assert.Equal(DeviceState.Connected, device.State); + Assert.Equal(1, transport.SyncConnectCalls); + + device.Disconnect(); + + Assert.False(device.IsConnected); + Assert.Equal(DeviceState.Disconnected, device.State); + Assert.Equal(1, transport.SyncDisconnectCalls); + Assert.Equal( + new[] { ConnectionStatus.Connecting, ConnectionStatus.Connected, ConnectionStatus.Disconnected }, + statusChanges); + } + + [Fact] + public void Connect_WhenTheTransportThrows_PropagatesTheOriginalExceptionUnwrapped() + { + using var transport = new GatedMockTransport { ConnectFailure = new InvalidOperationException("no route") }; + using var device = new DaqifiDevice("Mock Device", transport); + + var thrown = Assert.Throws(() => device.Connect()); + + Assert.Equal("no route", thrown.Message); + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + Assert.Equal(DeviceState.Disconnected, device.State); + } + + [Fact] + public void Dispose_StillDisconnectsAndDisposesTheTransport() + { + var transport = new GatedMockTransport(); + transport.OpenConnectGate(); + var device = new DaqifiDevice("Mock Device", transport); + device.Connect(); + + device.Dispose(); + + Assert.Equal(1, transport.SyncDisconnectCalls); + Assert.True(transport.IsDisposed); + } + + /// + /// Exposes the protected text-exchange seam so a test can hold the device-wide text exchange + /// lock while a disconnect is attempted. + /// + private sealed class TextExchangeProbeDevice(string name, IStreamTransport transport) + : DaqifiDevice(name, transport) + { + /// + /// Runs a text exchange that holds the lock for roughly + /// (nothing ever replies on the mock stream), signalling from + /// inside the lock. + /// + public Task HoldTextExchangeAsync(TaskCompletionSource entered, int responseTimeoutMs) => + ExecuteTextCommandAsync( + () => entered.TrySetResult(), + responseTimeoutMs: responseTimeoutMs, + completionTimeoutMs: 50); + } + + /// + /// Mock transport whose connect can be held open, made to fail, or observed — everything the + /// cancellation paths need without a real socket or serial port. + /// + private sealed class GatedMockTransport : IStreamTransport + { + private readonly MemoryStream _stream = new(); + private readonly TaskCompletionSource _connectGate = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _connectEntered = new(TaskCreationOptions.RunContinuationsAsynchronously); + private volatile bool _isConnected; + private volatile bool _disposed; + private int _connectAttempts; + private int _disconnectCount; + private int _syncConnectCalls; + private int _syncDisconnectCalls; + + /// When set, every connect attempt throws this instead of succeeding. + public Exception? ConnectFailure { get; init; } + + /// Invoked after the transport reports itself connected, before ConnectAsync returns. + public Action? OnConnected { get; set; } + + public int ConnectAttempts => Volatile.Read(ref _connectAttempts); + public int DisconnectCount => Volatile.Read(ref _disconnectCount); + public int SyncConnectCalls => Volatile.Read(ref _syncConnectCalls); + public int SyncDisconnectCalls => Volatile.Read(ref _syncDisconnectCalls); + public bool IsDisposed => _disposed; + + public Stream Stream => _disposed + ? throw new ObjectDisposedException(nameof(GatedMockTransport)) + : _stream; + + public bool IsConnected => _isConnected && !_disposed; + + public string ConnectionInfo => _isConnected ? "Gated: Connected" : "Gated: Disconnected"; + + public event EventHandler? StatusChanged; + + public void OpenConnectGate() => _connectGate.TrySetResult(); + + public Task WaitForConnectEnteredAsync() => WaitAsync(_connectEntered.Task); + + public async Task WaitForWrittenTextAsync(string expected) + { + var deadline = DateTime.UtcNow.AddSeconds(5); + while (DateTime.UtcNow < deadline) + { + try + { + if (System.Text.Encoding.UTF8.GetString(_stream.ToArray()).Contains(expected)) + { + return true; + } + } + catch (Exception) + { + // Snapshotting a MemoryStream the producer thread is writing to can tear; + // just look again on the next poll. + } + + await Task.Delay(25); + } + + return false; + } + + public Task ConnectAsync() => ConnectAsync(null, CancellationToken.None); + + public Task ConnectAsync(ConnectionRetryOptions? retryOptions) => + ConnectAsync(retryOptions, CancellationToken.None); + + public Task ConnectAsync(CancellationToken cancellationToken) => + ConnectAsync(null, cancellationToken); + + public async Task ConnectAsync(ConnectionRetryOptions? retryOptions, CancellationToken cancellationToken) + { + ObjectDisposedException.ThrowIf(_disposed, this); + + Interlocked.Increment(ref _connectAttempts); + _connectEntered.TrySetResult(); + + if (ConnectFailure != null) + { + throw ConnectFailure; + } + + await _connectGate.Task.WaitAsync(cancellationToken).ConfigureAwait(false); + + _isConnected = true; + StatusChanged?.Invoke(this, new TransportStatusEventArgs(true, ConnectionInfo)); + OnConnected?.Invoke(); + } + + public Task DisconnectAsync() + { + if (_isConnected) + { + Interlocked.Increment(ref _disconnectCount); + _isConnected = false; + StatusChanged?.Invoke(this, new TransportStatusEventArgs(false, ConnectionInfo)); + } + + return Task.CompletedTask; + } + + public void Connect() + { + Interlocked.Increment(ref _syncConnectCalls); + ConnectAsync().GetAwaiter().GetResult(); + } + + public void Disconnect() + { + Interlocked.Increment(ref _syncDisconnectCalls); + DisconnectAsync().GetAwaiter().GetResult(); + } + + public void Dispose() + { + if (_disposed) + { + return; + } + + _isConnected = false; + _disposed = true; + _stream.Dispose(); + } + + private static async Task WaitAsync(Task task) + { + return ReferenceEquals(task, await Task.WhenAny(task, Task.Delay(TimeSpan.FromSeconds(5)))); + } + } +} diff --git a/src/Daqifi.Core/Communication/Transport/ConnectRetryExecutor.cs b/src/Daqifi.Core/Communication/Transport/ConnectRetryExecutor.cs index 85ad0121..7dbef5c0 100644 --- a/src/Daqifi.Core/Communication/Transport/ConnectRetryExecutor.cs +++ b/src/Daqifi.Core/Communication/Transport/ConnectRetryExecutor.cs @@ -18,7 +18,8 @@ internal static class ConnectRetryExecutor /// Retry configuration, or null for a single no-retry attempt. /// /// Opens the transport handle. Receives the resolved options so it can honor the - /// connection timeout. Throwing signals a failed attempt. + /// connection timeout, and the caller's cancellation token so it can abandon an + /// attempt already in flight. Throwing signals a failed attempt. /// /// /// Disposes/nulls the transport handle after a failed attempt, before the next @@ -29,16 +30,27 @@ internal static class ConnectRetryExecutor /// (false, error) on each failure (a retry-in-progress exception between /// attempts, the real exception on the terminal failure). /// + /// + /// Observed between attempts, while waiting out a backoff delay, and by + /// itself. A cancellation is never treated as a retryable + /// attempt failure: the loop stops immediately and the + /// is surfaced to the caller, which is what lets an + /// auto-reconnect loop be torn down promptly rather than after the remaining attempts. + /// + /// Thrown when the operation is canceled. public static async Task ExecuteAsync( ConnectionRetryOptions? retryOptions, - Func connectAttempt, + Func connectAttempt, Action onAttemptFailed, - Action onStatusChanged) + Action onStatusChanged, + CancellationToken cancellationToken = default) { var options = retryOptions ?? ConnectionRetryOptions.NoRetry; var maxAttempts = options.Enabled ? options.MaxAttempts : 1; Exception? lastException = null; + cancellationToken.ThrowIfCancellationRequested(); + for (var attempt = 1; attempt <= maxAttempts; attempt++) { try @@ -49,14 +61,23 @@ public static async Task ExecuteAsync( var delay = options.CalculateDelay(attempt); if (delay > TimeSpan.Zero) { - await Task.Delay(delay); + await Task.Delay(delay, cancellationToken); } } - await connectAttempt(options); + await connectAttempt(options, cancellationToken); onStatusChanged(true, null); return; // Success! } + catch (OperationCanceledException ex) when (cancellationToken.IsCancellationRequested) + { + // The caller gave up on this connect. Clean up the half-built handle exactly as a + // failed attempt would, report the transport as disconnected, and stop — retrying + // after a cancellation would keep the device dialling long after the caller left. + onAttemptFailed(); + onStatusChanged(false, ex); + throw; + } catch (Exception ex) { lastException = ex; diff --git a/src/Daqifi.Core/Communication/Transport/IStreamTransport.cs b/src/Daqifi.Core/Communication/Transport/IStreamTransport.cs index ebeae07c..187b8485 100644 --- a/src/Daqifi.Core/Communication/Transport/IStreamTransport.cs +++ b/src/Daqifi.Core/Communication/Transport/IStreamTransport.cs @@ -44,6 +44,34 @@ public interface IStreamTransport : IDisposable /// A task representing the asynchronous connect operation. Task ConnectAsync(ConnectionRetryOptions? retryOptions); + /// + /// Establishes the transport connection, abandoning the attempt if + /// is signalled. + /// + /// A cancellation token to observe while connecting. + /// A task representing the asynchronous connect operation. + /// Thrown when the attempt is canceled. + Task ConnectAsync(CancellationToken cancellationToken) => ConnectAsync(null, cancellationToken); + + /// + /// Establishes the transport connection with retry support, abandoning the attempt if + /// is signalled — including while waiting out the + /// backoff delay between retries. + /// + /// + /// This is the cancellable form of . It has a + /// default implementation that simply forwards to the uncancellable overload, so an existing + /// implementation keeps compiling and working unchanged — it just + /// cannot honor the token. Implementations that can abandon an in-flight attempt should override + /// this member; the transports shipped in daqifi-core do. + /// + /// Configuration for retry behavior. If null, uses default single attempt. + /// A cancellation token to observe while connecting. + /// A task representing the asynchronous connect operation. + /// Thrown when the attempt is canceled. + Task ConnectAsync(ConnectionRetryOptions? retryOptions, CancellationToken cancellationToken) => + ConnectAsync(retryOptions); + /// /// Closes the transport connection. /// diff --git a/src/Daqifi.Core/Communication/Transport/SerialStreamTransport.cs b/src/Daqifi.Core/Communication/Transport/SerialStreamTransport.cs index 789b39dd..507104d9 100644 --- a/src/Daqifi.Core/Communication/Transport/SerialStreamTransport.cs +++ b/src/Daqifi.Core/Communication/Transport/SerialStreamTransport.cs @@ -237,6 +237,25 @@ public async Task ConnectAsync() /// Configuration for retry behavior. If null, uses default single attempt. /// A task representing the asynchronous connect operation. public async Task ConnectAsync(ConnectionRetryOptions? retryOptions) + { + await ConnectAsync(retryOptions, CancellationToken.None); + } + + /// + public async Task ConnectAsync(CancellationToken cancellationToken) + { + await ConnectAsync(null, cancellationToken); + } + + /// + /// + /// Cancellation is observed between attempts and while waiting out a backoff delay, and is + /// checked immediately before each . It cannot interrupt the + /// Open call itself — the framework offers no cancellable form of it — so on a port that + /// hangs open, cancellation takes effect when that call returns. In practice opening a serial + /// port either succeeds or fails quickly; the long wait worth cancelling is the retry loop. + /// + public async Task ConnectAsync(ConnectionRetryOptions? retryOptions, CancellationToken cancellationToken) { ThrowIfDisposed(); @@ -245,7 +264,7 @@ public async Task ConnectAsync(ConnectionRetryOptions? retryOptions) await ConnectRetryExecutor.ExecuteAsync( retryOptions, - connectAttempt: options => + connectAttempt: (options, attemptToken) => { var timeout = (int)options.ConnectionTimeout.TotalMilliseconds; _serialPort = new SerialPort(_portName, _baudRate, _parity, _dataBits, _stopBits) @@ -256,6 +275,11 @@ await ConnectRetryExecutor.ExecuteAsync( RtsEnable = _enableRts }; + // Last chance to bail before an uninterruptible open — on a serial port a + // DTR-triggered MCU reset also fires here, so not opening at all is the only way to + // honor a cancel that arrived while the previous attempt was backing off. + attemptToken.ThrowIfCancellationRequested(); + _serialPort.Open(); // After a successful open, swap the connect timeouts for the (shorter) @@ -269,7 +293,8 @@ await ConnectRetryExecutor.ExecuteAsync( _serialPort?.Dispose(); _serialPort = null; }, - onStatusChanged: OnStatusChanged); + onStatusChanged: OnStatusChanged, + cancellationToken: cancellationToken); StartDropDetection(); } diff --git a/src/Daqifi.Core/Communication/Transport/TcpStreamTransport.cs b/src/Daqifi.Core/Communication/Transport/TcpStreamTransport.cs index 8a5be806..03ce761f 100644 --- a/src/Daqifi.Core/Communication/Transport/TcpStreamTransport.cs +++ b/src/Daqifi.Core/Communication/Transport/TcpStreamTransport.cs @@ -212,6 +212,22 @@ public async Task ConnectAsync() /// . /// public async Task ConnectAsync(ConnectionRetryOptions? retryOptions) + { + await ConnectAsync(retryOptions, CancellationToken.None); + } + + /// + public async Task ConnectAsync(CancellationToken cancellationToken) + { + await ConnectAsync(null, cancellationToken); + } + + /// + /// + /// Thrown when the final connection attempt does not complete within + /// . + /// + public async Task ConnectAsync(ConnectionRetryOptions? retryOptions, CancellationToken cancellationToken) { ThrowIfDisposed(); @@ -220,7 +236,7 @@ public async Task ConnectAsync(ConnectionRetryOptions? retryOptions) await ConnectRetryExecutor.ExecuteAsync( retryOptions, - connectAttempt: async options => + connectAttempt: async (options, attemptToken) => { _tcpClient = _localInterface != null ? new TcpClient(new IPEndPoint(_localInterface, 0)) @@ -231,8 +247,13 @@ await ConnectRetryExecutor.ExecuteAsync( _tcpClient.ReceiveTimeout = timeout; _tcpClient.SendTimeout = timeout; - // Add connection timeout to prevent long waits - using var cts = new CancellationTokenSource(options.ConnectionTimeout); + // Two independent reasons to stop waiting — the connect timeout and the caller's + // token — are linked into one source so a caller can abandon a dial that would + // otherwise sit here for the full timeout. Which one fired is disambiguated below, + // because the two mean very different things to the caller. + using var cts = CancellationTokenSource.CreateLinkedTokenSource(attemptToken); + cts.CancelAfter(options.ConnectionTimeout); + var connectTask = ConnectTaskFactory != null ? ConnectTaskFactory(_tcpClient) : Hostname != null @@ -243,11 +264,14 @@ await ConnectRetryExecutor.ExecuteAsync( { await connectTask.WaitAsync(cts.Token); } - catch (OperationCanceledException oce) when (cts.IsCancellationRequested) + catch (OperationCanceledException oce) + when (cts.IsCancellationRequested && !attemptToken.IsCancellationRequested) { - // The only cancellation source here is the timeout token, so surface the - // failure as what it actually is — a connect timeout — rather than a - // misleading TaskCanceledException (daqifi-desktop#517). + // The timeout — not the caller — ended the wait, so surface the failure as what + // it actually is rather than a misleading TaskCanceledException + // (daqifi-desktop#517). A caller-driven cancel falls through this filter and + // propagates as OperationCanceledException, which is what callers expect from a + // token they signalled themselves. throw new TimeoutException( $"TCP connect to {Hostname ?? _endPoint.Address.ToString()}:{_endPoint.Port} " + $"timed out after {options.ConnectionTimeout.TotalSeconds:0.###}s.", oce); @@ -270,7 +294,8 @@ await ConnectRetryExecutor.ExecuteAsync( _tcpClient = null; _networkStream = null; }, - onStatusChanged: OnStatusChanged); + onStatusChanged: OnStatusChanged, + cancellationToken: cancellationToken); _watchdog.Arm(); } diff --git a/src/Daqifi.Core/Device/DaqifiDevice.cs b/src/Daqifi.Core/Device/DaqifiDevice.cs index 3e28d755..99bb3fe4 100644 --- a/src/Daqifi.Core/Device/DaqifiDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiDevice.cs @@ -25,7 +25,7 @@ namespace Daqifi.Core.Device /// Represents a DAQiFi device that can be connected to and communicated with. /// This is the base implementation of the IDevice interface. /// - public class DaqifiDevice : IDevice, IDisposable + public class DaqifiDevice : IDevice, IDisposable, IAsyncDisposable { /// /// Gets the name of the device. @@ -393,7 +393,15 @@ protected void WithChannelsLock(Action action) // Environment.CurrentManagedThreadId capture wouldn't work — the // value seen before await may not match the value seen after. private readonly AsyncLocal _isInsideTextExchange = new(); - + + /// + /// How long / wait to acquire + /// _textExchangeLock before tearing down anyway. See the remarks on + /// for how the budget is derived. + /// + private static readonly TimeSpan TextExchangeTeardownWait = TimeSpan.FromSeconds(10); + + /// /// Gets the current connection status of the device. /// @@ -505,53 +513,165 @@ public DaqifiDevice(string name, IStreamTransport transport, ILogger? logger = n /// /// Connects to the device. /// + /// + /// Opens the transport on the calling thread. is the + /// non-blocking, cancellable equivalent and is preferred on a UI thread; this overload is + /// kept for existing callers and behaves exactly as it always has. + /// public void Connect() { - Status = ConnectionStatus.Connecting; - State = DeviceState.Connecting; + BeginConnect(); try { // Connect transport if available _transport?.Connect(); - // Create message producer and consumer from transport if needed + CompleteConnect(); + } + catch + { + FailConnect(); + throw; + } + } + + /// + /// Connects to the device, abandoning the attempt if + /// is signalled. + /// + /// + /// The asynchronous counterpart to : it never blocks the calling + /// thread on the transport handshake, and the token is threaded all the way down to + /// , + /// so an attempt can be given up mid-flight — including between retries, where the retry + /// loop would otherwise keep dialling. A cancel that lands after the transport has come up + /// closes it again, so a cancelled attempt never leaves a half-open connection behind. + /// + /// A cancellation token to observe while connecting. + /// A task representing the asynchronous connect operation. + /// Thrown when the attempt is canceled. + public virtual async Task ConnectAsync(CancellationToken cancellationToken = default) + { + cancellationToken.ThrowIfCancellationRequested(); + + BeginConnect(); + + var transportConnected = false; + try + { if (_transport != null) { - // The reader/writer loops are the first thing to notice a device that has - // gone away; a transport that can act on that gets told (issue #382). - var healthSink = _transport as ITransportHealthSink; + await _transport.ConnectAsync(cancellationToken).ConfigureAwait(false); + transportConnected = true; + } - if (_messageProducer == null) - { - _messageProducer = new MessageProducer(_transport.Stream, healthSink: healthSink); - _messageProducer.SendFailed += OnMessageSendFailed; - } + cancellationToken.ThrowIfCancellationRequested(); - if (_messageConsumer == null) - { - _messageConsumer = new StreamMessageConsumer( - _transport.Stream, - new ProtobufMessageParser(), - healthSink: healthSink); - } - } + CompleteConnect(); + } + catch (OperationCanceledException) + { + FailConnect(); - // Start message producer and consumer if available - _messageProducer?.Start(); - _messageConsumer?.Start(); + // The transport opened and then the caller gave up — close it rather than leave a + // live connection owned by a device that reports itself disconnected. Matters most + // for a reconnect loop (issue #379), where the leaked handle would still be holding + // the serial port when the next attempt tries to open it. + if (transportConnected) + { + await SafeDisconnectTransportAsync().ConfigureAwait(false); + } - Status = ConnectionStatus.Connected; - State = DeviceState.Connected; + throw; } catch { - Status = ConnectionStatus.Disconnected; - State = DeviceState.Disconnected; + FailConnect(); throw; } } + /// + /// Marks the device as connecting. Shared entry point for and + /// . + /// + private void BeginConnect() + { + Status = ConnectionStatus.Connecting; + State = DeviceState.Connecting; + } + + /// + /// Builds (when needed) and starts the message pumps over the now-open transport, then + /// marks the device connected. Shared by and + /// so the sync and async paths cannot drift apart. + /// + private void CompleteConnect() + { + // Create message producer and consumer from transport if needed + if (_transport != null) + { + // The reader/writer loops are the first thing to notice a device that has + // gone away; a transport that can act on that gets told (issue #382). + var healthSink = _transport as ITransportHealthSink; + + if (_messageProducer == null) + { + _messageProducer = new MessageProducer(_transport.Stream, healthSink: healthSink); + _messageProducer.SendFailed += OnMessageSendFailed; + } + + if (_messageConsumer == null) + { + _messageConsumer = new StreamMessageConsumer( + _transport.Stream, + new ProtobufMessageParser(), + healthSink: healthSink); + } + } + + // Start message producer and consumer if available + _messageProducer?.Start(); + _messageConsumer?.Start(); + + Status = ConnectionStatus.Connected; + State = DeviceState.Connected; + } + + /// + /// Rolls the device's reported state back to disconnected after a failed or abandoned + /// connect attempt. + /// + private void FailConnect() + { + Status = ConnectionStatus.Disconnected; + State = DeviceState.Disconnected; + } + + /// + /// Best-effort transport close used on the cancellation path, where an exception must not + /// replace the the caller is waiting for. + /// + private async Task SafeDisconnectTransportAsync() + { + if (_transport == null) + { + return; + } + + try + { + await _transport.DisconnectAsync().ConfigureAwait(false); + } + catch (Exception ex) + { + SafeLog(() => _logger.LogDebug( + ex, + "Failed to close the transport after a cancelled connect attempt.")); + } + } + /// /// Disconnects from the device. /// @@ -568,76 +688,178 @@ public void Connect() /// of responseTimeoutMs*5 = 5s by default + safety margin) and /// most custom-timeout callers; on timeout the in-flight exchange /// sees _isDisconnecting == true via the post-acquisition - /// validation and bails out cleanly. Callers wanting non-blocking - /// disconnect should drive this off a Task.Run. + /// validation and bails out cleanly. Callers wanting a non-blocking + /// disconnect should use . /// public void Disconnect() { _isDisconnecting = true; - // Best-effort coordination with ExecuteTextCommandAsync — - // acquire the lock so we don't tear the transport out from - // under an in-flight text exchange. The lock IS released in - // the finally below when acquired (so a future Connect() - // followed by ExecuteTextCommandAsync isn't blocked); a - // stuck exchange that holds past the timeout drops to the - // _isDisconnecting validation path inside the exchange. - var lockAcquired = false; + var lockAcquired = AcquireTextExchangeLockForTeardown(); + try { - lockAcquired = _textExchangeLock.Wait(TimeSpan.FromSeconds(10)); + StopMessagePumps(); + + // Disconnect transport if available + _transport?.Disconnect(); } - catch (ObjectDisposedException) + finally { - // Disconnect called after Dispose — nothing to coordinate. + FinishDisconnect(lockAcquired); } + } + + /// + /// Disconnects from the device without blocking the calling thread. + /// + /// + /// + /// The asynchronous counterpart to , and the one to use on a UI + /// thread: the courtesy wait for an in-flight text exchange (up to + /// ) is awaited rather than blocked on, and the + /// remaining teardown — joining the reader/writer threads and closing the transport — is + /// pushed off the caller's thread. + /// + /// + /// What the token does. It shortens the courtesy wait: cancelling stops waiting for + /// the in-flight exchange and proceeds straight to teardown, exactly as the timeout does. + /// It never aborts the disconnect itself, and this method does not throw + /// — a teardown abandoned half-way would leave + /// producers, consumers and the transport in an indeterminate state, which is strictly + /// worse than finishing. On return the device is always disconnected. + /// + /// + /// is therefore raised on a thread pool thread rather than the + /// caller's; marshal to your UI thread in the handler if that matters. + /// + /// + /// A cancellation token to observe while disconnecting. + /// A task representing the asynchronous disconnect operation. + public virtual async Task DisconnectAsync(CancellationToken cancellationToken = default) + { + _isDisconnecting = true; + var lockAcquired = await AcquireTextExchangeLockForTeardownAsync(cancellationToken) + .ConfigureAwait(false); try { - // Unsubscribe from message consumer/producer events - if (_messageConsumer != null) - { - _messageConsumer.MessageReceived -= OnInboundMessageReceived; - } + // StopSafely joins the reader/writer threads with a bounded timeout, and closing a + // serial port whose device has gone away can stall: neither belongs on a UI thread, + // so the whole teardown runs on the thread pool. The ConfigureAwait(false) below + // keeps the transport close and the finally block there too. + await Task.Run(StopMessagePumps, CancellationToken.None).ConfigureAwait(false); - if (_messageProducer != null) + if (_transport != null) { - _messageProducer.SendFailed -= OnMessageSendFailed; + await _transport.DisconnectAsync().ConfigureAwait(false); } + } + finally + { + FinishDisconnect(lockAcquired); + } + } - // Stop message consumer and producer safely if available - _messageConsumer?.StopSafely(); - _messageProducer?.StopSafely(); - - // Null the producer/consumer so a subsequent Connect() - // rebuilds them against the transport's current Stream. - // SerialStreamTransport.Stream returns _serialPort.BaseStream, - // which is a new instance after Disconnect() → Connect() - // reopens the port; reusing the old producer/consumer would - // leave them bound to the previous (disposed) BaseStream - // and any Send() would silently no-op. Surfaced by PR #200's - // post-reconnect readiness probe (LAN chip-info returning - // null on every attempt because Send went to a dead stream). - _messageConsumer = null; - _messageProducer = null; + /// + /// Best-effort coordination with ExecuteTextCommandAsync before teardown: acquire + /// the lock so the transport is not torn out from under an in-flight text exchange. The + /// lock IS released by when acquired (so a future + /// followed by ExecuteTextCommandAsync isn't blocked); a stuck + /// exchange that holds past the timeout drops to the _isDisconnecting validation + /// path inside the exchange. + /// + /// true when the lock was acquired and must be released after teardown. + private bool AcquireTextExchangeLockForTeardown() + { + try + { + return _textExchangeLock.Wait(TextExchangeTeardownWait); + } + catch (ObjectDisposedException) + { + // Disconnect called after Dispose — nothing to coordinate. + return false; + } + } - // Disconnect transport if available - _transport?.Disconnect(); + /// + /// + /// Shortens the wait. A cancellation is swallowed rather than propagated: teardown must + /// still run, and the in-flight exchange sees _isDisconnecting == true and bails out + /// on its own — the same outcome as letting the wait time out. + /// + private async Task AcquireTextExchangeLockForTeardownAsync(CancellationToken cancellationToken) + { + try + { + return await _textExchangeLock.WaitAsync(TextExchangeTeardownWait, cancellationToken) + .ConfigureAwait(false); } - finally + catch (ObjectDisposedException) + { + // DisconnectAsync called after Dispose — nothing to coordinate. + return false; + } + catch (OperationCanceledException) + { + return false; + } + } + + /// + /// Unsubscribes, stops and drops the message producer/consumer. Shared by + /// and . + /// + private void StopMessagePumps() + { + // Unsubscribe from message consumer/producer events + if (_messageConsumer != null) + { + _messageConsumer.MessageReceived -= OnInboundMessageReceived; + } + + if (_messageProducer != null) + { + _messageProducer.SendFailed -= OnMessageSendFailed; + } + + // Stop message consumer and producer safely if available + _messageConsumer?.StopSafely(); + _messageProducer?.StopSafely(); + + // Null the producer/consumer so a subsequent Connect() + // rebuilds them against the transport's current Stream. + // SerialStreamTransport.Stream returns _serialPort.BaseStream, + // which is a new instance after Disconnect() → Connect() + // reopens the port; reusing the old producer/consumer would + // leave them bound to the previous (disposed) BaseStream + // and any Send() would silently no-op. Surfaced by PR #200's + // post-reconnect readiness probe (LAN chip-info returning + // null on every attempt because Send went to a dead stream). + _messageConsumer = null; + _messageProducer = null; + } + + /// + /// Settles the device's reported state after teardown and releases the text-exchange lock + /// if it was taken. Runs from a finally in both disconnect paths, so the device can + /// never be left reporting Connected because teardown threw. + /// + /// Whether the text-exchange lock was acquired before teardown. + private void FinishDisconnect(bool lockAcquired) + { + Status = ConnectionStatus.Disconnected; + State = DeviceState.Disconnected; + _isInitialized = false; + _isDisconnecting = false; + if (lockAcquired) { - Status = ConnectionStatus.Disconnected; - State = DeviceState.Disconnected; - _isInitialized = false; - _isDisconnecting = false; - if (lockAcquired) + try + { + _textExchangeLock.Release(); + } + catch (ObjectDisposedException) { - try - { - _textExchangeLock.Release(); - } - catch (ObjectDisposedException) - { - } } } } @@ -1434,17 +1656,57 @@ private void OnMessageSendFailed(object? sender, MessageSendFailedEventArgs /// Disposes the device and releases resources. /// + /// + /// Disconnects first, which blocks the calling thread. is the + /// non-blocking equivalent — prefer await using over using on a UI thread. + /// public void Dispose() { - if (!_disposed) + if (_disposed) { - Disconnect(); - _messageConsumer?.Dispose(); - _messageProducer?.Dispose(); - _transport?.Dispose(); - _textExchangeLock.Dispose(); - _disposed = true; + return; } + + Disconnect(); + ReleaseResources(); + } + + /// + /// Disposes the device and releases resources without blocking the calling thread, so + /// await using var device = ... is safe on a UI thread. + /// + /// + /// Equivalent to except that the disconnect it performs first runs + /// through . Safe to call more than once, and safe to mix with + /// — whichever runs first wins and the other becomes a no-op. + /// + /// A task representing the asynchronous dispose operation. + public async ValueTask DisposeAsync() + { + if (_disposed) + { + return; + } + + await DisconnectAsync().ConfigureAwait(false); + ReleaseResources(); + } + + /// + /// Releases everything the device owns once it is already disconnected. Shared tail of + /// and . + /// + /// + /// The transport is already closed by the preceding disconnect, so + /// on it does not block here. + /// + private void ReleaseResources() + { + _messageConsumer?.Dispose(); + _messageProducer?.Dispose(); + _transport?.Dispose(); + _textExchangeLock.Dispose(); + _disposed = true; } /// diff --git a/src/Daqifi.Core/Device/DaqifiDeviceFactory.cs b/src/Daqifi.Core/Device/DaqifiDeviceFactory.cs index 743588a5..54ad756a 100644 --- a/src/Daqifi.Core/Device/DaqifiDeviceFactory.cs +++ b/src/Daqifi.Core/Device/DaqifiDeviceFactory.cs @@ -463,8 +463,9 @@ private static async Task ConnectWithTransportAsync( { cancellationToken.ThrowIfCancellationRequested(); - // Step 1: Connect the transport - await transport.ConnectAsync(options.ConnectionRetry).ConfigureAwait(false); + // Step 1: Connect the transport. The token goes all the way down now, so a caller who + // gives up mid-dial stops the attempt instead of waiting out the retry loop (#341). + await transport.ConnectAsync(options.ConnectionRetry, cancellationToken).ConfigureAwait(false); cancellationToken.ThrowIfCancellationRequested(); @@ -473,7 +474,7 @@ private static async Task ConnectWithTransportAsync( device = new DaqifiStreamingDevice(options.DeviceName, transport, options.Logger); // Step 3: Connect the device (starts message producers/consumers) - device.Connect(); + await device.ConnectAsync(cancellationToken).ConfigureAwait(false); cancellationToken.ThrowIfCancellationRequested(); @@ -492,7 +493,7 @@ await device.InitializeAsync(options.ChannelPopulationTimeout, cancellationToken // If device was created, it owns the transport and will dispose it if (device != null) { - device.Dispose(); + await device.DisposeAsync().ConfigureAwait(false); } else { diff --git a/src/Daqifi.Core/Device/IDevice.cs b/src/Daqifi.Core/Device/IDevice.cs index 672d8b51..dadd041a 100644 --- a/src/Daqifi.Core/Device/IDevice.cs +++ b/src/Daqifi.Core/Device/IDevice.cs @@ -1,6 +1,8 @@ using Daqifi.Core.Communication.Messages; using System; using System.Net; +using System.Threading; +using System.Threading.Tasks; #nullable enable @@ -42,15 +44,67 @@ public interface IDevice event EventHandler MessageReceived; /// - /// Connects to the device. + /// Connects to the device, blocking the calling thread until the connection is open. /// + /// + /// Prefer on a UI thread, or whenever the attempt needs to be + /// abandonable. + /// void Connect(); /// - /// Disconnects from the device. + /// Disconnects from the device, blocking the calling thread until teardown completes. /// + /// + /// Prefer on a UI thread — teardown waits for any in-flight + /// command exchange to finish, which can take seconds. + /// void Disconnect(); + /// + /// Connects to the device, abandoning the attempt if + /// is signalled. + /// + /// + /// The default implementation simply calls on the calling thread, so + /// an existing implementation keeps compiling and working unchanged — + /// it just cannot honor the token beyond the check made before the attempt starts. + /// overrides it with a genuinely asynchronous, cancellable + /// implementation. + /// + /// A cancellation token to observe while connecting. + /// A task representing the asynchronous connect operation. + /// Thrown when the attempt is canceled. + Task ConnectAsync(CancellationToken cancellationToken = default) + { + cancellationToken.ThrowIfCancellationRequested(); + Connect(); + return Task.CompletedTask; + } + + /// + /// Disconnects from the device without blocking the calling thread. + /// + /// + /// The default implementation simply calls on the calling thread, + /// so an existing implementation keeps compiling and working + /// unchanged. overrides it with a genuinely asynchronous + /// implementation; see that override for what the token does — teardown always runs to + /// completion, so cancellation shortens the wait rather than aborting the disconnect. + /// + /// + /// A cancellation token to observe while disconnecting. Ignored by this default + /// implementation: aborting a teardown part-way would leave the device in an + /// indeterminate state, which is worse than finishing it. + /// + /// A task representing the asynchronous disconnect operation. + Task DisconnectAsync(CancellationToken cancellationToken = default) + { + _ = cancellationToken; + Disconnect(); + return Task.CompletedTask; + } + /// /// Sends a message to the device. /// From 52c99e5b606b501c1d7efb24773e1e5fc35dc5bf Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Fri, 31 Jul 2026 13:31:54 -0600 Subject: [PATCH 2/5] refactor(device): keep the async connect/disconnect pair non-virtual, harden the retry cancel check Review follow-ups: a virtual async twin of a non-virtual sync method invites a subclass to override one and silently bypass the other, and a retry policy with zero backoff had no Task.Delay in which to notice a cancellation. Co-Authored-By: Claude Opus 5 --- .../Transport/ConnectRetryExecutorTests.cs | 24 +++++++++++++++++++ .../Device/DaqifiDeviceAsyncLifecycleTests.cs | 12 ++++++++++ .../Transport/ConnectRetryExecutor.cs | 5 ++++ src/Daqifi.Core/Device/DaqifiDevice.cs | 8 +++++-- 4 files changed, 47 insertions(+), 2 deletions(-) diff --git a/src/Daqifi.Core.Tests/Communication/Transport/ConnectRetryExecutorTests.cs b/src/Daqifi.Core.Tests/Communication/Transport/ConnectRetryExecutorTests.cs index d98917bf..6120b0b2 100644 --- a/src/Daqifi.Core.Tests/Communication/Transport/ConnectRetryExecutorTests.cs +++ b/src/Daqifi.Core.Tests/Communication/Transport/ConnectRetryExecutorTests.cs @@ -250,6 +250,30 @@ await Assert.ThrowsAnyAsync(() => Assert.IsAssignableFrom(status.Error); } + [Fact] + public async Task ExecuteAsync_CanceledWhileAnAttemptFailedNormally_DoesNotDialAgain() + { + // Zero backoff means there is no Task.Delay to notice the cancellation, so the per-iteration + // check is the only thing standing between a cancelled caller and another connection attempt. + var attempts = 0; + using var cts = new CancellationTokenSource(); + + await Assert.ThrowsAnyAsync(() => + ConnectRetryExecutor.ExecuteAsync( + FastRetry(5), + connectAttempt: (_, _) => + { + attempts++; + cts.Cancel(); + throw new InvalidOperationException("boom"); + }, + onAttemptFailed: () => { }, + onStatusChanged: (_, _) => { }, + cancellationToken: cts.Token)); + + Assert.Equal(1, attempts); + } + [Fact] public async Task ExecuteAsync_CanceledDuringBackoffDelay_StopsWaiting() { diff --git a/src/Daqifi.Core.Tests/Device/DaqifiDeviceAsyncLifecycleTests.cs b/src/Daqifi.Core.Tests/Device/DaqifiDeviceAsyncLifecycleTests.cs index e7ed7037..cfc17dfd 100644 --- a/src/Daqifi.Core.Tests/Device/DaqifiDeviceAsyncLifecycleTests.cs +++ b/src/Daqifi.Core.Tests/Device/DaqifiDeviceAsyncLifecycleTests.cs @@ -218,6 +218,18 @@ public async Task DisposeAsync_IsIdempotentAndInterchangeableWithDispose() Assert.True(transport.IsDisposed); } + [Fact] + public async Task DisposeAsync_OnANeverConnectedDevice_StillDisposesTheTransport() + { + var transport = new GatedMockTransport(); + var device = new DaqifiDevice("Mock Device", transport); + + await device.DisposeAsync(); + + Assert.True(transport.IsDisposed); + Assert.Equal(0, transport.DisconnectCount); + } + [Fact] public async Task DisposeAsync_AfterSyncDispose_IsANoOp() { diff --git a/src/Daqifi.Core/Communication/Transport/ConnectRetryExecutor.cs b/src/Daqifi.Core/Communication/Transport/ConnectRetryExecutor.cs index 7dbef5c0..d3cd39b7 100644 --- a/src/Daqifi.Core/Communication/Transport/ConnectRetryExecutor.cs +++ b/src/Daqifi.Core/Communication/Transport/ConnectRetryExecutor.cs @@ -55,6 +55,11 @@ public static async Task ExecuteAsync( { try { + // Checked every iteration, not just before the delay: a retry policy configured + // with no backoff skips the delay entirely, and without this a cancellation that + // arrived during the previous attempt would be answered with another dial. + cancellationToken.ThrowIfCancellationRequested(); + // Calculate delay for this attempt if (attempt > 1) { diff --git a/src/Daqifi.Core/Device/DaqifiDevice.cs b/src/Daqifi.Core/Device/DaqifiDevice.cs index 99bb3fe4..350fbb46 100644 --- a/src/Daqifi.Core/Device/DaqifiDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiDevice.cs @@ -551,7 +551,10 @@ public void Connect() /// A cancellation token to observe while connecting. /// A task representing the asynchronous connect operation. /// Thrown when the attempt is canceled. - public virtual async Task ConnectAsync(CancellationToken cancellationToken = default) + // Deliberately not virtual, matching Connect(). A virtual async twin of a non-virtual sync + // method is a trap: a subclass would override this one, leave Connect() unintercepted, and + // get different behavior depending on which entry point the caller reached for. + public async Task ConnectAsync(CancellationToken cancellationToken = default) { cancellationToken.ThrowIfCancellationRequested(); @@ -735,7 +738,8 @@ public void Disconnect() /// /// A cancellation token to observe while disconnecting. /// A task representing the asynchronous disconnect operation. - public virtual async Task DisconnectAsync(CancellationToken cancellationToken = default) + // Not virtual, for the same reason as ConnectAsync. + public async Task DisconnectAsync(CancellationToken cancellationToken = default) { _isDisconnecting = true; var lockAcquired = await AcquireTextExchangeLockForTeardownAsync(cancellationToken) From 66d2ff5eed5ae91ebcdde1109063492191433c14 Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Fri, 31 Jul 2026 13:51:40 -0600 Subject: [PATCH 3/5] fix(device): honor pre-cancellation in the transport default impl, make disposal single-shot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both from Qodo review on #416. The default IStreamTransport.ConnectAsync(retryOptions, token) ignored the token outright, so a legacy transport would open a socket — or pulse DTR and reset the MCU on a serial open — for a caller that had already cancelled. It now refuses to start, matching IDevice's default shim. Dispose() and DisposeAsync() both gated on a flag published only at the end of teardown, while DisposeAsync spends real awaited time inside DisconnectAsync. A concurrent Dispose() could enter that window and run a second teardown, which the XML docs explicitly promised would not happen. An interlocked claim taken at the start of disposal makes that promise true; teardown also moved into a finally so a throwing disconnect no longer leaks the handles on a device that can never be disposed again. Co-Authored-By: Claude Opus 5 --- .../StreamTransportCancellationTests.cs | 26 ++++ .../Device/DaqifiDeviceAsyncLifecycleTests.cs | 124 +++++++++++++++++- .../Transport/IStreamTransport.cs | 24 +++- src/Daqifi.Core/Device/DaqifiDevice.cs | 59 ++++++++- 4 files changed, 214 insertions(+), 19 deletions(-) diff --git a/src/Daqifi.Core.Tests/Communication/Transport/StreamTransportCancellationTests.cs b/src/Daqifi.Core.Tests/Communication/Transport/StreamTransportCancellationTests.cs index f25051a9..0fb8a02d 100644 --- a/src/Daqifi.Core.Tests/Communication/Transport/StreamTransportCancellationTests.cs +++ b/src/Daqifi.Core.Tests/Communication/Transport/StreamTransportCancellationTests.cs @@ -106,6 +106,28 @@ public async Task LegacyTransport_WrittenBeforeTheTokenExisted_StillConnectsThro transport.Dispose(); } + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task LegacyTransport_WithAnAlreadyCanceledToken_RefusesToOpenAnything(bool withRetryOptions) + { + // A transport that cannot abandon an in-flight dial can still decline to start one, and it + // must: opening a connection the caller has already given up on has real side effects — a + // serial open pulses DTR and resets the MCU. Both cancellable overloads share the default + // implementation, so both are checked. + using var transport = new LegacyStreamTransport(); + IStreamTransport asInterface = transport; + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + + await Assert.ThrowsAnyAsync(() => withRetryOptions + ? asInterface.ConnectAsync(null, cts.Token) + : asInterface.ConnectAsync(cts.Token)); + + Assert.Equal(0, transport.ConnectCalls); + Assert.False(transport.IsConnected); + } + /// /// An implementation frozen at the pre-#341 shape — it does not /// override either cancellable overload, so it exercises their default implementations. @@ -115,6 +137,9 @@ private sealed class LegacyStreamTransport : IStreamTransport private readonly MemoryStream _stream = new(); private bool _isConnected; + /// How many times the transport actually tried to open something. + public int ConnectCalls { get; private set; } + public Stream Stream => _stream; public bool IsConnected => _isConnected; public string ConnectionInfo => "Legacy"; @@ -125,6 +150,7 @@ private sealed class LegacyStreamTransport : IStreamTransport public Task ConnectAsync(ConnectionRetryOptions? retryOptions) { + ConnectCalls++; _isConnected = true; StatusChanged?.Invoke(this, new TransportStatusEventArgs(true, ConnectionInfo)); return Task.CompletedTask; diff --git a/src/Daqifi.Core.Tests/Device/DaqifiDeviceAsyncLifecycleTests.cs b/src/Daqifi.Core.Tests/Device/DaqifiDeviceAsyncLifecycleTests.cs index cfc17dfd..7bb01f7f 100644 --- a/src/Daqifi.Core.Tests/Device/DaqifiDeviceAsyncLifecycleTests.cs +++ b/src/Daqifi.Core.Tests/Device/DaqifiDeviceAsyncLifecycleTests.cs @@ -230,6 +230,54 @@ public async Task DisposeAsync_OnANeverConnectedDevice_StillDisposesTheTransport Assert.Equal(0, transport.DisconnectCount); } + [Fact] + public async Task Dispose_CalledWhileDisposeAsyncIsStillTearingDown_DoesNotStartASecondTeardown() + { + // Regression for the disposal overlap race. DisposeAsync spends real awaited time inside + // DisconnectAsync, and a flag published only at the end of teardown leaves that whole + // window open — a concurrent Dispose() would sail through and dispose the transport and the + // text-exchange semaphore a second time. + var transport = new GatedMockTransport(); + transport.OpenConnectGate(); + var device = new DaqifiDevice("Mock Device", transport); + await device.ConnectAsync(); + + transport.HoldFirstDisconnect(); + var disposeAsync = device.DisposeAsync(); + Assert.True(await transport.WaitForDisconnectEnteredAsync(), + "DisposeAsync never reached the transport disconnect."); + + // The async teardown is parked mid-flight. A Dispose() arriving now must bounce off the + // gate immediately rather than run its own teardown. + var sw = Stopwatch.StartNew(); + device.Dispose(); + sw.Stop(); + + transport.ReleaseDisconnect(); + await disposeAsync; + + Assert.Equal(1, transport.DisconnectCount); + Assert.Equal(1, transport.DisposeCount); + Assert.Equal(0, transport.SyncDisconnectCalls); + Assert.True(sw.Elapsed < TimeSpan.FromSeconds(1), + $"Dispose() spent {sw.ElapsedMilliseconds}ms; it should have returned at once as the loser."); + } + + [Fact] + public async Task Dispose_WhenTheDisconnectThrows_StillReleasesTheResources() + { + // Claiming the gate up front means there is no second chance at disposal, so teardown has + // to release the handles even when the disconnect fails on the way out. + var transport = new ThrowOnDisconnectTransport(); + var device = new DaqifiDevice("Mock Device", transport); + device.Connect(); + + Assert.Throws(() => device.Dispose()); + + Assert.True(transport.IsDisposed); + await Task.CompletedTask; + } + [Fact] public async Task DisposeAsync_AfterSyncDispose_IsANoOp() { @@ -299,6 +347,45 @@ public void Dispose_StillDisconnectsAndDisposesTheTransport() Assert.True(transport.IsDisposed); } + /// + /// Transport whose close fails, standing in for a serial port that throws on the way out. + /// + private sealed class ThrowOnDisconnectTransport : IStreamTransport + { + private readonly MemoryStream _stream = new(); + private bool _isConnected; + + public bool IsDisposed { get; private set; } + + public Stream Stream => _stream; + public bool IsConnected => _isConnected; + public string ConnectionInfo => "Throwing"; + + public event EventHandler? StatusChanged; + + public Task ConnectAsync() => ConnectAsync(null); + + public Task ConnectAsync(ConnectionRetryOptions? retryOptions) + { + _isConnected = true; + StatusChanged?.Invoke(this, new TransportStatusEventArgs(true, ConnectionInfo)); + return Task.CompletedTask; + } + + public Task DisconnectAsync() => throw new IOException("the port went away"); + + public void Connect() => ConnectAsync().GetAwaiter().GetResult(); + + public void Disconnect() => DisconnectAsync().GetAwaiter().GetResult(); + + public void Dispose() + { + IsDisposed = true; + _isConnected = false; + _stream.Dispose(); + } + } + /// /// Exposes the protected text-exchange seam so a test can hold the device-wide text exchange /// lock while a disconnect is attempted. @@ -327,12 +414,16 @@ private sealed class GatedMockTransport : IStreamTransport private readonly MemoryStream _stream = new(); private readonly TaskCompletionSource _connectGate = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly TaskCompletionSource _connectEntered = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _disconnectGate = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _disconnectEntered = new(TaskCreationOptions.RunContinuationsAsynchronously); private volatile bool _isConnected; private volatile bool _disposed; private int _connectAttempts; private int _disconnectCount; private int _syncConnectCalls; private int _syncDisconnectCalls; + private int _disposeCount; + private int _disconnectGateArmed; /// When set, every connect attempt throws this instead of succeeding. public Exception? ConnectFailure { get; init; } @@ -344,8 +435,20 @@ private sealed class GatedMockTransport : IStreamTransport public int DisconnectCount => Volatile.Read(ref _disconnectCount); public int SyncConnectCalls => Volatile.Read(ref _syncConnectCalls); public int SyncDisconnectCalls => Volatile.Read(ref _syncDisconnectCalls); + public int DisposeCount => Volatile.Read(ref _disposeCount); public bool IsDisposed => _disposed; + /// + /// Parks the FIRST disconnect until is called. Only the + /// first, so a regression that starts a second teardown fails an assertion instead of + /// deadlocking the test run. + /// + public void HoldFirstDisconnect() => Interlocked.Exchange(ref _disconnectGateArmed, 1); + + public void ReleaseDisconnect() => _disconnectGate.TrySetResult(); + + public Task WaitForDisconnectEnteredAsync() => WaitAsync(_disconnectEntered.Task); + public Stream Stream => _disposed ? throw new ObjectDisposedException(nameof(GatedMockTransport)) : _stream; @@ -411,16 +514,23 @@ public async Task ConnectAsync(ConnectionRetryOptions? retryOptions, Cancellatio OnConnected?.Invoke(); } - public Task DisconnectAsync() + public async Task DisconnectAsync() { - if (_isConnected) + if (!_isConnected) { - Interlocked.Increment(ref _disconnectCount); - _isConnected = false; - StatusChanged?.Invoke(this, new TransportStatusEventArgs(false, ConnectionInfo)); + return; } - return Task.CompletedTask; + _disconnectEntered.TrySetResult(); + + if (Interlocked.Exchange(ref _disconnectGateArmed, 0) == 1) + { + await _disconnectGate.Task.ConfigureAwait(false); + } + + Interlocked.Increment(ref _disconnectCount); + _isConnected = false; + StatusChanged?.Invoke(this, new TransportStatusEventArgs(false, ConnectionInfo)); } public void Connect() @@ -437,6 +547,8 @@ public void Disconnect() public void Dispose() { + Interlocked.Increment(ref _disposeCount); + if (_disposed) { return; diff --git a/src/Daqifi.Core/Communication/Transport/IStreamTransport.cs b/src/Daqifi.Core/Communication/Transport/IStreamTransport.cs index 187b8485..192e5dd0 100644 --- a/src/Daqifi.Core/Communication/Transport/IStreamTransport.cs +++ b/src/Daqifi.Core/Communication/Transport/IStreamTransport.cs @@ -59,18 +59,30 @@ public interface IStreamTransport : IDisposable /// backoff delay between retries. /// /// + /// /// This is the cancellable form of . It has a - /// default implementation that simply forwards to the uncancellable overload, so an existing - /// implementation keeps compiling and working unchanged — it just - /// cannot honor the token. Implementations that can abandon an in-flight attempt should override - /// this member; the transports shipped in daqifi-core do. + /// default implementation, so an existing implementation keeps + /// compiling and working unchanged. That default honors the token only before the attempt + /// starts and then forwards to the uncancellable overload, which cannot be interrupted once it + /// is running. + /// + /// + /// The pre-check is not a formality: opening a connection the caller has already given up on has + /// real side effects — a serial open pulses DTR and resets the MCU — so refusing to start is the + /// one part of this contract every implementation can keep. Implementations that can also + /// abandon an attempt already in flight should override this member; the transports shipped in + /// daqifi-core do. + /// /// /// Configuration for retry behavior. If null, uses default single attempt. /// A cancellation token to observe while connecting. /// A task representing the asynchronous connect operation. /// Thrown when the attempt is canceled. - Task ConnectAsync(ConnectionRetryOptions? retryOptions, CancellationToken cancellationToken) => - ConnectAsync(retryOptions); + Task ConnectAsync(ConnectionRetryOptions? retryOptions, CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + return ConnectAsync(retryOptions); + } /// /// Closes the transport connection. diff --git a/src/Daqifi.Core/Device/DaqifiDevice.cs b/src/Daqifi.Core/Device/DaqifiDevice.cs index 350fbb46..7b56e232 100644 --- a/src/Daqifi.Core/Device/DaqifiDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiDevice.cs @@ -335,7 +335,15 @@ protected void WithChannelsLock(Action action) protected IStreamTransport? Transport => _transport; private IProtocolHandler? _protocolHandler; + + // "Teardown has finished" — read by ExecuteTextCommandCoreAsync to reject work on a dead + // device. Distinct from _disposeClaimed below, which marks teardown as *started*. private bool _disposed; + + // Disposal gate, 0 until a caller claims teardown. Interlocked rather than a bool because + // Dispose() and DisposeAsync() are documented as interchangeable and DisposeAsync spends + // real awaited time inside the window a plain flag would leave open. See TryClaimDisposal. + private int _disposeClaimed; private bool _isDisconnecting; private bool _isInitialized; private readonly List _channels = new(); @@ -1666,13 +1674,19 @@ private void OnMessageSendFailed(object? sender, MessageSendFailedEventArgs public void Dispose() { - if (_disposed) + if (!TryClaimDisposal()) { return; } - Disconnect(); - ReleaseResources(); + try + { + Disconnect(); + } + finally + { + ReleaseResources(); + } } /// @@ -1687,22 +1701,53 @@ public void Dispose() /// A task representing the asynchronous dispose operation. public async ValueTask DisposeAsync() { - if (_disposed) + if (!TryClaimDisposal()) { return; } - await DisconnectAsync().ConfigureAwait(false); - ReleaseResources(); + try + { + await DisconnectAsync().ConfigureAwait(false); + } + finally + { + ReleaseResources(); + } } + /// + /// Claims the right to tear this device down, atomically and exactly once. + /// + /// + /// + /// The gate is taken at the start of disposal rather than published at the end, + /// because spends real time awaiting + /// . A plain "have we finished disposing?" flag leaves that + /// entire window open, so a concurrent — an await using scope + /// unwinding while another shutdown path fires, say — would sail through the check and run + /// a second teardown, disposing the transport and the text-exchange semaphore twice. + /// + /// + /// The loser returns immediately rather than waiting for the winner to finish. That is the + /// normal contract for a redundant call, and the + /// alternative — having block on the in-flight + /// — would reintroduce on the calling thread exactly the stall + /// this class now exists to avoid. + /// + /// + /// true for the first caller only. + private bool TryClaimDisposal() => Interlocked.Exchange(ref _disposeClaimed, 1) == 0; + /// /// Releases everything the device owns once it is already disconnected. Shared tail of /// and . /// /// /// The transport is already closed by the preceding disconnect, so - /// on it does not block here. + /// on it does not block here. Runs from a + /// finally so that a disconnect which throws — a serial close can — still releases + /// the handles rather than leaking them on a device that can never be disposed again. /// private void ReleaseResources() { From 17f39a68dda722bdd4b0ed2fd393229cf12f14f2 Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Fri, 31 Jul 2026 17:42:20 -0600 Subject: [PATCH 4/5] test(device): make the throttle-reset guard timing-independent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Qodo review on #416: the reconnect test bounded nothing, so on a slow or loaded runner the two errors could fall more than five seconds apart, the second raise would be due regardless of the reset, and the test would pass with _errorThrottle.Reset() deleted — the exact regression it exists to catch. The assertion now rests on SuppressedCount rather than on elapsed time. A reset clears the bucket, so a fresh session's first raise has nothing collapsed behind it; a bucket that survived the reconnect carries the previous session's count forward whenever it eventually fires. That holds however long the run took. With the reset dropped the count is 46, not a marginal one or two. The five-second window is still checked, but only as a precondition and only after the count is known clean: if the errors did fall outside one window the run cannot distinguish the two cases, so it fails loudly rather than banking a pass that guards nothing. Ordered second so the common regression reports the real cause instead of "inconclusive". Verified: five consecutive passes with the reset in place (~256ms each, the gap being a fraction of the window), three consecutive failures without it. Co-Authored-By: Claude Opus 5 --- .../Device/DaqifiDeviceAsyncLifecycleTests.cs | 96 +++++++++++++++---- 1 file changed, 80 insertions(+), 16 deletions(-) diff --git a/src/Daqifi.Core.Tests/Device/DaqifiDeviceAsyncLifecycleTests.cs b/src/Daqifi.Core.Tests/Device/DaqifiDeviceAsyncLifecycleTests.cs index 860bea20..0c40e2f9 100644 --- a/src/Daqifi.Core.Tests/Device/DaqifiDeviceAsyncLifecycleTests.cs +++ b/src/Daqifi.Core.Tests/Device/DaqifiDeviceAsyncLifecycleTests.cs @@ -324,37 +324,101 @@ public async Task ConnectAsync_WiresUpTheBackgroundErrorSurface() [Fact] public async Task ConnectAsync_OpensAFreshErrorThrottleSessionOnReconnect() { - // The throttle collapses repeats of the same (source, exception type) for five seconds. - // A reconnect is a new session and must report its first failure at once — so the second - // session's error has to arrive well inside that five-second window to prove the reset ran. + // The throttle collapses repeats of the same (source, exception type) for five seconds, so + // a reconnect must clear it or the new session's first failure is swallowed. + // + // The load-bearing assertion here is SuppressedCount, not elapsed time. "The second error + // arrived quickly" only proves the reset ran if the two errors fell inside one throttle + // window, which a slow machine can break — and then the test passes for the wrong reason. + // SuppressedCount cannot: a reset clears the bucket, so a fresh session's first raise has + // nothing collapsed behind it, while a surviving bucket carries the previous session's + // count forward no matter when it finally fires. The window is still checked, but as a + // precondition that fails loudly rather than as the proof itself. using var transport = new ScriptedErrorTransport(); using var device = new DaqifiDevice("Erroring Device", transport); - var raises = 0; - var raised = new ManualResetEventSlim(false); - device.ErrorOccurred += (_, _) => + var gate = new object(); + var firstSessionRaised = new ManualResetEventSlim(false); + var secondSessionRaised = new ManualResetEventSlim(false); + var inSecondSession = false; + long firstRaisedAt = 0; + long secondRaisedAt = 0; + DeviceErrorEventArgs? secondSessionFirstError = null; + + device.ErrorOccurred += (_, e) => { - Interlocked.Increment(ref raises); - raised.Set(); + lock (gate) + { + if (inSecondSession) + { + // Only the session's *first* raise carries the "nothing suppressed behind me" + // claim; later ones legitimately report collapsed occurrences. + if (secondSessionFirstError != null) + { + return; + } + + secondSessionFirstError = e; + secondRaisedAt = Stopwatch.GetTimestamp(); + secondSessionRaised.Set(); + return; + } + + if (firstRaisedAt == 0) + { + firstRaisedAt = Stopwatch.GetTimestamp(); + firstSessionRaised.Set(); + } + } }; + // Session one: fail reads until the throttle has opened a window and collapsed at least one + // repeat behind it, so a surviving bucket would have a non-zero count to carry forward. await device.ConnectAsync(); transport.ScriptedStream.FailReads = true; - Assert.True(raised.Wait(TimeSpan.FromSeconds(10)), "the first session never reported an error."); + Assert.True(firstSessionRaised.Wait(TimeSpan.FromSeconds(10)), + "the first session never reported an error."); + // Quiesce and tear down. Disconnect stops the consumer, so no session-one raise can still + // be in flight when the flag flips below. transport.ScriptedStream.FailReads = false; await device.DisconnectAsync(); - raised.Reset(); - var raisesAfterFirstSession = Volatile.Read(ref raises); - + // Session two: same failure, same bucket key. await device.ConnectAsync(); + lock (gate) + { + inSecondSession = true; + } + transport.ScriptedStream.FailReads = true; - Assert.True(raised.Wait(TimeSpan.FromSeconds(3)), - "the reconnected session's first failure was collapsed into the previous session's " - + "throttle window — ConnectAsync did not reset the error throttle."); - Assert.True(Volatile.Read(ref raises) > raisesAfterFirstSession); + Assert.True(secondSessionRaised.Wait(TimeSpan.FromSeconds(10)), + "the reconnected session never reported an error at all — its first failure was " + + "collapsed into the previous session's throttle window."); + + // The guard itself, checked first because it holds however long the run took: a reset + // clears the bucket, so the new session's first raise has nothing collapsed behind it. A + // surviving bucket carries session one's suppressed occurrences across the reconnect and + // reports them here. + var gap = Stopwatch.GetElapsedTime(firstRaisedAt, secondRaisedAt); + Assert.True( + secondSessionFirstError!.SuppressedCount == 0, + $"The reconnected session's first error reported {secondSessionFirstError.SuppressedCount} " + + "suppressed occurrence(s), so the throttle bucket survived the reconnect — ConnectAsync " + + $"did not reset the error throttle. (The raise was also held back {gap.TotalSeconds:0.##}s, " + + "which is the window expiring rather than a fresh session reporting immediately.)"); + + // Only reachable with the count clean. If the two errors still fell outside one throttle + // window, the machine was slow enough that the second raise was due anyway — the run + // happens to agree with the guard without having exercised it. Fail loudly rather than + // bank a pass that proves nothing. + Assert.True( + gap < DeviceErrorThrottle.DefaultInterval, + $"Inconclusive run: {gap.TotalSeconds:0.##}s separated the two errors, but the throttle " + + $"window is {DeviceErrorThrottle.DefaultInterval.TotalSeconds:0.##}s. The second raise " + + "would have been due even without the reset, so this run cannot distinguish the two. " + + "Failing rather than reporting a pass that guards nothing."); } /// From 1895c1e94ff7b9a4ccd5dfe2d9bcc1193ec5f73b Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Fri, 31 Jul 2026 17:52:24 -0600 Subject: [PATCH 5/5] test(device): take the clock out of the throttle-reset guard entirely MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Qodo review on #416: the precondition added last round hard-fails when the two errors land more than DeviceErrorThrottle.DefaultInterval apart, so a loaded runner could fail a correct implementation. Both that finding and the previous one are right, and reverting either way just oscillates between them: without a bound a slow run passes while proving nothing, with one a slow run fails while nothing is wrong. The bound was never the problem — depending on wall-clock time at all was. DaqifiDevice now takes an injectable error throttle (internal, matching the existing SetSerialPortForTesting / ConnectTaskFactory seams), and the test installs a ten-minute window. A reset clears the bucket and the new session raises immediately; without the reset the bucket stays shut for ten minutes, so the wait times out and names the defect. Neither outcome can be reached by the machine being fast or slow. The window check survives as a backstop on the test's own premise, but against ten minutes it is unreachable in practice. Verified: 5 consecutive passes with the reset (~250ms each), 5 consecutive failures without it, each reporting that the reconnected session's first failure was collapsed into the previous session's window. Co-Authored-By: Claude Opus 5 --- .../Device/DaqifiDeviceAsyncLifecycleTests.cs | 61 ++++++++++--------- src/Daqifi.Core/Device/DaqifiDevice.cs | 20 +++++- 2 files changed, 51 insertions(+), 30 deletions(-) diff --git a/src/Daqifi.Core.Tests/Device/DaqifiDeviceAsyncLifecycleTests.cs b/src/Daqifi.Core.Tests/Device/DaqifiDeviceAsyncLifecycleTests.cs index 0c40e2f9..10fca68e 100644 --- a/src/Daqifi.Core.Tests/Device/DaqifiDeviceAsyncLifecycleTests.cs +++ b/src/Daqifi.Core.Tests/Device/DaqifiDeviceAsyncLifecycleTests.cs @@ -324,19 +324,21 @@ public async Task ConnectAsync_WiresUpTheBackgroundErrorSurface() [Fact] public async Task ConnectAsync_OpensAFreshErrorThrottleSessionOnReconnect() { - // The throttle collapses repeats of the same (source, exception type) for five seconds, so - // a reconnect must clear it or the new session's first failure is swallowed. + // The throttle collapses repeats of the same (source, exception type), so a reconnect must + // clear it or the new session's first failure is swallowed. // - // The load-bearing assertion here is SuppressedCount, not elapsed time. "The second error - // arrived quickly" only proves the reset ran if the two errors fell inside one throttle - // window, which a slow machine can break — and then the test passes for the wrong reason. - // SuppressedCount cannot: a reset clears the bucket, so a fresh session's first raise has - // nothing collapsed behind it, while a surviving bucket carries the previous session's - // count forward no matter when it finally fires. The window is still checked, but as a - // precondition that fails loudly rather than as the proof itself. + // Nothing here depends on how fast the machine is. Against the default five-second window + // it would: the test has to observe two errors inside one window, so a slow run makes the + // second error due anyway (passing while proving nothing), and a tight bound fails a + // correct implementation on a loaded runner. Widening the window to ten minutes takes the + // clock out of it — a fresh session raises immediately, and a surviving bucket stays shut + // for far longer than any run of this test. using var transport = new ScriptedErrorTransport(); using var device = new DaqifiDevice("Erroring Device", transport); + var throttle = new DeviceErrorThrottle(TimeSpan.FromMinutes(10)); + device.SetErrorThrottleForTesting(throttle); + var gate = new object(); var firstSessionRaised = new ManualResetEventSlim(false); var secondSessionRaised = new ManualResetEventSlim(false); @@ -393,32 +395,33 @@ public async Task ConnectAsync_OpensAFreshErrorThrottleSessionOnReconnect() transport.ScriptedStream.FailReads = true; + // Primary guard. Without the reset the bucket is still shut — for another ten minutes — so + // this failure never arrives and the wait is what trips. Assert.True(secondSessionRaised.Wait(TimeSpan.FromSeconds(10)), - "the reconnected session never reported an error at all — its first failure was " - + "collapsed into the previous session's throttle window."); + "the reconnected session never reported an error at all: its first failure was " + + "collapsed into the throttle window the previous session opened, so ConnectAsync " + + "did not reset the error throttle."); - // The guard itself, checked first because it holds however long the run took: a reset - // clears the bucket, so the new session's first raise has nothing collapsed behind it. A - // surviving bucket carries session one's suppressed occurrences across the reconnect and - // reports them here. - var gap = Stopwatch.GetElapsedTime(firstRaisedAt, secondRaisedAt); + // Corroborating guard, independent of the clock in a different way: a reset clears the + // bucket, so a fresh session's first raise has nothing collapsed behind it. A bucket that + // survived would report the previous session's count here whenever it eventually fired. Assert.True( secondSessionFirstError!.SuppressedCount == 0, $"The reconnected session's first error reported {secondSessionFirstError.SuppressedCount} " - + "suppressed occurrence(s), so the throttle bucket survived the reconnect — ConnectAsync " - + $"did not reset the error throttle. (The raise was also held back {gap.TotalSeconds:0.##}s, " - + "which is the window expiring rather than a fresh session reporting immediately.)"); - - // Only reachable with the count clean. If the two errors still fell outside one throttle - // window, the machine was slow enough that the second raise was due anyway — the run - // happens to agree with the guard without having exercised it. Fail loudly rather than - // bank a pass that proves nothing. + + "suppressed occurrence(s), so the throttle bucket survived the reconnect."); + + // Backstop for the test's own premise rather than for the code: if the two errors somehow + // landed a whole throttle window apart, the second raise was due regardless and the run + // proved nothing. Ten minutes makes that unreachable in practice — a run that slow has + // failed on the waits above long before — but assert it rather than assume it, so the test + // can never report a pass it did not earn. + var gap = Stopwatch.GetElapsedTime(firstRaisedAt, secondRaisedAt); Assert.True( - gap < DeviceErrorThrottle.DefaultInterval, - $"Inconclusive run: {gap.TotalSeconds:0.##}s separated the two errors, but the throttle " - + $"window is {DeviceErrorThrottle.DefaultInterval.TotalSeconds:0.##}s. The second raise " - + "would have been due even without the reset, so this run cannot distinguish the two. " - + "Failing rather than reporting a pass that guards nothing."); + gap < throttle.Interval, + $"Inconclusive run: {gap.TotalSeconds:0.##}s separated the two errors, which is beyond " + + $"the {throttle.Interval.TotalMinutes:0.##}-minute throttle window this test installs. " + + "The second raise would have been due even without the reset, so this run cannot " + + "distinguish the two cases. Failing rather than reporting a pass that guards nothing."); } /// diff --git a/src/Daqifi.Core/Device/DaqifiDevice.cs b/src/Daqifi.Core/Device/DaqifiDevice.cs index a0820dad..a817d504 100644 --- a/src/Daqifi.Core/Device/DaqifiDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiDevice.cs @@ -556,7 +556,25 @@ private set /// Collapses repeated background failures so a systematic fault stays visible without /// storming . See that event for the documented policy. /// - private readonly DeviceErrorThrottle _errorThrottle = new(); + private DeviceErrorThrottle _errorThrottle = new(); + + /// + /// Test seam: replaces the background-error throttle, so a test can widen the collapsing + /// window far beyond the default instead of racing it. Never called in production. + /// + /// + /// A test that wants to prove a reconnect clears the throttle otherwise has to observe two + /// errors inside the default five-second window — which makes the result depend on how + /// quickly the machine happened to run, in both directions: too slow and the second error + /// is due anyway (the test passes without proving anything), tighten the bound and a loaded + /// CI box fails a correct implementation. Widening the window removes the clock from the + /// question entirely. Call before connecting; the field is read from background threads. + /// + /// The throttle to use. + internal void SetErrorThrottleForTesting(DeviceErrorThrottle throttle) + { + _errorThrottle = throttle ?? throw new ArgumentNullException(nameof(throttle)); + } /// /// Initializes a new instance of the class.