From f47212c598fdd0851e49fd380cacc9038bc63af1 Mon Sep 17 00:00:00 2001 From: Dietmar Borgards <2646931+dborgards@users.noreply.github.com> Date: Thu, 1 Oct 2026 19:16:11 +0200 Subject: [PATCH 1/5] fix(j1939tp): bind BAM steps to their session, keep T3 on a CTS hold, validate timers Three defects from the 2026-09-30 deep review (docs/reviews/2026-09-30-deep-review.md, findings T1, T2 and the J1939-TP half of I8). A BAM chain was addressed by its (destination, PGN) key only. Cancelling a BAM and sending the same PGN again inside the packet spacing left the first session's timer armed; it then drove the new session, so TP.DT 1 went out twice, a conforming receiver aborted, and the sender reported success. Each step now carries its session and acts only while that instance is the registered one. A CTS with numPackets = 0 replaced the running timer with T4 in every state, but T4's expiry is ignored outside WaitCts. After the last TP.DT that left the send with no timer at all, holding its slot and blocking every later send to that destination. The hold now only takes effect while the sender waits for a CTS; otherwise T3 keeps running. T1..T4 and BamPacketSpacing were not validated. A negative value made Arm throw after the session was registered and the RTS sent, again leaving a send without a deadline. T1..T4 must now be greater than zero and BamPacketSpacing not negative, rejected at Open. Each fix has a regression test that was seen to fail with that fix removed. Co-Authored-By: Claude Sonnet 5.5 --- src/CanKit.Pro.J1939Tp/J1939TpChannel.cs | 36 ++-- src/CanKit.Pro.J1939Tp/J1939TpOptions.cs | 15 ++ .../TestCases/J1939TpTests.cs | 166 ++++++++++++++++++ 3 files changed, 206 insertions(+), 11 deletions(-) diff --git a/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs b/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs index b9871ef..cc5d49c 100644 --- a/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs +++ b/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs @@ -935,17 +935,26 @@ private void StartTx(TxSessionKey key, byte[] pdu, TxCompletion tcs, bool isCm) // Do not schedule TP.DT until the BAM announce is TX-confirmed. Otherwise a rejected // BAM can still complete SendBamAsync after the packet spacing once DTs finish. SendTpCm(bam, destinationAddress: J1939TpFrames.GlobalDestinationAddress, session, - onConfirmed: () => OnBamAnnounceConfirmed(key)); + onConfirmed: () => OnBamAnnounceConfirmed(session)); } } - private void OnBamAnnounceConfirmed(TxSessionKey key) + // A BAM chain is driven by timers and send confirmations that outlive a cancel: the caller can + // cancel and send the same PGN again within the packet spacing, and the same key then names a + // different session. Every step therefore carries the session it belongs to and acts only + // while that very instance is the registered one -- the key alone cannot tell them apart. + private bool IsCurrentBamSession(TxSession session) + => !session.IsCm + && _txSessions.TryGetValue(session.Key, out var current) + && ReferenceEquals(current, session); + + private void OnBamAnnounceConfirmed(TxSession session) { - if (!_txSessions.TryGetValue(key, out var session) || session.IsCm) return; + if (!IsCurrentBamSession(session)) return; // BAM sender: the packet spacing between BAM and first DT, then between subsequent DTs. session.State = TxStage.SendingDt; session.NextSn = 1; - _actor.Schedule(_options.BamPacketSpacing, () => TrySendNextBamDt(key)); + _actor.Schedule(_options.BamPacketSpacing, () => TrySendNextBamDt(session)); } private void HandleRxTxSideResponse(byte sa, uint dataPgn, byte[] payload) @@ -966,6 +975,11 @@ private void HandleRxTxSideResponse(byte sa, uint dataPgn, byte[] payload) // Stay in WaitCts until a non-zero CTS resumes the transfer. if (numPackets == 0) { + // Only a sender waiting for a CTS has anything to hold. After the last DT (T3 + // for the EndOfMsgAck) or mid-block (T3 follows the block) the running timer + // stays: swapping it for a T4 whose expiry is ignored in those states left the + // send without any timer at all. + if (session.State != TxStage.WaitCts) return; session.Deadline?.Dispose(); session.Deadline = _deadlines.Arm(_options.T4, () => OnTxT4Expired(key)); return; @@ -1083,19 +1097,19 @@ private void HandleRxTxSideResponse(byte sa, uint dataPgn, byte[] payload) } } - private void TrySendNextBamDt(TxSessionKey key) + private void TrySendNextBamDt(TxSession session) { - if (!_txSessions.TryGetValue(key, out var session) || session.IsCm) return; + if (!IsCurrentBamSession(session)) return; int offset = (session.NextSn - 1) * J1939TpFrames.DtDataBytes; byte sn = session.NextSn; var dt = J1939TpFrames.BuildDt(sn, session.Pdu, offset); SendControlFrame(J1939Pgn.TpDt, dt, destinationAddress: J1939TpFrames.GlobalDestinationAddress, - session, onConfirmed: () => OnBamDtConfirmed(key, sn)); + session, onConfirmed: () => OnBamDtConfirmed(session, sn)); } - private void OnBamDtConfirmed(TxSessionKey key, byte confirmedSn) + private void OnBamDtConfirmed(TxSession session, byte confirmedSn) { - if (!_txSessions.TryGetValue(key, out var session) || session.IsCm) return; + if (!IsCurrentBamSession(session)) return; if (session.NextSn != confirmedSn) return; // stale confirmation // Compute the next SN as an int first: for a maximum-length PDU (TotalPackets=255) the @@ -1106,13 +1120,13 @@ private void OnBamDtConfirmed(TxSessionKey key, byte confirmedSn) { // BAM has no ack -- complete once every DT has been transmitted. session.Tcs.TrySetResult(null); - EndTx(key); + EndTx(session.Key); return; } session.NextSn = (byte)nextSn; // The spacing between two consecutive BAM DTs (J1939-21 §5.10.3, 50..200 ms). - _actor.Schedule(_options.BamPacketSpacing, () => TrySendNextBamDt(key)); + _actor.Schedule(_options.BamPacketSpacing, () => TrySendNextBamDt(session)); } private void TrySendNextCmDt(TxSessionKey key) diff --git a/src/CanKit.Pro.J1939Tp/J1939TpOptions.cs b/src/CanKit.Pro.J1939Tp/J1939TpOptions.cs index d2643fa..022e71e 100644 --- a/src/CanKit.Pro.J1939Tp/J1939TpOptions.cs +++ b/src/CanKit.Pro.J1939Tp/J1939TpOptions.cs @@ -152,6 +152,15 @@ public J1939TpOptions With( /// internal void Validate() { + // A negative timer makes Arm throw after the session is registered and the RTS is on the + // wire, which leaves a send with no deadline and its destination blocked; reject it here. + RequirePositive(nameof(T1), T1); + RequirePositive(nameof(T2), T2); + RequirePositive(nameof(T3), T3); + RequirePositive(nameof(T4), T4); + if (BamPacketSpacing < TimeSpan.Zero) + throw new ArgumentOutOfRangeException(nameof(BamPacketSpacing), BamPacketSpacing, + "BamPacketSpacing must not be negative."); if (MaxPacketsPerCts == 0) throw new ArgumentOutOfRangeException(nameof(MaxPacketsPerCts), MaxPacketsPerCts, "MaxPacketsPerCts must be in [1, 255]; 0 is not a valid CTS grant size."); @@ -168,4 +177,10 @@ internal void Validate() throw new ArgumentOutOfRangeException(nameof(MaxQueuedSendsPerDestination), MaxQueuedSendsPerDestination, "MaxQueuedSendsPerDestination must be >= 0 (0 admits no waiting send)."); } + + private static void RequirePositive(string name, TimeSpan value) + { + if (value <= TimeSpan.Zero) + throw new ArgumentOutOfRangeException(name, value, name + " must be greater than zero."); + } } diff --git a/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs b/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs index 235fb01..a131c86 100644 --- a/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs +++ b/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs @@ -1462,6 +1462,91 @@ public async Task Cm_Sender_T4Timeout_WhenPeerHoldsWithCtsZero() ex.Message.Should().Contain("T4"); } + // T4 is the hold timer after a CTS with numPackets = 0, and it only means something while the + // sender waits for a CTS. A hold that arrives after the last DT must leave the running T3 for + // the EndOfMsgAck alone: swapping it for a T4 whose expiry is ignored in that state left the + // send without any timer, and every later send to that destination queued behind it. + [Fact] + public async Task A_Cts_Hold_After_The_Last_Dt_Leaves_The_EndOfMsgAck_Timer_Running() + { + var session = NewSession(); + using var senderBus = Open(session, 0); + using var peerBus = Open(session, 1); + + const byte senderSa = 0x05; + const byte peerSa = 0x06; + const uint pgn = 0xFF05u; + const uint probePgn = 0xFF06u; + var payload = RandomPayload(21, seed: 41); // 3 packets + + // Distinct values, so the armed timer names which one is running. + var options = new J1939TpOptions().With( + t2: TimeSpan.FromSeconds(5), + t3: TimeSpan.FromSeconds(2), + t4: TimeSpan.FromMilliseconds(1050)); + using var clock = new VirtualClock(); + var actor = clock.NewActor(); + using var sender = new J1939TpChannel(new CanBusService(senderBus), sourceAddress: senderSa, + options, ownsService: true, actor); + + var rtsSeen = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var lastDtSeen = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var probeCtsSeen = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + peerBus.FrameObserved += (_, e) => + { + if (!e.CanFrame.IsExtendedFrame) return; + var fields = J1939Id.Decompose((uint)e.CanFrame.ID); + if (fields.SourceAddress != senderSa) return; + var data = e.CanFrame.Data.Span; + if (J1939Pgn.IsTransportCm(fields.Pgn) && data.Length >= 8) + { + if (data[0] == J1939TpFrames.ControlRts && J1939TpFrames.ReadDataPgn(data) == pgn) + rtsSeen.TrySetResult(null); + if (data[0] == J1939TpFrames.ControlCts && J1939TpFrames.ReadDataPgn(data) == probePgn) + probeCtsSeen.TrySetResult(null); + } + else if (fields.Pgn == J1939Pgn.TpDt && data.Length >= 1 && data[0] == 3) + { + lastDtSeen.TrySetResult(null); + } + }; + + var send = sender.SendCmAsync(pgn, destinationAddress: peerSa, payload); + await rtsSeen.Task.AsTaskWithTimeout(ShortTimeout); + + var cmId = (int)J1939Id.ComposePgn(7, J1939Pgn.TpCm, peerSa, senderSa); + peerBus.Transmit(CanFrame.Classic(cmId, J1939TpFrames.BuildCts(numPackets: 3, nextPacketSn: 1, dataPgn: pgn), + isExtendedFrame: true)); + await lastDtSeen.Task.AsTaskWithTimeout(ShortTimeout); + + // The last DT on the wire is not proof the timer for the EndOfMsgAck is armed yet. + await clock.WaitUntilTimerArmedAsync(actor, options.T3, ShortTimeout); + + var hold = new byte[8]; + hold[0] = J1939TpFrames.ControlCts; + hold[1] = 0x00; + hold[2] = 0x04; + hold[3] = 0xFF; + hold[4] = 0xFF; + hold[5] = (byte)(pgn & 0xFF); + hold[6] = (byte)((pgn >> 8) & 0xFF); + hold[7] = (byte)((pgn >> 16) & 0xFF); + peerBus.Transmit(CanFrame.Classic(cmId, hold, isExtendedFrame: true)); + + // The channel reads its frames in order: its answer to this RTS is proof the hold above + // has been handled, without any wall time standing in for it. + var probe = J1939TpFrames.BuildRts(totalBytes: 14, totalPackets: 2, maxPacketsPerCts: 0xFF, dataPgn: probePgn); + peerBus.Transmit(CanFrame.Classic(cmId, probe, isExtendedFrame: true)); + await probeCtsSeen.Task.AsTaskWithTimeout(ShortTimeout); + + await clock.AdvanceAsync(options.T3); + + Func act = async () => await send.WithTimeout(ShortTimeout); + var ex = (await act.Should().ThrowAsync()).Which; + ex.Reason.Should().Be(J1939TpAbortReason.Timeout); + ex.Message.Should().Contain("EndOfMsgAck"); + } + // FR-TP-031: the receiver must cap its CTS grant at the originator's RTS-advertised // maximum (here 2), even though its own MaxPacketsPerCts (16) is larger — and keep the // cap on every subsequent block's CTS. @@ -3099,6 +3184,87 @@ public void Options_Default_The_Queue_Limit_To_Eight_And_Reject_A_Negative_One() new J1939TpOptions().With(maxQueuedSendsPerDestination: 0).MaxQueuedSendsPerDestination.Should().Be(0); } + // The BAM chain is driven by timers and confirmations that outlive a cancel. Cancelling and + // sending the same PGN again inside the packet spacing puts a second session under the same + // key while the first one's timer is still armed; that timer must not drive the new session. + [Fact] + public async Task A_Cancelled_Bam_Does_Not_Drive_The_Next_Bam_Of_The_Same_Pgn() + { + const uint pgn = 0xFECBu; + var payload = RandomPayload(21, seed: 248); // 3 TP.DT + using var clock = new VirtualClock(); + var actor = clock.NewActor(); + using var bus = ControllableBus.EchoCapable(NewSession()); + var dtSequence = new List(); + bus.OnTransmitting = frame => + { + var fields = J1939Id.Decompose((uint)frame.ID); + if (fields.Pgn != J1939Pgn.TpDt) return; + lock (dtSequence) dtSequence.Add(frame.Data.Span[0]); + }; + var background = new List(); + using var service = new CanBusService(bus); + using var sender = BamSenderOn(actor, service); + sender.BackgroundExceptionOccurred += (_, ex) => { lock (background) background.Add(ex); }; + + using var cts = new CancellationTokenSource(); + var first = sender.SendBamAsync(pgn, payload, cts.Token); + await clock.WaitUntilTimerArmedAsync(actor, InFlightSpacing, ShortTimeout); + + cts.Cancel(); + Func cancelled = async () => await first.WithTimeout(ShortTimeout); + await cancelled.Should().ThrowAsync(); + + var second = sender.SendBamAsync(pgn, payload); + // The first session's timer is still armed; the second one's joins it once its announce + // is confirmed. Only then do both fire on the same advance. + while (await actor.PostAsync(() => actor.PendingTimerCount) < 2) + await Task.Yield(); + + await DrainAsync(clock, actor, second); + + lock (dtSequence) + dtSequence.Should().Equal(new byte[] { 1, 2, 3 }, + "the stale timer of the cancelled BAM must not send a TP.DT for the new one"); + lock (background) background.Should().BeEmpty(); + } + + [Theory] + [InlineData(nameof(J1939TpOptions.T1), 0)] + [InlineData(nameof(J1939TpOptions.T1), -1)] + [InlineData(nameof(J1939TpOptions.T2), 0)] + [InlineData(nameof(J1939TpOptions.T2), -1)] + [InlineData(nameof(J1939TpOptions.T3), 0)] + [InlineData(nameof(J1939TpOptions.T3), -1)] + [InlineData(nameof(J1939TpOptions.T4), 0)] + [InlineData(nameof(J1939TpOptions.T4), -1)] + [InlineData(nameof(J1939TpOptions.BamPacketSpacing), -1)] + public void A_Timer_Option_That_Cannot_Arm_Is_Rejected_At_Open(string option, int milliseconds) + { + var value = TimeSpan.FromMilliseconds(milliseconds); + var options = option switch + { + nameof(J1939TpOptions.T1) => new J1939TpOptions { T1 = value }, + nameof(J1939TpOptions.T2) => new J1939TpOptions { T2 = value }, + nameof(J1939TpOptions.T3) => new J1939TpOptions { T3 = value }, + nameof(J1939TpOptions.T4) => new J1939TpOptions { T4 = value }, + _ => new J1939TpOptions { BamPacketSpacing = value }, + }; + + using var bus = Open(NewSession(), 0); + Action open = () => J1939TpFactory.Open(bus, sourceAddress: 0x85, options: options); + open.Should().Throw().Which.ParamName.Should().Be(option); + } + + [Fact] + public void A_Zero_Bam_Packet_Spacing_Is_Still_Accepted() + { + using var bus = Open(NewSession(), 0); + using var channel = J1939TpFactory.Open(bus, sourceAddress: 0x86, + options: new J1939TpOptions { BamPacketSpacing = TimeSpan.Zero }); + channel.Should().NotBeNull(); + } + private static J1939TpChannel BamSenderOn(ProtocolActor actor, ICanBusService service) => new(service, sourceAddress: 0x10, new J1939TpOptions().With(bamPacketSpacing: InFlightSpacing), ownsService: false, actor); From ece32923c0756adcdfd3386a37152cdd68736343 Mon Sep 17 00:00:00 2001 From: Dietmar Borgards <2646931+dborgards@users.noreply.github.com> Date: Thu, 1 Oct 2026 21:56:13 +0200 Subject: [PATCH 2/5] fix(j1939tp): apply a CTS hold that races the block-end confirmation A CTS(0) that arrives right after the last DT of a block, before its confirmation has run, was dropped by the SendingDt guard, and the confirmation then armed T3. Remember the hold (not for the message's last packet, where T3 for the EndOfMsgAck is right) and arm T4 when the block ends; a CTS that grants packets clears it. Co-Authored-By: Claude Sonnet 5.5 --- src/CanKit.Pro.J1939Tp/J1939TpChannel.cs | 25 ++++- .../TestCases/J1939TpTests.cs | 99 +++++++++++++++++++ 2 files changed, 122 insertions(+), 2 deletions(-) diff --git a/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs b/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs index cc5d49c..20db2c9 100644 --- a/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs +++ b/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs @@ -979,12 +979,23 @@ private void HandleRxTxSideResponse(byte sa, uint dataPgn, byte[] payload) // for the EndOfMsgAck) or mid-block (T3 follows the block) the running timer // stays: swapping it for a T4 whose expiry is ignored in those states left the // send without any timer at all. - if (session.State != TxStage.WaitCts) return; + if (session.State != TxStage.WaitCts) + { + // A fast peer can hold right after the last DT of a block, before our + // confirmation of that DT has run -- the same early-response race a + // non-zero CTS is stashed for. The hold is remembered and takes effect when + // the block ends: T4 then, not T3. + if (session.State == TxStage.SendingDt && session.BlockRemaining == 1 + && session.NextSn != session.TotalPackets) + session.HoldPending = true; + return; + } session.Deadline?.Dispose(); session.Deadline = _deadlines.Arm(_options.T4, () => OnTxT4Expired(key)); return; } + session.HoldPending = false; // a CTS that grants packets ends any hold // What has gone out: the highest packet ever confirmed (HighestSentSn -- an int, so a // 255-packet message's last packet counts, where the byte NextSn wraps to 0), and, // while a block drains, the one outstanding (NextSn, unconfirmed). A CTS for a @@ -1187,7 +1198,15 @@ private void OnCmDtConfirmed(TxSessionKey key, byte confirmedSn) session.State = TxStage.WaitCts; session.Deadline?.Dispose(); - session.Deadline = _deadlines.Arm(_options.T3, () => OnTxT3Expired(key)); + if (session.HoldPending) + { + session.HoldPending = false; + session.Deadline = _deadlines.Arm(_options.T4, () => OnTxT4Expired(key)); + } + else + { + session.Deadline = _deadlines.Arm(_options.T3, () => OnTxT3Expired(key)); + } return; } @@ -1523,6 +1542,8 @@ public TxSession(TxSessionKey key, byte[] pdu, int totalPackets, TxCompletion tc /// been confirmed (Virtual-loopback race). Applied in . /// public bool HasPendingCts { get; set; } + /// A CTS(0) hold arrived before the last DT of the block was confirmed: T4 starts when it is. + public bool HoldPending { get; set; } public byte PendingCtsNumPackets { get; set; } /// The stashed CTS asks for a packet already sent: applied as soon as the outstanding DT is confirmed. public bool PendingCtsIsRetransmit { get; set; } diff --git a/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs b/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs index a131c86..e7e6e19 100644 --- a/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs +++ b/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs @@ -1547,6 +1547,105 @@ public async Task A_Cts_Hold_After_The_Last_Dt_Leaves_The_EndOfMsgAck_Timer_Runn ex.Message.Should().Contain("EndOfMsgAck"); } + // A hold can arrive right after the last DT of a block, before this side has seen the + // confirmation of that DT -- a fast peer on a loopback bus. It is the same early response a + // non-zero CTS is stashed for, so the hold takes effect when the block ends: T4, not T3. + [Fact] + public async Task A_Cts_Hold_Before_The_Block_End_Confirmation_Starts_T4_Not_T3() + { + using var bus = ControllableBus.DeferredEchoCapable(NewSession()); + using var service = new FrameConsumptionCountingBusService(new CanBusService(bus)); + const byte subjectSa = 0x10, peerSa = 0x20; + const uint pgn = 0xFEC7u; + var payload = RandomPayload(21, seed: 8); // three packets, granted one at a time + var options = new J1939TpOptions().With( + t2: TimeSpan.FromSeconds(5), t3: TimeSpan.FromSeconds(2), t4: TimeSpan.FromMilliseconds(1050)); + using var clock = new VirtualClock(); + var actor = clock.NewActor(); + using var sender = new J1939TpChannel(service, sourceAddress: subjectSa, options, + ownsService: false, actor); + static CanFrame PeerCm(byte peerSa, byte subjectSa, byte[] data) + => CanFrame.Classic((int)J1939Id.ComposePgn(7, J1939Pgn.TpCm, peerSa, subjectSa), data, isExtendedFrame: true); + + var send = sender.SendCmAsync(pgn, peerSa, payload); + await bus.DeferredEchoes.WaitForEnqueuedAsync(1, ShortTimeout); // the RTS + bus.DeferredEchoes.ReleaseNext(); + bus.RaiseObserved(PeerCm(peerSa, subjectSa, + J1939TpFrames.BuildCts(numPackets: 1, nextPacketSn: 1, dataPgn: pgn)), isEcho: false); + await bus.DeferredEchoes.WaitForEnqueuedAsync(2, ShortTimeout); // DT 1, its confirmation held + + var hold = new byte[8]; + hold[0] = J1939TpFrames.ControlCts; + hold[2] = 0x02; + hold[3] = hold[4] = 0xFF; + hold[5] = (byte)(pgn & 0xFF); + hold[6] = (byte)((pgn >> 8) & 0xFF); + hold[7] = (byte)((pgn >> 16) & 0xFF); + var handedOver = service.WaitUntilConsumedAsync(e => e.Frame.Data.ToArray().SequenceEqual(hold)); + bus.RaiseObserved(PeerCm(peerSa, subjectSa, hold), isEcho: false); + await handedOver.WaitAsync(ShortTimeout); + await actor.PostAsync(() => 0); + + bus.DeferredEchoes.ReleaseNext(); // the block ends now + await clock.WaitUntilTimerArmedAsync(actor, options.T4, ShortTimeout); + await clock.AdvanceAsync(options.T4); + + Func act = async () => await send.WithTimeout(ShortTimeout); + var ex = (await act.Should().ThrowAsync()).Which; + ex.Message.Should().Contain("T4"); + } + + // The same hold in the middle of a block has nothing to hold yet: the block runs on, and the + // timer after it is T3 as usual. + [Fact] + public async Task A_Cts_Hold_In_The_Middle_Of_A_Block_Is_Ignored() + { + using var bus = ControllableBus.DeferredEchoCapable(NewSession()); + using var service = new FrameConsumptionCountingBusService(new CanBusService(bus)); + const byte subjectSa = 0x10, peerSa = 0x20; + const uint pgn = 0xFEC5u; + var payload = RandomPayload(21, seed: 9); // three packets, granted in one block + var options = new J1939TpOptions().With( + t2: TimeSpan.FromSeconds(5), t3: TimeSpan.FromSeconds(2), t4: TimeSpan.FromMilliseconds(1050)); + using var clock = new VirtualClock(); + var actor = clock.NewActor(); + using var sender = new J1939TpChannel(service, sourceAddress: subjectSa, options, + ownsService: false, actor); + static CanFrame PeerCm(byte peerSa, byte subjectSa, byte[] data) + => CanFrame.Classic((int)J1939Id.ComposePgn(7, J1939Pgn.TpCm, peerSa, subjectSa), data, isExtendedFrame: true); + + var send = sender.SendCmAsync(pgn, peerSa, payload); + await bus.DeferredEchoes.WaitForEnqueuedAsync(1, ShortTimeout); + bus.DeferredEchoes.ReleaseNext(); + bus.RaiseObserved(PeerCm(peerSa, subjectSa, + J1939TpFrames.BuildCts(numPackets: 3, nextPacketSn: 1, dataPgn: pgn)), isEcho: false); + await bus.DeferredEchoes.WaitForEnqueuedAsync(2, ShortTimeout); // DT 1, its confirmation held + + var hold = new byte[8]; + hold[0] = J1939TpFrames.ControlCts; + hold[2] = 0x02; + hold[3] = hold[4] = 0xFF; + hold[5] = (byte)(pgn & 0xFF); + hold[6] = (byte)((pgn >> 8) & 0xFF); + hold[7] = (byte)((pgn >> 16) & 0xFF); + var handedOver = service.WaitUntilConsumedAsync(e => e.Frame.Data.ToArray().SequenceEqual(hold)); + bus.RaiseObserved(PeerCm(peerSa, subjectSa, hold), isEcho: false); + await handedOver.WaitAsync(ShortTimeout); + await actor.PostAsync(() => 0); + + bus.DeferredEchoes.ReleaseNext(); // DT 1 + await bus.DeferredEchoes.WaitForEnqueuedAsync(3, ShortTimeout); + bus.DeferredEchoes.ReleaseNext(); // DT 2 + await bus.DeferredEchoes.WaitForEnqueuedAsync(4, ShortTimeout); + bus.DeferredEchoes.ReleaseNext(); // DT 3: the last packet, so T3 for the EndOfMsgAck + await clock.WaitUntilTimerArmedAsync(actor, options.T3, ShortTimeout); + await clock.AdvanceAsync(options.T3); + + Func act = async () => await send.WithTimeout(ShortTimeout); + var ex = (await act.Should().ThrowAsync()).Which; + ex.Message.Should().Contain("EndOfMsgAck"); + } + // FR-TP-031: the receiver must cap its CTS grant at the originator's RTS-advertised // maximum (here 2), even though its own MaxPacketsPerCts (16) is larger — and keep the // cap on every subsequent block's CTS. From 0f1152781c2e9598cf3f9bfbc38f60fd8972512d Mon Sep 17 00:00:00 2001 From: Dietmar Borgards <2646931+dborgards@users.noreply.github.com> Date: Thu, 1 Oct 2026 22:22:34 +0200 Subject: [PATCH 3/5] fix(j1939tp): release a BAM's spacing timer with its session The spacing timer of a BAM captures its session and so its payload; the handle was discarded, so a cancelled BAM stayed reachable until the timer fired, which with a long BamPacketSpacing is as long as that value. The session now holds the handle and EndTx releases it. Tests: the timer is gone after a cancel; a late announce confirmation of a cancelled BAM cannot start the next BAM of the same PGN (the instance check, alone); a CTS hold while the last packet is outstanding leaves T3 for the EndOfMsgAck. The hold bookkeeping from the previous commit drops a condition that never had an effect. Co-Authored-By: Claude Sonnet 5.5 --- src/CanKit.Pro.J1939Tp/J1939TpChannel.cs | 15 +- .../TestCases/J1939TpTests.cs | 136 +++++++++++++++++- 2 files changed, 141 insertions(+), 10 deletions(-) diff --git a/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs b/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs index 20db2c9..f1018b1 100644 --- a/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs +++ b/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs @@ -895,6 +895,9 @@ internal void RemoveQueued(TxCompletion tcs) /// private void EndTx(TxSessionKey key) { + // The spacing timer of a BAM holds the session -- and its payload -- until it fires, which + // with a long spacing outlives a cancel by that long. + if (_txSessions.TryGetValue(key, out var ended)) ended.SpacingTimer?.Dispose(); _txSessions.Remove(key); if (_disposed != 0 || !_txQueues.TryGetValue(key.DestinationAddress, out var queue)) return; PendingTx? start = null; @@ -954,7 +957,7 @@ private void OnBamAnnounceConfirmed(TxSession session) // BAM sender: the packet spacing between BAM and first DT, then between subsequent DTs. session.State = TxStage.SendingDt; session.NextSn = 1; - _actor.Schedule(_options.BamPacketSpacing, () => TrySendNextBamDt(session)); + session.SpacingTimer = _actor.Schedule(_options.BamPacketSpacing, () => TrySendNextBamDt(session)); } private void HandleRxTxSideResponse(byte sa, uint dataPgn, byte[] payload) @@ -984,9 +987,9 @@ private void HandleRxTxSideResponse(byte sa, uint dataPgn, byte[] payload) // A fast peer can hold right after the last DT of a block, before our // confirmation of that DT has run -- the same early-response race a // non-zero CTS is stashed for. The hold is remembered and takes effect when - // the block ends: T4 then, not T3. - if (session.State == TxStage.SendingDt && session.BlockRemaining == 1 - && session.NextSn != session.TotalPackets) + // the block ends: T4 then, not T3. After the message's last packet the + // confirmation waits for the EndOfMsgAck under T3 and never reads it. + if (session.State == TxStage.SendingDt && session.BlockRemaining == 1) session.HoldPending = true; return; } @@ -1137,7 +1140,7 @@ private void OnBamDtConfirmed(TxSession session, byte confirmedSn) session.NextSn = (byte)nextSn; // The spacing between two consecutive BAM DTs (J1939-21 §5.10.3, 50..200 ms). - _actor.Schedule(_options.BamPacketSpacing, () => TrySendNextBamDt(session)); + session.SpacingTimer = _actor.Schedule(_options.BamPacketSpacing, () => TrySendNextBamDt(session)); } private void TrySendNextCmDt(TxSessionKey key) @@ -1536,6 +1539,8 @@ public TxSession(TxSessionKey key, byte[] pdu, int totalPackets, TxCompletion tc public byte NextSn { get; set; } public int BlockRemaining { get; set; } public IDeadline? Deadline { get; set; } + /// The armed packet-spacing timer of a BAM; released when the session ends. + public IDisposable? SpacingTimer { get; set; } /// /// Set when a CTS for the next block arrives before the last DT of the current block has diff --git a/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs b/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs index e7e6e19..ffd2b4b 100644 --- a/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs +++ b/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs @@ -3285,7 +3285,9 @@ public void Options_Default_The_Queue_Limit_To_Eight_And_Reject_A_Negative_One() // The BAM chain is driven by timers and confirmations that outlive a cancel. Cancelling and // sending the same PGN again inside the packet spacing puts a second session under the same - // key while the first one's timer is still armed; that timer must not drive the new session. + // key; nothing of the first one -- its timer, or a confirmation still in flight -- may drive + // the new session. Two lines of defence hold that (the timer is released with the session, and + // every step checks it belongs to the registered instance); the test fails only without both. [Fact] public async Task A_Cancelled_Bam_Does_Not_Drive_The_Next_Bam_Of_The_Same_Pgn() { @@ -3315,10 +3317,9 @@ public async Task A_Cancelled_Bam_Does_Not_Drive_The_Next_Bam_Of_The_Same_Pgn() await cancelled.Should().ThrowAsync(); var second = sender.SendBamAsync(pgn, payload); - // The first session's timer is still armed; the second one's joins it once its announce - // is confirmed. Only then do both fire on the same advance. - while (await actor.PostAsync(() => actor.PendingTimerCount) < 2) - await Task.Yield(); + // The second announce is out; its spacing timer is armed once that is confirmed. + await WaitForTransmitCount(bus, 2); + await clock.WaitUntilTimerArmedAsync(actor, InFlightSpacing, ShortTimeout); await DrainAsync(clock, actor, second); @@ -3328,6 +3329,131 @@ public async Task A_Cancelled_Bam_Does_Not_Drive_The_Next_Bam_Of_The_Same_Pgn() lock (background) background.Should().BeEmpty(); } + // The spacing timer of a BAM holds its session and payload until it fires. Cancelling must + // release it: with a long spacing, every cancelled BAM would otherwise stay reachable for + // that long. + [Fact] + public async Task Cancelling_A_Bam_Releases_Its_Spacing_Timer() + { + var spacing = TimeSpan.FromMinutes(10); + using var clock = new VirtualClock(); + var actor = clock.NewActor(); + using var bus = ControllableBus.EchoCapable(NewSession()); + using var service = new CanBusService(bus); + using var sender = new J1939TpChannel(service, sourceAddress: 0x10, + new J1939TpOptions().With(bamPacketSpacing: spacing), ownsService: false, actor); + + using var cts = new CancellationTokenSource(); + var send = sender.SendBamAsync(0xFECBu, RandomPayload(21, seed: 249), cts.Token); + await clock.WaitUntilTimerArmedAsync(actor, spacing, ShortTimeout); + + cts.Cancel(); + Func cancelled = async () => await send.WithTimeout(ShortTimeout); + await cancelled.Should().ThrowAsync(); + + (await actor.NextTimerDelayAsync()).Should().BeNull( + "the cancelled BAM's spacing timer is released, not left to fire in ten minutes"); + } + + // The other half of the same race: the announce of a cancelled BAM is still unconfirmed when + // the same PGN is sent again, and its confirmation arrives after the new session is + // registered under the same key. It belongs to the old session and must not start the new + // session's chain a second time. + [Fact] + public async Task A_Late_Announce_Confirmation_Of_A_Cancelled_Bam_Does_Not_Start_The_Next_Bam() + { + const uint pgn = 0xFECAu; + var payload = RandomPayload(21, seed: 250); // 3 TP.DT + using var clock = new VirtualClock(); + var actor = clock.NewActor(); + using var bus = ControllableBus.DeferredEchoCapable(NewSession()); + var dtSequence = new List(); + bus.OnTransmitting = frame => + { + if (J1939Id.Decompose((uint)frame.ID).Pgn != J1939Pgn.TpDt) return; + lock (dtSequence) dtSequence.Add(frame.Data.Span[0]); + }; + using var service = new CanBusService(bus); + using var sender = BamSenderOn(actor, service); + + using var cts = new CancellationTokenSource(); + var first = sender.SendBamAsync(pgn, payload, cts.Token); + await bus.DeferredEchoes.WaitForEnqueuedAsync(1, ShortTimeout); // its announce, confirmation held + cts.Cancel(); + Func cancelled = async () => await first.WithTimeout(ShortTimeout); + await cancelled.Should().ThrowAsync(); + + var second = sender.SendBamAsync(pgn, payload); + await bus.DeferredEchoes.WaitForEnqueuedAsync(2, ShortTimeout); // the new announce + + bus.DeferredEchoes.ReleaseNext(); // the cancelled BAM's announce + await actor.PostAsync(() => 0); + bus.DeferredEchoes.ReleaseNext(); // the new one + await clock.WaitUntilTimerArmedAsync(actor, InFlightSpacing, ShortTimeout); + + // Every TP.DT confirmation is held too; release them as they appear. + for (int sent = 0; sent < 3; sent++) + { + await clock.AdvanceAsync(InFlightSpacing); + await bus.DeferredEchoes.WaitForEnqueuedAsync(3 + sent, ShortTimeout); + bus.DeferredEchoes.ReleaseNext(); + if (sent < 2) await clock.WaitUntilTimerArmedAsync(actor, InFlightSpacing, ShortTimeout); + } + await second.WaitAsync(ShortTimeout); + + lock (dtSequence) + dtSequence.Should().Equal(new byte[] { 1, 2, 3 }); + } + + // A hold while the last packet of the message is outstanding has nothing to hold: the + // EndOfMsgAck is what the sender waits for next, under T3. + [Fact] + public async Task A_Cts_Hold_While_The_Last_Packet_Is_Outstanding_Leaves_T3_For_The_EndOfMsgAck() + { + using var bus = ControllableBus.DeferredEchoCapable(NewSession()); + using var service = new FrameConsumptionCountingBusService(new CanBusService(bus)); + const byte subjectSa = 0x10, peerSa = 0x20; + const uint pgn = 0xFEC4u; + var payload = RandomPayload(14, seed: 11); // two packets, one block + var options = new J1939TpOptions().With( + t2: TimeSpan.FromSeconds(5), t3: TimeSpan.FromSeconds(2), t4: TimeSpan.FromMilliseconds(1050)); + using var clock = new VirtualClock(); + var actor = clock.NewActor(); + using var sender = new J1939TpChannel(service, sourceAddress: subjectSa, options, + ownsService: false, actor); + static CanFrame PeerCm(byte peerSa, byte subjectSa, byte[] data) + => CanFrame.Classic((int)J1939Id.ComposePgn(7, J1939Pgn.TpCm, peerSa, subjectSa), data, isExtendedFrame: true); + + var send = sender.SendCmAsync(pgn, peerSa, payload); + await bus.DeferredEchoes.WaitForEnqueuedAsync(1, ShortTimeout); + bus.DeferredEchoes.ReleaseNext(); + bus.RaiseObserved(PeerCm(peerSa, subjectSa, + J1939TpFrames.BuildCts(numPackets: 2, nextPacketSn: 1, dataPgn: pgn)), isEcho: false); + await bus.DeferredEchoes.WaitForEnqueuedAsync(2, ShortTimeout); // DT 1 + bus.DeferredEchoes.ReleaseNext(); + await bus.DeferredEchoes.WaitForEnqueuedAsync(3, ShortTimeout); // DT 2, the last, held + + var hold = new byte[8]; + hold[0] = J1939TpFrames.ControlCts; + hold[2] = 0x03; + hold[3] = hold[4] = 0xFF; + hold[5] = (byte)(pgn & 0xFF); + hold[6] = (byte)((pgn >> 8) & 0xFF); + hold[7] = (byte)((pgn >> 16) & 0xFF); + var handedOver = service.WaitUntilConsumedAsync(e => e.Frame.Data.ToArray().SequenceEqual(hold)); + bus.RaiseObserved(PeerCm(peerSa, subjectSa, hold), isEcho: false); + await handedOver.WaitAsync(ShortTimeout); + await actor.PostAsync(() => 0); + + bus.DeferredEchoes.ReleaseNext(); + await clock.WaitUntilTimerArmedAsync(actor, options.T3, ShortTimeout); + await clock.AdvanceAsync(options.T3); + + Func act = async () => await send.WithTimeout(ShortTimeout); + var ex = (await act.Should().ThrowAsync()).Which; + ex.Message.Should().Contain("EndOfMsgAck"); + } + [Theory] [InlineData(nameof(J1939TpOptions.T1), 0)] [InlineData(nameof(J1939TpOptions.T1), -1)] From 34ecd1907d5bd63b86886404f9140ce13c77f7b8 Mon Sep 17 00:00:00 2001 From: Dietmar Borgards <2646931+dborgards@users.noreply.github.com> Date: Thu, 1 Oct 2026 22:38:35 +0200 Subject: [PATCH 4/5] fix(j1939tp): release the spacing timer on channel dispose too The channel ends its sessions through TxSession.Fail on dispose, not through EndTx, so a BAM in flight on a borrowed actor kept its spacing timer, session and payload until the interval elapsed. Release it in Cancel and Fail as well, behind one method. Also drops the redundant IsCm test from the current-session check (only BAM steps call it) and adds coverage for a late TP.DT confirmation of a cancelled BAM. Co-Authored-By: Claude Sonnet 5.5 --- src/CanKit.Pro.J1939Tp/J1939TpChannel.cs | 19 +++-- .../TestCases/J1939TpTests.cs | 76 +++++++++++++++++++ 2 files changed, 90 insertions(+), 5 deletions(-) diff --git a/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs b/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs index f1018b1..90dd6ad 100644 --- a/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs +++ b/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs @@ -897,8 +897,11 @@ private void EndTx(TxSessionKey key) { // The spacing timer of a BAM holds the session -- and its payload -- until it fires, which // with a long spacing outlives a cancel by that long. - if (_txSessions.TryGetValue(key, out var ended)) ended.SpacingTimer?.Dispose(); - _txSessions.Remove(key); + if (_txSessions.TryGetValue(key, out var ended)) + { + ended.ReleaseSpacingTimer(); + _txSessions.Remove(key); + } if (_disposed != 0 || !_txQueues.TryGetValue(key.DestinationAddress, out var queue)) return; PendingTx? start = null; while (start is null && queue.Count > 0) @@ -947,9 +950,7 @@ private void StartTx(TxSessionKey key, byte[] pdu, TxCompletion tcs, bool isCm) // different session. Every step therefore carries the session it belongs to and acts only // while that very instance is the registered one -- the key alone cannot tell them apart. private bool IsCurrentBamSession(TxSession session) - => !session.IsCm - && _txSessions.TryGetValue(session.Key, out var current) - && ReferenceEquals(current, session); + => _txSessions.TryGetValue(session.Key, out var current) && ReferenceEquals(current, session); private void OnBamAnnounceConfirmed(TxSession session) { @@ -1572,10 +1573,17 @@ public TxSession(TxSessionKey key, byte[] pdu, int totalPackets, TxCompletion tc public bool LastDtQueued { get; set; } public byte PendingCtsNextSn { get; set; } + public void ReleaseSpacingTimer() + { + SpacingTimer?.Dispose(); + SpacingTimer = null; + } + public void Cancel() { Deadline?.Dispose(); Deadline = null; + ReleaseSpacingTimer(); Tcs.TrySetCanceled(); } @@ -1583,6 +1591,7 @@ public void Fail(Exception ex) { Deadline?.Dispose(); Deadline = null; + ReleaseSpacingTimer(); Tcs.TrySetException(ex); } } diff --git a/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs b/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs index ffd2b4b..ba2bca5 100644 --- a/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs +++ b/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs @@ -3454,6 +3454,82 @@ static CanFrame PeerCm(byte peerSa, byte subjectSa, byte[] data) ex.Message.Should().Contain("EndOfMsgAck"); } + // The same release when the whole channel goes: a BAM in flight on a borrowed actor must not + // leave its spacing timer -- and so its session and payload -- behind. + [Fact] + public async Task Disposing_The_Channel_Releases_The_Spacing_Timer_Of_A_Bam_In_Flight() + { + var spacing = TimeSpan.FromMinutes(10); + using var clock = new VirtualClock(); + var actor = clock.NewActor(); + using var bus = ControllableBus.EchoCapable(NewSession()); + using var service = new CanBusService(bus); + var sender = new J1939TpChannel(service, sourceAddress: 0x10, + new J1939TpOptions().With(bamPacketSpacing: spacing), ownsService: false, actor); + + var send = sender.SendBamAsync(0xFECBu, RandomPayload(21, seed: 251)); + await clock.WaitUntilTimerArmedAsync(actor, spacing, ShortTimeout); + + sender.Dispose(); + Func failed = async () => await send.WithTimeout(ShortTimeout); + await failed.Should().ThrowAsync(); + + (await actor.NextTimerDelayAsync()).Should().BeNull( + "the spacing timer is released with the session, not left to fire in ten minutes"); + } + + // And the TP.DT confirmation: a DT of a cancelled BAM can still be in flight when the same PGN + // is sent again, and its confirmation, arriving afterwards, belongs to the old session. The + // sequence-number check in the handler also holds this one, so the test pins the outcome, not + // the instance check alone (the announce test above does that). + [Fact] + public async Task A_Late_Dt_Confirmation_Of_A_Cancelled_Bam_Does_Not_Advance_The_Next_Bam() + { + const uint pgn = 0xFEC9u; + var payload = RandomPayload(21, seed: 252); // 3 TP.DT + using var clock = new VirtualClock(); + var actor = clock.NewActor(); + using var bus = ControllableBus.DeferredEchoCapable(NewSession()); + var dtSequence = new List(); + bus.OnTransmitting = frame => + { + if (J1939Id.Decompose((uint)frame.ID).Pgn != J1939Pgn.TpDt) return; + lock (dtSequence) dtSequence.Add(frame.Data.Span[0]); + }; + using var service = new CanBusService(bus); + using var sender = BamSenderOn(actor, service); + + using var cts = new CancellationTokenSource(); + var first = sender.SendBamAsync(pgn, payload, cts.Token); + await bus.DeferredEchoes.WaitForEnqueuedAsync(1, ShortTimeout); // announce + bus.DeferredEchoes.ReleaseNext(); + await clock.WaitUntilTimerArmedAsync(actor, InFlightSpacing, ShortTimeout); + await clock.AdvanceAsync(InFlightSpacing); + await bus.DeferredEchoes.WaitForEnqueuedAsync(2, ShortTimeout); // DT 1, its confirmation held + + cts.Cancel(); + Func cancelled = async () => await first.WithTimeout(ShortTimeout); + await cancelled.Should().ThrowAsync(); + var second = sender.SendBamAsync(pgn, payload); + await bus.DeferredEchoes.WaitForEnqueuedAsync(3, ShortTimeout); // the new announce + + bus.DeferredEchoes.ReleaseNext(); // the cancelled BAM's DT 1 + await actor.PostAsync(() => 0); + bus.DeferredEchoes.ReleaseNext(); // the new announce + for (int sent = 0; sent < 3; sent++) + { + await clock.WaitUntilTimerArmedAsync(actor, InFlightSpacing, ShortTimeout); + await clock.AdvanceAsync(InFlightSpacing); + await bus.DeferredEchoes.WaitForEnqueuedAsync(4 + sent, ShortTimeout); + bus.DeferredEchoes.ReleaseNext(); + } + await second.WaitAsync(ShortTimeout); + + lock (dtSequence) + dtSequence.Should().Equal(new byte[] { 1, 1, 2, 3 }, + "the old session's DT 1, then the new session's own three"); + } + [Theory] [InlineData(nameof(J1939TpOptions.T1), 0)] [InlineData(nameof(J1939TpOptions.T1), -1)] From 570ab7b9b2e9c0b875b88e34a86188376d391306 Mon Sep 17 00:00:00 2001 From: Dietmar Borgards <2646931+dborgards@users.noreply.github.com> Date: Thu, 1 Oct 2026 22:53:44 +0200 Subject: [PATCH 5/5] test(j1939tp): dispose the channel in the dispose test, cover a late confirmation with no successor Takes back the release line in TxSession.Cancel, which nothing calls (only RxSession.Cancel is used), so it only added an uncovered line. Co-Authored-By: Claude Sonnet 5.5 --- src/CanKit.Pro.J1939Tp/J1939TpChannel.cs | 1 - .../TestCases/J1939TpTests.cs | 33 ++++++++++++++++++- 2 files changed, 32 insertions(+), 2 deletions(-) diff --git a/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs b/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs index 90dd6ad..ed28716 100644 --- a/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs +++ b/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs @@ -1583,7 +1583,6 @@ public void Cancel() { Deadline?.Dispose(); Deadline = null; - ReleaseSpacingTimer(); Tcs.TrySetCanceled(); } diff --git a/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs b/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs index ba2bca5..c21fdbc 100644 --- a/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs +++ b/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs @@ -3464,7 +3464,7 @@ public async Task Disposing_The_Channel_Releases_The_Spacing_Timer_Of_A_Bam_In_F var actor = clock.NewActor(); using var bus = ControllableBus.EchoCapable(NewSession()); using var service = new CanBusService(bus); - var sender = new J1939TpChannel(service, sourceAddress: 0x10, + using var sender = new J1939TpChannel(service, sourceAddress: 0x10, new J1939TpOptions().With(bamPacketSpacing: spacing), ownsService: false, actor); var send = sender.SendBamAsync(0xFECBu, RandomPayload(21, seed: 251)); @@ -3530,6 +3530,37 @@ public async Task A_Late_Dt_Confirmation_Of_A_Cancelled_Bam_Does_Not_Advance_The "the old session's DT 1, then the new session's own three"); } + // The same late confirmation with no successor: nothing is registered under the key any more, + // and the cancelled BAM's announce confirmation must not start a chain for it. + [Fact] + public async Task A_Late_Announce_Confirmation_Of_A_Cancelled_Bam_With_No_Successor_Sends_Nothing() + { + using var clock = new VirtualClock(); + var actor = clock.NewActor(); + using var bus = ControllableBus.DeferredEchoCapable(NewSession()); + var dtCount = 0; + bus.OnTransmitting = frame => + { + if (J1939Id.Decompose((uint)frame.ID).Pgn == J1939Pgn.TpDt) Interlocked.Increment(ref dtCount); + }; + using var service = new CanBusService(bus); + using var sender = BamSenderOn(actor, service); + + using var cts = new CancellationTokenSource(); + var send = sender.SendBamAsync(0xFECAu, RandomPayload(21, seed: 253), cts.Token); + await bus.DeferredEchoes.WaitForEnqueuedAsync(1, ShortTimeout); // its announce, confirmation held + cts.Cancel(); + Func cancelled = async () => await send.WithTimeout(ShortTimeout); + await cancelled.Should().ThrowAsync(); + + bus.DeferredEchoes.ReleaseNext(); + await actor.PostAsync(() => 0); + await clock.AdvanceAsync(InFlightSpacing); + + (await actor.NextTimerDelayAsync()).Should().BeNull("no chain was started for the cancelled BAM"); + Volatile.Read(ref dtCount).Should().Be(0); + } + [Theory] [InlineData(nameof(J1939TpOptions.T1), 0)] [InlineData(nameof(J1939TpOptions.T1), -1)]