From a5816cd48c17ee114b656f25b8026e97e2a79662 Mon Sep 17 00:00:00 2001 From: Dietmar Borgards Date: Sun, 25 Oct 2026 16:03:16 +0100 Subject: [PATCH] feat(rawcan): expose ordered subscription capture loss --- docs/requirements/SRS-CanKit.Pro-en.md | 1 + docs/requirements/SRS-CanKit.Pro.md | 1 + .../ILossAwareSubscription.cs | 57 ++++ src/CanKit.Pro.RawCan/ISubscription.cs | 3 + src/CanKit.Pro.RawCan/README.md | 47 +++ src/CanKit.Pro.RawCan/Subscription.cs | 74 ++++- src/CanKit.Pro.RawCan/SubscriptionFrame.cs | 27 ++ .../CanKit.Pro.RawCan.approved.txt | 12 + .../TestCases/SubscriptionLossTests.cs | 268 ++++++++++++++++++ 9 files changed, 481 insertions(+), 9 deletions(-) create mode 100644 src/CanKit.Pro.RawCan/ILossAwareSubscription.cs create mode 100644 src/CanKit.Pro.RawCan/SubscriptionFrame.cs create mode 100644 tests/CanKit.Pro.Tests/TestCases/SubscriptionLossTests.cs diff --git a/docs/requirements/SRS-CanKit.Pro-en.md b/docs/requirements/SRS-CanKit.Pro-en.md index b968b0eb..ee12d00e 100644 --- a/docs/requirements/SRS-CanKit.Pro-en.md +++ b/docs/requirements/SRS-CanKit.Pro-en.md @@ -192,6 +192,7 @@ Background: Today several protocol instances (e.g. ISO-TP + CANopen) cannot obta | FR-RAW-013 | The system SHOULD provide a default demultiplex strategy for the common case "one 11/29-bit CAN-ID range per protocol instance" (fast path without generic predicate) to avoid overhead at high bus frequency. | Should | Performance test: throughput comparison ID-range filter vs. generic predicate at ≥ 1000 frames/s. | NFR performance, section 5.1 | | FR-RAW-014 | The system MAY allow the protocol developer to reconfigure subscriptions at runtime (change filter criteria) without recreating the subscription. | Could | Unit test: change filter criterion at runtime, subsequent frames follow the new criterion. | — | | FR-RAW-015 | The system MUST attach to every subscription element the echo marking reported by the bus and the receive timestamp captured by the adapter, and it MUST deliver echo frames only to subscriptions that explicitly requested them (default: no echoes). | Must | Unit test: bus reports a receive frame and an echo frame; the default subscription receives only the receive frame, `includeEcho: true` receives both with correct `IsEcho`; the timestamp reaches the subscriber unchanged. | [#23](https://github.com/dborgards/CanKit.Pro/issues/23) (review finding Q5) | +| FR-RAW-016 | Built-in subscriptions MUST expose known local buffer overflow through an optional extension compatible with existing `ISubscription` implementations: an independent drop count and ordered frame sequence for conservatively locating capture gaps. Filter/echo exclusions, reads, cancellation and disposal do not count as overflow. Dispatch does not wait for readers; overflow cannot erase the loss metadata itself. Zero local drops do not prove loss-free adapter/driver capture. | Must | Deterministic small-buffer tests: exact evictions, ordering/gaps, independent subscriptions, concurrent writers, exclusions, lifetime and counter boundaries. | [#402](https://github.com/dborgards/CanKit.Pro/issues/402) | #### 4.1.3 Threading/actor model per protocol instance diff --git a/docs/requirements/SRS-CanKit.Pro.md b/docs/requirements/SRS-CanKit.Pro.md index a55218bc..67a5811d 100644 --- a/docs/requirements/SRS-CanKit.Pro.md +++ b/docs/requirements/SRS-CanKit.Pro.md @@ -190,6 +190,7 @@ Hintergrund: Heute können mehrere Protokollinstanzen (z. B. ISO-TP + CANopen) n | FR-RAW-013 | Das System SOLLTE eine Standard-Demultiplex-Strategie für den häufigen Fall „ein 11/29-Bit-CAN-ID-Bereich pro Protokollinstanz“ bereitstellen (Fast-Path ohne generisches Prädikat), um Overhead bei hoher Busfrequenz zu vermeiden. | Should | Performance-Test: Durchsatzvergleich ID-Bereichsfilter vs. generisches Prädikat bei ≥ 1000 Frames/s. | NFR-Performance, Abschnitt 5.1 | | FR-RAW-014 | Das System KANN dem Protokollentwickler erlauben, Subscriptions zur Laufzeit umzukonfigurieren (Filterkriterien ändern), ohne die Subscription neu zu erzeugen. | Could | Unit-Test: Filterkriterium zur Laufzeit ändern, nachfolgende Frames folgen neuem Kriterium. | — | | FR-RAW-015 | Das System MUSS jedem Subscription-Element die vom Bus gemeldete Echo-Kennzeichnung und den vom Adapter erfassten Empfangszeitstempel mitgeben, und es MUSS Echo-Frames nur an Subscriptions ausliefern, die dies ausdrücklich angefordert haben (Standard: keine Echos). | Must | Unit-Test: Bus meldet Empfangs- und Echo-Frame; Standard-Subscription erhält nur den Empfangsframe, `includeEcho: true` erhält beide mit korrektem `IsEcho`; Zeitstempel erreicht den Subscriber unverändert. | [#23](https://github.com/dborgards/CanKit.Pro/issues/23) (Review-Befund Q5) | +| FR-RAW-016 | Die integrierten Subscriptions MÜSSEN bekannte lokale Pufferüberläufe über eine optionale, zu bestehenden `ISubscription`-Implementierungen kompatible Erweiterung sichtbar machen: unabhängiger Drop-Zähler und geordnete Frame-Sequenz zur konservativen Einordnung von Capture-Lücken. Filter-/Echo-Ausschlüsse, Lesen, Abbruch und Dispose zählen nicht als Überlauf. Der Dispatch wartet nicht auf Leser; Metadaten gehen bei Überlauf nicht selbst verloren. Null lokale Drops sind kein Nachweis verlustfreier Adapter-/Treibererfassung. | Must | Deterministische Tests mit kleinen Puffern: genaue Eviktionen, Reihenfolge/Lücken, unabhängige Subscriptions, konkurrierende Writer, Ausschlüsse, Lifetime und Zählergrenzen. | [#402](https://github.com/dborgards/CanKit.Pro/issues/402) | #### 4.1.3 Threading-/Aktor-Modell pro Protokollinstanz diff --git a/src/CanKit.Pro.RawCan/ILossAwareSubscription.cs b/src/CanKit.Pro.RawCan/ILossAwareSubscription.cs new file mode 100644 index 00000000..f4e111a4 --- /dev/null +++ b/src/CanKit.Pro.RawCan/ILossAwareSubscription.cs @@ -0,0 +1,57 @@ +using System.Collections.Generic; + +namespace CanKit.Pro.RawCan +{ + /// + /// Optional subscription-local capture continuity information (FR-RAW-016). + /// Subscriptions created by implement this interface; + /// existing implementations of need not do so. + /// + /// + /// These diagnostics describe this subscription's bounded buffer only. Zero drops do not + /// prove that the adapter, driver or upstream path lost nothing. + /// + /// Use one reader and or + /// to order a discontinuity relative to delivered frames. The first accepted frame has + /// sequence number 1; each accepted enqueue advances it, modulo 2^64. A number other than + /// unchecked(previous + 1) identifies missing input before that frame, including + /// overflow before the first read. Discard pending pairing/timing evidence across such a + /// gap and qualify the capture separately from protocol findings. + /// + /// + /// Both projections and both read methods consume the same buffer. Competing readers can + /// also create gaps in a reader's sequence; their reads do not increase the drop counter. + /// Sequence numbers belong to one subscription's enqueue order, not hardware order, and + /// cannot be compared across subscriptions. Filtering and echo exclusion allocate no number. + /// + /// + public interface ILossAwareSubscription : ISubscription + { + /// + /// Number of frames actually evicted by this subscription's bounded buffer. Monotonic + /// for its lifetime, readable after disposal, and independent of other subscriptions. + /// + /// + /// Filter/echo exclusions, reads, cancellation and disposal do not increment it. + /// This is a separate instantaneous sample, not a timestamped boundary: a reader may + /// obtain a retained frame before a concurrent enqueue finishes updating the counter. + /// Use sequence numbers for ordering, rather than assuming a counter sample describes + /// a particular frame. The counter saturates at ; then its + /// value is a lower bound and a fresh subscription is needed for exact counting. + /// + long DroppedFrameCount { get; } + + /// + /// The same buffered frames as , with their accepted + /// enqueue numbers. Completes after disposal and draining; cancellation ends the reader. + /// + IAsyncEnumerable FramesWithSequence { get; } + + /// + /// Removes one buffered frame with its enqueue number without waiting. Returns false + /// and the default value when the buffer is empty. + /// can be used before this method as with . + /// + bool TryReadWithSequence(out SubscriptionFrame frame); + } +} diff --git a/src/CanKit.Pro.RawCan/ISubscription.cs b/src/CanKit.Pro.RawCan/ISubscription.cs index d2dd1ca1..03cfc1a4 100644 --- a/src/CanKit.Pro.RawCan/ISubscription.cs +++ b/src/CanKit.Pro.RawCan/ISubscription.cs @@ -13,6 +13,9 @@ namespace CanKit.Pro.RawCan /// Each subscription owns its own bounded buffer, so a slow (or never-drained) consumer only /// drops its own oldest frames and can never delay delivery to other subscriptions or to the /// underlying bus's own FrameObserved event (FR-RAW-011). + /// Subscriptions created by additionally implement the optional + /// for known buffer losses and ordered capture gaps. + /// Other implementations need not implement that extension. /// /// Disposing the subscription deterministically deregisters it from the service (it stops /// receiving frames) and completes , so any in-flight diff --git a/src/CanKit.Pro.RawCan/README.md b/src/CanKit.Pro.RawCan/README.md index c05b1eea..b042eae2 100644 --- a/src/CanKit.Pro.RawCan/README.md +++ b/src/CanKit.Pro.RawCan/README.md @@ -50,6 +50,53 @@ runtime without recreating the subscription (FR-RAW-014); only frames observed a the new criterion. This layer is built purely on the public `ICanBus.FrameObserved` surface, so it works identically for every adapter. +### Known capture gaps + +Built-in subscriptions also implement the optional `ILossAwareSubscription` (FR-RAW-016). +`DroppedFrameCount` counts only frames evicted by that subscription's bounded buffer. A slow +subscriber cannot increase another subscriber's count; filters, echo exclusion, reads, +cancellation and disposal do not increase it. Zero local drops say nothing about losses in an +adapter, driver or upstream pipe. Other `ISubscription` implementations remain compatible and +may omit this interface: missing diagnostics mean unavailable information, not zero losses. + +`SubscriptionFrame` contains the original `CanFrameEvent` and a subscription-local +`SequenceNumber`. The first accepted enqueue is 1; numbers continue across reconfiguration and +wrap modulo 2^64. For one exclusive reader, a number other than the expected successor marks +missing input before that frame, including loss before the first read. Numbers describe enqueue +order under concurrent dispatch, not hardware timestamp order, and cannot be compared across +subscriptions. Both projections consume the same buffer; competing readers can create sequence +gaps without incrementing the overflow counter. + +```csharp +using var capture = service.Subscribe(bufferCapacity: 64); +if (capture is ILossAwareSubscription loss) +{ + ulong expected = 1; + await foreach (var item in loss.FramesWithSequence.WithCancellation(token)) + { + if (item.SequenceNumber != expected) + { + // Invalidate pending pair/timing evidence here; qualify capture separately. + Console.WriteLine($"Capture gap before #{item.SequenceNumber}; known local drops: {loss.DroppedFrameCount}"); + } + expected = unchecked(item.SequenceNumber + 1); + Console.WriteLine($"#{item.SequenceNumber}: CAN-ID {item.FrameEvent.Frame.ID:X}"); + } +} +``` + +`TryReadWithSequence` and the inherited `WaitToReadAsync` provide the same metadata to a +pull-style consumer. The lifetime count is a separate instantaneous sample: an enqueue can +finish updating it after a concurrent reader has obtained a retained frame. Use sequence +numbers to order a gap. The count saturates at `long.MaxValue`, after which it is a lower bound; +start a new subscription when exact counting is required. + +Across a known gap, discard any pending pair and timing evidence. A fresh contiguous run can +start a new observation period under the consuming protocol's recovery rules; the lifetime +counter does not reset. A locally observed timeout is not proof of remote failure when capture +is incomplete. Continuous SRDO supervision and its recovery policy are separate from this +RawCan capture contract (#401). + ### Echoes A bus opened with `WorkMode == ChannelWorkMode.Echo` reports the host's own transmissions back diff --git a/src/CanKit.Pro.RawCan/Subscription.cs b/src/CanKit.Pro.RawCan/Subscription.cs index b059a9df..51a9536a 100644 --- a/src/CanKit.Pro.RawCan/Subscription.cs +++ b/src/CanKit.Pro.RawCan/Subscription.cs @@ -13,7 +13,7 @@ namespace CanKit.Pro.RawCan /// its independent RX buffer (FR-RAW-011) and applies this subscription's filter on the /// service's dispatch hot path. /// - internal sealed class Subscription : ISubscription + internal sealed class Subscription : ILossAwareSubscription { private sealed class FilterCriteria { @@ -30,7 +30,10 @@ public FilterCriteria(CanIdFilter? idFilter, Func? predicat } private readonly CanBusService _service; - private readonly Channel _channel; + private readonly Channel _channel; + private readonly object _writeGate = new(); + private long _droppedFrameCount; + private ulong _sequenceNumber; // Fixed at Subscribe time and never reconfigurable: "do I want to see my own // transmissions?" is a property of what the subscriber *is*, not of which IDs it happens @@ -64,10 +67,16 @@ internal Subscription( CanIdFilter? idFilter, Func? predicate, int capacity, - bool includeEcho) + bool includeEcho, + long initialDroppedFrameCount = 0, + ulong initialSequenceNumber = 0) { _service = service; _includeEcho = includeEcho; + // Internal seeds let tests exercise saturation and wrap without an impractical + // number of frames. Every service-created subscription starts both at zero. + _droppedFrameCount = initialDroppedFrameCount; + _sequenceNumber = initialSequenceNumber; _criteria = idFilter is { } filter ? new FilterCriteria(filter, null) : predicate is { } p @@ -83,11 +92,23 @@ internal Subscription( { SingleReader = false, SingleWriter = false, + AllowSynchronousContinuations = false, FullMode = BoundedChannelFullMode.DropOldest, }; - _channel = Channel.CreateBounded(options); + _channel = Channel.CreateBounded(options, _ => RecordDrop()); } + private void RecordDrop() + { + // Called only by TryWrite under _writeGate, so writers cannot race this saturation + // check. This callback records a fact; no consumer handler runs on dispatch. + if (Interlocked.Read(ref _droppedFrameCount) < long.MaxValue) + Interlocked.Increment(ref _droppedFrameCount); + } + + /// + public long DroppedFrameCount => Interlocked.Read(ref _droppedFrameCount); + // A plain assignment to a volatile field, not Interlocked.Exchange: the whole criteria // object is swapped in one reference write, no caller reads the previous value, and the // dispatch path only ever reads. Volatile is exactly the guarantee that buys -- the write @@ -179,15 +200,39 @@ internal void TryDeliver(in CanFrameView view, bool isEcho, TimeSpan receiveTime ownedPayload ??= view.Data.ToArray(); var owned = new CanFrameView(view.FrameKind, view.ID, ownedPayload, view.Flags); - _channel.Writer.TryWrite(new CanFrameEvent(owned, isEcho, receiveTimestamp, - hostArrivalTimestamp)); + var frameEvent = new CanFrameEvent(owned, isEcho, receiveTimestamp, hostArrivalTimestamp); + // Assign and publish in the same short writer section. Assigning with Interlocked + // before TryWrite would let concurrent writers enqueue 2 before 1, hiding which + // buffered frames an overflow removed. No predicate or consumer runs in this section; + // a full buffer evicts rather than waiting for a reader. + lock (_writeGate) + { + var sequence = unchecked(_sequenceNumber + 1); + if (_channel.Writer.TryWrite(new SubscriptionFrame(frameEvent, sequence))) + _sequenceNumber = sequence; + } } /// public IAsyncEnumerable Frames => ReadAsync(); /// - public bool TryRead(out CanFrameEvent frameEvent) => _channel.Reader.TryRead(out frameEvent); + public bool TryRead(out CanFrameEvent frameEvent) + { + if (_channel.Reader.TryRead(out var captured)) + { + frameEvent = captured.FrameEvent; + return true; + } + frameEvent = default; + return false; + } + + /// + public IAsyncEnumerable FramesWithSequence => ReadWithSequenceAsync(); + + /// + public bool TryReadWithSequence(out SubscriptionFrame frame) => _channel.Reader.TryRead(out frame); /// public ValueTask WaitToReadAsync(CancellationToken cancellationToken = default) @@ -199,8 +244,19 @@ private async IAsyncEnumerable ReadAsync( var reader = _channel.Reader; while (await reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false)) { - while (reader.TryRead(out var frameEvent)) - yield return frameEvent; + while (reader.TryRead(out var captured)) + yield return captured.FrameEvent; + } + } + + private async IAsyncEnumerable ReadWithSequenceAsync( + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + var reader = _channel.Reader; + while (await reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false)) + { + while (reader.TryRead(out var captured)) + yield return captured; } } diff --git a/src/CanKit.Pro.RawCan/SubscriptionFrame.cs b/src/CanKit.Pro.RawCan/SubscriptionFrame.cs new file mode 100644 index 00000000..058ce79c --- /dev/null +++ b/src/CanKit.Pro.RawCan/SubscriptionFrame.cs @@ -0,0 +1,27 @@ +namespace CanKit.Pro.RawCan +{ + /// + /// A buffered frame and its subscription-local enqueue number (FR-RAW-016). + /// See for continuity and competing-reader semantics. + /// + public readonly struct SubscriptionFrame + { + /// Creates a frame envelope. The supplied event retains its payload lifetime. + /// The frame, echo flag and original arrival timestamps. + /// Accepted enqueue number, modulo 2^64. + public SubscriptionFrame(CanFrameEvent frameEvent, ulong sequenceNumber) + { + FrameEvent = frameEvent; + SequenceNumber = sequenceNumber; + } + + /// The original event, with independently owned payload when read from the demux. + public CanFrameEvent FrameEvent { get; } + + /// + /// This subscription's accepted enqueue number. Starts at 1 and wraps from + /// to 0; excluded or refused frames consume no number. + /// + public ulong SequenceNumber { get; } + } +} diff --git a/tests/CanKit.Pro.Tests/ApiApprovals/CanKit.Pro.RawCan.approved.txt b/tests/CanKit.Pro.Tests/ApiApprovals/CanKit.Pro.RawCan.approved.txt index a759f7f8..5037cd28 100644 --- a/tests/CanKit.Pro.Tests/ApiApprovals/CanKit.Pro.RawCan.approved.txt +++ b/tests/CanKit.Pro.Tests/ApiApprovals/CanKit.Pro.RawCan.approved.txt @@ -58,6 +58,12 @@ namespace CanKit.Pro.RawCan CanKit.Pro.RawCan.ISubscription Subscribe(CanKit.Pro.RawCan.CanIdFilter filter, int? bufferCapacity = default, bool includeEcho = false); CanKit.Pro.RawCan.ISubscription Subscribe(System.Func? predicate = null, int? bufferCapacity = default, bool includeEcho = false); } + public interface ILossAwareSubscription : CanKit.Pro.RawCan.ISubscription, System.IDisposable + { + long DroppedFrameCount { get; } + System.Collections.Generic.IAsyncEnumerable FramesWithSequence { get; } + bool TryReadWithSequence(out CanKit.Pro.RawCan.SubscriptionFrame frame); + } public interface ISubscription : System.IDisposable { System.Collections.Generic.IAsyncEnumerable Frames { get; } @@ -73,6 +79,12 @@ namespace CanKit.Pro.RawCan public const CanKit.Core.Exceptions.CanKitErrorCode ProtocolPeerAbort = 6003; public const CanKit.Core.Exceptions.CanKitErrorCode ProtocolTimeout = 6002; } + public readonly struct SubscriptionFrame + { + public SubscriptionFrame(CanKit.Pro.RawCan.CanFrameEvent frameEvent, ulong sequenceNumber) { } + public CanKit.Pro.RawCan.CanFrameEvent FrameEvent { get; } + public ulong SequenceNumber { get; } + } public enum TxConfirmFailureReason { None = 0, diff --git a/tests/CanKit.Pro.Tests/TestCases/SubscriptionLossTests.cs b/tests/CanKit.Pro.Tests/TestCases/SubscriptionLossTests.cs new file mode 100644 index 00000000..53dc9121 --- /dev/null +++ b/tests/CanKit.Pro.Tests/TestCases/SubscriptionLossTests.cs @@ -0,0 +1,268 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using AwesomeAssertions; +using CanKit.Abstractions.API.Can; +using CanKit.Abstractions.API.Can.Definitions; +using CanKit.Abstractions.API.Common.Definitions; +using CanKit.Pro.RawCan; +using CanKit.Pro.Tests.Infrastructure; +using Xunit; + +namespace CanKit.Pro.Tests.TestCases; + +/// Subscription-local capture continuity (FR-RAW-011, FR-RAW-016; #402). +public class SubscriptionLossTests : IClassFixture +{ + private static readonly TimeSpan Watchdog = TimeSpan.FromSeconds(30); + + private static string Session() => $"loss-{Guid.NewGuid():N}"; + private static CanFrame Frame(int id) => CanFrame.Classic(id, new[] { (byte)id }); + + private static List Drain(ILossAwareSubscription subscription) + { + var frames = new List(); + while (subscription.TryReadWithSequence(out var frame)) frames.Add(frame); + return frames; + } + + [Fact] + public void Overflow_Counts_Exactly_The_Evicted_Frames_And_Orders_The_Gap_Before_The_Retained_Ones() + { + using var bus = ControllableBus.EchoCapable(Session()); + using var service = new CanBusService(bus); + using var subscription = service.Subscribe(bufferCapacity: 2); + var loss = (ILossAwareSubscription)subscription; + + for (var id = 1; id <= 5; id++) bus.RaiseObserved(Frame(id), isEcho: false); + + loss.DroppedFrameCount.Should().Be(3); + var retained = Drain(loss); + retained.Select(f => f.SequenceNumber).Should().Equal(4UL, 5UL); + retained.Select(f => f.FrameEvent.Frame.ID).Should().Equal(4, 5); + subscription.TryRead(out _).Should().BeFalse(); + loss.TryReadWithSequence(out var empty).Should().BeFalse(); + empty.SequenceNumber.Should().Be(0); + } + + [Fact] + public void A_Slow_Subscription_Does_Not_Lose_Or_Count_The_Fast_Subscriptions_Frames() + { + using var bus = ControllableBus.EchoCapable(Session()); + using var service = new CanBusService(bus); + using var slow = service.Subscribe(bufferCapacity: 1); + using var fast = service.Subscribe(bufferCapacity: 1); + var slowLoss = (ILossAwareSubscription)slow; + var fastLoss = (ILossAwareSubscription)fast; + + for (var id = 1; id <= 20; id++) + { + bus.RaiseObserved(Frame(id), isEcho: false); + fast.TryRead(out var frame).Should().BeTrue(); + frame.Frame.ID.Should().Be(id); + } + + slowLoss.DroppedFrameCount.Should().Be(19); + fastLoss.DroppedFrameCount.Should().Be(0); + Drain(slowLoss).Single().SequenceNumber.Should().Be(20); + } + + [Fact] + public void Filter_And_Echo_Exclusions_Consume_Neither_A_Number_Nor_A_Drop() + { + using var bus = ControllableBus.EchoCapable(Session()); + using var service = new CanBusService(bus); + using var subscription = service.Subscribe( + CanIdFilter.Range(0x123, 0x123, CanFilterIDType.Standard), bufferCapacity: 1); + var loss = (ILossAwareSubscription)subscription; + + bus.RaiseObserved(Frame(0x124), isEcho: false); + bus.RaiseObserved(Frame(0x123), isEcho: true); + loss.DroppedFrameCount.Should().Be(0); + loss.TryReadWithSequence(out _).Should().BeFalse(); + + bus.RaiseObserved(Frame(0x123), isEcho: false); + loss.TryReadWithSequence(out var first).Should().BeTrue(); + first.SequenceNumber.Should().Be(1); + subscription.Reconfigure(e => e.Frame.ID == 0x125); + bus.RaiseObserved(Frame(0x123), isEcho: false); + bus.RaiseObserved(Frame(0x125), isEcho: false); + loss.TryReadWithSequence(out var second).Should().BeTrue(); + second.SequenceNumber.Should().Be(2); + loss.DroppedFrameCount.Should().Be(0); + } + + [Fact] + public async Task Concurrent_Overflow_Retains_The_Last_Enqueue_Numbers_In_Order() + { + using var bus = ControllableBus.EchoCapable(Session()); + using var service = new CanBusService(bus); + using var subscription = service.Subscribe(bufferCapacity: 3); + var loss = (ILossAwareSubscription)subscription; + var frame = Frame(0x123); + var writers = Enumerable.Range(0, 4).Select(_ => Task.Run(() => + { + for (var i = 0; i < 250; i++) bus.RaiseObserved(frame, isEcho: false); + })); + await Task.WhenAll(writers).WaitAsync(Watchdog); + + loss.DroppedFrameCount.Should().Be(997); + Drain(loss).Select(f => f.SequenceNumber).Should().Equal(998UL, 999UL, 1000UL); + } + + [Fact] + public async Task Cancelling_A_Reader_Does_Not_Stop_The_Subscription_Or_Count_A_Drop() + { + using var bus = ControllableBus.EchoCapable(Session()); + using var service = new CanBusService(bus); + using var subscription = service.Subscribe(bufferCapacity: 2); + var loss = (ILossAwareSubscription)subscription; + using var cancellation = new CancellationTokenSource(); + await using (var reader = loss.FramesWithSequence.GetAsyncEnumerator(cancellation.Token)) + { + var pending = reader.MoveNextAsync().AsTask(); + cancellation.Cancel(); + await Assert.ThrowsAnyAsync(() => pending.WaitAsync(Watchdog)); + } + loss.DroppedFrameCount.Should().Be(0); + + for (var id = 1; id <= 3; id++) bus.RaiseObserved(Frame(id), isEcho: false); + subscription.TryRead(out var legacy).Should().BeTrue(); + legacy.Frame.ID.Should().Be(2); + loss.TryReadWithSequence(out var sequenced).Should().BeTrue(); + sequenced.SequenceNumber.Should().Be(3, "both projections consume one shared buffer"); + loss.DroppedFrameCount.Should().Be(1); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task Disposal_Drains_The_Retained_Frames_Without_Counting_Another_Loss(bool disposeService) + { + using var bus = ControllableBus.EchoCapable(Session()); + using var service = new CanBusService(bus); + using var subscription = service.Subscribe(bufferCapacity: 2); + var loss = (ILossAwareSubscription)subscription; + for (var id = 1; id <= 3; id++) bus.RaiseObserved(Frame(id), isEcho: false); + + if (disposeService) service.Dispose(); + else subscription.Dispose(); + bus.RaiseObserved(Frame(4), isEcho: false); + var retained = new List(); + using var watchdog = new CancellationTokenSource(Watchdog); + await foreach (var frame in loss.FramesWithSequence.WithCancellation(watchdog.Token)) retained.Add(frame); + retained.Select(f => f.SequenceNumber).Should().Equal(2UL, 3UL); + loss.DroppedFrameCount.Should().Be(1); + (await subscription.WaitToReadAsync()).Should().BeFalse(); + } + + [Fact] + public void A_Gap_Is_Visible_Before_A_Retained_Partner_And_A_Fresh_Contiguous_Pair_Can_Follow() + { + using var bus = ControllableBus.EchoCapable(Session()); + using var service = new CanBusService(bus); + using var subscription = service.Subscribe(bufferCapacity: 1); + var loss = (ILossAwareSubscription)subscription; + bus.RaiseObserved(Frame(1), isEcho: false); + loss.TryReadWithSequence(out var pending).Should().BeTrue(); + bus.RaiseObserved(Frame(2), isEcho: false); + bus.RaiseObserved(Frame(3), isEcho: false); + loss.TryReadWithSequence(out var retained).Should().BeTrue(); + retained.SequenceNumber.Should().NotBe(unchecked(pending.SequenceNumber + 1), + "a consumer must invalidate pending pairing/timing evidence before using this frame"); + loss.DroppedFrameCount.Should().Be(1); + + bus.RaiseObserved(Frame(4), isEcho: false); + loss.TryReadWithSequence(out var fresh).Should().BeTrue(); + bus.RaiseObserved(Frame(5), isEcho: false); + loss.TryReadWithSequence(out var partner).Should().BeTrue(); + partner.SequenceNumber.Should().Be(unchecked(fresh.SequenceNumber + 1)); + loss.DroppedFrameCount.Should().Be(1, "fresh evidence can be contiguous without resetting the lifetime count"); + } + + private static void Deliver(Subscription subscription, byte[] payload, bool echo = false) + { + var view = new CanFrameView(CanFrameType.Can20, 0x123, payload, FrameFlags.None); + byte[]? owned = null; + subscription.TryDeliver(view, echo, TimeSpan.FromTicks(123), 456, ref owned); + } + + [Fact] + public void Sequenced_Frames_Keep_The_Original_Metadata_And_Own_Their_Payload() + { + using var bus = ControllableBus.EchoCapable(Session()); + using var service = new CanBusService(bus); + using var subscription = new Subscription(service, null, null, 2, includeEcho: true); + var payload = new byte[] { 1, 2 }; + Deliver(subscription, payload, echo: true); + payload[0] = 99; + subscription.TryReadWithSequence(out var frame).Should().BeTrue(); + frame.SequenceNumber.Should().Be(1); + frame.FrameEvent.Frame.Data.ToArray().Should().Equal(1, 2); + frame.FrameEvent.IsEcho.Should().BeTrue(); + frame.FrameEvent.ReceiveTimestamp.Should().Be(TimeSpan.FromTicks(123)); + frame.FrameEvent.HostArrivalTimestamp.Should().Be(456); + } + + [Fact] + public void The_Counter_Saturates_And_Disposal_Refuses_Later_Frames() + { + using var bus = ControllableBus.EchoCapable(Session()); + using var service = new CanBusService(bus); + using var subscription = new Subscription(service, null, null, 1, false, + initialDroppedFrameCount: long.MaxValue - 1); + Deliver(subscription, new byte[] { 1 }); + Deliver(subscription, new byte[] { 2 }); + subscription.DroppedFrameCount.Should().Be(long.MaxValue); + Deliver(subscription, new byte[] { 3 }); + subscription.DroppedFrameCount.Should().Be(long.MaxValue); + subscription.Dispose(); + Deliver(subscription, new byte[] { 4 }); + subscription.TryReadWithSequence(out var last).Should().BeTrue(); + last.FrameEvent.Frame.Data.ToArray().Should().Equal(3); + subscription.TryReadWithSequence(out _).Should().BeFalse(); + subscription.DroppedFrameCount.Should().Be(long.MaxValue); + } + + [Fact] + public void Sequence_Numbers_Wrap_Without_Inventing_A_Loss() + { + using var bus = ControllableBus.EchoCapable(Session()); + using var service = new CanBusService(bus); + using var subscription = new Subscription(service, null, null, 2, false, + initialSequenceNumber: ulong.MaxValue - 1); + Deliver(subscription, new byte[] { 1 }); + Deliver(subscription, new byte[] { 2 }); + Drain(subscription).Select(f => f.SequenceNumber).Should().Equal(ulong.MaxValue, 0UL); + subscription.DroppedFrameCount.Should().Be(0); + } + + [Fact] + public async Task A_Waiting_Stream_Reader_Does_Not_Run_Its_Consumer_Inside_Dispatch() + { + using var bus = ControllableBus.EchoCapable(Session()); + using var service = new CanBusService(bus); + using var subscription = service.Subscribe(bufferCapacity: 1); + var loss = (ILossAwareSubscription)subscription; + await using var reader = loss.FramesWithSequence.GetAsyncEnumerator(); + var moving = reader.MoveNextAsync().AsTask(); + moving.IsCompleted.Should().BeFalse(); + var writerThread = Environment.CurrentManagedThreadId; + var dispatchActive = 1; + var inline = 0; + var observed = moving.ContinueWith(completed => + { + if (Environment.CurrentManagedThreadId == writerThread && Volatile.Read(ref dispatchActive) != 0) + Interlocked.Exchange(ref inline, 1); + return completed.GetAwaiter().GetResult(); + }, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); + + bus.RaiseObserved(Frame(1), isEcho: false); + Volatile.Write(ref dispatchActive, 0); + (await observed.WaitAsync(Watchdog)).Should().BeTrue(); + reader.Current.SequenceNumber.Should().Be(1); + inline.Should().Be(0, "publishing a frame must not invoke a waiting consumer on the dispatch thread"); + } +}