feat(j1939tp)!: serve retransmit requests, normalise PDU1 PGNs, refuse an RTS allowing no packet, deliver before the event, and rename Th to BamPacketSpacing

src/CanKit.Pro.J1939Tp/IJ1939TpChannel.cs

@@ -13,7 +13,7 @@ namespace CanKit.Pro.J1939Tp;
 /// <remarks>
 /// <para>
 /// Threading model (SRS FR-TP-034 = FR-TP-016/017 applied to J1939-TP): every session's state
-/// (sequence numbers, remaining bytes, block counters, T1..T4/Th deadlines) lives inside a
+/// (sequence numbers, remaining bytes, block counters, T1..T4 deadlines and the BAM packet spacing) lives inside a
 /// single <see cref="Actor.IProtocolActor"/> mailbox and is only ever read/written on the
 /// actor's loop thread. Callers may invoke <see cref="SendBamAsync"/> or
 /// <see cref="SendCmAsync"/> concurrently from any thread; the channel serializes them per
@@ -79,10 +79,13 @@ Task SendCmAsync(uint pgn, byte destinationAddress, ReadOnlyMemory<byte> payload
     IAsyncEnumerable<J1939TpDatagram> ReceiveAllAsync(CancellationToken cancellationToken = default);
 
     /// <summary>
-    /// Raised on the actor's loop thread every time a full PDU is reassembled. The same datagram
-    /// is also enqueued for <see cref="ReceiveAsync"/> / <see cref="ReceiveAllAsync"/>. Handlers
-    /// must be lightweight and non-throwing; a throwing handler is caught and surfaced via
-    /// <see cref="BackgroundExceptionOccurred"/>.
+    /// Raised every time a full PDU is reassembled, after the datagram has been enqueued for
+    /// <see cref="ReceiveAsync"/> / <see cref="ReceiveAllAsync"/>, and on the thread pool, not
+    /// on the actor's loop -- so a handler that waits on this channel synchronously gets the
+    /// datagram rather than deadlocking the actor (#58; as ISO-TP delivers). Two datagrams
+    /// reassembled close together may therefore reach their handlers concurrently, and not
+    /// necessarily in the order they completed; <see cref="ReceiveAsync"/> keeps that order.
+    /// A throwing handler is caught and surfaced via <see cref="BackgroundExceptionOccurred"/>.
     /// </summary>
     event EventHandler<J1939TpDatagram>? DatagramReceived;
 

src/CanKit.Pro.J1939Tp/J1939TpException.cs

@@ -23,7 +23,7 @@ protected J1939TpException(CanKitErrorCode errorCode, string message) : base(err
 
 /// <summary>
 /// Raised when a J1939-21 §5.10.5 Connection Abort is issued or received on a TP.CM session,
-/// or when a BAM/TP.CM session gives up because one of T1/T2/T3/T4/Th expired.
+/// or when a BAM/TP.CM session gives up because one of T1/T2/T3/T4 expired.
 /// </summary>
 public sealed class J1939TpAbortException : J1939TpException
 {

src/CanKit.Pro.J1939Tp/J1939TpFrames.cs

@@ -170,8 +170,14 @@ public static byte[] BuildDt(byte sn, ReadOnlySpan<byte> pdu, int offset)
     public static uint ReadDataPgn(ReadOnlySpan<byte> tpCmPayload)
     {
         if (tpCmPayload.Length < 8) throw new ArgumentException("TP.CM payload must be 8 bytes.", nameof(tpCmPayload));
-        return ((uint)tpCmPayload[5] | ((uint)tpCmPayload[6] << 8) | ((uint)tpCmPayload[7] << 16))
-               & J1939Pgn.MaxValue;
+        uint pgn = ((uint)tpCmPayload[5] | ((uint)tpCmPayload[6] << 8) | ((uint)tpCmPayload[7] << 16))
+                   & J1939Pgn.MaxValue;
+        // A PDU1 PGN (PDU Format < 240) has a PDU Specific byte of 0 (SAE J1939-21); a stack
+        // that writes the destination address into it instead names the same group, and its
+        // CTS or EndOfMsgAck must still find the session keyed on the PGN. Normalised here, so
+        // every reader of the field agrees (#58).
+        if (((pgn >> 8) & 0xFF) < 240) pgn &= 0x3FF00u;
+        return pgn;
     }
 
     private static void WriteDataPgn(byte[] payload, uint dataPgn)

src/CanKit.Pro.J1939Tp/J1939TpOptions.cs

@@ -15,10 +15,12 @@ namespace CanKit.Pro.J1939Tp;
 ///   <item><description><see cref="T2"/> = 1250 ms — CTS→first TP.DT timeout at the receiver.</description></item>
 ///   <item><description><see cref="T3"/> = 1250 ms — RTS→CTS, block→next CTS and last DT→EndOfMsgAck timeout at the originator.</description></item>
 ///   <item><description><see cref="T4"/> = 1050 ms — TP.CM hold timeout at the originator after CTS(0) (§5.10.2.4).</description></item>
-///   <item><description><see cref="Th"/> = 50 ms — hold-off between two consecutive BAM DTs (50..200 ms).</description></item>
 /// </list>
 /// The standard's Tr (200 ms) is the time a node has to <em>send</em> a response it owes, not
-/// a timer a peer is held to; this stack answers at once and has no option for it (#31).
+/// a timer a peer is held to; this stack answers at once and has no option for it (#31). Its
+/// Th (500 ms) is the holding time between two CTS(0) messages a responder sends; this stack
+/// sends none and has no option for it either (#144). <see cref="BamPacketSpacing"/> (50 ms)
+/// is the spacing between two BAM packets, §5.10.3's 50..200 ms -- not a timer of §5.10.2.4.
 /// </remarks>
 public sealed class J1939TpOptions
 {
@@ -46,11 +48,22 @@ public sealed class J1939TpOptions
     public TimeSpan T4 { get; init; } = TimeSpan.FromMilliseconds(1050);
 
     /// <summary>
-    /// Th — minimum hold-off between two consecutive BAM TP.DT frames on the wire (§5.10.3
+    /// Minimum spacing between two consecutive BAM TP.DT frames on the wire (§5.10.3
     /// "50..200 ms"). Default 50 ms to stay at the lower recommended bound while still gating
-    /// against a receiver that cannot keep up.
+    /// against a receiver that cannot keep up. Not the standard's Th, which is the holding time
+    /// between CTS(0) messages and which this stack does not use; this option was named Th
+    /// before #144.
     /// </summary>
-    public TimeSpan Th { get; init; } = TimeSpan.FromMilliseconds(50);
+    public TimeSpan BamPacketSpacing { get; init; } = TimeSpan.FromMilliseconds(50);
+
+    /// <summary>
+    /// How many times per TP.CM session this originator serves a CTS that asks for a packet it
+    /// has already sent -- a retransmit request (§5.10.2.4); the next one is answered with
+    /// Connection Abort reason 5, "maximum retransmit request limit reached" (table 7). The
+    /// standard names the limit and leaves its value to the implementation. Default 2; 0 serves
+    /// none (#58).
+    /// </summary>
+    public int MaxRetransmitRequests { get; init; } = 2;
 
     /// <summary>
     /// TX priority for TP.CM / TP.DT frames sent by this channel (0..7, 0 = highest). J1939-21
@@ -85,25 +98,30 @@ public J1939TpOptions With(
         TimeSpan? t2 = null,
         TimeSpan? t3 = null,
         TimeSpan? t4 = null,
-        TimeSpan? th = null,
+        TimeSpan? bamPacketSpacing = null,
         byte? priority = null,
         byte? maxPacketsPerCts = null,
-        int? receiveBufferCapacity = null)
+        int? receiveBufferCapacity = null,
+        int? maxRetransmitRequests = null)
     {
         if (maxPacketsPerCts is 0)
             throw new ArgumentOutOfRangeException(nameof(maxPacketsPerCts), maxPacketsPerCts,
                 "MaxPacketsPerCts must be in [1, 255]; 0 is not a valid CTS grant size.");
+        if (maxRetransmitRequests < 0)
+            throw new ArgumentOutOfRangeException(nameof(maxRetransmitRequests), maxRetransmitRequests,
+                "MaxRetransmitRequests must be >= 0 (0 serves none).");
 
         return new()
         {
             T1 = t1 ?? T1,
             T2 = t2 ?? T2,
             T3 = t3 ?? T3,
             T4 = t4 ?? T4,
-            Th = th ?? Th,
+            BamPacketSpacing = bamPacketSpacing ?? BamPacketSpacing,
             Priority = priority ?? Priority,
             MaxPacketsPerCts = maxPacketsPerCts ?? MaxPacketsPerCts,
             ReceiveBufferCapacity = receiveBufferCapacity ?? ReceiveBufferCapacity,
+            MaxRetransmitRequests = maxRetransmitRequests ?? MaxRetransmitRequests,
         };
     }
 
@@ -122,5 +140,8 @@ internal void Validate()
         if (ReceiveBufferCapacity < 1)
             throw new ArgumentOutOfRangeException(nameof(ReceiveBufferCapacity), ReceiveBufferCapacity,
                 "ReceiveBufferCapacity must be >= 1.");
+        if (MaxRetransmitRequests < 0)
+            throw new ArgumentOutOfRangeException(nameof(MaxRetransmitRequests), MaxRetransmitRequests,
+                "MaxRetransmitRequests must be >= 0 (0 serves none).");
     }
 }

src/CanKit.Pro.J1939Tp/README.md

@@ -13,13 +13,13 @@ can still change until then. See [Versioning](https://github.com/dborgards/CanKi
 
 TP.DT (Data Transfer) frames carry the segmented payload for both flavors, sequence-numbered from 1 (FR-TP-033). Every session runs on its own actor-owned state and its own set of `IDeadline`s (T1, T2, T3, T4 — FR-TP-032), so multiple sessions can execute in parallel over the same physical bus (FR-TP-034/035) without interfering with each other.
 
-The timers carry J1939-21 §5.10.2.4's meanings: T1 (750 ms) between TP.DTs at the receiver, T2 (1250 ms) from the receiver's CTS to the first TP.DT of the block, T3 (1250 ms) at the originator for the response it is owed — CTS after RTS, the next CTS after a block, EndOfMsgAck after the last packet — and T4 (1050 ms) after a CTS(0) hold. Tr (200 ms) is the time a node has to *send* a response, not a timer a peer is held to, so there is no option for it (#31); the `Th` option is the BAM inter-packet spacing, not the standard's holding time (#144). Connection Abort carries table 7's reason codes and nothing outside the table (`J1939TpAbortReason`), so a peer stack reads the abort as what happened (#33).
+The timers carry J1939-21 §5.10.2.4's meanings: T1 (750 ms) between TP.DTs at the receiver, T2 (1250 ms) from the receiver's CTS to the first TP.DT of the block, T3 (1250 ms) at the originator for the response it is owed — CTS after RTS, the next CTS after a block, EndOfMsgAck after the last packet — and T4 (1050 ms) after a CTS(0) hold. Tr (200 ms) is the time a node has to *send* a response, not a timer a peer is held to, so there is no option for it (#31); Th (500 ms), the holding time between two CTS(0) messages, is not used by this stack, which sends no CTS(0), so there is no option for it either — the BAM inter-packet spacing of §5.10.3 is `BamPacketSpacing`, named `Th` before #144. Connection Abort carries table 7's reason codes and nothing outside the table (`J1939TpAbortReason`), so a peer stack reads the abort as what happened (#33). A CTS that asks for a packet already sent is served, up to `MaxRetransmitRequests` (default 2) times per session, after which reason 5 goes out; a PDU1 PGN in a TP.CM is read with its low byte cleared, so a stack that writes the destination address there still reaches the session; and an RTS allowing no packet per CTS opens no session (#58).
 
 The channel ships on nuget.org alongside the other `CanKit.Pro.*` L2/L3 building blocks. It re-uses:
 
 - `CanKit.Pro.RawCan` — one `ICanBusService` per channel to demultiplex the TP.CM / TP.DT frames back out of the shared bus stream and to confirm outbound frames.
 - `CanKit.Pro.Actor` — one `IProtocolActor` mailbox for single-writer session state.
-- `CanKit.Pro.Reliability` — `IDeadlineScheduler` for T1/T2/T3/T4/Th, cancelled/re-armed on the actor's loop.
+- `CanKit.Pro.Reliability` — `IDeadlineScheduler` for T1/T2/T3/T4 and the BAM packet spacing, cancelled/re-armed on the actor's loop.
 - `CanKit.Pro.Addressing` — `J1939Id` / `J1939Pgn` for composing the TP.CM (PGN 0xEC00) and TP.DT (PGN 0xEB00) 29-bit IDs.
 
 ## Basic usage

tests/CanKit.Pro.Tests/ApiApprovals/CanKit.Pro.J1939Tp.approved.txt

@@ -58,15 +58,16 @@ namespace CanKit.Pro.J1939Tp
     public sealed class J1939TpOptions
     {
         public J1939TpOptions() { }
+        public System.TimeSpan BamPacketSpacing { get; init; }
         public byte MaxPacketsPerCts { get; init; }
+        public int MaxRetransmitRequests { get; init; }
         public byte Priority { get; init; }
         public int ReceiveBufferCapacity { get; init; }
         public System.TimeSpan T1 { get; init; }
         public System.TimeSpan T2 { get; init; }
         public System.TimeSpan T3 { get; init; }
         public System.TimeSpan T4 { get; init; }
-        public System.TimeSpan Th { get; init; }
-        public CanKit.Pro.J1939Tp.J1939TpOptions With(System.TimeSpan? t1 = default, System.TimeSpan? t2 = default, System.TimeSpan? t3 = default, System.TimeSpan? t4 = default, System.TimeSpan? th = default, byte? priority = default, byte? maxPacketsPerCts = default, int? receiveBufferCapacity = default) { }
+        public CanKit.Pro.J1939Tp.J1939TpOptions With(System.TimeSpan? t1 = default, System.TimeSpan? t2 = default, System.TimeSpan? t3 = default, System.TimeSpan? t4 = default, System.TimeSpan? bamPacketSpacing = default, byte? priority = default, byte? maxPacketsPerCts = default, int? receiveBufferCapacity = default, int? maxRetransmitRequests = default) { }
     }
     public sealed class J1939TpSendRejectedException : CanKit.Pro.J1939Tp.J1939TpException
     {

tests/CanKit.Pro.Tests/TestCases/J1939/J1939NodeTests.cs

@@ -623,11 +623,11 @@ public async Task Send_LargePayload_UsesJ1939TpBamPath()
         // Shorten Th so the multi-frame test runs in <1s while still exercising the timer.
         var senderOpts = new J1939NodeOptions(Name(1))
         {
-            TransportOptions = new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(5)),
+            TransportOptions = new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(5)),
         };
         var receiverOpts = new J1939NodeOptions(Name(2))
         {
-            TransportOptions = new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(5)),
+            TransportOptions = new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(5)),
         };
 
         using var sender = J1939Node.Open(busA, senderOpts);
@@ -680,7 +680,7 @@ public async Task DirectedTpCm_ToClaimedAddress_IsReceivedAfterClaim()
         // and surfaces the directed multi-frame PDU on MessageReceived.
         using var receiver = J1939Node.Open(busB, new J1939NodeOptions(Name(2))
         {
-            TransportOptions = new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(5)),
+            TransportOptions = new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(5)),
         });
         await receiver.ClaimAddressAsync(0xA0).WithTimeout(ShortTimeout);
         receiver.ClaimState.Should().Be(J1939ClaimState.Claimed);
@@ -690,7 +690,7 @@ public async Task DirectedTpCm_ToClaimedAddress_IsReceivedAfterClaim()
         // J1939-TP channel from a different SA so the frames actually travel across the
         // virtual bus and hit the node's transport RX filter.
         using var peerTp = CanKit.Pro.J1939Tp.J1939Tp.Open(busA, sourceAddress: 0x55,
-            new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(5)));
+            new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(5)));
 
         var payload = new byte[24];
         for (int i = 0; i < payload.Length; i++) payload[i] = (byte)(0xB0 + i);
@@ -1014,7 +1014,7 @@ public async Task RebindTransport_DoesNotDeliverBamMoreThanOncePerRebind()
         var opts = new J1939NodeOptions(Name(1))
         {
             ClaimAnnounceTimeout = TimeSpan.FromMilliseconds(40),
-            TransportOptions = new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(2)),
+            TransportOptions = new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(2)),
         };
         using var node = J1939Node.Open(busNode, opts);
 
@@ -1042,11 +1042,11 @@ static byte[] Datagram(int seq)
         }
 
         using var peerTp = CanKit.Pro.J1939Tp.J1939Tp.Open(busPeer, sourceAddress: 0x77,
-            new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(2)));
+            new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(2)));
         // A second source address for the probes: one SA may only run one BAM session at a
         // time, and the probe must not have to queue behind the background stream.
         using var probeTp = CanKit.Pro.J1939Tp.J1939Tp.Open(busProbe, sourceAddress: 0x78,
-            new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(2)));
+            new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(2)));
 
         int sent = 0;
         using var peerCts = new CancellationTokenSource();
@@ -1156,7 +1156,7 @@ public async Task Send_InFlightAcrossReclaim_FailsWithNoAddressException()
         var opts = new J1939NodeOptions(Name(1))
         {
             ClaimAnnounceTimeout = TimeSpan.FromMilliseconds(200),
-            TransportOptions = new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(60)),
+            TransportOptions = new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(60)),
         };
         using var node = J1939Node.Open(busA, opts);
         await node.ClaimAddressAsync(0x11).WithTimeout(ShortTimeout);
@@ -1393,7 +1393,7 @@ public async Task StartPeriodicSend_MultiFrame_Emits_On_An_Exact_Grid_On_A_Clock
 
         var nodeOptions = new J1939NodeOptions(Name(1))
         {
-            TransportOptions = new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(1)),
+            TransportOptions = new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(1)),
         };
         var senderActor = clock.NewActor();
         using var sender = new J1939NodeImpl(service, nodeOptions, ownsService: false, senderActor);