From ad0cb0ea752dfc071ddc94e09451b72484908970 Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Sun, 2 Aug 2026 10:55:23 -0600 Subject: [PATCH 1/3] fix(device): stop broadcasting the malformed first stream frame (closes #425) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Firmware up to and including 3.7.2 emits a leading frame at stream start whose analog payload holds a single value regardless of the enabled channel mask. Core's per-channel decode has guarded against it since #351, but the raw MessageReceived event was still handed the frame verbatim — and that is the path most callers use, including the example CLI, whose offline export inferred a channel count of one from it and truncated every sample that followed (daqifi-core-example-app#34). Both consumer paths are now gated together, and every drop is reported through the new StreamFrameDiscarded event and DiscardedStreamFrameCount so a suppressed frame is never invisible. Also adds the cross-session leftover-frame guard #351 asked for (daqifi-nyquist-firmware #533) as a StreamFrameGate collaborator: the frame the device latches across a stop is recognised by its device-tick counter using wrap-safe modular arithmetic, measured against a reference fixed at session start so a quick restart cannot cascade, and capped so a stream can never be withheld indefinitely. Co-Authored-By: Claude Opus 5 --- .../DaqifiStreamingDeviceDecodeTests.cs | 255 +++++++++++++++++- .../Device/Internal/StreamFrameGateTests.cs | 177 ++++++++++++ .../Device/DaqifiStreamingDevice.cs | 205 +++++++++++--- .../Device/Internal/StreamFrameGate.cs | 168 ++++++++++++ .../Device/StreamFrameDiscardReason.cs | 39 +++ .../Device/StreamFrameDiscardedEventArgs.cs | 57 ++++ 6 files changed, 860 insertions(+), 41 deletions(-) create mode 100644 src/Daqifi.Core.Tests/Device/Internal/StreamFrameGateTests.cs create mode 100644 src/Daqifi.Core/Device/Internal/StreamFrameGate.cs create mode 100644 src/Daqifi.Core/Device/StreamFrameDiscardReason.cs create mode 100644 src/Daqifi.Core/Device/StreamFrameDiscardedEventArgs.cs diff --git a/src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceDecodeTests.cs b/src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceDecodeTests.cs index 404abc88..c4d5269b 100644 --- a/src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceDecodeTests.cs +++ b/src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceDecodeTests.cs @@ -607,22 +607,117 @@ public void Decode_CombinedWarmupFrame_SuppressesAnalogButKeepsDigital() } [Fact] - public void Decode_WarmupFrame_StillReRaisesRawMessage() + public void Decode_WarmupFrame_NotHandedToRawFrameConsumers() { - // Suppression skips only the per-channel decode; raw-frame consumers still see the frame. + // Issue #425: the malformed frame must not reach raw-frame consumers either. They read + // AnalogInDataFloat straight off the frame, so a partial payload is exactly as harmful + // there as in the decoded path — the example CLI's offline export inferred a channel count + // of one from it and truncated every sample that followed. var device = CreateStreamingDevice(analogCount: 2); AnalogChannel(device, 0).IsEnabled = true; AnalogChannel(device, 1).IsEnabled = true; device.StartStreaming(); - var rawFrames = 0; - device.MessageReceived += (_, _) => rawFrames++; + var rawFrames = new List(); + device.MessageReceived += (_, e) => + { + if (e.Message.Data is DaqifiOutMessage frame) rawFrames.Add(frame); + }; + var classified = 0; + device.StreamMessageReceived += _ => classified++; var warmup = new DaqifiOutMessage { MsgTimeStamp = 1 }; warmup.AnalogInDataFloat.Add(0.1f); device.InvokeStreamMessage(warmup); + Assert.Empty(rawFrames); + Assert.Equal(0, classified); + + // The next full frame reaches raw consumers untouched. + var full = new DaqifiOutMessage { MsgTimeStamp = 2 }; + full.AnalogInDataFloat.Add(1f); + full.AnalogInDataFloat.Add(2f); + device.InvokeStreamMessage(full); + + Assert.Equal(new[] { 1f, 2f }, Assert.Single(rawFrames).AnalogInDataFloat); + Assert.Equal(1, classified); + } + + [Fact] + public void Decode_WarmupFrame_ReportsDiscardWithCounts() + { + // A suppressed frame must be observable: a consumer counting samples has to be able to + // tell "Core dropped a malformed frame" from "the device sent nothing". + var device = CreateStreamingDevice(analogCount: 4); + for (var n = 0; n < 4; n++) AnalogChannel(device, n).IsEnabled = true; + device.StartStreaming(); + + var discards = new List(); + device.StreamFrameDiscarded += (_, e) => discards.Add(e); + + var warmup = new DaqifiOutMessage { MsgTimeStamp = 1836224389 }; + warmup.AnalogInDataFloat.Add(2f); // the bench evidence: 1 value for 4 enabled channels + device.InvokeStreamMessage(warmup); + + var discarded = Assert.Single(discards); + Assert.Equal(StreamFrameDiscardReason.PartialAnalogFrame, discarded.Reason); + Assert.Equal(1836224389u, discarded.DeviceTimestamp); + Assert.Equal(1, discarded.AnalogValueCount); + Assert.Equal(4, discarded.EnabledAnalogChannelCount); + Assert.Equal(1, device.DiscardedStreamFrameCount); + } + + [Fact] + public void Decode_WellFormedFirstFrame_PassesThroughCompletelyUnchanged() + { + // The guard is meant to be safe to leave in permanently, including on firmware that no + // longer emits the malformed frame: a well-formed first frame must reach both consumer + // paths, with the same object and the same values, and report no discard at all. + var device = CreateStreamingDevice(analogCount: 2); + var ai0 = AnalogChannel(device, 0); + var ai1 = AnalogChannel(device, 1); + ai0.IsEnabled = true; + ai1.IsEnabled = true; + device.StartStreaming(); + + var rawFrames = new List(); + device.MessageReceived += (_, e) => rawFrames.Add(e.Message.Data); + var discards = 0; + device.StreamFrameDiscarded += (_, _) => discards++; + + var first = new DaqifiOutMessage { MsgTimeStamp = 1000 }; + first.AnalogInDataFloat.Add(4f); + first.AnalogInDataFloat.Add(8f); + device.InvokeStreamMessage(first); + + Assert.Same(first, Assert.Single(rawFrames)); + Assert.Equal(4.0, ai0.ActiveSample!.Value); + Assert.Equal(8.0, ai1.ActiveSample!.Value); + Assert.Equal(0, discards); + Assert.Equal(0, device.DiscardedStreamFrameCount); + } + + [Fact] + public void Decode_SingleEnabledAnalogChannel_FirstFrameNotSuppressed() + { + // One enabled channel and one analog value is a *complete* frame, not a partial one. This + // is the case the "analog count < enabled count" rule must never get wrong, because it is + // indistinguishable from the malformed frame by value count alone. + var device = CreateStreamingDevice(analogCount: 4); + var ai0 = AnalogChannel(device, 0); + ai0.IsEnabled = true; // exactly one enabled + device.StartStreaming(); + + var rawFrames = 0; + device.MessageReceived += (_, _) => rawFrames++; + + var first = new DaqifiOutMessage { MsgTimeStamp = 1000 }; + first.AnalogInDataFloat.Add(3.5f); + device.InvokeStreamMessage(first); + Assert.Equal(1, rawFrames); + Assert.Equal(3.5, ai0.ActiveSample!.Value); + Assert.Equal(0, device.DiscardedStreamFrameCount); } [Fact] @@ -674,20 +769,23 @@ public void Decode_WarmupGuardReArmsForEachSession() ai0.IsEnabled = true; ai1.IsEnabled = true; + device.StreamingFrequency = 100; // 500_000 ticks per sample at the 50 MHz default + // Session 1: warmup + a full frame. device.StartStreaming(); - var w1 = new DaqifiOutMessage { MsgTimeStamp = 1 }; + var w1 = new DaqifiOutMessage { MsgTimeStamp = 500_000 }; w1.AnalogInDataFloat.Add(0.1f); device.InvokeStreamMessage(w1); - var f1 = new DaqifiOutMessage { MsgTimeStamp = 2 }; + var f1 = new DaqifiOutMessage { MsgTimeStamp = 1_000_000 }; f1.AnalogInDataFloat.Add(1f); f1.AnalogInDataFloat.Add(2f); device.InvokeStreamMessage(f1); device.StopStreaming(); - // Session 2: a fresh warmup frame must again be suppressed. + // Session 2 starts well past the leftover window (a real stop/start gap), so its first + // frame is genuine — and, being partial, must again be suppressed as a warmup frame. device.StartStreaming(); - var w2 = new DaqifiOutMessage { MsgTimeStamp = 3 }; + var w2 = new DaqifiOutMessage { MsgTimeStamp = 60_000_000 }; w2.AnalogInDataFloat.Add(5f); // single value -> partial again device.InvokeStreamMessage(w2); @@ -753,6 +851,147 @@ public void Decode_PersistentShortFrames_ReleasedAfterCap() #endregion + #region Cross-session leftover frames (firmware #533) + + [Fact] + public void Decode_LeftoverFrameFromPreviousSession_DiscardedFromBothConsumerPaths() + { + // The device latches the last frame of a stopped session and emits it as the first frame of + // the next one, one sample period after the session it actually belongs to. It must not + // reach consumers or anchor the new session's clock. + var device = CreateStreamingDevice(analogCount: 2); + var ai0 = AnalogChannel(device, 0); + var ai1 = AnalogChannel(device, 1); + ai0.IsEnabled = true; + ai1.IsEnabled = true; + device.StreamingFrequency = 100; // 500_000 ticks per sample + + // Session 1 ends with a frame at tick 10_000_000. + device.StartStreaming(); + var last = new DaqifiOutMessage { MsgTimeStamp = 10_000_000 }; + last.AnalogInDataFloat.Add(1f); + last.AnalogInDataFloat.Add(2f); + device.InvokeStreamMessage(last); + device.StopStreaming(); + + // Session 2 starts a long while later, but the first frame that arrives carries a counter + // one sample period past session 1's last frame — the latched leftover. + device.StartStreaming(); + var rawFrames = 0; + device.MessageReceived += (_, _) => rawFrames++; + var discards = new List(); + device.StreamFrameDiscarded += (_, e) => discards.Add(e); + + var leftover = new DaqifiOutMessage { MsgTimeStamp = 10_500_000 }; + leftover.AnalogInDataFloat.Add(98f); + leftover.AnalogInDataFloat.Add(99f); + device.InvokeStreamMessage(leftover); + + Assert.Equal(0, rawFrames); + Assert.Equal(StreamFrameDiscardReason.StaleLeftoverFrame, Assert.Single(discards).Reason); + Assert.Equal(1.0, ai0.ActiveSample!.Value); // still session 1's values + Assert.Equal(2.0, ai1.ActiveSample!.Value); + + // The genuine first frame of session 2 follows and is delivered normally. + var genuine = new DaqifiOutMessage { MsgTimeStamp = 400_000_000 }; + genuine.AnalogInDataFloat.Add(5f); + genuine.AnalogInDataFloat.Add(6f); + device.InvokeStreamMessage(genuine); + + Assert.Equal(1, rawFrames); + Assert.Equal(5.0, ai0.ActiveSample!.Value); + Assert.Equal(6.0, ai1.ActiveSample!.Value); + Assert.Equal(1, device.DiscardedStreamFrameCount); + } + + [Fact] + public void Decode_QuickRestart_DoesNotDiscardGenuineFrames() + { + // The leftover window scales with the sample period, so a restart that takes longer than a + // couple of sample periods is never mistaken for a leftover. At 100 Hz the window is 25 ms; + // this restart gap is 200 ms. + var device = CreateStreamingDevice(analogCount: 2); + var ai0 = AnalogChannel(device, 0); + ai0.IsEnabled = true; + AnalogChannel(device, 1).IsEnabled = true; + device.StreamingFrequency = 100; + + device.StartStreaming(); + var last = new DaqifiOutMessage { MsgTimeStamp = 10_000_000 }; + last.AnalogInDataFloat.Add(1f); + last.AnalogInDataFloat.Add(2f); + device.InvokeStreamMessage(last); + device.StopStreaming(); + + device.StartStreaming(); + var discards = 0; + device.StreamFrameDiscarded += (_, _) => discards++; + + // 200 ms later at 50 MHz = 10_000_000 ticks on. + var genuine = new DaqifiOutMessage { MsgTimeStamp = 20_000_000 }; + genuine.AnalogInDataFloat.Add(7f); + genuine.AnalogInDataFloat.Add(8f); + device.InvokeStreamMessage(genuine); + + Assert.Equal(0, discards); + Assert.Equal(7.0, ai0.ActiveSample!.Value); + } + + [Fact] + public void Decode_LeftoverGuardInactiveOnFirstSession() + { + // With no counter value from before the session there is nothing to compare against, so the + // first session after connect delivers its first frame untouched rather than guessing. + var device = CreateStreamingDevice(analogCount: 1); + var ai0 = AnalogChannel(device, 0); + ai0.IsEnabled = true; + device.StartStreaming(); + + var discards = 0; + device.StreamFrameDiscarded += (_, _) => discards++; + + var first = new DaqifiOutMessage { MsgTimeStamp = 7 }; + first.AnalogInDataFloat.Add(3f); + device.InvokeStreamMessage(first); + + Assert.Equal(0, discards); + Assert.Equal(3.0, ai0.ActiveSample!.Value); + } + + [Fact] + public void Decode_FrameArrivingWhileStopped_SeedsTheLeftoverReference() + { + // The device can emit a final frame after the stop command lands, and the frame latched for + // the next session follows *that* one. A frame received while stopped is still re-raised to + // raw consumers, but it must also update the reference the next session is checked against. + var device = CreateStreamingDevice(analogCount: 1); + var ai0 = AnalogChannel(device, 0); + ai0.IsEnabled = true; + device.StreamingFrequency = 100; + + var trailing = new DaqifiOutMessage { MsgTimeStamp = 10_000_000 }; + trailing.AnalogInDataFloat.Add(1f); + + var rawFrames = 0; + device.MessageReceived += (_, _) => rawFrames++; + device.InvokeStreamMessage(trailing); // arrives while not streaming + Assert.Equal(1, rawFrames); + Assert.Null(ai0.ActiveSample); // re-raised but not decoded + + device.StartStreaming(); + var discards = 0; + device.StreamFrameDiscarded += (_, _) => discards++; + + var leftover = new DaqifiOutMessage { MsgTimeStamp = 10_500_000 }; + leftover.AnalogInDataFloat.Add(99f); + device.InvokeStreamMessage(leftover); + + Assert.Equal(1, discards); + Assert.Null(ai0.ActiveSample); + } + + #endregion + #region Helpers private static DecodableStreamingDevice CreateStreamingDevice( diff --git a/src/Daqifi.Core.Tests/Device/Internal/StreamFrameGateTests.cs b/src/Daqifi.Core.Tests/Device/Internal/StreamFrameGateTests.cs new file mode 100644 index 00000000..db448574 --- /dev/null +++ b/src/Daqifi.Core.Tests/Device/Internal/StreamFrameGateTests.cs @@ -0,0 +1,177 @@ +using Daqifi.Core.Communication.Messages; +using Daqifi.Core.Device.Internal; +using Xunit; + +namespace Daqifi.Core.Tests.Device.Internal; + +/// +/// Unit tests for , the cross-session leftover-frame guard +/// (daqifi-nyquist-firmware #533). A tick frequency of 1000 Hz is used throughout so one tick is +/// one millisecond and the window arithmetic can be read off the timestamps directly. +/// +public class StreamFrameGateTests +{ + private const uint TicksPerSecond = 1000; + + [Fact] + public void IsValidating_FalseWithoutACounterReference() + { + // Nothing was ever seen, so there is nothing to compare a first frame against: the gate + // stands aside rather than guessing. + var gate = new StreamFrameGate(); + gate.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); + + Assert.False(gate.IsValidating); + } + + [Fact] + public void IsValidating_TrueOnceAFrameHasBeenSeen() + { + var gate = new StreamFrameGate(); + gate.TrackFrame(1000); + gate.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); + + Assert.True(gate.IsValidating); + } + + [Fact] + public void LeftoverInsideTheWindow_IsRejected_AndAGenuineFrameOpensTheGate() + { + // 1 Hz -> 1000-tick sample period -> 2500-tick window. + var gate = new StreamFrameGate(); + gate.TrackFrame(10_000); + gate.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); + + // Latched frame: one sample period past the previous session's last frame. + Assert.True(gate.IsLeftoverFromPreviousSession(Frame(11_000))); + Assert.Equal(1, gate.DiscardedFrameCount); + Assert.True(gate.IsValidating); + + // Genuine first frame: a full stop-to-start gap on. + Assert.False(gate.IsLeftoverFromPreviousSession(Frame(40_000))); + Assert.False(gate.IsValidating); // gate steps aside for the rest of the session + } + + [Fact] + public void CounterWrap_GenuineFrameAfterTheWrapIsNotMistakenForALeftover() + { + // The counter is a uint that wraps. Measured with modular subtraction this frame is 11_001 + // ticks past the reference and therefore genuine; measured naively it looks like a large + // negative delta, which would put it inside the window and drop real data. + var gate = new StreamFrameGate(); + gate.TrackFrame(uint.MaxValue - 1000); + gate.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); + + Assert.False(gate.IsLeftoverFromPreviousSession(Frame(10_000))); + Assert.Equal(0, gate.DiscardedFrameCount); + } + + [Fact] + public void CounterWrap_LeftoverStraddlingTheWrapIsStillRejected() + { + // Reference sits 1001 ticks below the wrap; the latched frame lands 500 ticks past it, + // 1501 ticks on in modular terms — inside the 2500-tick window. + var gate = new StreamFrameGate(); + gate.TrackFrame(uint.MaxValue - 1000); + gate.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); + + Assert.True(gate.IsLeftoverFromPreviousSession(Frame(500))); + Assert.Equal(1, gate.DiscardedFrameCount); + } + + [Fact] + public void WindowScalesWithTheSampleRate() + { + // The same 300-tick delta is well inside one sample period at 1 Hz and several periods on + // at 10 Hz. A fixed seconds-based window would drop the second case. + var slow = new StreamFrameGate(); + slow.TrackFrame(10_000); + slow.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); // window 2500 ticks + Assert.True(slow.IsLeftoverFromPreviousSession(Frame(10_300))); + + var fast = new StreamFrameGate(); + fast.TrackFrame(10_000); + fast.BeginSession(TicksPerSecond, streamingFrequencyHz: 10); // window 250 ticks + Assert.False(fast.IsLeftoverFromPreviousSession(Frame(10_300))); + } + + [Fact] + public void UnknownRate_FallsBackToTheWidestWindow() + { + // A session started without a usable rate assumes the device's slowest, which is the + // conservative choice: the widest window. + var gate = new StreamFrameGate(); + gate.TrackFrame(10_000); + gate.BeginSession(TicksPerSecond, streamingFrequencyHz: 0); + + Assert.True(gate.IsLeftoverFromPreviousSession(Frame(12_000))); // inside 2500 + Assert.False(gate.IsLeftoverFromPreviousSession(Frame(13_000))); // 3000 ticks on: genuine + } + + [Fact] + public void UnknownTickFrequency_FallsBackToTheDefault() + { + // 50 MHz default tick rate at 1 Hz -> a 125_000_000-tick window. + var gate = new StreamFrameGate(); + gate.TrackFrame(0); + gate.BeginSession(timestampFrequency: 0, streamingFrequencyHz: 1); + + Assert.True(gate.IsLeftoverFromPreviousSession(Frame(100_000_000))); + Assert.False(gate.IsLeftoverFromPreviousSession(Frame(130_000_000))); + } + + [Fact] + public void QuickRestart_DiscardsABoundedPrefixInsteadOfCascading() + { + // Worst case for the heuristic: a restart so fast that genuine frames land inside the + // window. Because every frame is measured against the counter fixed at session start, the + // deltas grow by one period each time and the run ends on its own — two discards here, not + // an unbounded cascade. (Measuring against the *previous* frame instead would discard every + // frame in the session, each sitting one period from its discarded predecessor.) + var gate = new StreamFrameGate(); + gate.TrackFrame(0); + gate.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); // window 2500 ticks + + Assert.True(gate.IsLeftoverFromPreviousSession(Frame(1000))); + Assert.True(gate.IsLeftoverFromPreviousSession(Frame(2000))); + Assert.False(gate.IsLeftoverFromPreviousSession(Frame(3000))); + Assert.Equal(2, gate.DiscardedFrameCount); + } + + [Fact] + public void DiscardsAreCappedSoAStreamCanNeverBeWithheldIndefinitely() + { + // A device whose counter barely advances would otherwise be discarded forever. After the + // cap the gate gives up and lets everything through. + var gate = new StreamFrameGate(); + gate.TrackFrame(0); + gate.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); + + for (uint ts = 1; ts <= StreamFrameGate.MaxDiscardedFrames; ts++) + { + Assert.True(gate.IsLeftoverFromPreviousSession(Frame(ts))); + } + + Assert.Equal(StreamFrameGate.MaxDiscardedFrames, gate.DiscardedFrameCount); + Assert.False(gate.IsLeftoverFromPreviousSession(Frame(StreamFrameGate.MaxDiscardedFrames + 1))); + Assert.False(gate.IsValidating); + } + + [Fact] + public void BeginSession_ResetsTheDiscardCountAndRe_AnchorsTheReference() + { + var gate = new StreamFrameGate(); + gate.TrackFrame(0); + gate.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); + Assert.True(gate.IsLeftoverFromPreviousSession(Frame(1000))); + Assert.Equal(1, gate.DiscardedFrameCount); + + // The next session is anchored to the last frame seen (1000), not to the old reference. + gate.BeginSession(TicksPerSecond, streamingFrequencyHz: 1); + Assert.Equal(0, gate.DiscardedFrameCount); + Assert.True(gate.IsLeftoverFromPreviousSession(Frame(2000))); // 1000 ticks past 1000 + Assert.False(gate.IsLeftoverFromPreviousSession(Frame(60_000))); + } + + private static DaqifiOutMessage Frame(uint timestamp) => new() { MsgTimeStamp = timestamp }; +} diff --git a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs index 4166eed2..2099ad65 100644 --- a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs @@ -62,6 +62,12 @@ public class DaqifiStreamingDevice : DaqifiDevice, IStreamingDevice, INetworkCon /// private readonly TimestampGapDetector _gapDetector = new(); + /// + /// Keeps frames the device latched from the previous streaming session out of this one + /// (daqifi-nyquist-firmware #533). Re-armed by . + /// + private readonly StreamFrameGate _frameGate = new(); + /// /// The maximum number of leading short-analog frames suppressed at stream start /// (see ). Bounds the warmup-frame guard so a @@ -89,6 +95,37 @@ public class DaqifiStreamingDevice : DaqifiDevice, IStreamingDevice, INetworkCon /// private int _suppressedWarmupFrameCount; + /// + /// Backing counter for . + /// + private long _discardedStreamFrameCount; + + /// + /// Raised when a stream frame was withheld from consumers because the device should not have + /// sent it — a malformed leading frame, or one latched from the previous session. + /// + /// + /// + /// Suppressing a bad frame is the right thing to do, but doing it invisibly is not: a + /// consumer counting samples or reconciling against the device's own frame count needs to + /// tell "Core dropped a bad frame" apart from "the device sent nothing". This event, and the + /// running , are that signal. + /// + /// + /// A subscriber exception is caught and traced rather than propagated, so a misbehaving + /// handler cannot disturb the frame that follows. + /// + /// + public event EventHandler? StreamFrameDiscarded; + + /// + /// Gets the number of stream frames withheld from consumers since the current streaming + /// session began. A healthy stream on firmware without the device-side defects leaves it at + /// zero; on firmware 3.7.2 it is typically one, for the malformed leading frame. + /// + /// Reset by , so it describes the current session. + public long DiscardedStreamFrameCount => Interlocked.Read(ref _discardedStreamFrameCount); + /// /// Backing counter for . /// @@ -366,6 +403,11 @@ private void BeginStreamingSession() _awaitingFirstFullAnalogFrame = CountEnabledAnalogChannels(SnapshotChannels()) > 0; _suppressedWarmupFrameCount = 0; Interlocked.Exchange(ref _decodeFailureCount, 0); + + // Re-arm the cross-session leftover guard against the counter value this session + // inherits, and at this session's rate — the window is measured in sample periods. + _frameGate.BeginSession(TimestampFrequency, StreamingFrequency); + Interlocked.Exchange(ref _discardedStreamFrameCount, 0); } /// @@ -811,27 +853,74 @@ void OnSample(object? sender, SampleReceivedEventArgs e) => } /// - /// Handles a streaming data frame: re-raises it for raw-frame consumers (via the base - /// implementation) and, while streaming, decodes it into per-channel samples that drive + /// Handles a streaming data frame: screens out the frames the device should not have sent, + /// then re-raises the frame for raw-frame consumers (via the base implementation) and, while + /// streaming, decodes it into per-channel samples that drive /// . /// + /// + /// Screening covers both consumer paths, which is the whole point of issue #425. The + /// per-channel decode has guarded against the firmware's malformed leading frame since + /// issue #351, but the raw event was still handed + /// the frame verbatim — and that is the path most callers actually use, including the + /// example CLI, whose offline export inferred a channel count of one from it and truncated + /// every sample that followed. Whatever is unfit for the decoded path is unfit for the raw + /// one; both are now gated together, and every drop is reported through + /// . + /// /// The streaming message from the device. protected override void OnStreamMessageReceived(DaqifiOutMessage message) { - // Preserve the raw-frame MessageReceived event so existing consumers that hand-demux - // the protobuf frame keep working unchanged. - base.OnStreamMessageReceived(message); + if (IsStreaming && _frameGate.IsValidating && _frameGate.IsLeftoverFromPreviousSession(message)) + { + RaiseStreamFrameDiscarded(StreamFrameDiscardReason.StaleLeftoverFrame, message); + return; + } + + _frameGate.TrackFrame(message.MsgTimeStamp); // Only decode into channel samples while an app-driven stream is active. A stray frame - // that arrives outside a streaming session is still re-raised above but not decoded. + // that arrives outside a streaming session is still re-raised but not decoded. if (!IsStreaming) { + base.OnStreamMessageReceived(message); return; } + EmitStreamFrame(message); + } + + /// + /// Delivers a frame that has cleared : re-raises it for raw-frame + /// consumers and decodes it into per-channel samples. + /// + /// + /// The firmware's malformed leading frame (issue #351) is caught here rather than by the + /// gate, because it needs the enabled-channel count and because what it costs is narrower: + /// only the analog payload is unusable. Such a frame is withheld from raw consumers — they + /// read AnalogInData straight off it, so there is no way to hand it over safely — but + /// its digital payload is still decoded and its timestamp still anchors the session clock, + /// exactly as before. + /// + /// The frame to deliver. + private void EmitStreamFrame(DaqifiOutMessage message) + { + var suppressAnalog = _awaitingFirstFullAnalogFrame && ShouldSuppressPartialAnalog(message); + + if (suppressAnalog) + { + RaiseStreamFrameDiscarded(StreamFrameDiscardReason.PartialAnalogFrame, message); + } + else + { + // Preserve the raw-frame MessageReceived event so existing consumers that hand-demux + // the protobuf frame keep working unchanged. + base.OnStreamMessageReceived(message); + } + try { - DecodeStreamFrame(message); + DecodeStreamFrame(message, suppressAnalog); } catch (Exception ex) { @@ -847,13 +936,87 @@ protected override void OnStreamMessageReceived(DaqifiOutMessage message) } } + /// + /// Decides whether a leading frame's analog payload is the firmware's malformed warmup frame + /// (issue #351): fewer analog values than there are enabled analog channels. Disarms the + /// guard on the first analog-bearing frame that is not short, or once + /// have been suppressed. + /// + /// The frame about to be delivered. + /// true when the frame's analog values must be withheld. + private bool ShouldSuppressPartialAnalog(DaqifiOutMessage message) + { + var analogValueCount = message.AnalogInDataFloat.Count > 0 + ? message.AnalogInDataFloat.Count + : message.AnalogInData.Count; + + // A frame with no analog payload says nothing about the warmup frame either way, so the + // guard stays armed for the first analog-bearing frame. + if (analogValueCount == 0) + { + return false; + } + + var enabledAnalogCount = CountEnabledAnalogChannels(SnapshotChannels()); + if (enabledAnalogCount > 0 + && analogValueCount < enabledAnalogCount + && _suppressedWarmupFrameCount < MaxSuppressedWarmupFrames) + { + _suppressedWarmupFrameCount++; + return true; + } + + _awaitingFirstFullAnalogFrame = false; + return false; + } + + /// + /// Raises for a frame that was withheld, and counts it. + /// Subscriber exceptions are isolated, mirroring . + /// + /// Why the frame was withheld. + /// The frame that was withheld. + private void RaiseStreamFrameDiscarded(StreamFrameDiscardReason reason, DaqifiOutMessage frame) + { + Interlocked.Increment(ref _discardedStreamFrameCount); + + var handler = StreamFrameDiscarded; + if (handler == null) + { + return; + } + + var analogValueCount = frame.AnalogInDataFloat.Count > 0 + ? frame.AnalogInDataFloat.Count + : frame.AnalogInData.Count; + + try + { + handler(this, new StreamFrameDiscardedEventArgs( + reason, + frame.MsgTimeStamp, + analogValueCount, + CountEnabledAnalogChannels(SnapshotChannels()))); + } + catch (Exception ex) + { + Trace.WriteLine($"[{nameof(StreamFrameDiscarded)}] Subscriber threw: {ex}"); + } + } + /// /// Decodes a streaming frame into per-channel samples: selects the active channels in /// device order, chooses the correct value source (USB pre-scaled float vs. WiFi raw ADC /// count scaled via calibration), unpacks digital bits, and pushes a sample to each channel. /// /// The streaming message to decode. - private void DecodeStreamFrame(DaqifiOutMessage message) + /// + /// When true, the frame's analog payload is the firmware's malformed warmup frame + /// (issue #351) and is skipped. Only the analog values are withheld — a combined frame's + /// digital payload is still decoded, and the frame's (normal one-period) timestamp still + /// anchors the session clock, so digital state and edges are not lost. + /// + private void DecodeStreamFrame(DaqifiOutMessage message, bool suppressAnalog) { var hasFloat = message.AnalogInDataFloat.Count > 0; var hasRawAnalog = message.AnalogInData.Count > 0; @@ -868,30 +1031,6 @@ private void DecodeStreamFrame(DaqifiOutMessage message) // thread that runs this decode, so the structure is stable for the duration of the call. var channels = SnapshotChannels(); - // Suppress the firmware's malformed warmup frame at stream start (issue #351): its fast - // streaming encoder can emit a leading analog-bearing frame with fewer values than the - // enabled channel mask. Only the malformed *analog* values are withheld — a combined - // frame's digital payload is still decoded, and the frame's (normal one-period) - // timestamp still anchors the session clock, so digital state/edges are not lost. Only - // leading short frames are suppressed (mid-stream short frames stay best-effort mapped), - // bounded so a genuinely short stream is never withheld indefinitely. - var suppressWarmupAnalog = false; - if (_awaitingFirstFullAnalogFrame && (hasFloat || hasRawAnalog)) - { - var analogValueCount = hasFloat ? message.AnalogInDataFloat.Count : message.AnalogInData.Count; - var enabledAnalogCount = CountEnabledAnalogChannels(channels); - if (enabledAnalogCount > 0 && analogValueCount < enabledAnalogCount - && _suppressedWarmupFrameCount < MaxSuppressedWarmupFrames) - { - _suppressedWarmupFrameCount++; - suppressWarmupAnalog = true; - } - else - { - _awaitingFirstFullAnalogFrame = false; - } - } - // Reconstruct a host timestamp from the device tick counter (rollover-aware) and carry // the raw device tick value through to each decoded sample. var deviceTimestamp = message.MsgTimeStamp; @@ -908,7 +1047,7 @@ private void DecodeStreamFrame(DaqifiOutMessage message) hostTimestamp, timestampResult.SecondsBetweenMessages, deviceTimestamp)); } - if ((hasFloat || hasRawAnalog) && !suppressWarmupAnalog) + if ((hasFloat || hasRawAnalog) && !suppressAnalog) { DecodeAnalog(message, channels, hostTimestamp, deviceTimestamp, hasFloat); } diff --git a/src/Daqifi.Core/Device/Internal/StreamFrameGate.cs b/src/Daqifi.Core/Device/Internal/StreamFrameGate.cs new file mode 100644 index 00000000..bcea3e27 --- /dev/null +++ b/src/Daqifi.Core/Device/Internal/StreamFrameGate.cs @@ -0,0 +1,168 @@ +using Daqifi.Core.Communication.Messages; + +#nullable enable + +namespace Daqifi.Core.Device.Internal +{ + /// + /// Keeps the frame the device latched from the previous streaming session out of the current one + /// (daqifi-nyquist-firmware #533). + /// + /// + /// + /// The device holds the final frame of a stopped session in its transmit path and emits it as + /// the first frame of the next session. Its free-running tick counter is never reset, so that + /// frame arrives carrying a counter value from the session before — and left alone it prepends a + /// stale sample to the capture and anchors the new session's clock to a time that never + /// happened. + /// + /// + /// The counter is what tells the two apart. A latched frame sits about one sample period past + /// the last counter value seen before the session began; a genuine first frame is offset by the + /// whole stop-to-start gap. Comparisons use modular subtraction, so they stay + /// correct across the counter's wrap (about 86 s at the 50 MHz default tick rate). + /// + /// + /// Known limitation. This needs a counter value from before the session, which means the + /// very first session after a connect is unprotected: the device's latched frame survives a + /// disconnect, but a freshly connected instance has nothing to recognize it against. Every + /// session from the second onward — which is the stop/start case #533 describes — is covered. + /// The alternative, holding the first frame of the first session until a second frame can vouch + /// for it, was considered and rejected: it delays every consumer's first sample to defend + /// against a frame that has never been observed on current firmware. + /// + /// + /// This type is not thread-safe by itself. It is driven from the message-consumer thread that + /// decodes frames and re-armed from the caller thread that starts a session — the same + /// arrangement, and the same trade, as the timestamp processor and gap detector it sits beside. + /// + /// + internal sealed class StreamFrameGate + { + /// + /// How far past the session-start counter, measured in sample periods, a frame can sit and + /// still be treated as belonging to the previous session. + /// + /// + /// The latched frame sits one sample period past the previous session's last frame, so the + /// window has to clear one period with margin. Scaling it by the sample period rather than + /// fixing it in seconds is what keeps a quick stop/start from being mistaken for a leftover: + /// at 20 Hz the window is 125 ms, not the 2.5 s a fixed window would impose. + /// + internal const double LeftoverWindowSamplePeriods = 2.5; + + /// + /// Rate assumed when the streaming frequency is unknown, in Hz. The device's slowest rate, + /// which makes the window its widest — the conservative choice when there is nothing better + /// to go on. + /// + internal const int FallbackStreamingFrequencyHz = 1; + + /// + /// Hard cap on discards per session, so a device whose counter behaves in a way this gate + /// did not anticipate loses a bounded prefix of one session rather than the whole stream. + /// + /// + /// The window arithmetic already bounds discards on its own: comparisons are made against a + /// counter reference fixed at session start, so successive frames march steadily out of the + /// window and at most of them can fall inside it. + /// Measuring against the last frame seen instead would let a quick restart cascade — every + /// genuine frame sitting one period from its discarded predecessor, and discarded in turn — + /// which is the trap this cap exists to backstop. + /// + internal const int MaxDiscardedFrames = 5; + + private uint _lastSeenDeviceTimestamp; + private bool _hasDeviceTimestampReference; + + private uint _sessionStartReference; + private bool _checkForLeftoverFrames; + private int _discardedFrameCount; + private double _leftoverWindowTicks; + + /// + /// Gets a value indicating whether the gate still has something to decide. False for the + /// overwhelming majority of frames, which lets the caller skip evaluation entirely. + /// + public bool IsValidating => _checkForLeftoverFrames; + + /// + /// Gets the number of frames discarded since the current session began. + /// + public int DiscardedFrameCount => _discardedFrameCount; + + /// + /// Re-arms the gate for a streaming session that is about to begin. + /// + /// + /// The device's tick frequency in Hz; zero falls back to + /// . + /// + /// + /// The rate the session is starting at; zero or negative falls back to + /// . + /// + public void BeginSession(uint timestampFrequency, int streamingFrequencyHz) + { + var ticksPerSecond = timestampFrequency != 0 + ? timestampFrequency + : TimestampProcessor.DefaultTimestampFrequency; + var rate = streamingFrequencyHz > 0 ? streamingFrequencyHz : FallbackStreamingFrequencyHz; + _leftoverWindowTicks = LeftoverWindowSamplePeriods * (ticksPerSecond / (double)rate); + + _discardedFrameCount = 0; + _sessionStartReference = _lastSeenDeviceTimestamp; + _checkForLeftoverFrames = _hasDeviceTimestampReference; + } + + /// + /// Records a frame's counter value without evaluating it. Used for frames that arrive + /// outside a streaming session and for frames that arrive once the gate is satisfied. + /// + /// + /// Frames that arrive while no session is running matter as much as any other: the device + /// can emit a final frame after a stop command lands, and the frame latched for the next + /// session follows that one by a sample period. Leaving it out of the reference would + /// aim the next session's window at the wrong counter value. + /// + /// The frame's raw 32-bit counter value. + public void TrackFrame(uint deviceTimestamp) + { + _lastSeenDeviceTimestamp = deviceTimestamp; + _hasDeviceTimestampReference = true; + } + + /// + /// Decides whether a frame that arrived while is true belongs to + /// the previous streaming session. + /// + /// The frame that just arrived. + /// + /// true when the frame is a leftover and must be dropped; false when it is + /// genuine, in which case the gate steps out of the way for the rest of the session. + /// + public bool IsLeftoverFromPreviousSession(DaqifiOutMessage message) + { + var deviceTimestamp = message.MsgTimeStamp; + + if (_discardedFrameCount < MaxDiscardedFrames + && TicksSince(_sessionStartReference, deviceTimestamp) < _leftoverWindowTicks) + { + _discardedFrameCount++; + TrackFrame(deviceTimestamp); + return true; + } + + _checkForLeftoverFrames = false; + TrackFrame(deviceTimestamp); + return false; + } + + /// + /// Device ticks from to , using modular + /// subtraction so the result stays correct when the counter wraps between + /// the two. + /// + private static double TicksSince(uint from, uint to) => unchecked(to - from); + } +} diff --git a/src/Daqifi.Core/Device/StreamFrameDiscardReason.cs b/src/Daqifi.Core/Device/StreamFrameDiscardReason.cs new file mode 100644 index 00000000..38d140e1 --- /dev/null +++ b/src/Daqifi.Core/Device/StreamFrameDiscardReason.cs @@ -0,0 +1,39 @@ +namespace Daqifi.Core.Device +{ + /// + /// Why withheld a stream frame from its consumers, reported + /// by . + /// + /// + /// Every value here describes a device-side defect that host-side protection papers over, so a + /// discard is never a Core bug report — it is Core telling you the wire carried something the + /// device should not have sent. + /// + public enum StreamFrameDiscardReason + { + /// + /// The frame carried fewer analog values than the number of enabled analog channels. + /// + /// + /// Firmware up to and including 3.7.2 emits a leading frame at stream start whose analog + /// payload holds a single value regardless of the enabled channel mask (daqifi-core #351, + /// daqifi-nyquist-firmware #707). Consumers that infer channel width from the first sample + /// would silently truncate the whole capture, so the malformed analog payload is withheld. + /// Any digital payload in the same frame is still delivered, and the frame's timestamp still + /// anchors the session clock. + /// + PartialAnalogFrame, + + /// + /// The frame belonged to the previous streaming session. + /// + /// + /// The device latches the final frame of a stopped session in its transmit path and emits it + /// as the first frame of the next one (daqifi-nyquist-firmware #533). Its device-tick counter + /// sits about one sample period after the last frame of the previous session rather than at + /// the new session's start, so it would anchor the session clock to a time that never + /// happened. + /// + StaleLeftoverFrame, + } +} diff --git a/src/Daqifi.Core/Device/StreamFrameDiscardedEventArgs.cs b/src/Daqifi.Core/Device/StreamFrameDiscardedEventArgs.cs new file mode 100644 index 00000000..4c6928ad --- /dev/null +++ b/src/Daqifi.Core/Device/StreamFrameDiscardedEventArgs.cs @@ -0,0 +1,57 @@ +using System; + +namespace Daqifi.Core.Device +{ + /// + /// Carries information about a stream frame that withheld + /// from its consumers because the device should not have sent it. + /// + /// + /// This exists so a discard is never invisible. Silently dropping a frame is the right thing to + /// do with a malformed one, but a consumer counting samples, watching for dropouts, or + /// reconciling against the device's own frame count needs to be able to tell the difference + /// between "Core suppressed a bad frame" and "the device sent nothing". + /// + public sealed class StreamFrameDiscardedEventArgs : EventArgs + { + /// + /// Gets the reason the frame was withheld. + /// + public StreamFrameDiscardReason Reason { get; } + + /// + /// Gets the raw device tick counter value carried by the discarded frame. + /// + public uint DeviceTimestamp { get; } + + /// + /// Gets the number of analog values the discarded frame carried. + /// + public int AnalogValueCount { get; } + + /// + /// Gets the number of analog channels enabled when the frame arrived — the number of values + /// a well-formed frame would have carried. + /// + public int EnabledAnalogChannelCount { get; } + + /// + /// Initializes a new instance of the class. + /// + /// Why the frame was withheld. + /// The discarded frame's raw device tick counter value. + /// How many analog values the discarded frame carried. + /// How many analog channels were enabled at the time. + public StreamFrameDiscardedEventArgs( + StreamFrameDiscardReason reason, + uint deviceTimestamp, + int analogValueCount, + int enabledAnalogChannelCount) + { + Reason = reason; + DeviceTimestamp = deviceTimestamp; + AnalogValueCount = analogValueCount; + EnabledAnalogChannelCount = enabledAnalogChannelCount; + } + } +} From e256014424b885eb7dddb083c308ecc9500b831f Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Sun, 2 Aug 2026 12:52:05 -0600 Subject: [PATCH 2/3] fix(device): keep trace listeners out of the isolation path, report discard counts from the decision MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses both Qodo findings on #428. Trace dispatches to consumer-installed listeners, so logging from inside the catch that contains a bad subscriber is itself consumer code — a throwing listener escapes the containment and takes down the frame pipeline the catch was protecting. Routed through a new SafeTrace helper, mirroring the guarantee DaqifiDevice.SafeLog already gives RaiseClassifiedEvent. Applied to the two other Trace sites in the file as well: RaiseGapDetected had the same unguarded pattern I copied, and TrackStreamingStart documents that tracking a command must never fail the send that carried it. StreamFrameDiscarded now reports the analog and enabled-channel counts the suppression decision was actually made on, instead of re-reading channel state that another thread may have changed in between. Self-inconsistent telemetry would undermine the observability this PR exists to add. Co-Authored-By: Claude Opus 5 --- .../DaqifiStreamingDeviceDecodeTests.cs | 67 +++++++++- .../Device/DaqifiStreamingDevice.cs | 123 ++++++++++++++---- 2 files changed, 166 insertions(+), 24 deletions(-) diff --git a/src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceDecodeTests.cs b/src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceDecodeTests.cs index c4d5269b..62d39c01 100644 --- a/src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceDecodeTests.cs +++ b/src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceDecodeTests.cs @@ -4,6 +4,7 @@ using Google.Protobuf; using System; using System.Collections.Generic; +using System.Diagnostics; using System.Linq; using System.Net; using Xunit; @@ -667,6 +668,43 @@ public void Decode_WarmupFrame_ReportsDiscardWithCounts() Assert.Equal(1, device.DiscardedStreamFrameCount); } + [Fact] + public void Decode_ThrowingDiscardSubscriberAndThrowingTraceListener_DoesNotBreakTheStream() + { + // The catch around the discard event exists to contain a bad subscriber. Trace dispatches + // to listeners the consumer installed, so logging from inside that catch is itself consumer + // code — a throwing listener there would escape the containment and take down the frame + // pipeline it was protecting. + var device = CreateStreamingDevice(analogCount: 2); + AnalogChannel(device, 0).IsEnabled = true; + AnalogChannel(device, 1).IsEnabled = true; + device.StartStreaming(); + + device.StreamFrameDiscarded += (_, _) => throw new InvalidOperationException("bad subscriber"); + + var listener = new ThrowOnMarkerTraceListener("StreamFrameDiscarded"); + Trace.Listeners.Add(listener); + try + { + var warmup = new DaqifiOutMessage { MsgTimeStamp = 1000 }; + warmup.AnalogInDataFloat.Add(0.1f); + + Assert.Null(Record.Exception(() => device.InvokeStreamMessage(warmup))); + Assert.True(listener.Threw, "the trace listener should have been reached and thrown"); + + // The stream carries on: the next full frame decodes normally. + var full = new DaqifiOutMessage { MsgTimeStamp = 2000 }; + full.AnalogInDataFloat.Add(4f); + full.AnalogInDataFloat.Add(8f); + Assert.Null(Record.Exception(() => device.InvokeStreamMessage(full))); + Assert.Equal(4.0, AnalogChannel(device, 0).ActiveSample!.Value); + } + finally + { + Trace.Listeners.Remove(listener); + } + } + [Fact] public void Decode_WellFormedFirstFrame_PassesThroughCompletelyUnchanged() { @@ -888,7 +926,11 @@ public void Decode_LeftoverFrameFromPreviousSession_DiscardedFromBothConsumerPat device.InvokeStreamMessage(leftover); Assert.Equal(0, rawFrames); - Assert.Equal(StreamFrameDiscardReason.StaleLeftoverFrame, Assert.Single(discards).Reason); + var discarded = Assert.Single(discards); + Assert.Equal(StreamFrameDiscardReason.StaleLeftoverFrame, discarded.Reason); + Assert.Equal(10_500_000u, discarded.DeviceTimestamp); + Assert.Equal(2, discarded.AnalogValueCount); // a leftover is a *full* frame + Assert.Equal(2, discarded.EnabledAnalogChannelCount); Assert.Equal(1.0, ai0.ActiveSample!.Value); // still session 1's values Assert.Equal(2.0, ai1.ActiveSample!.Value); @@ -1032,6 +1074,29 @@ private static DaqifiOutMessage AnalogFrame(uint timestamp, float value) return frame; } + /// + /// A trace listener that throws, but only for lines containing , so it + /// cannot disturb unrelated tests running in parallel against the global + /// collection. + /// + private sealed class ThrowOnMarkerTraceListener(string marker) : TraceListener + { + public bool Threw { get; private set; } + + public override void Write(string? message) => WriteLine(message); + + public override void WriteLine(string? message) + { + if (message?.Contains(marker, StringComparison.Ordinal) != true) + { + return; + } + + Threw = true; + throw new InvalidOperationException("trace listener failure"); + } + } + /// /// A that captures sent SCPI commands (so streaming /// setup does not require a real transport) and exposes the protected stream handler so a diff --git a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs index 2099ad65..7fc7829d 100644 --- a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs @@ -123,7 +123,20 @@ public class DaqifiStreamingDevice : DaqifiDevice, IStreamingDevice, INetworkCon /// session began. A healthy stream on firmware without the device-side defects leaves it at /// zero; on firmware 3.7.2 it is typically one, for the malformed leading frame. /// - /// Reset by , so it describes the current session. + /// + /// + /// Reset whenever a streaming session begins, so it describes the current session — that + /// includes a session started by a raw SYSTem:StartStreamData through + /// , not only by . + /// + /// + /// Every drop is counted whether or not anyone is subscribed to + /// , so a consumer that subscribes after streaming has + /// begun will see a total larger than the number of events it received. Read inside a + /// handler, the count already includes the frame being + /// reported. + /// + /// public long DiscardedStreamFrameCount => Interlocked.Read(ref _discardedStreamFrameCount); /// @@ -590,7 +603,7 @@ private void TrackStreamingStart(ReadOnlySpan argument) || frequency < 1 || frequency > maxSamplingRate) { - Trace.WriteLine( + SafeTrace( $"[{nameof(TrackStreamingStart)}] Ignoring a start-streaming command with an unusable rate " + $"('{rate.ToString()}'); the session state is unchanged."); return; @@ -873,7 +886,11 @@ protected override void OnStreamMessageReceived(DaqifiOutMessage message) { if (IsStreaming && _frameGate.IsValidating && _frameGate.IsLeftoverFromPreviousSession(message)) { - RaiseStreamFrameDiscarded(StreamFrameDiscardReason.StaleLeftoverFrame, message); + RaiseStreamFrameDiscarded( + StreamFrameDiscardReason.StaleLeftoverFrame, + message, + CountAnalogValues(message), + CountEnabledAnalogChannels(SnapshotChannels())); return; } @@ -905,11 +922,27 @@ protected override void OnStreamMessageReceived(DaqifiOutMessage message) /// The frame to deliver. private void EmitStreamFrame(DaqifiOutMessage message) { - var suppressAnalog = _awaitingFirstFullAnalogFrame && ShouldSuppressPartialAnalog(message); + var suppressAnalog = false; + var analogValueCount = 0; + var enabledAnalogChannelCount = 0; + + if (_awaitingFirstFullAnalogFrame) + { + suppressAnalog = ShouldSuppressPartialAnalog( + message, out analogValueCount, out enabledAnalogChannelCount); + } if (suppressAnalog) { - RaiseStreamFrameDiscarded(StreamFrameDiscardReason.PartialAnalogFrame, message); + // The counts reported here are the very ones the suppression decision was made on, + // not a fresh reading: channel enablement can change from another thread, and a + // discard whose reported numbers disagree with the reason it was discarded would + // make the telemetry harder to trust than no telemetry at all. + RaiseStreamFrameDiscarded( + StreamFrameDiscardReason.PartialAnalogFrame, + message, + analogValueCount, + enabledAnalogChannelCount); } else { @@ -943,12 +976,20 @@ private void EmitStreamFrame(DaqifiOutMessage message) /// have been suppressed. /// /// The frame about to be delivered. + /// The number of analog values the frame carried. + /// + /// The number of enabled analog channels the decision was made against. Handed back so the + /// discard event reports the same numbers the decision used rather than re-reading channel + /// state that another thread may have changed in between. + /// /// true when the frame's analog values must be withheld. - private bool ShouldSuppressPartialAnalog(DaqifiOutMessage message) + private bool ShouldSuppressPartialAnalog( + DaqifiOutMessage message, + out int analogValueCount, + out int enabledAnalogChannelCount) { - var analogValueCount = message.AnalogInDataFloat.Count > 0 - ? message.AnalogInDataFloat.Count - : message.AnalogInData.Count; + analogValueCount = CountAnalogValues(message); + enabledAnalogChannelCount = 0; // A frame with no analog payload says nothing about the warmup frame either way, so the // guard stays armed for the first analog-bearing frame. @@ -957,9 +998,9 @@ private bool ShouldSuppressPartialAnalog(DaqifiOutMessage message) return false; } - var enabledAnalogCount = CountEnabledAnalogChannels(SnapshotChannels()); - if (enabledAnalogCount > 0 - && analogValueCount < enabledAnalogCount + enabledAnalogChannelCount = CountEnabledAnalogChannels(SnapshotChannels()); + if (enabledAnalogChannelCount > 0 + && analogValueCount < enabledAnalogChannelCount && _suppressedWarmupFrameCount < MaxSuppressedWarmupFrames) { _suppressedWarmupFrameCount++; @@ -970,13 +1011,31 @@ private bool ShouldSuppressPartialAnalog(DaqifiOutMessage message) return false; } + /// + /// The number of analog values a frame carries, from whichever payload the transport used — + /// USB streams pre-scaled floats, WiFi streams raw ADC counts. + /// + private static int CountAnalogValues(DaqifiOutMessage message) => + message.AnalogInDataFloat.Count > 0 + ? message.AnalogInDataFloat.Count + : message.AnalogInData.Count; + /// /// Raises for a frame that was withheld, and counts it. /// Subscriber exceptions are isolated, mirroring . /// /// Why the frame was withheld. /// The frame that was withheld. - private void RaiseStreamFrameDiscarded(StreamFrameDiscardReason reason, DaqifiOutMessage frame) + /// The number of analog values the frame carried. + /// + /// The number of enabled analog channels to report. Passed in rather than re-derived so it + /// is the same reading the discard decision was made against. + /// + private void RaiseStreamFrameDiscarded( + StreamFrameDiscardReason reason, + DaqifiOutMessage frame, + int analogValueCount, + int enabledAnalogChannelCount) { Interlocked.Increment(ref _discardedStreamFrameCount); @@ -986,21 +1045,39 @@ private void RaiseStreamFrameDiscarded(StreamFrameDiscardReason reason, DaqifiOu return; } - var analogValueCount = frame.AnalogInDataFloat.Count > 0 - ? frame.AnalogInDataFloat.Count - : frame.AnalogInData.Count; - try { handler(this, new StreamFrameDiscardedEventArgs( - reason, - frame.MsgTimeStamp, - analogValueCount, - CountEnabledAnalogChannels(SnapshotChannels()))); + reason, frame.MsgTimeStamp, analogValueCount, enabledAnalogChannelCount)); } catch (Exception ex) { - Trace.WriteLine($"[{nameof(StreamFrameDiscarded)}] Subscriber threw: {ex}"); + SafeTrace($"[{nameof(StreamFrameDiscarded)}] Subscriber threw: {ex}"); + } + } + + /// + /// Writes a diagnostic line, swallowing anything a misbehaving + /// throws. + /// + /// + /// dispatches to listeners the consumer installed, so it is consumer + /// code and can throw like any other. That matters most in the places that exist purely to + /// isolate the frame pipeline from faults: a listener throwing out of the catch that + /// was containing a bad subscriber would defeat the containment and take down the very + /// frame processing it was protecting. Same reasoning, and the same guarantee, as + /// DaqifiDevice.SafeLog — which is private to the base class, hence this local twin. + /// + /// The diagnostic line to write. + private static void SafeTrace(string message) + { + try + { + Trace.WriteLine(message); + } + catch + { + // A trace listener that throws is not permitted to affect device operation. } } @@ -1078,7 +1155,7 @@ private void RaiseGapDetected(TimestampGapEventArgs args) } catch (Exception ex) { - Trace.WriteLine($"[{nameof(GapDetected)}] Subscriber threw: {ex}"); + SafeTrace($"[{nameof(GapDetected)}] Subscriber threw: {ex}"); } } From 4165dffd4bac8506726a99423f2a4c9f442959b0 Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Sun, 2 Aug 2026 13:08:28 -0600 Subject: [PATCH 3/3] test(device): prove discard-subscriber isolation without touching global Trace state MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Trace.Listeners is process-global, so a test that installs a throwing listener can be reached by anything else running in the same process. This was the only Trace.Listeners usage in the repo, and it was added to a suite that is concurrently being destabilised by exactly that class of problem (#430). The test now proves the property that actually ships — a throwing StreamFrameDiscarded subscriber does not break the frame pipeline, and the next frame still decodes — using only a throwing subscriber, with a call counter so it cannot pass vacuously. SafeTrace stays in production, where it is the real fix; the codebase already treats its twin, DaqifiDevice.SafeLog, as covered through an injected logger rather than global state. Co-Authored-By: Claude Opus 5 --- .../DaqifiStreamingDeviceDecodeTests.cs | 76 ++++++------------- 1 file changed, 25 insertions(+), 51 deletions(-) diff --git a/src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceDecodeTests.cs b/src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceDecodeTests.cs index 62d39c01..3a1bd443 100644 --- a/src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceDecodeTests.cs +++ b/src/Daqifi.Core.Tests/Device/DaqifiStreamingDeviceDecodeTests.cs @@ -4,7 +4,6 @@ using Google.Protobuf; using System; using System.Collections.Generic; -using System.Diagnostics; using System.Linq; using System.Net; using Xunit; @@ -669,40 +668,38 @@ public void Decode_WarmupFrame_ReportsDiscardWithCounts() } [Fact] - public void Decode_ThrowingDiscardSubscriberAndThrowingTraceListener_DoesNotBreakTheStream() + public void Decode_ThrowingDiscardSubscriber_DoesNotBreakTheStream() { - // The catch around the discard event exists to contain a bad subscriber. Trace dispatches - // to listeners the consumer installed, so logging from inside that catch is itself consumer - // code — a throwing listener there would escape the containment and take down the frame - // pipeline it was protecting. + // The catch around the discard event exists to contain a bad subscriber: the frame it was + // reporting is still dropped, and — the part that matters — the frames after it are still + // decoded. Deliberately no throwing TraceListener here: Trace.Listeners is process-global, + // and a test that installs a throwing listener can be reached by anything else running in + // the same process. SafeTrace covers the listener case in production. var device = CreateStreamingDevice(analogCount: 2); AnalogChannel(device, 0).IsEnabled = true; AnalogChannel(device, 1).IsEnabled = true; device.StartStreaming(); - device.StreamFrameDiscarded += (_, _) => throw new InvalidOperationException("bad subscriber"); - - var listener = new ThrowOnMarkerTraceListener("StreamFrameDiscarded"); - Trace.Listeners.Add(listener); - try - { - var warmup = new DaqifiOutMessage { MsgTimeStamp = 1000 }; - warmup.AnalogInDataFloat.Add(0.1f); - - Assert.Null(Record.Exception(() => device.InvokeStreamMessage(warmup))); - Assert.True(listener.Threw, "the trace listener should have been reached and thrown"); - - // The stream carries on: the next full frame decodes normally. - var full = new DaqifiOutMessage { MsgTimeStamp = 2000 }; - full.AnalogInDataFloat.Add(4f); - full.AnalogInDataFloat.Add(8f); - Assert.Null(Record.Exception(() => device.InvokeStreamMessage(full))); - Assert.Equal(4.0, AnalogChannel(device, 0).ActiveSample!.Value); - } - finally + var calls = 0; + device.StreamFrameDiscarded += (_, _) => { - Trace.Listeners.Remove(listener); - } + calls++; + throw new InvalidOperationException("bad subscriber"); + }; + + var warmup = new DaqifiOutMessage { MsgTimeStamp = 1000 }; + warmup.AnalogInDataFloat.Add(0.1f); + + Assert.Null(Record.Exception(() => device.InvokeStreamMessage(warmup))); + Assert.Equal(1, calls); // the subscriber really did run and really did throw + + // The stream carries on: the next full frame decodes normally. + var full = new DaqifiOutMessage { MsgTimeStamp = 2000 }; + full.AnalogInDataFloat.Add(4f); + full.AnalogInDataFloat.Add(8f); + Assert.Null(Record.Exception(() => device.InvokeStreamMessage(full))); + Assert.Equal(4.0, AnalogChannel(device, 0).ActiveSample!.Value); + Assert.Equal(8.0, AnalogChannel(device, 1).ActiveSample!.Value); } [Fact] @@ -1074,29 +1071,6 @@ private static DaqifiOutMessage AnalogFrame(uint timestamp, float value) return frame; } - /// - /// A trace listener that throws, but only for lines containing , so it - /// cannot disturb unrelated tests running in parallel against the global - /// collection. - /// - private sealed class ThrowOnMarkerTraceListener(string marker) : TraceListener - { - public bool Threw { get; private set; } - - public override void Write(string? message) => WriteLine(message); - - public override void WriteLine(string? message) - { - if (message?.Contains(marker, StringComparison.Ordinal) != true) - { - return; - } - - Threw = true; - throw new InvalidOperationException("trace listener failure"); - } - } - /// /// A that captures sent SCPI commands (so streaming /// setup does not require a real transport) and exposes the protected stream handler so a