feat(rawcan): expose ordered subscription capture loss

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
 

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
 

src/CanKit.Pro.RawCan/ILossAwareSubscription.cs

@@ -0,0 +1,57 @@
+using System.Collections.Generic;
+
+namespace CanKit.Pro.RawCan
+{
+    /// <summary>
+    /// Optional subscription-local capture continuity information (FR-RAW-016).
+    /// Subscriptions created by <see cref="CanBusService"/> implement this interface;
+    /// existing implementations of <see cref="ISubscription"/> need not do so.
+    /// </summary>
+    /// <remarks>
+    /// These diagnostics describe this subscription's bounded buffer only. Zero drops do not
+    /// prove that the adapter, driver or upstream path lost nothing.
+    /// <para>
+    /// Use one reader and <see cref="FramesWithSequence"/> or <see cref="TryReadWithSequence"/>
+    /// 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
+    /// <c>unchecked(previous + 1)</c> 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.
+    /// </para>
+    /// <para>
+    /// 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.
+    /// </para>
+    /// </remarks>
+    public interface ILossAwareSubscription : ISubscription
+    {
+        /// <summary>
+        /// Number of frames actually evicted by this subscription's bounded buffer. Monotonic
+        /// for its lifetime, readable after disposal, and independent of other subscriptions.
+        /// </summary>
+        /// <remarks>
+        /// 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 <see cref="long.MaxValue"/>; then its
+        /// value is a lower bound and a fresh subscription is needed for exact counting.
+        /// </remarks>
+        long DroppedFrameCount { get; }
+
+        /// <summary>
+        /// The same buffered frames as <see cref="ISubscription.Frames"/>, with their accepted
+        /// enqueue numbers. Completes after disposal and draining; cancellation ends the reader.
+        /// </summary>
+        IAsyncEnumerable<SubscriptionFrame> FramesWithSequence { get; }
+
+        /// <summary>
+        /// Removes one buffered frame with its enqueue number without waiting. Returns false
+        /// and the default value when the buffer is empty. <see cref="ISubscription.WaitToReadAsync"/>
+        /// can be used before this method as with <see cref="ISubscription.TryRead"/>.
+        /// </summary>
+        bool TryReadWithSequence(out SubscriptionFrame frame);
+    }
+}

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 <c>FrameObserved</c> event (FR-RAW-011).
+    /// Subscriptions created by <see cref="CanBusService"/> additionally implement the optional
+    /// <see cref="ILossAwareSubscription"/> for known buffer losses and ordered capture gaps.
+    /// Other implementations need not implement that extension.
     /// <para>
     /// Disposing the subscription deterministically deregisters it from the service (it stops
     /// receiving frames) and completes <see cref="Frames"/>, so any in-flight

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

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.
     /// </summary>
-    internal sealed class Subscription : ISubscription
+    internal sealed class Subscription : ILossAwareSubscription
     {
         private sealed class FilterCriteria
         {
@@ -30,7 +30,10 @@ public FilterCriteria(CanIdFilter? idFilter, Func<CanFrameEvent, bool>? predicat
         }
 
         private readonly CanBusService _service;
-        private readonly Channel<CanFrameEvent> _channel;
+        private readonly Channel<SubscriptionFrame> _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<CanFrameEvent, bool>? 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<CanFrameEvent>(options);
+            _channel = Channel.CreateBounded<SubscriptionFrame>(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);
+        }
+
+        /// <inheritdoc />
+        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;
+            }
         }
 
         /// <inheritdoc/>
         public IAsyncEnumerable<CanFrameEvent> Frames => ReadAsync();
 
         /// <inheritdoc/>
-        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;
+        }
+
+        /// <inheritdoc />
+        public IAsyncEnumerable<SubscriptionFrame> FramesWithSequence => ReadWithSequenceAsync();
+
+        /// <inheritdoc />
+        public bool TryReadWithSequence(out SubscriptionFrame frame) => _channel.Reader.TryRead(out frame);
 
         /// <inheritdoc/>
         public ValueTask<bool> WaitToReadAsync(CancellationToken cancellationToken = default)
@@ -199,8 +244,19 @@ private async IAsyncEnumerable<CanFrameEvent> 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<SubscriptionFrame> 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;
             }
         }
 

src/CanKit.Pro.RawCan/SubscriptionFrame.cs

@@ -0,0 +1,27 @@
+namespace CanKit.Pro.RawCan
+{
+    /// <summary>
+    /// A buffered frame and its subscription-local enqueue number (FR-RAW-016).
+    /// See <see cref="ILossAwareSubscription"/> for continuity and competing-reader semantics.
+    /// </summary>
+    public readonly struct SubscriptionFrame
+    {
+        /// <summary>Creates a frame envelope. The supplied event retains its payload lifetime.</summary>
+        /// <param name="frameEvent">The frame, echo flag and original arrival timestamps.</param>
+        /// <param name="sequenceNumber">Accepted enqueue number, modulo 2^64.</param>
+        public SubscriptionFrame(CanFrameEvent frameEvent, ulong sequenceNumber)
+        {
+            FrameEvent = frameEvent;
+            SequenceNumber = sequenceNumber;
+        }
+
+        /// <summary>The original event, with independently owned payload when read from the demux.</summary>
+        public CanFrameEvent FrameEvent { get; }
+
+        /// <summary>
+        /// This subscription's accepted enqueue number. Starts at 1 and wraps from
+        /// <see cref="ulong.MaxValue"/> to 0; excluded or refused frames consume no number.
+        /// </summary>
+        public ulong SequenceNumber { get; }
+    }
+}

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<CanKit.Pro.RawCan.CanFrameEvent, bool>? predicate = null, int? bufferCapacity = default, bool includeEcho = false);
     }
+    public interface ILossAwareSubscription : CanKit.Pro.RawCan.ISubscription, System.IDisposable
+    {
+        long DroppedFrameCount { get; }
+        System.Collections.Generic.IAsyncEnumerable<CanKit.Pro.RawCan.SubscriptionFrame> FramesWithSequence { get; }
+        bool TryReadWithSequence(out CanKit.Pro.RawCan.SubscriptionFrame frame);
+    }
     public interface ISubscription : System.IDisposable
     {
         System.Collections.Generic.IAsyncEnumerable<CanKit.Pro.RawCan.CanFrameEvent> 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,