CanBusService.cs

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using CanKit.Abstractions.API.Can;
using CanKit.Abstractions.API.Can.Definitions;
using CanKit.Abstractions.API.Common.Definitions;

namespace CanKit.Pro.RawCan
{
    /// <summary>
    /// Default <see cref="ICanBusService"/>: attaches once to <see cref="ICanBus.FrameObserved"/>
    /// and fans each observed <see cref="CanFrameView"/> out to every registered
    /// <see cref="Subscription"/> (arc42 §5.3 "Multi-Protokoll-Demux", ADR-5), and implements
    /// <see cref="SendConfirmed"/> (arc42 §6.3, ADR-7; FR-RAW-030..034) on the same single
    /// <see cref="ICanBus.FrameObserved"/> subscription.
    /// </summary>
    public sealed class CanBusService : ICanBusService
    {
        /// <summary>
        /// Default per-subscription bounded buffer capacity when none is specified.
        /// </summary>
        public const int DefaultBufferCapacity = 1024;

        /// <summary>
        /// Default <see cref="SendConfirmed"/> echo-wait timeout when none is specified
        /// (FR-RAW-034).
        /// </summary>
        public static readonly TimeSpan DefaultConfirmTimeout = TimeSpan.FromSeconds(1);

        private readonly ICanBus _bus;

        // Guards the mutable registry and the rebuild of _snapshot; only entered on
        // subscribe/dispose (setup/teardown), never on the per-frame dispatch path. Mirrors the
        // registry-lock discipline of VirtualBusHub.Join/Detach.
        private readonly object _gate = new();
        private readonly List<Subscription> _subscriptions = new();

        // Copy-on-write snapshot read lock-free by OnFrameObserved, so the dispatch hot path takes
        // no lock and allocates nothing per frame — same reasoning as VirtualBusHub.Broadcast not
        // holding _hubsGate while delivering.
        private volatile Subscription[] _snapshot = Array.Empty<Subscription>();

        // Pending SendConfirmed calls awaiting an echo match, keyed by everything that identifies
        // the frame on the wire (see PendingKey) so multiple concurrent identical sends are matched
        // FIFO instead of crashing/cross-matching, and two sends that merely *look* alike -- a
        // standard and an extended 0x100 with the same payload -- do not share one FIFO at all
        // (FR-RAW-031). Guarded by its own lock, separate from _gate, so TX-confirm churn never
        // contends with subscription registry churn (and vice versa).
        private readonly object _pendingGate = new();
        private readonly Dictionary<PendingKey, LinkedList<PendingSend>> _pending = new();

        // Guarded by _pendingGate; set once by Dispose so a racing SendConfirmed call can never
        // register a pending entry after Dispose's final sweep has already canceled everything.
        private bool _pendingDisposed;

        // Cheap lock-free fast path: skip the _pendingGate lock (and the PendingKey hashing) in
        // OnFrameObserved entirely for services where nobody has ever called SendConfirmed. Same
        // "no cost for callers who don't use the feature" discipline as the subscription snapshot.
        private int _pendingCount;

        private int _disposed;

        // Host arrival and transmit stamps. Production reads Stopwatch; a test passes the same
        // counter the protocol window is measured on, so a frame and the deadline that admits it
        // are not two clocks (#171).
        private readonly Func<long> _hostTimestamp;

        /// <summary>
        /// Creates a service that demultiplexes <paramref name="bus"/>. Attaches to the bus's
        /// <see cref="ICanBus.FrameObserved"/> event immediately.
        /// </summary>
        public CanBusService(ICanBus bus)
            : this(bus, hostTimestamp: null)
        {
        }

        /// <summary>
        /// As the public constructor, but <paramref name="hostTimestamp"/> supplies the host
        /// arrival and transmit stamps. Null keeps <see cref="Stopwatch.GetTimestamp"/>.
        /// </summary>
        /// <remarks>
        /// Internal on purpose. The stamps are facts about one clock, and the only caller that
        /// needs to choose which clock is a test driving a functional window by hand (#171).
        /// </remarks>
        internal CanBusService(ICanBus bus, Func<long>? hostTimestamp)
        {
            _bus = bus ?? throw new ArgumentNullException(nameof(bus));
            _hostTimestamp = hostTimestamp ?? Stopwatch.GetTimestamp;
            _bus.FrameObserved += OnFrameObserved;
            _bus.FaultOccurred += OnFaultOccurred;
        }

        /// <inheritdoc />
        public ICanBus Bus => _bus;

        /// <inheritdoc />
        public int SubscriptionCount
        {
            get
            {
                lock (_gate)
                {
                    return _subscriptions.Count;
                }
            }
        }

        /// <inheritdoc />
        public ISubscription Subscribe(Func<CanFrameEvent, bool>? predicate = null, int? bufferCapacity = null, bool includeEcho = false)
            => AddSubscription(idFilter: null, predicate: predicate, bufferCapacity, includeEcho);

        /// <inheritdoc />
        public ISubscription Subscribe(CanIdFilter filter, int? bufferCapacity = null, bool includeEcho = false)
            => AddSubscription(idFilter: filter, predicate: null, bufferCapacity, includeEcho);

        /// <inheritdoc />
        public IReadOnlyList<FilterOverlap> FindOverlappingFilterSubscriptions()
        {
            // Snapshot read, same lock-free discipline as the dispatch hot path -- this is a
            // diagnostic call, not something exercised per-frame, but there's no reason to take
            // _gate for a read when the existing snapshot already gives a consistent view.
            var subscriptions = _snapshot;
            var overlaps = new List<FilterOverlap>();

            for (var i = 0; i < subscriptions.Length; i++)
            {
                if (subscriptions[i].IsDisposed || subscriptions[i].IdFilter is not { } filterI) continue;
                for (var j = i + 1; j < subscriptions.Length; j++)
                {
                    if (subscriptions[j].IsDisposed || subscriptions[j].IdFilter is not { } filterJ) continue;
                    if (filterI.TryGetSharedIdRange(filterJ, out var lowest, out var highest))
                        overlaps.Add(new FilterOverlap(subscriptions[i], subscriptions[j], lowest, highest));
                }
            }

            return overlaps;
        }

        private ISubscription AddSubscription(CanIdFilter? idFilter, Func<CanFrameEvent, bool>? predicate, int? bufferCapacity, bool includeEcho)
        {
            var capacity = bufferCapacity ?? DefaultBufferCapacity;
            if (capacity <= 0)
                throw new ArgumentOutOfRangeException(nameof(bufferCapacity), "Buffer capacity must be positive.");

            var subscription = new Subscription(this, idFilter, predicate, capacity, includeEcho);
            lock (_gate)
            {
                if (_disposed != 0)
                    throw new ObjectDisposedException(nameof(CanBusService));

                _subscriptions.Add(subscription);
                _snapshot = _subscriptions.ToArray();
            }

            return subscription;
        }

        /// <summary>
        /// Deregisters <paramref name="subscription"/> so it stops receiving frames. Called from
        /// <see cref="Subscription.Dispose"/>. Held under <see cref="_gate"/> together with
        /// <see cref="AddSubscription"/>, mirroring VirtualBusHub.Detach.
        /// </summary>
        internal void Remove(Subscription subscription)
        {
            lock (_gate)
            {
                if (_subscriptions.Remove(subscription))
                    _snapshot = _subscriptions.ToArray();
            }
        }

        private void OnFrameObserved(object? sender, CanReceiveDataView e)
        {
            // Taken once per frame, here, and not per subscription: every matching subscriber must
            // agree on when the frame arrived. Taken on the very first line for the same reason --
            // everything below it can block. TryMatchEcho takes _pendingGate, which
            // SendWithEchoConfirmAsync deliberately holds across _bus.Transmit (#102), and the
            // per-subscription buffers below can hold a frame while their reader is descheduled.
            // Either would make a punctual frame look late to whoever is enforcing a deadline on
            // it (Codex on #112).
            var hostArrival = _hostTimestamp();

            // Independent of subscription dispatch below: echo frames must be checked against
            // outstanding SendConfirmed calls regardless of whether anyone also has a
            // subscription open. Guarded by the same lock-free fast path as subscriptions.
            if (e.IsEcho && Volatile.Read(ref _pendingCount) > 0)
                TryMatchEcho(e.CanFrame);

            var subscriptions = _snapshot; // volatile read; no lock, no per-frame allocation
            if (subscriptions.Length == 0) return;

            // All three facts travel together from here on. The demux used to forward only
            // e.CanFrame and drop e.IsEcho and e.ReceiveTimestamp on the floor, so no subscriber
            // could see either, however much it wanted to (#23). Reading them costs nothing: the
            // event argument already carries them.
            //
            // This does not make the protocol layers' own self-traffic checks redundant -- the
            // flag is host-scoped and not every adapter sets it. See the remarks on
            // ICanBusService for why they are retained.
            var view = e.CanFrame;
            var isEcho = e.IsEcho;
            var receiveTimestamp = e.ReceiveTimestamp;

            // One owned payload copy per *frame*, created by the first subscription that actually
            // buffers it and reused by every later one, instead of one copy per matching
            // subscription. The copy exists because the view aliases the adapter's RX lease (see
            // Subscription.TryDeliver); nothing about that reason is per-subscriber, and what the
            // subscribers get handed is a ReadOnlyMemory they may only read. A frame nobody
            // matches still allocates nothing at all.
            byte[]? ownedPayload = null;

            foreach (var subscription in subscriptions)
            {
                // A subscription's filter predicate is caller-supplied and may throw. Isolate each
                // delivery so one broken predicate can never suppress delivery to the *other*
                // subscriptions for this frame, nor escape into the bus's FrameObserved multicast
                // (which would abort dispatch to every subscription still pending in this loop) —
                // that would violate the independence every subscription is guaranteed under
                // FR-RAW-010. The fault is surfaced through BackgroundExceptionOccurred instead
                // of being silently swallowed.
                try
                {
                    subscription.TryDeliver(view, isEcho, receiveTimestamp, hostArrival,
                        ref ownedPayload);
                }
                catch (Exception ex)
                {
                    RaiseBackgroundException(ex);
                }
            }
        }

        /// <inheritdoc />
        public event EventHandler<Exception>? BackgroundExceptionOccurred;

        /// <summary>
        /// Routes <paramref name="ex"/> through <see cref="BackgroundExceptionOccurred"/>,
        /// isolating a misbehaving listener from the caller. Internal (not part of
        /// <see cref="ICanBusService"/>) because events can only be raised from their declaring
        /// type; exposed so <see cref="CanBusServiceExtensions.Subscribe"/> can report a failing
        /// callback handler through this same fault channel instead of a second, parallel one.
        /// </summary>
        internal void RaiseBackgroundException(Exception ex)
        {
            try { BackgroundExceptionOccurred?.Invoke(this, ex); }
            catch { /* a fault listener must not break dispatch either */ }
        }

        /// <inheritdoc />
        public void Dispose()
        {
            if (Interlocked.Exchange(ref _disposed, 1) != 0) return; // idempotent

            // Detach first so no further frames are dispatched into subscriptions we're tearing
            // down (no leaked FrameObserved handler — the exact class of leak the ownership PR
            // fixed for VirtualBusHub._hubs, here for the subscription registry).
            _bus.FrameObserved -= OnFrameObserved;
            _bus.FaultOccurred -= OnFaultOccurred;

            Subscription[] outstanding;
            lock (_gate)
            {
                outstanding = _subscriptions.ToArray();
                _subscriptions.Clear();
                _snapshot = Array.Empty<Subscription>();
            }

            // Complete each channel outside the lock (CompleteFromService does not re-enter the
            // registry), so a slow subscriber can never turn teardown into a lock convoy.
            foreach (var subscription in outstanding)
                subscription.CompleteFromService();

            // Cancel every outstanding SendConfirmed call rather than leaving it to time out on
            // its own -- otherwise disposing the service while sends are in flight would make
            // their tasks hang until each one's individual timeout, not "no leaked resources"
            // (same reasoning as unwinding subscriptions above; standard .NET convention is that
            // disposing an in-flight operation's owner cancels it, hence TrySetCanceled rather
            // than a TxConfirmation result -- there's no SRS-defined FailureReason for "disposed").
            PendingSend[] pending;
            lock (_pendingGate)
            {
                // Set before clearing, under the same lock SendWithEchoConfirmAsync checks before
                // registering: closes the race where a call passes SendConfirmed's eager disposed
                // check but hasn't registered yet -- it now either registers-and-transmits fully
                // before this line runs (and gets swept up below like any other pending entry), or
                // sees _pendingDisposed=true and throws ObjectDisposedException instead of silently
                // leaving an orphaned entry that would otherwise sit unmatched until its own timeout.
                _pendingDisposed = true;
                pending = _pending.Values.SelectMany(list => list).ToArray();
                // Null out Node before clearing: SendWithEchoConfirmAsync's `finally` still calls
                // RemovePending once its WaitForPendingAsync unblocks below, and RemovePending's
                // "already removed" no-op check (Node == null) is what stops it from decrementing
                // _pendingCount a second time for the entries we're resolving right here.
                foreach (var p in pending)
                    p.Node = null;
                _pending.Clear();
                Volatile.Write(ref _pendingCount, 0);
            }
            foreach (var p in pending)
                p.Tcs.TrySetCanceled();
        }

        /// <inheritdoc />
        public async Task<TxConfirmation> SendConfirmed(CanFrame frame, TimeSpan? timeout = null, CancellationToken cancellationToken = default)
        {
            if (Volatile.Read(ref _disposed) != 0)
                throw new ObjectDisposedException(nameof(CanBusService));

            var effectiveTimeout = timeout ?? DefaultConfirmTimeout;
            if (effectiveTimeout <= TimeSpan.Zero)
                throw new ArgumentOutOfRangeException(nameof(timeout), "Timeout must be positive.");

            // FR-RAW-030: one API, two internal strategies, chosen by whether the bus both
            // declares the hardware capability *and* has actually enabled it for this session --
            // CanFeature.Echo alone only means "this adapter type is capable of it", exactly like
            // every other CanFeature flag; WorkMode == Echo is the existing cross-adapter opt-in
            // that turns real echo delivery on for a given bus (see VirtualBusHub.Broadcast).
            var useEcho = _bus.Options.Features.HasFlag(CanFeature.Echo)
                          && _bus.Options.WorkMode == ChannelWorkMode.Echo;

            return useEcho
                ? await SendWithEchoConfirmAsync(frame, effectiveTimeout, cancellationToken).ConfigureAwait(false)
                : await SendApproximatedAsync(frame, cancellationToken).ConfigureAwait(false);
        }

        private async Task<TxConfirmation> SendApproximatedAsync(CanFrame frame, CancellationToken cancellationToken)
        {
            // FR-RAW-032: best-effort approximation -- confirmed as soon as the driver accepts the
            // frame, explicitly marked IsApproximated so callers can never mistake this for a real
            // hardware acknowledgment.
            //
            // The hand-off instant is taken by a continuation on the thread that completes the
            // driver's task, not after awaiting it. Resuming this method is a scheduling event:
            // for an adapter whose TransmitAsync completes asynchronously, the driver has
            // accepted the frame -- and the peer may already be answering -- before this method
            // runs again, and a reading taken there starts a caller's response deadline late
            // (Codex on #112). ExecuteSynchronously is what observes completion closest; when the
            // runtime declines to inline it the reading is what it would have been anyway.
            var stamp = new HandoffStamp(_hostTimestamp);
            var handoffStart = stamp.Now();
            var accepted = await _bus.TransmitAsync(frame, cancellationToken)
                .ContinueWith(
                    static (completed, state) =>
                    {
                        ((HandoffStamp)state!).Value = ((HandoffStamp)state!).Now();

                        // GetResult rather than .Result: it surfaces a driver fault or a
                        // cancellation as itself instead of wrapping it in an AggregateException,
                        // so this continuation is invisible to callers apart from the stamp.
                        return completed.GetAwaiter().GetResult();
                    },
                    stamp,
                    CancellationToken.None,
                    TaskContinuationOptions.ExecuteSynchronously,
                    TaskScheduler.Default)
                .ConfigureAwait(false);
            var handoff = stamp.Value;
            return accepted > 0
                ? new TxConfirmation { Confirmed = true, IsApproximated = true, Timestamp = DateTime.UtcNow, FailureReason = TxConfirmFailureReason.None, HostTransmitTimestamp = handoff, HostHandoffTimestamp = handoffStart }
                : new TxConfirmation { Confirmed = false, IsApproximated = false, Timestamp = DateTime.UtcNow, FailureReason = TxConfirmFailureReason.Rejected };
        }

        private async Task<TxConfirmation> SendWithEchoConfirmAsync(CanFrame frame, TimeSpan timeout, CancellationToken cancellationToken)
        {
            var pending = new PendingSend(PendingKey.ForPendingSend(frame.ID, frame.Data, frame.Flags, frame.FrameKind));

            int accepted;
            long handoff = 0, handoffStart = 0;
            try
            {
                // Register and transmit as one atomic step under _pendingGate: this is what makes
                // TryMatchEcho's FIFO order equal actual transmission order rather than mere
                // registration order. Without it, two threads sending byte-identical frames could
                // register in one order but transmit in the other, so the oldest *pending* entry
                // is not necessarily the oldest *sent* one -- an echo could then confirm the wrong
                // caller, or confirm a send before it was even transmitted (FR-RAW-031). A
                // synchronous echo delivered inside Transmit itself (e.g. Virtual's
                // WorkMode=Echo) re-enters this same lock on this same thread -- Monitor is
                // reentrant, so TryMatchEcho can only ever see this thread's own just-registered
                // entry at that point, never a different, unrelated in-flight one. This also
                // closes the dispose race: Dispose sets _pendingDisposed under this same lock, so
                // a call can never register after Dispose has already swept and canceled every
                // pending entry -- it throws ObjectDisposedException instead, matching the eager
                // check at the top of SendConfirmed. The lock is held only across the register +
                // enqueue step (Transmit is expected to be a fast, non-blocking enqueue, same
                // assumption every other caller of ICanBus.Transmit already makes), never across
                // the echo wait, so unrelated sends are not serialized against each other.
                //
                // A review finding (#53) asked for Transmit to move out of this lock, so that a
                // blocking vendor driver cannot stall the adapter's RX thread in TryMatchEcho. It
                // is deliberately not done: the atomicity above is the whole reason the FIFO order
                // means anything, an echo-mode adapter re-enters this lock from inside Transmit
                // anyway, and no driver that blocks in Transmit could be used with this service in
                // any case -- the same call is on the dispatch path of every other consumer of
                // ICanBus. If one ever has to be, the fix is a queue in front of the driver, not a
                // pending list whose order no longer matches the wire.
                lock (_pendingGate)
                {
                    if (_pendingDisposed)
                        throw new ObjectDisposedException(nameof(CanBusService));

                    RegisterPending(pending);
                    // The other end of the driver call, inside the lock: a caller's cutoff for
                    // what can still be a response to this frame (Codex on #147).
                    handoffStart = _hostTimestamp();
                    accepted = _bus.Transmit(in frame);

                    // Taken here, inside the lock and immediately after the driver call returns:
                    // this is the closest observable instant to the frame reaching the wire.
                    // Anything earlier is before the frame was handed over -- including the wait
                    // for this very lock, which another send holds across its own Transmit -- and
                    // would start a caller's response deadline while the request was still
                    // queued behind it (Codex on #112).
                    handoff = _hostTimestamp();
                }
            }
            catch
            {
                RemovePending(pending);
                throw;
            }

            if (accepted <= 0)
            {
                RemovePending(pending);
                return new TxConfirmation { Confirmed = false, IsApproximated = false, Timestamp = DateTime.UtcNow, FailureReason = TxConfirmFailureReason.Rejected };
            }

            try
            {
                var confirmation = await WaitForPendingAsync(pending, timeout, cancellationToken)
                    .ConfigureAwait(false);
                return confirmation with { HostTransmitTimestamp = handoff, HostHandoffTimestamp = handoffStart };
            }
            finally
            {
                // No-op if TryMatchEcho/OnFaultOccurred/Dispose already removed it; defensive
                // cleanup for the timeout/cancellation paths, which don't remove it themselves.
                RemovePending(pending);
            }
        }

        private async Task<TxConfirmation> WaitForPendingAsync(PendingSend pending, TimeSpan timeout, CancellationToken cancellationToken)
        {
            using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
            // Registration fires on whichever comes first: caller cancellation or our own timeout.
            using var registration = timeoutCts.Token.Register(static state =>
            {
                var (p, ct) = ((PendingSend, CancellationToken))state!;

                // Complete the Tcs *without touching _pendingGate*. An earlier revision of the #24
                // fix unlinked here first, reasoning that an entry must stop being matchable the
                // instant it is resolved. It does -- but taking the lock to achieve that runs on
                // whichever thread trips the token, and SendWithEchoConfirmAsync holds that lock
                // across _bus.Transmit. A caller's own CancellationTokenSource.Cancel() therefore
                // blocked until an unrelated send's driver call returned. Cancelling one send is
                // not something that should wait on another send's adapter.
                //
                // Unlinking here is also unnecessary. What makes the still-linked expired entry
                // harmless is TryMatchEcho skipping (and unlinking) entries whose Tcs is already
                // completed -- see the loop there, which is the mechanism that fixes #24. An echo
                // arriving in the window before the `finally` unlinks walks past this entry to the
                // live one behind it instead of being swallowed.
                //
                // What this does *not* change: the caller's SendConfirmed task still completes only
                // after that `finally`, and the unlink there does take the lock. That coupling is
                // older than this fix and follows from holding _pendingGate across Transmit at all
                // -- see the note on that lock, and #53.
                if (ct.IsCancellationRequested)
                {
                    p.Tcs.TrySetCanceled(ct);
                }
                else
                {
                    p.Tcs.TrySetResult(new TxConfirmation
                    {
                        Confirmed = false,
                        IsApproximated = false,
                        Timestamp = DateTime.UtcNow,
                        FailureReason = TxConfirmFailureReason.Timeout,
                    });
                }
            }, (pending, cancellationToken));

            timeoutCts.CancelAfter(timeout);
            return await pending.Tcs.Task.ConfigureAwait(false);
        }

        private void RegisterPending(PendingSend pending)
        {
            lock (_pendingGate)
            {
                if (!_pending.TryGetValue(pending.Key, out var list))
                {
                    list = new LinkedList<PendingSend>();
                    _pending[pending.Key] = list;
                }
                pending.Node = list.AddLast(pending);
                Interlocked.Increment(ref _pendingCount);
            }
        }

        private void RemovePending(PendingSend pending)
        {
            lock (_pendingGate)
            {
                var node = pending.Node;
                if (node is null) return; // already removed by a match, fault, or dispose

                var list = node.List;
                list?.Remove(node);
                pending.Node = null;
                Interlocked.Decrement(ref _pendingCount);

                if (list is { Count: 0 })
                    _pending.Remove(pending.Key);
            }
        }

        private void TryMatchEcho(in CanFrameView echoView)
        {
            // Aliases the echo frame's payload rather than copying it: this runs for every echo
            // frame the adapter reports, and the key is dropped again before the lock is released.
            var key = PendingKey.ForEchoLookup(echoView.ID, echoView.Data, echoView.Flags, echoView.FrameKind);

            lock (_pendingGate)
            {
                if (_pending.TryGetValue(key, out var list))
                {
                    var confirmation = new TxConfirmation
                    {
                        Confirmed = true,
                        IsApproximated = false,
                        Timestamp = DateTime.UtcNow,
                        FailureReason = TxConfirmFailureReason.None,
                    };

                    // FIFO: the oldest pending send for this key gets the echo -- but only if it
                    // is still waiting. An entry already resolved by some other path can no longer
                    // consume anything, and handing it the echo would drop it silently while the
                    // send it actually belonged to waits for one that has already come and gone.
                    //
                    // Walking past such entries is the mechanism that fixes #24, not a redundant
                    // guard. The timeout and cancellation path completes its Tcs without taking
                    // this lock -- deliberately, so a deadline cannot be held up by an unrelated
                    // send sitting in a slow _bus.Transmit -- and the entry it resolved stays
                    // linked until SendWithEchoConfirmAsync's `finally` runs a scheduling turn
                    // later. Inside that window the expired entry is still the FIFO head for its
                    // key, and without this walk it would swallow the next byte-identical send's
                    // echo, turning one timeout into a cascade of them.
                    //
                    // Unlinking what it skips matters as much as skipping it: otherwise the same
                    // dead entry would block the FIFO for every later echo rather than only this
                    // one.
                    for (var node = list.First; node is not null;)
                    {
                        var next = node.Next;
                        var candidate = node.Value;

                        list.Remove(node);
                        candidate.Node = null;
                        Interlocked.Decrement(ref _pendingCount);

                        // Claim by completing, not by asking first. An `IsCompleted` test followed
                        // by a TrySetResult is check-then-act: the timeout and cancellation path
                        // completes without this lock, so it can land between the two, and then
                        // the echo is consumed by an entry that lost the race while a live send
                        // behind it in the FIFO waits for an echo that has already arrived. That
                        // is the same defect as #24 wearing different clothes.
                        //
                        // TrySetResult is the only test that cannot be raced, because it *is* the
                        // transition. A false return means some other path got there first, so
                        // this candidate never owned the echo and the walk continues to the next.
                        //
                        // Safe under the lock precisely because the Tcs is created with
                        // RunContinuationsAsynchronously: completing it queues the awaiting
                        // continuation rather than running it inline, so no caller code executes
                        // while _pendingGate is held.
                        if (candidate.Tcs.TrySetResult(confirmation)) break;

                        node = next;
                    }

                    if (list.Count == 0)
                        _pending.Remove(key);
                }
            }
        }

        private void OnFaultOccurred(object? sender, Exception ex)
        {
            // Scoped strictly to FR-RAW-033's named "Bus-Off" failure mode: FaultOccurred fires for
            // other fault severities too (see CanBusExceptionDispatcher), which aren't necessarily
            // a reason to fail every outstanding confirmation -- BusState is the authoritative
            // signal for whether this specific fault means the bus actually went off.
            if (_bus.BusState != BusState.BusOff) return;

            PendingSend[] pending;
            lock (_pendingGate)
            {
                pending = _pending.Values.SelectMany(list => list).ToArray();
                // See the identical comment in Dispose(): null Node before clearing so the
                // pending SendWithEchoConfirmAsync calls' own `finally`-triggered RemovePending
                // no-ops instead of double-decrementing _pendingCount.
                foreach (var p in pending)
                    p.Node = null;
                _pending.Clear();
                Volatile.Write(ref _pendingCount, 0);
            }

            foreach (var p in pending)
            {
                p.Tcs.TrySetResult(new TxConfirmation
                {
                    Confirmed = false,
                    IsApproximated = false,
                    Timestamp = DateTime.UtcNow,
                    FailureReason = TxConfirmFailureReason.BusOff,
                });
            }
        }

        /// <summary>
        /// The stamp taken on the thread that completes the driver's task. A static continuation
        /// cannot close over the service, so the delegate travels with the box.
        /// </summary>
        private sealed class HandoffStamp
        {
            private readonly Func<long> _now;

            public HandoffStamp(Func<long> now) => _now = now;

            public long Value;

            public long Now() => _now();
        }
    }
}