From 3c5eaf1013d2ad96adc9605fe91d88b65fb8a48e Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Wed, 5 Aug 2026 19:14:48 -0600 Subject: [PATCH 1/2] refactor(device): extract the live-sample async stream into a collaborator (part of #344) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DaqifiStreamingDevice hosted the whole pull-based live-sample path itself: the bounded channel, the drop-oldest overflow callback, the per-enumeration subscribe/unsubscribe bookkeeping and the device-wide drop counter. None of it is device state — it is an adapter over the IChannel.SampleReceived events the decoder already raises. Move it to Device/Internal/LiveSampleStream, reached through the existing IDeviceOperationHost.SnapshotChannels(). No interface change, no new host member, no public API change: DefaultLiveSampleBufferCapacity, DroppedLiveSampleCount and StreamSamplesAsync keep their signatures and their semantics. StreamSamplesAsync now hands back the collaborator's async iterator directly rather than wrapping it in one of its own. That is what preserves the two deferred behaviors a caller can observe: WithCancellation still reaches the iterator's own [EnumeratorCancellation] parameter, and an invalid bufferCapacity still throws on the first MoveNextAsync rather than at the call. Both are now pinned by tests. Co-Authored-By: Claude Opus 5 --- .../DaqifiStreamingDeviceLiveStreamTests.cs | 25 ++ .../Device/Internal/LiveSampleStreamTests.cs | 292 ++++++++++++++++++ .../Device/DaqifiStreamingDevice.cs | 66 +--- .../Device/Internal/LiveSampleStream.cs | 97 ++++++ 4 files changed, 431 insertions(+), 49 deletions(-) create mode 100644 src/Daqifi.Core.Tests/Device/Internal/LiveSampleStreamTests.cs create mode 100644 src/Daqifi.Core/Device/Internal/LiveSampleStream.cs diff --git a/src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceLiveStreamTests.cs b/src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceLiveStreamTests.cs index 4ada0c8e..970941fa 100644 --- a/src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceLiveStreamTests.cs +++ b/src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceLiveStreamTests.cs @@ -84,6 +84,31 @@ public async Task StreamSamplesAsync_ConsumerFallsBehind_DropsOldest_AndCountsDr Assert.True(device.DroppedLiveSampleCount > 0, "drop-oldest should have dropped and counted overflow samples"); } + [Fact] + public async Task StreamSamplesAsync_WithCancellation_EndsEnumeration_ButNotDeviceStream() + { + var device = CreateStreaming(analogCount: 1); + AnalogChannel(device, 0).IsEnabled = true; + device.StartStreaming(); + + using var cts = new CancellationTokenSource(); + + // The token supplied by WithCancellation rather than by the argument. This device method + // hands back LiveSampleStream's async iterator as-is, so the token has to reach that + // iterator's [EnumeratorCancellation] parameter; re-wrapping the forward in another + // iterator without the attribute would drop it silently and hang here instead. + var enumeration = Task.Run(async () => + { + await foreach (var _ in device.StreamSamplesAsync().WithCancellation(cts.Token)) { } + }); + + cts.Cancel(); + + await Assert.ThrowsAnyAsync( + () => enumeration.WaitAsync(TimeSpan.FromSeconds(5))); + Assert.True(device.IsStreaming); + } + [Fact] public async Task StreamSamplesAsync_InvalidBufferCapacity_Throws() { diff --git a/src/Daqifi.Core.Tests/Device/Internal/LiveSampleStreamTests.cs b/src/Daqifi.Core.Tests/Device/Internal/LiveSampleStreamTests.cs new file mode 100644 index 00000000..0596118c --- /dev/null +++ b/src/Daqifi.Core.Tests/Device/Internal/LiveSampleStreamTests.cs @@ -0,0 +1,292 @@ +using Daqifi.Core.Channel; +using Daqifi.Core.Communication.Messages; +using Daqifi.Core.Device; +using Daqifi.Core.Device.Internal; +using Daqifi.Core.Device.SdCard; +using System; +using System.Collections.Generic; +using System.IO; +using System.Threading; +using System.Threading.Tasks; +using Xunit; + +namespace Daqifi.Core.Tests.Device.Internal; + +/// +/// Unit tests for , the pull-based live-sample view extracted from +/// (#344). +/// +/// +/// +/// What a consumer observes end to end — samples yielded in order, cancellation ending enumeration +/// without stopping the device, drop-oldest under backpressure, the deferred capacity throw — is +/// already pinned through the device by DaqifiStreamingDeviceLiveStreamTests. Those are +/// deliberately untouched: they are the evidence that the extraction changed nothing, so they are +/// not repeated here. +/// +/// +/// These add what only a direct test can see: the subscription bookkeeping (every channel +/// subscribed once at start and unsubscribed on every exit path, including cancellation), that the +/// channel set is snapshotted exactly once per enumeration, and that the drop counter is +/// device-wide rather than per-enumeration. The fake host below throws on every member outside +/// this block's remit, so a future change that sends a command, takes the channels lock, or +/// touches streaming state fails loudly rather than passing quietly. +/// +/// +public class LiveSampleStreamTests +{ + [Fact] + public void Constructor_NullHost_Throws() + { + Assert.Throws(() => new LiveSampleStream(null!)); + } + + [Fact] + public async Task Enumeration_SubscribesToEveryChannelInTheSnapshot() + { + var host = new FakeHost(new FakeChannel(0), new FakeChannel(1)); + var stream = new LiveSampleStream(host); + + await using var e = stream.StreamSamplesAsync(CancellationToken.None).GetAsyncEnumerator(); + var moveNext = e.MoveNextAsync(); // runs the body: subscribes synchronously, then awaits + + Assert.All(host.Channels, c => Assert.Equal(1, c.SubscriberCount)); + + // Every subscribed channel must reach the same buffer, not just the first. + host.Channels[1].RaiseSample(2.5); + Assert.True(await moveNext.AsTask().WaitAsync(TimeSpan.FromSeconds(5))); + Assert.Same(host.Channels[1], e.Current.Channel); + Assert.Equal(2.5, e.Current.Sample.Value); + } + + [Fact] + public async Task Enumeration_Disposed_UnsubscribesFromEveryChannel() + { + var host = new FakeHost(new FakeChannel(0), new FakeChannel(1)); + var stream = new LiveSampleStream(host); + + var e = stream.StreamSamplesAsync(CancellationToken.None).GetAsyncEnumerator(); + var moveNext = e.MoveNextAsync(); + host.Channels[0].RaiseSample(1.0); + Assert.True(await moveNext.AsTask().WaitAsync(TimeSpan.FromSeconds(5))); + + await e.DisposeAsync(); + + // A leaked handler would keep the decode path writing into a dead buffer for the rest of + // the device's life — one per enumeration a consumer ever started. + Assert.All(host.Channels, c => Assert.Equal(0, c.SubscriberCount)); + } + + [Fact] + public async Task Enumeration_EndedByCancellation_StillUnsubscribes() + { + var host = new FakeHost(new FakeChannel(0)); + var stream = new LiveSampleStream(host); + + using var cts = new CancellationTokenSource(); + var e = stream.StreamSamplesAsync(cts.Token).GetAsyncEnumerator(); + var moveNext = e.MoveNextAsync(); + cts.Cancel(); + await Assert.ThrowsAnyAsync(async () => await moveNext); + await e.DisposeAsync(); + + // The unsubscribe lives in a finally, so the throwing exit path has to clean up too. + Assert.Equal(0, host.Channels[0].SubscriberCount); + } + + [Fact] + public async Task Enumeration_SnapshotsTheChannelsOnce_AndIgnoresLaterArrivals() + { + var host = new FakeHost(new FakeChannel(0)); + var stream = new LiveSampleStream(host); + + await using var e = stream.StreamSamplesAsync(CancellationToken.None).GetAsyncEnumerator(); + var moveNext = e.MoveNextAsync(); + + Assert.Equal(1, host.SnapshotCalls); + + // A channel that appears after the enumeration started is not observed by it — that is the + // documented "observes the channels present when it starts" contract, and it is also what + // makes the unsubscribe list above exactly right. + var late = new FakeChannel(1); + host.Add(late); + late.RaiseSample(9.0); + Assert.Equal(0, late.SubscriberCount); + + host.Channels[0].RaiseSample(1.0); + Assert.True(await moveNext.AsTask().WaitAsync(TimeSpan.FromSeconds(5))); + Assert.Equal(1.0, e.Current.Sample.Value); + Assert.Equal(1, host.SnapshotCalls); + } + + [Fact] + public async Task ConcurrentEnumerations_EachGetTheirOwnBuffer() + { + var host = new FakeHost(new FakeChannel(0)); + var stream = new LiveSampleStream(host); + + await using var first = stream.StreamSamplesAsync(CancellationToken.None).GetAsyncEnumerator(); + await using var second = stream.StreamSamplesAsync(CancellationToken.None).GetAsyncEnumerator(); + var firstMove = first.MoveNextAsync(); + var secondMove = second.MoveNextAsync(); + + Assert.Equal(2, host.Channels[0].SubscriberCount); + + host.Channels[0].RaiseSample(4.0); + + // One sample, delivered to both consumers — neither steals it from the other. + Assert.True(await firstMove.AsTask().WaitAsync(TimeSpan.FromSeconds(5))); + Assert.True(await secondMove.AsTask().WaitAsync(TimeSpan.FromSeconds(5))); + Assert.Equal(4.0, first.Current.Sample.Value); + Assert.Equal(4.0, second.Current.Sample.Value); + } + + [Fact] + public async Task DroppedSampleCount_AccumulatesAcrossEnumerations() + { + var host = new FakeHost(new FakeChannel(0)); + var stream = new LiveSampleStream(host); + + Assert.Equal(0, stream.DroppedSampleCount); + + var afterFirst = await OverflowOnce(stream, host); + Assert.True(afterFirst > 0, "drop-oldest should have dropped and counted overflow samples"); + + var afterSecond = await OverflowOnce(stream, host); + + // The counter is a device-wide health signal, so a second enumeration adds to it rather + // than starting over. + Assert.True(afterSecond > afterFirst, $"expected the count to keep growing, got {afterFirst} then {afterSecond}"); + } + + [Fact] + public async Task InvalidBufferCapacity_ThrowsOnFirstMoveNext_NotAtTheCall() + { + var host = new FakeHost(new FakeChannel(0)); + var stream = new LiveSampleStream(host); + + // An async iterator defers its body, so nothing runs — and nothing is subscribed — until + // the first MoveNextAsync. The device forwards this iterator as-is to keep that timing. + var enumerable = stream.StreamSamplesAsync(CancellationToken.None, bufferCapacity: 0); + Assert.Equal(0, host.SnapshotCalls); + + var ex = await Assert.ThrowsAsync(async () => + { + await foreach (var _ in enumerable) { } + }); + Assert.Equal("bufferCapacity", ex.ParamName); + } + + #region Helpers + + /// + /// Runs one enumeration whose reader is parked, floods it past its two-slot buffer, then + /// returns the collaborator's drop count after that enumeration has been disposed. + /// + private static async Task OverflowOnce(LiveSampleStream stream, FakeHost host) + { + var e = stream.StreamSamplesAsync(CancellationToken.None, bufferCapacity: 2).GetAsyncEnumerator(); + var moveNext = e.MoveNextAsync(); // subscribes; reader is awaiting, not consuming synchronously + for (var i = 0; i < 20; i++) + { + host.Channels[0].RaiseSample(i); + } + + Assert.True(await moveNext.AsTask().WaitAsync(TimeSpan.FromSeconds(5))); + await e.DisposeAsync(); + return stream.DroppedSampleCount; + } + + private sealed class FakeChannel : IChannel + { + private EventHandler? _sampleReceived; + + public FakeChannel(int channelNumber) + { + ChannelNumber = channelNumber; + Name = "ch" + channelNumber; + } + + public int SubscriberCount { get; private set; } + + public event EventHandler? SampleReceived + { + add { _sampleReceived += value; SubscriberCount++; } + remove { _sampleReceived -= value; SubscriberCount--; } + } + + public void RaiseSample(double value) + { + var sample = new DataSample(DateTime.UtcNow, value); + ActiveSample = sample; + _sampleReceived?.Invoke(this, new SampleReceivedEventArgs(this, sample)); + } + + public int ChannelNumber { get; } + public string Name { get; set; } + public bool IsEnabled { get; set; } = true; + public ChannelType Type => ChannelType.Analog; + public ChannelDirection Direction { get; set; } = ChannelDirection.Input; + public IDataSample? ActiveSample { get; private set; } + + public void SetActiveSample(double value, DateTime timestamp) => throw new NotSupportedException(); + public void SetActiveSample(IDataSample sample) => throw new NotSupportedException(); + } + + private sealed class FakeHost : IDeviceOperationHost + { + private readonly List _channels; + + public FakeHost(params FakeChannel[] channels) + { + _channels = new List(channels); + } + + public IReadOnlyList Channels => _channels; + + public int SnapshotCalls { get; private set; } + + public void Add(FakeChannel channel) => _channels.Add(channel); + + public IReadOnlyList SnapshotChannels() + { + SnapshotCalls++; + return _channels.ToArray(); + } + + // Outside this block's remit — reaching for any of these is a regression, not a refinement. + // In particular the live path must never send a command or take the channels lock: it is an + // adapter over events the decoder already raises, and it runs on the decode thread. + public bool IsConnected => throw new NotSupportedException(); + public bool IsUsbConnection => throw new NotSupportedException(); + public bool IsStreaming { get => throw new NotSupportedException(); set => throw new NotSupportedException(); } + public int StreamingFrequency => throw new NotSupportedException(); + public DeviceMetadata Metadata => throw new NotSupportedException(); + public TimeSpan SdCardDownloadTimeout => throw new NotSupportedException(); + public TimeSpan SdCardTransferIdleTimeout => throw new NotSupportedException(); + public void StopStreaming() => throw new NotSupportedException(); + public void Send(IOutboundMessage message) => throw new NotSupportedException(); + public void Disconnect() => throw new NotSupportedException(); + public void WithChannelsLock(Action action) => throw new NotSupportedException(); + public Task> ExecuteTextCommandAsync( + Action setupAction, + int responseTimeoutMs = 1000, + int completionTimeoutMs = 250, + CancellationToken cancellationToken = default, + Func? prepareAsync = null, + Func? finalizeAsync = null) => throw new NotSupportedException(); + public Task ExecuteRawCaptureAsync( + Func rawAction, + CancellationToken cancellationToken = default) => throw new NotSupportedException(); + public void EnsureSupported(DeviceFeature feature) => throw new NotSupportedException(); + public FeatureNotSupportedException CreateFeatureNotSupportedException(DeviceFeature feature) + => throw new NotSupportedException(); + public void RaiseLowSdSpaceWarning(LowSdSpaceWarningEventArgs e) => throw new NotSupportedException(); + public void RaiseStreamFrameDiscarded(StreamFrameDiscardedEventArgs e) => throw new NotSupportedException(); + public void RaiseGapDetected(TimestampGapEventArgs e) => throw new NotSupportedException(); + public void RaiseRawStreamFrame(DaqifiOutMessage message) => throw new NotSupportedException(); + public void RaiseStreamDecodeFailure(Exception error) => throw new NotSupportedException(); + } + + #endregion +} diff --git a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs index 4d77a4bc..c1c2fc47 100644 --- a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs @@ -16,9 +16,7 @@ using System.IO; using System.Linq; using System.Net; -using System.Runtime.CompilerServices; using System.Threading; -using System.Threading.Channels; using System.Threading.Tasks; #nullable enable @@ -199,6 +197,7 @@ private void InitializeStreamingDevice() // which a field initializer cannot reference. Every constructor routes through this // method, so they are always in place before the device is handed to a caller. _frameDecoder = new StreamFrameDecoder(this); + _liveSampleStream = new LiveSampleStream(this); _channelControl = new ChannelControlOperations(this); _administration = new DeviceAdministrationOperations(this); _networkOperations = new NetworkConfigurationOperations(this); @@ -708,14 +707,12 @@ protected override Task RestoreSessionSnapshotAsync( /// public const int DefaultLiveSampleBufferCapacity = 4096; - private long _droppedLiveSampleCount; - /// /// Gets the cumulative number of live samples dropped across all /// enumerations because a consumer could not keep up with the incoming rate (drop-oldest policy). /// A non-zero and growing value means a live consumer is too slow for the current stream rate. /// - public long DroppedLiveSampleCount => Interlocked.Read(ref _droppedLiveSampleCount); + public long DroppedLiveSampleCount => _liveSampleStream.DroppedSampleCount; /// /// Exposes decoded live samples as an for pull-based @@ -724,6 +721,7 @@ protected override Task RestoreSessionSnapshotAsync( /// per-channel and raw-frame events are unaffected. /// /// + /// /// Samples are buffered in a bounded channel with a drop-oldest overflow policy: if the /// consumer falls behind, the oldest buffered samples are discarded (memory never grows /// unbounded) and is incremented — the decode thread that @@ -731,6 +729,14 @@ protected override Task RestoreSessionSnapshotAsync( /// cancelling ends it promptly (surfaced as /// ) and unsubscribes, but does not stop the /// device's stream — call for that. + /// + /// + /// This returns 's async iterator directly rather than wrapping + /// it in one of its own, which is what keeps the two deferred behaviors a caller can observe + /// exactly as they were: WithCancellation still reaches the iterator's own + /// [EnumeratorCancellation] parameter, and an invalid + /// still throws on the first MoveNextAsync rather than at the call. + /// /// /// Ends enumeration when cancelled. /// @@ -738,51 +744,10 @@ protected override Task RestoreSessionSnapshotAsync( /// /// An async stream of (channel + decoded sample). /// is less than 1. - public async IAsyncEnumerable StreamSamplesAsync( - [EnumeratorCancellation] CancellationToken cancellationToken = default, + public IAsyncEnumerable StreamSamplesAsync( + CancellationToken cancellationToken = default, int? bufferCapacity = null) - { - var capacity = bufferCapacity ?? DefaultLiveSampleBufferCapacity; - if (capacity < 1) - { - throw new ArgumentOutOfRangeException( - nameof(bufferCapacity), capacity, "Buffer capacity must be at least 1."); - } - - var buffer = System.Threading.Channels.Channel.CreateBounded( - new BoundedChannelOptions(capacity) - { - FullMode = BoundedChannelFullMode.DropOldest, - SingleReader = true, - SingleWriter = false, - }, - _ => Interlocked.Increment(ref _droppedLiveSampleCount)); - - void OnSample(object? sender, SampleReceivedEventArgs e) => - buffer.Writer.TryWrite(new LiveSample(e.Channel, e.Sample)); - - var channels = SnapshotChannels(); - foreach (var channel in channels) - { - channel.SampleReceived += OnSample; - } - - try - { - await foreach (var sample in buffer.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) - { - yield return sample; - } - } - finally - { - foreach (var channel in channels) - { - channel.SampleReceived -= OnSample; - } - buffer.Writer.TryComplete(); - } - } + => _liveSampleStream.StreamSamplesAsync(cancellationToken, bufferCapacity); /// /// Handles a streaming data frame by handing it to , which @@ -1017,6 +982,9 @@ public void SetAdcCalibrationOffset(int channelNumber, double calB) /// The streaming hot path: frame screening, timestamps, gaps, per-channel decode. private StreamFrameDecoder _frameDecoder = null!; + /// The pull-based live-sample view: bounded buffer, drop-oldest, drop counter. + private LiveSampleStream _liveSampleStream = null!; + /// Channel enable/disable, DIO, PWM and analog output (). private ChannelControlOperations _channelControl = null!; diff --git a/src/Daqifi.Core/Device/Internal/LiveSampleStream.cs b/src/Daqifi.Core/Device/Internal/LiveSampleStream.cs new file mode 100644 index 00000000..43e39089 --- /dev/null +++ b/src/Daqifi.Core/Device/Internal/LiveSampleStream.cs @@ -0,0 +1,97 @@ +using Daqifi.Core.Channel; +using System; +using System.Collections.Generic; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Channels; +using System.Threading.Tasks; + +#nullable enable + +namespace Daqifi.Core.Device.Internal +{ + /// + /// The pull-based live-sample view of a streaming device — the bounded buffer behind + /// and the drop counter that goes with + /// it — extracted from (#344) so the device delegates + /// rather than hosts it. + /// + /// + /// + /// This is an adapter over the push-based events, not a + /// second decode path: the decoder still raises those events exactly as before, and each + /// enumeration simply subscribes to them for its lifetime. Nothing here runs on the decode + /// thread beyond a non-blocking . + /// + /// + /// The channels to subscribe to are read through , so the + /// enumeration observes the same channel collection — under the same lock — that the rest of + /// the device does. + /// + /// + internal sealed class LiveSampleStream + { + private readonly IDeviceOperationHost _host; + + /// + /// Cumulative drop count across every enumeration this collaborator has served. Lives here + /// rather than per-enumeration because the device exposes it as a device-wide health signal. + /// + private long _droppedSampleCount; + + internal LiveSampleStream(IDeviceOperationHost host) + { + _host = host ?? throw new ArgumentNullException(nameof(host)); + } + + /// + internal long DroppedSampleCount => Interlocked.Read(ref _droppedSampleCount); + + /// + internal async IAsyncEnumerable StreamSamplesAsync( + [EnumeratorCancellation] CancellationToken cancellationToken = default, + int? bufferCapacity = null) + { + var capacity = bufferCapacity ?? DaqifiStreamingDevice.DefaultLiveSampleBufferCapacity; + if (capacity < 1) + { + throw new ArgumentOutOfRangeException( + nameof(bufferCapacity), capacity, "Buffer capacity must be at least 1."); + } + + var buffer = System.Threading.Channels.Channel.CreateBounded( + new BoundedChannelOptions(capacity) + { + FullMode = BoundedChannelFullMode.DropOldest, + SingleReader = true, + SingleWriter = false, + }, + _ => Interlocked.Increment(ref _droppedSampleCount)); + + void OnSample(object? sender, SampleReceivedEventArgs e) => + buffer.Writer.TryWrite(new LiveSample(e.Channel, e.Sample)); + + var channels = _host.SnapshotChannels(); + foreach (var channel in channels) + { + channel.SampleReceived += OnSample; + } + + try + { + await foreach (var sample in buffer.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) + { + yield return sample; + } + } + finally + { + foreach (var channel in channels) + { + channel.SampleReceived -= OnSample; + } + buffer.Writer.TryComplete(); + } + } + } +} From 2cf3bd1ee875e85f07d749f08b7295174f233718 Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Wed, 5 Aug 2026 19:30:34 -0600 Subject: [PATCH 2/2] test(device): bound every wait in LiveSampleStreamTests so a regression fails instead of hanging Two waits in the new tests were unbounded: the post-cancellation await in Enumeration_EndedByCancellation_StillUnsubscribes, and the await foreach inside InvalidBufferCapacity_ThrowsOnFirstMoveNext_NotAtTheCall. A live stream's read is unbounded by design, so if cancellation or the capacity validation ever regressed, those two would park forever and stall the whole run rather than failing it. Both now go through the same .WaitAsync(...) bound the rest of the file already used, hoisted into a named MoveNextTimeout so the intent is stated once and the next test added here inherits it. Verified by simulating both regressions (dropping the Cancel() call, and passing a valid capacity): each test now fails in ~5s with TimeoutException instead of hanging. Co-Authored-By: Claude Opus 5 --- .../Device/Internal/LiveSampleStreamTests.cs | 40 ++++++++++++++----- 1 file changed, 30 insertions(+), 10 deletions(-) diff --git a/src/Daqifi.Core.Tests/Device/Internal/LiveSampleStreamTests.cs b/src/Daqifi.Core.Tests/Device/Internal/LiveSampleStreamTests.cs index 0596118c..e5303eae 100644 --- a/src/Daqifi.Core.Tests/Device/Internal/LiveSampleStreamTests.cs +++ b/src/Daqifi.Core.Tests/Device/Internal/LiveSampleStreamTests.cs @@ -35,6 +35,15 @@ namespace Daqifi.Core.Tests.Device.Internal; /// public class LiveSampleStreamTests { + /// + /// Upper bound on every await in this file. The collaborator's reads are unbounded by design — + /// a live stream waits for the next sample — so a regression in cancellation, in delivery, or + /// in argument validation would otherwise park a test forever and stall the whole run. Every + /// wait here is bounded so such a regression surfaces as a fast + /// instead. Generous enough not to flake on a loaded CI agent. + /// + private static readonly TimeSpan MoveNextTimeout = TimeSpan.FromSeconds(5); + [Fact] public void Constructor_NullHost_Throws() { @@ -54,7 +63,7 @@ public async Task Enumeration_SubscribesToEveryChannelInTheSnapshot() // Every subscribed channel must reach the same buffer, not just the first. host.Channels[1].RaiseSample(2.5); - Assert.True(await moveNext.AsTask().WaitAsync(TimeSpan.FromSeconds(5))); + Assert.True(await moveNext.AsTask().WaitAsync(MoveNextTimeout)); Assert.Same(host.Channels[1], e.Current.Channel); Assert.Equal(2.5, e.Current.Sample.Value); } @@ -68,7 +77,7 @@ public async Task Enumeration_Disposed_UnsubscribesFromEveryChannel() var e = stream.StreamSamplesAsync(CancellationToken.None).GetAsyncEnumerator(); var moveNext = e.MoveNextAsync(); host.Channels[0].RaiseSample(1.0); - Assert.True(await moveNext.AsTask().WaitAsync(TimeSpan.FromSeconds(5))); + Assert.True(await moveNext.AsTask().WaitAsync(MoveNextTimeout)); await e.DisposeAsync(); @@ -87,7 +96,12 @@ public async Task Enumeration_EndedByCancellation_StillUnsubscribes() var e = stream.StreamSamplesAsync(cts.Token).GetAsyncEnumerator(); var moveNext = e.MoveNextAsync(); cts.Cancel(); - await Assert.ThrowsAnyAsync(async () => await moveNext); + + // Bounded on purpose: if cancellation ever stops ending the read, an unbounded await here + // would park forever and hang the whole run. The timeout turns that into a fast, named + // failure (TimeoutException instead of the expected OperationCanceledException). + await Assert.ThrowsAnyAsync( + () => moveNext.AsTask().WaitAsync(MoveNextTimeout)); await e.DisposeAsync(); // The unsubscribe lives in a finally, so the throwing exit path has to clean up too. @@ -114,7 +128,7 @@ public async Task Enumeration_SnapshotsTheChannelsOnce_AndIgnoresLaterArrivals() Assert.Equal(0, late.SubscriberCount); host.Channels[0].RaiseSample(1.0); - Assert.True(await moveNext.AsTask().WaitAsync(TimeSpan.FromSeconds(5))); + Assert.True(await moveNext.AsTask().WaitAsync(MoveNextTimeout)); Assert.Equal(1.0, e.Current.Sample.Value); Assert.Equal(1, host.SnapshotCalls); } @@ -135,8 +149,8 @@ public async Task ConcurrentEnumerations_EachGetTheirOwnBuffer() host.Channels[0].RaiseSample(4.0); // One sample, delivered to both consumers — neither steals it from the other. - Assert.True(await firstMove.AsTask().WaitAsync(TimeSpan.FromSeconds(5))); - Assert.True(await secondMove.AsTask().WaitAsync(TimeSpan.FromSeconds(5))); + Assert.True(await firstMove.AsTask().WaitAsync(MoveNextTimeout)); + Assert.True(await secondMove.AsTask().WaitAsync(MoveNextTimeout)); Assert.Equal(4.0, first.Current.Sample.Value); Assert.Equal(4.0, second.Current.Sample.Value); } @@ -170,11 +184,17 @@ public async Task InvalidBufferCapacity_ThrowsOnFirstMoveNext_NotAtTheCall() var enumerable = stream.StreamSamplesAsync(CancellationToken.None, bufferCapacity: 0); Assert.Equal(0, host.SnapshotCalls); - var ex = await Assert.ThrowsAsync(async () => + // Bounded on purpose: were the validation to stop throwing, this enumeration would block + // on an empty buffer that nothing ever writes to, so an unbounded await would hang the run + // rather than fail it. + var ex = await Assert.ThrowsAsync( + () => ConsumeAsync().WaitAsync(MoveNextTimeout)); + Assert.Equal("bufferCapacity", ex.ParamName); + + async Task ConsumeAsync() { await foreach (var _ in enumerable) { } - }); - Assert.Equal("bufferCapacity", ex.ParamName); + } } #region Helpers @@ -192,7 +212,7 @@ private static async Task OverflowOnce(LiveSampleStream stream, FakeHost h host.Channels[0].RaiseSample(i); } - Assert.True(await moveNext.AsTask().WaitAsync(TimeSpan.FromSeconds(5))); + Assert.True(await moveNext.AsTask().WaitAsync(MoveNextTimeout)); await e.DisposeAsync(); return stream.DroppedSampleCount; }