feat(uds): the collected UDS findings -- busy repeat, functional addressing, and four corrections

src/CanKit.Pro.IsoTp/IIsoTpChannel.cs

@@ -126,6 +126,19 @@ Task<IsoTpTransmitStamps> SendWithTransmitStampAsync(ReadOnlyMemory<byte> pdu,
     /// </summary>
     IReadOnlyList<IsoTpReceptionInProgress> GetReceptionsInProgress();
 
+    /// <summary>
+    /// Completes once every frame the bus had delivered before the call has been taken through
+    /// the channel: a PDU it completed is in the inbox, a reception it began is in
+    /// <see cref="GetReceptionsInProgress"/>. For a decision taken at a deadline — is a
+    /// response there, was there a 0x78 — what the inbox does not hold after this did not
+    /// arrive before the call; without it, a frame stamped in time can still be on its way
+    /// through the channel's actor when the deadline fires (Codex on #150). Called on the
+    /// channel's own actor — from a <c>BackgroundExceptionOccurred</c> handler — it returns at
+    /// once and the frames on their way are handled after the current work item: they queue
+    /// behind frames already in the mailbox, and handling them inline would reorder the two.
+    /// </summary>
+    Task SettleAsync();
+
     /// <summary>
     /// Drains every buffered inbox item — both completed PDUs and pending reassembly-abort
     /// faults enqueued by <c>AbortRx</c> — and returns how many were dropped. Also silently
@@ -137,6 +150,17 @@ Task<IsoTpTransmitStamps> SendWithTransmitStampAsync(ReadOnlyMemory<byte> pdu,
     /// </summary>
     int DiscardPendingPdus();
 
+    /// <summary>
+    /// As <see cref="DiscardPendingPdus()"/>, dropping what arrived before
+    /// <paramref name="arrivedBefore"/> (a <see cref="System.Diagnostics.Stopwatch.GetTimestamp"/>
+    /// reading) rather than before now, and keeping what arrived since. For a caller that
+    /// inspects the inbox before discarding — routing an NRC 0x78 to its window — the stamp
+    /// taken before the inspection makes the two one step: everything the discard drops was
+    /// inspected, and a frame arriving between the inspection and the discard is kept for the
+    /// caller's next read instead of vanishing (Codex on #150).
+    /// </summary>
+    int DiscardPendingPdus(long arrivedBefore);
+
     /// <summary>
     /// Enumerates every fully reassembled inbound PDU as it becomes available. The enumeration
     /// ends when the channel is disposed. A reassembly abort (N_Cr / CF sequence mismatch /
@@ -150,7 +174,7 @@ Task<IsoTpTransmitStamps> SendWithTransmitStampAsync(ReadOnlyMemory<byte> pdu,
     /// Raised (on a thread-pool thread) every time a full PDU is reassembled. The same PDU is
     /// enqueued for <see cref="ReceiveAsync"/>/<see cref="ReceiveAllAsync"/> before the event
     /// fires, so a handler that synchronously waits on those APIs — or on
-    /// <see cref="DiscardPendingPdus"/> — cannot deadlock the protocol actor. Handlers must be
+    /// <see cref="DiscardPendingPdus()"/> — cannot deadlock the protocol actor. Handlers must be
     /// non-throwing; a throwing handler is caught and surfaced via
     /// <see cref="BackgroundExceptionOccurred"/>.
     /// </summary>

src/CanKit.Pro.IsoTp/IsoTpChannel.cs

@@ -344,9 +344,37 @@ public bool TryReceiveWithArrival(out IsoTpReceivedPdu pdu)
     }
 
     /// <inheritdoc />
-    public int DiscardPendingPdus()
+    /// <inheritdoc />
+    public Task SettleAsync()
+    {
+        if (Volatile.Read(ref _disposed) != 0) return Task.CompletedTask;
+        // What the demux has buffered goes to the actor now rather than after the reader's
+        // next scheduling; then a no-op posted behind it completes once the actor has taken
+        // everything queued so far. Called from the actor itself, the post would wait for the
+        // loop it is on, and the pumped frames cannot be handled inline either: frames the
+        // reader task posted earlier sit ahead of them in the mailbox, and reassembly does not
+        // survive the reordering (Bugbot on #150). They are handled after the current work
+        // item, which the contract says.
+        PumpSubscription();
+        if (_actor is ProtocolActor { IsOnCurrentActor: true }) return Task.CompletedTask;
+        try
+        {
+            return _actor.PostAsync(() => { });
+        }
+        catch (ObjectDisposedException)
+        {
+            return Task.CompletedTask; // channel tearing down: nothing left on its way
+        }
+    }
+
+    public int DiscardPendingPdus() => DiscardPendingPdus(arrivedBefore: 0);
+
+    /// <inheritdoc />
+    public int DiscardPendingPdus(long arrivedBefore)
     {
-        // Everything that arrived up to now is pending. The stamp goes first, so a frame the
+        // Everything that arrived up to now is pending -- or up to the caller's stamp, which
+        // may be earlier (Codex on #150); the discard stamp never moves back, so a frame
+        // already admitted stays admitted. The stamp goes first, so a frame the
         // pump posts -- or the reader task posts concurrently -- is dropped by the actor if it
         // arrived before it, and answered with no Flow Control that would invite the rest of
         // a transfer the caller has given up on (Bugbot on #143). Whatever the demux has
@@ -358,8 +386,8 @@ public int DiscardPendingPdus()
         long stamp;
         lock (_pumpGate)
         {
-            stamp = Stopwatch.GetTimestamp();
-            Volatile.Write(ref _discardStamp, stamp);
+            stamp = arrivedBefore > 0 ? arrivedBefore : Stopwatch.GetTimestamp();
+            if (stamp > Volatile.Read(ref _discardStamp)) Volatile.Write(ref _discardStamp, stamp);
             PumpSubscription(unstampedArrival: stamp - 1);
         }
 

src/CanKit.Pro.IsoTp/IsoTpFunctionalClient.cs

@@ -1,5 +1,7 @@
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using CanKit.Abstractions.API.Can.Definitions;
@@ -56,6 +58,9 @@ public sealed class IsoTpFunctionalClient : IDisposable
     // Shared dummy parsing endpoint for Normal addressing (no AE byte) — used in TryParsePci.
     private static readonly IsoTpEndpoint NormalParseEndpoint = IsoTpEndpoint.Normal(0, 0);
 
+    // CancellationTokenSource.CancelAfter's bound.
+    private static readonly TimeSpan MaxWindow = TimeSpan.FromMilliseconds(uint.MaxValue - 1);
+
     private int _disposed;
 
     internal IsoTpFunctionalClient(
@@ -120,10 +125,30 @@ public async Task<IReadOnlyList<IsoTpFunctionalResponse>> SendAndCollectAsync(
         ReadOnlyMemory<byte> pdu,
         TimeSpan window,
         CancellationToken cancellationToken = default)
+        => (await SendAndCollectWithTransmitStampAsync(pdu, window, cancellationToken).ConfigureAwait(false)).Responses;
+
+    /// <summary>
+    /// As <see cref="SendAndCollectAsync"/>, also returning when the request was handed to the
+    /// driver and when it was transmitted (<see cref="IsoTpTransmitStamps"/>, zero where the
+    /// driver reports neither). A response that arrived before the handoff answers something
+    /// else -- the subscription is made before the send, and a frame from between the two is
+    /// not this request's -- and is left out (Codex on #150); a caller keeping its own deadline
+    /// from a response, such as UDS P2* from an NRC 0x78, anchors it no earlier than the
+    /// handoff for the same reason.
+    /// </summary>
+    public async Task<IsoTpFunctionalCollection> SendAndCollectWithTransmitStampAsync(
+        ReadOnlyMemory<byte> pdu,
+        TimeSpan window,
+        CancellationToken cancellationToken = default)
     {
         ThrowIfDisposed();
         if (pdu.Length == 0)
             throw new ArgumentException("ISO-TP PDU must be non-empty.", nameof(pdu));
+        // A window the collector's timer would refuse is refused here, before the frame goes
+        // out: the send is not undone by the argument error that would follow it (Codex on #150).
+        if (window < TimeSpan.Zero || window > MaxWindow)
+            throw new ArgumentOutOfRangeException(nameof(window), window,
+                "The collection window must be between zero and what a timer can measure (about 49 days).");
 
         // Drain-before-send: subscribe, synchronously discard whatever is already buffered
         // (background chatter, a previous request's late reply, an unrelated broadcast, …),
@@ -144,9 +169,13 @@ public async Task<IReadOnlyList<IsoTpFunctionalResponse>> SendAndCollectAsync(
         using var sub = _service.Subscribe(_responseFilter, includeEcho: true);
         DrainBuffered(sub);
 
-        await SendSingleFrameAsync(pdu, cancellationToken).ConfigureAwait(false);
+        var stamps = await SendSingleFrameAsync(pdu, cancellationToken).ConfigureAwait(false);
 
-        return await CollectFromSubscriptionAsync(sub, window, cancellationToken).ConfigureAwait(false);
+        var collected = await CollectFromSubscriptionAsync(sub, window, cancellationToken).ConfigureAwait(false);
+        long cutoff = stamps.LastFrameHandoffTimestamp;
+        if (cutoff > 0 && collected.Any(r => r.HostArrivalTimestamp < cutoff))
+            collected = collected.Where(r => r.HostArrivalTimestamp >= cutoff).ToList().AsReadOnly();
+        return new IsoTpFunctionalCollection(collected, stamps);
     }
 
     /// <summary>
@@ -164,6 +193,14 @@ public async Task<IReadOnlyList<IsoTpFunctionalResponse>> SendAndCollectAsync(
     /// </exception>
     /// <exception cref="IsoTpException">TX-confirm failed.</exception>
     public Task SendAsync(ReadOnlyMemory<byte> pdu, CancellationToken cancellationToken = default)
+        => SendWithTransmitStampAsync(pdu, cancellationToken);
+
+    /// <summary>
+    /// As <see cref="SendAsync"/>, returning when the frame was handed to the driver and when
+    /// it was transmitted (<see cref="IsoTpTransmitStamps"/>, zero where the driver reports
+    /// neither), for a caller that keeps a deadline from the transmission (Codex on #150).
+    /// </summary>
+    public Task<IsoTpTransmitStamps> SendWithTransmitStampAsync(ReadOnlyMemory<byte> pdu, CancellationToken cancellationToken = default)
     {
         ThrowIfDisposed();
         if (pdu.Length == 0)
@@ -193,6 +230,19 @@ public async Task<IReadOnlyList<IsoTpFunctionalResponse>> CollectResponsesAsync(
         return await CollectFromSubscriptionAsync(sub, window, cancellationToken).ConfigureAwait(false);
     }
 
+    /// <summary>
+    /// Subscribes to the response range now and returns a listener that keeps the subscription
+    /// across collections. Obtained before a <see cref="SendAsync"/>, it closes the gap between
+    /// send and subscribe that <see cref="CollectResponsesAsync"/> leaves, and a response that
+    /// arrives between two of its collections is buffered for the next one.
+    /// </summary>
+    /// <exception cref="ObjectDisposedException">The client was disposed.</exception>
+    public IsoTpFunctionalListener Listen()
+    {
+        ThrowIfDisposed();
+        return new IsoTpFunctionalListener(_service.Subscribe(_responseFilter, includeEcho: true));
+    }
+
     /// <inheritdoc/>
     public void Dispose()
     {
@@ -209,7 +259,7 @@ public void Dispose()
     /// Sends <paramref name="pdu"/> as a functional Single Frame and awaits the CAN driver's
     /// TX confirmation.
     /// </summary>
-    private async Task SendSingleFrameAsync(ReadOnlyMemory<byte> pdu, CancellationToken ct)
+    private async Task<IsoTpTransmitStamps> SendSingleFrameAsync(ReadOnlyMemory<byte> pdu, CancellationToken ct)
     {
         int sfMax = IsoTpFrameCodec.SingleFrameMaxDataLength(_options.UseCanFd,
             _txEndpoint.UsesAddressExtension);
@@ -249,6 +299,7 @@ private async Task SendSingleFrameAsync(ReadOnlyMemory<byte> pdu, CancellationTo
                     new IsoTpException("Functional Single Frame TX confirmation failed with unknown reason."),
             };
         }
+        return new IsoTpTransmitStamps(confirmation.HostHandoffTimestamp, confirmation.HostTransmitTimestamp);
     }
 
     private static async Task<IReadOnlyList<IsoTpFunctionalResponse>> CollectFromSubscriptionAsync(
@@ -257,6 +308,10 @@ private static async Task<IReadOnlyList<IsoTpFunctionalResponse>> CollectFromSub
         var responses = new List<IsoTpFunctionalResponse>();
 
         // Combine the caller's token with a deadline token so the window bounds the collection.
+        // The window's end is also held as an arrival stamp: the timer's callback and this
+        // method's continuations are scheduling, and a frame that arrived after the deadline
+        // but before they ran is not the window's (Codex on #150).
+        long deadline = Stopwatch.GetTimestamp() + (long)(window.TotalSeconds * Stopwatch.Frequency);
         using var windowCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
         windowCts.CancelAfter(window);
         var windowToken = windowCts.Token;
@@ -265,8 +320,8 @@ private static async Task<IReadOnlyList<IsoTpFunctionalResponse>> CollectFromSub
         {
             await foreach (var frameEvent in sub.Frames.WithCancellation(windowToken).ConfigureAwait(false))
             {
-                if (TryParseFunctionalResponse(frameEvent.Frame, out var response))
-                    responses.Add(response!);
+                if (TryParseFunctionalResponse(frameEvent, out var response) && response!.HostArrivalTimestamp <= deadline)
+                    responses.Add(response);
             }
         }
         catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
@@ -285,8 +340,8 @@ private static async Task<IReadOnlyList<IsoTpFunctionalResponse>> CollectFromSub
             sub.Dispose();
             while (sub.TryRead(out var frameEvent))
             {
-                if (TryParseFunctionalResponse(frameEvent.Frame, out var response))
-                    responses.Add(response!);
+                if (TryParseFunctionalResponse(frameEvent, out var response) && response!.HostArrivalTimestamp <= deadline)
+                    responses.Add(response);
             }
         }
 
@@ -303,9 +358,10 @@ private static void DrainBuffered(ISubscription sub)
         while (sub.TryRead(out _)) { }
     }
 
-    private static bool TryParseFunctionalResponse(CanFrameView frame,
+    internal static bool TryParseFunctionalResponse(in CanFrameEvent frameEvent,
         out IsoTpFunctionalResponse? response)
     {
+        var frame = frameEvent.Frame;
         var payload = frame.Data.ToArray();
         bool isCanFd = frame.FrameKind == CanFrameType.CanFd;
 
@@ -333,7 +389,9 @@ private static bool TryParseFunctionalResponse(CanFrameView frame,
 
         var pdu = new byte[pci.Length];
         Array.Copy(payload, pci.DataOffset, pdu, 0, pci.Length);
-        response = new IsoTpFunctionalResponse((uint)frame.ID, pdu);
+        // Stamped by the demux at arrival; "now" only for an event built without a stamp.
+        var arrival = frameEvent.HostArrivalTimestamp > 0 ? frameEvent.HostArrivalTimestamp : Stopwatch.GetTimestamp();
+        response = new IsoTpFunctionalResponse((uint)frame.ID, pdu, arrival);
         return true;
     }
 

src/CanKit.Pro.IsoTp/IsoTpFunctionalCollection.cs

@@ -0,0 +1,22 @@
+using System.Collections.Generic;
+
+namespace CanKit.Pro.IsoTp;
+
+/// <summary>
+/// What <see cref="IsoTpFunctionalClient.SendAndCollectWithTransmitStampAsync"/> returns: the
+/// responses collected within the window, and when the request went out.
+/// </summary>
+public readonly struct IsoTpFunctionalCollection
+{
+    internal IsoTpFunctionalCollection(IReadOnlyList<IsoTpFunctionalResponse> responses, IsoTpTransmitStamps transmitStamps)
+    {
+        Responses = responses;
+        TransmitStamps = transmitStamps;
+    }
+
+    /// <summary>The Single-Frame responses collected within the window, in arrival order.</summary>
+    public IReadOnlyList<IsoTpFunctionalResponse> Responses { get; }
+
+    /// <summary>When the request was handed to the driver and transmitted; zero where unknown.</summary>
+    public IsoTpTransmitStamps TransmitStamps { get; }
+}