indexbus_route/internal/

router.rs

use core::hint::spin_loop;
use core::time::Duration;

use core::sync::atomic::{AtomicU64, Ordering};

use indexbus_abi::IndexbusAtomicU64;

use indexbus_abi::layouts::{IndexQueue, MpscQueue, SharedFanoutLayout};
use indexbus_abi::INDEXBUS_QUEUE_CAPACITY;
use indexbus_blocking::fanout_router_ref_blocking;
use indexbus_core::{FanoutRouter, RouterMode, RouterSource, WaitStrategy};

/// Backpressure policy to apply when routing to consumers.
///
/// ## Contract
///
/// - Backpressure policies only affect the router loop (this crate). Producers and consumers are
///   still nonblocking and may observe drops depending on mode and policy.
/// - In v1, only [`RouterMode::WorkQueue`] can apply backpressure without changing the ABI layout.
///   [`RouterMode::Broadcast`] remains best-effort and uses drop-on-full semantics.
///
/// ## Notes
///
/// - [`BackpressurePolicy::SpinThenBlock`] is a CPU-side wait strategy; it does not use wake
///   sections.
/// - [`BackpressurePolicy::Block`] uses optional v1 wake sections when available; if wake-backed
///   waits fail at runtime, the loop falls back to the provided [`WaitStrategy`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BackpressurePolicy {
    /// Drop when a destination queue is full.
    Drop,

    /// When routing in `RouterMode::WorkQueue`, avoid dequeuing a message if all consumer queues
    /// are full; instead, apply the configured wait strategy until at least one consumer has
    /// capacity.
    ///
    /// Notes:
    /// - This is a v1 best-effort policy; it does not use wake sections.
    /// - In `RouterMode::Broadcast`, this policy behaves like `Drop` (broadcast cannot be made
    ///   "all-or-nothing" without a different layout).
    SpinThenBlock,

    /// Use the optional v1 wake sections (`INDEXBUS_CAP_SUPPORTS_BLOCKING`) to block the router
    /// thread when idle or when all consumers are full.
    ///
    /// Notes:
    /// - Implemented only for `RouterMode::WorkQueue`.
    /// - If wake sections are unavailable, this falls back to the configured wait strategy.
    Block,
}

/// Credit policy (router-enforced credits).
///
/// Credits are an overload-control mechanism, not a reliability mechanism.
///
/// ## Contract
///
/// - Credit enforcement is implemented in the router loop; it does not change the underlying
///   ABI layout.
/// - Credits are based on an estimated per-consumer queue depth and are best-effort.
/// - Credit policies are primarily intended for [`RouterMode::WorkQueue`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CreditPolicy {
    /// Drop when a message cannot be delivered due to credit exhaustion.
    Drop,

    /// Temporarily detach consumers that remain over the credit limit for long enough.
    ///
    /// Detached consumers are not considered for delivery until they catch up.
    ///
    /// Requires the `std` feature to measure elapsed time for the detach threshold. Without
    /// `std`, this behaves like [`CreditPolicy::Park`] (no detaches will occur).
    Detach,

    /// When no consumers are eligible (work-queue only), wait instead of dropping.
    Park,
}

/// Credit configuration (router-enforced credits).
///
/// The credit limit is an approximate in-flight depth bound per consumer. It is computed as a
/// `tail - head` distance in the destination queue and may be stale.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct CreditConfig {
    /// Maximum in-flight depth allowed per consumer.
    ///
    /// Depth is estimated as `tail - head` of the consumer queue.
    ///
    /// Values of `0` are treated as `1`.
    pub credit_max: u32,

    /// Credit enforcement behavior when consumers exceed `credit_max`.
    pub policy: CreditPolicy,

    /// Optional detach threshold. Only meaningful with `CreditPolicy::Detach`.
    ///
    /// If set, consumers that remain over `credit_max` for longer than this duration are
    /// detached (temporarily ignored).
    ///
    /// Only available with the `std` feature.
    pub detach_after_ms: Option<u32>,
}

/// Configuration for a routing loop.
///
/// ## Batching
///
/// The loop routes messages sequentially. Batching repeats single-message routing up to
/// [`batch_max`](RouterLoopConfig::batch_max) times per iteration; it does not reorder messages.
///
/// ## Feature notes
///
/// - [`batch_time_us`](RouterLoopConfig::batch_time_us) and
///   [`yield_every`](RouterLoopConfig::yield_every) are only effective with the `std` feature.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RouterLoopConfig {
    /// Producer queue to drain from (`Spsc` or `Mpsc`).
    pub source: RouterSource,
    /// Routing mode (`Broadcast` or `WorkQueue`).
    pub mode: RouterMode,
    /// Backpressure behavior when routing cannot make progress.
    pub policy: BackpressurePolicy,

    /// Optional router-enforced credit enforcement.
    pub credit: Option<CreditConfig>,

    /// Maximum number of messages to route per loop iteration.
    ///
    /// Notes:
    /// - `0` is treated as `1`.
    /// - Batching never reorders messages; it just repeats `route_once_with_stats`.
    pub batch_max: u32,

    /// Optional time cap for a single batch.
    ///
    /// This is best-effort and only enforced when the `std` feature is enabled.
    pub batch_time_us: Option<u32>,

    /// Fairness knob: yield the current thread every N loop iterations.
    ///
    /// Only effective with the `std` feature; otherwise ignored.
    pub yield_every: u32,

    /// When idle, spin up to this many times before invoking the wait strategy.
    ///
    /// This provides a bounded low-latency polling window without requiring a custom
    /// `WaitStrategy` implementation.
    pub idle_spin_limit: u32,
}

impl Default for RouterLoopConfig {
    fn default() -> Self {
        Self {
            source: RouterSource::Spsc,
            mode: RouterMode::Broadcast,
            policy: BackpressurePolicy::Drop,

            credit: None,

            batch_max: 1,
            batch_time_us: None,
            yield_every: 0,
            idle_spin_limit: 0,
        }
    }
}

/// Counters returned by routing loops.
///
/// These counters are best-effort observability signals. In particular, some drop attribution is
/// only approximate for work-queue routing because the router may fail to find any eligible
/// destination.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct RouterCounters {
    /// Number of messages routed (i.e. consumed from the producer queue).
    pub routed: u64,
    /// Total number of consumer enqueues performed.
    pub delivered: u64,
    /// Total number of consumer enqueue failures.
    pub dropped: u64,

    /// Drops attributable to destination queue being full (best-effort, v1 fanout).
    pub dropped_queue_full: u64,

    /// Drops attributable to no eligible consumer having space (work-queue: all full).
    pub dropped_all_full: u64,

    /// Drops attributable to no eligible consumer having credit (best-effort).
    pub dropped_no_credit: u64,

    /// Number of times routing was throttled due to credit exhaustion.
    pub credit_waits: u64,

    /// Number of consumer detach events.
    pub detaches: u64,
    /// Number of times routing was throttled because consumer queues had no capacity.
    pub pressure_waits: u64,
    /// Number of times the loop observed "no work" and invoked the wait strategy.
    pub idle_waits: u64,

    /// Number of batches routed.
    pub batches: u64,

    /// Sum of batch sizes (messages per batch).
    pub batch_sum: u64,

    /// Maximum batch size observed.
    pub batch_max: u32,

    /// Number of OS-backed wake waits performed (when `BackpressurePolicy::Block` is active).
    pub wake_waits: u64,

    /// Number of OS-backed wake waits that timed out.
    pub wake_timeouts: u64,
}

/// Per-consumer credit stats (router-enforced credits).
///
/// Notes:
/// - For `RouterMode::WorkQueue`, `dropped_full` is not attributable to a single consumer when
///   the router fails to find any eligible destination; in that case per-consumer drops are
///   best-effort.
/// - If `N > 64`, per-consumer masks are not computed and attribution is disabled.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CreditStats<const N: usize> {
    /// Per-consumer count of delivered messages.
    pub delivered: [u64; N],
    /// Per-consumer drops attributable to credit exhaustion.
    pub dropped_no_credit: [u64; N],
    /// Per-consumer drops attributable to destination queues being full.
    pub dropped_full: [u64; N],
    /// Per-consumer detached state (for `CreditPolicy::Detach`).
    pub detached: [bool; N],
    /// Per-consumer number of detach events.
    pub detach_count: [u64; N],
    /// Per-consumer estimated queue depth (`tail - head`).
    ///
    /// This is sampled from shared state and may be stale.
    pub depth: [u64; N],

    #[cfg(feature = "std")]
    #[doc(hidden)]
    pub(crate) exhausted_since: [Option<std::time::Instant>; N],
}

impl<const N: usize> Default for CreditStats<N> {
    fn default() -> Self {
        Self {
            delivered: [0; N],
            dropped_no_credit: [0; N],
            dropped_full: [0; N],
            detached: [false; N],
            detach_count: [0; N],
            depth: [0; N],

            #[cfg(feature = "std")]
            exhausted_since: [None; N],
        }
    }
}

const PRESSURE_WAIT_TIMEOUT: Duration = Duration::from_millis(1);
const IDLE_WAIT_TIMEOUT: Duration = Duration::from_millis(10);

#[inline]
fn load_u64(addr: *const IndexbusAtomicU64) -> u64 {
    unsafe { (&*(addr as *const AtomicU64)).load(Ordering::Acquire) }
}

#[inline]
fn index_queue_depth(q: &IndexQueue) -> u64 {
    let head = load_u64(core::ptr::addr_of!(q.head));
    let tail = load_u64(core::ptr::addr_of!(q.tail));
    tail.wrapping_sub(head)
}

#[inline]
fn mpsc_queue_depth(q: &MpscQueue) -> u64 {
    let w = load_u64(core::ptr::addr_of!(q.write));
    let r = load_u64(core::ptr::addr_of!(q.read));
    w.wrapping_sub(r)
}

#[inline]
fn has_pending_message<const N: usize>(
    shared: *const SharedFanoutLayout<N>,
    source: RouterSource,
) -> bool {
    let s = unsafe { &*shared };
    match source {
        RouterSource::Spsc => index_queue_depth(&s.producer_queue) != 0,
        RouterSource::Mpsc => mpsc_queue_depth(&s.producer_queue_mpsc) != 0,
    }
}

#[inline]
fn any_consumer_has_space<const N: usize>(shared: *const SharedFanoutLayout<N>) -> bool {
    if N == 0 {
        return true;
    }

    let s = unsafe { &*shared };
    let cap = INDEXBUS_QUEUE_CAPACITY as u64;
    for q in &s.consumer_queues {
        if index_queue_depth(q) < cap {
            return true;
        }
    }
    false
}

#[inline]
fn consumer_depths<const N: usize>(shared: *const SharedFanoutLayout<N>) -> [u64; N] {
    let s = unsafe { &*shared };
    let mut out = [0u64; N];
    for (i, q) in s.consumer_queues.iter().enumerate() {
        out[i] = index_queue_depth(q);
    }
    out
}

#[inline]
fn compute_credit_masks<const N: usize>(
    shared: *const SharedFanoutLayout<N>,
    credit_max: u32,
    detached: &[bool; N],
) -> (u64, u64, u64, [u64; N]) {
    let depths = consumer_depths(shared);
    if N == 0 {
        return (0, 0, 0, depths);
    }

    let cap = INDEXBUS_QUEUE_CAPACITY as u64;
    let mut eligible_mask: u64 = 0;
    let mut no_credit_mask: u64 = 0;
    let mut full_mask: u64 = 0;

    if N > 64 {
        // Best-effort: no masks for N>64.
        return (0, 0, 0, depths);
    }

    let credit_max = credit_max as u64;
    for (i, &is_detached) in detached.iter().enumerate() {
        if is_detached {
            continue;
        }
        let d = depths[i];
        let bit = 1u64 << (i as u32);
        if d >= cap {
            full_mask |= bit;
        } else if d >= credit_max {
            no_credit_mask |= bit;
        } else {
            eligible_mask |= bit;
        }
    }

    (eligible_mask, no_credit_mask, full_mask, depths)
}

#[inline]
fn any_eligible_consumer_has_space_and_credit<const N: usize>(
    shared: *const SharedFanoutLayout<N>,
    credit_max: u32,
    detached: &[bool; N],
) -> bool {
    if N == 0 {
        return true;
    }
    let cap = INDEXBUS_QUEUE_CAPACITY as u64;
    let credit_max = credit_max as u64;

    let s = unsafe { &*shared };
    for (i, &is_detached) in detached.iter().enumerate() {
        if is_detached {
            continue;
        }
        let d = index_queue_depth(&s.consumer_queues[i]);
        if d < cap && d < credit_max {
            return true;
        }
    }
    false
}

#[inline]
fn apply_masks_to_stats<const N: usize>(
    stats: &mut CreditStats<N>,
    delivered_mask: u64,
    dropped_full_mask: u64,
    dropped_no_credit_mask: u64,
) {
    if N > 64 {
        return;
    }
    for i in 0..N {
        let bit = 1u64 << (i as u32);
        if (delivered_mask & bit) != 0 {
            stats.delivered[i] = stats.delivered[i].wrapping_add(1);
        }
        if (dropped_full_mask & bit) != 0 {
            stats.dropped_full[i] = stats.dropped_full[i].wrapping_add(1);
        }
        if (dropped_no_credit_mask & bit) != 0 {
            stats.dropped_no_credit[i] = stats.dropped_no_credit[i].wrapping_add(1);
        }
    }
}

/// Run a fanout router until `should_stop()` returns true.
///
/// ## Contract
///
/// - `should_stop` should be cheap and side-effect free; it may be called frequently.
/// - The provided [`WaitStrategy`] is used for idle/backpressure waits and is reset after any
///   successful routing step.
#[inline]
pub fn route_until<const N: usize, W: WaitStrategy>(
    router: &FanoutRouter<N>,
    cfg: RouterLoopConfig,
    wait: &mut W,
    mut should_stop: impl FnMut() -> bool,
) -> RouterCounters {
    let mut credit_stats = CreditStats::<N>::default();
    route_until_with_credit_stats(router, cfg, wait, &mut credit_stats, &mut should_stop)
}

/// Run a fanout router until `should_stop()` returns true, periodically reporting counters.
///
/// Notes:
/// - Reporting is best-effort and uses `std::time::Instant`, so this is `std`-only.
/// - `report` is invoked at most once per `report_period`, and once at the end.
#[cfg(feature = "std")]
#[inline]
pub fn route_until_reporting<const N: usize, W: WaitStrategy>(
    router: &FanoutRouter<N>,
    cfg: RouterLoopConfig,
    wait: &mut W,
    mut should_stop: impl FnMut() -> bool,
    report_period: Duration,
    mut report: impl FnMut(RouterCounters),
) -> RouterCounters {
    let mut credit_stats = CreditStats::<N>::default();
    let mut last_report = std::time::Instant::now();

    let mut c = RouterCounters::default();

    let batch_max = cfg.batch_max.max(1);
    let mut loop_iters: u64 = 0;

    let blocking = if cfg.policy == BackpressurePolicy::Block {
        fanout_router_ref_blocking(router).ok()
    } else {
        None
    };

    let mut pressure_cursor: usize = 0;

    while !should_stop() {
        loop_iters = loop_iters.wrapping_add(1);
        if cfg.yield_every != 0 && loop_iters.is_multiple_of(cfg.yield_every as u64) {
            std::thread::yield_now();
        }

        if report_period.as_nanos() != 0 && last_report.elapsed() >= report_period {
            report(c);
            last_report = std::time::Instant::now();
        }

        if cfg.mode == RouterMode::WorkQueue && cfg.policy == BackpressurePolicy::SpinThenBlock {
            let shared = router.as_ptr();
            let blocked = if let Some(credit) = cfg.credit {
                has_pending_message(shared, cfg.source)
                    && !any_eligible_consumer_has_space_and_credit(
                        shared,
                        credit.credit_max.max(1),
                        &credit_stats.detached,
                    )
            } else {
                has_pending_message(shared, cfg.source) && !any_consumer_has_space(shared)
            };

            if blocked {
                if cfg.credit.is_some() {
                    c.credit_waits += 1;
                } else {
                    c.pressure_waits += 1;
                }
                wait.wait();
                continue;
            }
        }

        if cfg.mode == RouterMode::WorkQueue && cfg.policy == BackpressurePolicy::Block {
            let shared = router.as_ptr();
            let blocked = if let Some(credit) = cfg.credit {
                has_pending_message(shared, cfg.source)
                    && !any_eligible_consumer_has_space_and_credit(
                        shared,
                        credit.credit_max.max(1),
                        &credit_stats.detached,
                    )
            } else {
                has_pending_message(shared, cfg.source) && !any_consumer_has_space(shared)
            };

            if blocked {
                if cfg.credit.is_some() {
                    c.credit_waits += 1;
                } else {
                    c.pressure_waits += 1;
                }

                if let Some(br) = &blocking {
                    if N != 0 {
                        let i = pressure_cursor % N;
                        pressure_cursor = pressure_cursor.wrapping_add(1);

                        if let Some(expected) = br.consumer_seq(i) {
                            c.wake_waits += 1;
                            match br.wait_consumer_seq_ne(i, expected, Some(PRESSURE_WAIT_TIMEOUT))
                            {
                                Ok(true) => {}
                                Ok(false) => c.wake_timeouts += 1,
                                Err(_) => {
                                    wait.wait();
                                }
                            }
                        } else {
                            wait.wait();
                        }
                    } else {
                        wait.wait();
                    }
                } else {
                    wait.wait();
                }

                continue;
            }
        }

        // Credits: if configured to park/detach, avoid dequeuing when nobody is eligible.
        if cfg.mode == RouterMode::WorkQueue {
            if let Some(credit) = cfg.credit {
                if credit.policy == CreditPolicy::Park || credit.policy == CreditPolicy::Detach {
                    let shared = router.as_ptr();
                    if has_pending_message(shared, cfg.source)
                        && !any_eligible_consumer_has_space_and_credit(
                            shared,
                            credit.credit_max.max(1),
                            &credit_stats.detached,
                        )
                    {
                        c.credit_waits += 1;

                        if cfg.policy == BackpressurePolicy::Block {
                            if let Some(br) = &blocking {
                                if N != 0 {
                                    let i = pressure_cursor % N;
                                    pressure_cursor = pressure_cursor.wrapping_add(1);
                                    if let Some(expected) = br.consumer_seq(i) {
                                        c.wake_waits += 1;
                                        match br.wait_consumer_seq_ne(
                                            i,
                                            expected,
                                            Some(PRESSURE_WAIT_TIMEOUT),
                                        ) {
                                            Ok(true) => {}
                                            Ok(false) => c.wake_timeouts += 1,
                                            Err(_) => wait.wait(),
                                        }
                                    } else {
                                        wait.wait();
                                    }
                                } else {
                                    wait.wait();
                                }
                            } else {
                                wait.wait();
                            }
                        } else {
                            wait.wait();
                        }

                        continue;
                    }
                }
            }
        }

        // Credit detach bookkeeping (best-effort).
        if let Some(credit) = cfg.credit {
            let shared = router.as_ptr();
            let (eligible_mask, _no_credit_mask, _full_mask, depths) =
                compute_credit_masks(shared, credit.credit_max.max(1), &credit_stats.detached);
            credit_stats.depth = depths;

            if credit.policy == CreditPolicy::Detach {
                let now = std::time::Instant::now();
                let detach_after = credit
                    .detach_after_ms
                    .map(|ms| std::time::Duration::from_millis(ms as u64));
                let credit_max = credit.credit_max.max(1) as u64;
                let reattach_at = (credit_max / 2).min(credit_max.saturating_sub(1));

                for i in 0..N {
                    if credit_stats.detached[i] {
                        if credit_stats.depth[i] <= reattach_at {
                            credit_stats.detached[i] = false;
                            credit_stats.exhausted_since[i] = None;
                        }
                        continue;
                    }

                    if credit_stats.depth[i] >= credit_max {
                        if credit_stats.exhausted_since[i].is_none() {
                            credit_stats.exhausted_since[i] = Some(now);
                        }

                        if let (Some(since), Some(threshold)) =
                            (credit_stats.exhausted_since[i], detach_after)
                        {
                            if now.duration_since(since) >= threshold {
                                credit_stats.detached[i] = true;
                                credit_stats.detach_count[i] =
                                    credit_stats.detach_count[i].wrapping_add(1);
                                c.detaches += 1;
                                credit_stats.exhausted_since[i] = None;
                            }
                        }
                    } else {
                        credit_stats.exhausted_since[i] = None;
                    }
                }
            }

            let _ = eligible_mask;
        }

        // Attempt at least one route.
        let first_routed = if let Some(credit) = cfg.credit {
            let shared = router.as_ptr();
            let (eligible, no_credit, full, depths) =
                compute_credit_masks(shared, credit.credit_max.max(1), &credit_stats.detached);
            credit_stats.depth = depths;

            let step = router
                .route_once_with_credit_masks(cfg.source, cfg.mode, eligible, no_credit, full);
            if !step.routed {
                false
            } else {
                c.routed += 1;
                c.delivered += step.delivered as u64;
                c.dropped += (step.dropped_full + step.dropped_no_credit) as u64;
                match cfg.mode {
                    RouterMode::Broadcast => c.dropped_queue_full += step.dropped_full as u64,
                    RouterMode::WorkQueue => c.dropped_all_full += step.dropped_full as u64,
                }
                c.dropped_no_credit += step.dropped_no_credit as u64;

                if cfg.mode == RouterMode::Broadcast {
                    apply_masks_to_stats(
                        &mut credit_stats,
                        step.delivered_mask,
                        step.dropped_full_mask,
                        step.dropped_no_credit_mask,
                    );
                } else {
                    apply_masks_to_stats(&mut credit_stats, step.delivered_mask, 0, 0);

                    if step.dropped_no_credit != 0 {
                        apply_masks_to_stats(&mut credit_stats, 0, 0, no_credit);
                    }
                    if step.dropped_full != 0 {
                        apply_masks_to_stats(&mut credit_stats, 0, full, 0);
                    }
                }

                true
            }
        } else {
            let first = router.route_once_with_stats(cfg.source, cfg.mode);
            if !first.routed {
                false
            } else {
                c.routed += 1;
                c.delivered += first.delivered as u64;
                c.dropped += first.dropped as u64;
                if first.dropped != 0 {
                    match cfg.mode {
                        RouterMode::Broadcast => c.dropped_queue_full += first.dropped as u64,
                        RouterMode::WorkQueue => c.dropped_all_full += first.dropped as u64,
                    }
                }
                true
            }
        };

        if !first_routed {
            c.idle_waits += 1;

            if cfg.idle_spin_limit != 0 {
                let shared = router.as_ptr();
                for _ in 0..cfg.idle_spin_limit {
                    if should_stop() {
                        report(c);
                        return c;
                    }
                    if has_pending_message(shared, cfg.source) {
                        break;
                    }
                    spin_loop();
                }
                if has_pending_message(shared, cfg.source) {
                    continue;
                }
            }

            if cfg.policy == BackpressurePolicy::Block {
                if let Some(br) = &blocking {
                    let expected = br.producer_seq();
                    c.wake_waits += 1;
                    match br.wait_producer_seq_ne(expected, Some(IDLE_WAIT_TIMEOUT)) {
                        Ok(true) => {}
                        Ok(false) => c.wake_timeouts += 1,
                        Err(_) => wait.wait(),
                    }
                } else {
                    wait.wait();
                }
            } else {
                wait.wait();
            }

            continue;
        }

        let mut batch_size: u32 = 0;
        let batch_start = std::time::Instant::now();
        batch_size += 1;

        while batch_size < batch_max {
            if let Some(us) = cfg.batch_time_us {
                if batch_start.elapsed() >= std::time::Duration::from_micros(us as u64) {
                    break;
                }
            }

            if cfg.mode == RouterMode::WorkQueue
                && (cfg.policy == BackpressurePolicy::SpinThenBlock
                    || cfg.policy == BackpressurePolicy::Block)
            {
                let shared = router.as_ptr();
                let blocked = if let Some(credit) = cfg.credit {
                    has_pending_message(shared, cfg.source)
                        && !any_eligible_consumer_has_space_and_credit(
                            shared,
                            credit.credit_max.max(1),
                            &credit_stats.detached,
                        )
                } else {
                    has_pending_message(shared, cfg.source) && !any_consumer_has_space(shared)
                };

                if blocked {
                    break;
                }
            }

            if let Some(credit) = cfg.credit {
                let shared = router.as_ptr();
                let (eligible, no_credit, full, depths) =
                    compute_credit_masks(shared, credit.credit_max.max(1), &credit_stats.detached);
                credit_stats.depth = depths;

                let step = router
                    .route_once_with_credit_masks(cfg.source, cfg.mode, eligible, no_credit, full);
                if !step.routed {
                    break;
                }

                c.routed += 1;
                c.delivered += step.delivered as u64;
                c.dropped += (step.dropped_full + step.dropped_no_credit) as u64;
                match cfg.mode {
                    RouterMode::Broadcast => c.dropped_queue_full += step.dropped_full as u64,
                    RouterMode::WorkQueue => c.dropped_all_full += step.dropped_full as u64,
                }
                c.dropped_no_credit += step.dropped_no_credit as u64;

                if cfg.mode == RouterMode::Broadcast {
                    apply_masks_to_stats(
                        &mut credit_stats,
                        step.delivered_mask,
                        step.dropped_full_mask,
                        step.dropped_no_credit_mask,
                    );
                } else {
                    apply_masks_to_stats(&mut credit_stats, step.delivered_mask, 0, 0);
                    if step.dropped_no_credit != 0 {
                        apply_masks_to_stats(&mut credit_stats, 0, 0, no_credit);
                    }
                    if step.dropped_full != 0 {
                        apply_masks_to_stats(&mut credit_stats, 0, full, 0);
                    }
                }
            } else {
                let step = router.route_once_with_stats(cfg.source, cfg.mode);
                if !step.routed {
                    break;
                }
                c.routed += 1;
                c.delivered += step.delivered as u64;
                c.dropped += step.dropped as u64;
                if step.dropped != 0 {
                    match cfg.mode {
                        RouterMode::Broadcast => c.dropped_queue_full += step.dropped as u64,
                        RouterMode::WorkQueue => c.dropped_all_full += step.dropped as u64,
                    }
                }
            }
            batch_size += 1;
        }

        c.batches += 1;
        c.batch_sum += batch_size as u64;
        c.batch_max = c.batch_max.max(batch_size);
        wait.reset();
    }

    report(c);
    c
}

/// Run a fanout router until `should_stop()` returns true, returning aggregate counters.
///
/// If `cfg.credit` is enabled, `credit_stats` is updated in place.
///
/// ## Backpressure and blocking
///
/// - For broadcast routing, destination-full remains drop-on-full.
/// - For work-queue routing, backpressure policies can avoid dequeuing the next source message
///   when no consumer can accept it.
/// - With [`BackpressurePolicy::Block`], the router uses wake sections when present; if wake
///   waits fail at runtime, the loop falls back to `wait.wait()`.
#[inline]
pub fn route_until_with_credit_stats<const N: usize, W: WaitStrategy>(
    router: &FanoutRouter<N>,
    cfg: RouterLoopConfig,
    wait: &mut W,
    credit_stats: &mut CreditStats<N>,
    mut should_stop: impl FnMut() -> bool,
) -> RouterCounters {
    let mut c = RouterCounters::default();

    let batch_max = cfg.batch_max.max(1);
    let mut loop_iters: u64 = 0;

    let blocking = if cfg.policy == BackpressurePolicy::Block {
        fanout_router_ref_blocking(router).ok()
    } else {
        None
    };

    let mut pressure_cursor: usize = 0;

    while !should_stop() {
        loop_iters = loop_iters.wrapping_add(1);
        if cfg.yield_every != 0 && loop_iters.is_multiple_of(cfg.yield_every as u64) {
            #[cfg(feature = "std")]
            std::thread::yield_now();
        }

        if cfg.mode == RouterMode::WorkQueue && cfg.policy == BackpressurePolicy::SpinThenBlock {
            let shared = router.as_ptr();
            let blocked = if let Some(credit) = cfg.credit {
                has_pending_message(shared, cfg.source)
                    && !any_eligible_consumer_has_space_and_credit(
                        shared,
                        credit.credit_max.max(1),
                        &credit_stats.detached,
                    )
            } else {
                has_pending_message(shared, cfg.source) && !any_consumer_has_space(shared)
            };

            if blocked {
                if cfg.credit.is_some() {
                    c.credit_waits += 1;
                } else {
                    c.pressure_waits += 1;
                }
                wait.wait();
                continue;
            }
        }

        if cfg.mode == RouterMode::WorkQueue && cfg.policy == BackpressurePolicy::Block {
            let shared = router.as_ptr();
            let blocked = if let Some(credit) = cfg.credit {
                has_pending_message(shared, cfg.source)
                    && !any_eligible_consumer_has_space_and_credit(
                        shared,
                        credit.credit_max.max(1),
                        &credit_stats.detached,
                    )
            } else {
                has_pending_message(shared, cfg.source) && !any_consumer_has_space(shared)
            };

            if blocked {
                if cfg.credit.is_some() {
                    c.credit_waits += 1;
                } else {
                    c.pressure_waits += 1;
                }

                if let Some(br) = &blocking {
                    if N != 0 {
                        let i = pressure_cursor % N;
                        pressure_cursor = pressure_cursor.wrapping_add(1);

                        if let Some(expected) = br.consumer_seq(i) {
                            c.wake_waits += 1;
                            match br.wait_consumer_seq_ne(i, expected, Some(PRESSURE_WAIT_TIMEOUT))
                            {
                                Ok(true) => {}
                                Ok(false) => c.wake_timeouts += 1,
                                Err(_) => {
                                    // Fall back if OS wait fails.
                                    wait.wait();
                                }
                            }
                        } else {
                            wait.wait();
                        }
                    } else {
                        wait.wait();
                    }
                } else {
                    wait.wait();
                }

                continue;
            }
        }

        // Credits: if configured to park/detach, avoid dequeuing when nobody is eligible.
        if cfg.mode == RouterMode::WorkQueue {
            if let Some(credit) = cfg.credit {
                if credit.policy == CreditPolicy::Park || credit.policy == CreditPolicy::Detach {
                    let shared = router.as_ptr();
                    if has_pending_message(shared, cfg.source)
                        && !any_eligible_consumer_has_space_and_credit(
                            shared,
                            credit.credit_max.max(1),
                            &credit_stats.detached,
                        )
                    {
                        c.credit_waits += 1;

                        // Prefer wake-backed blocking if enabled.
                        if cfg.policy == BackpressurePolicy::Block {
                            if let Some(br) = &blocking {
                                if N != 0 {
                                    let i = pressure_cursor % N;
                                    pressure_cursor = pressure_cursor.wrapping_add(1);
                                    if let Some(expected) = br.consumer_seq(i) {
                                        c.wake_waits += 1;
                                        match br.wait_consumer_seq_ne(
                                            i,
                                            expected,
                                            Some(PRESSURE_WAIT_TIMEOUT),
                                        ) {
                                            Ok(true) => {}
                                            Ok(false) => c.wake_timeouts += 1,
                                            Err(_) => wait.wait(),
                                        }
                                    } else {
                                        wait.wait();
                                    }
                                } else {
                                    wait.wait();
                                }
                            } else {
                                wait.wait();
                            }
                        } else {
                            wait.wait();
                        }

                        continue;
                    }
                }
            }
        }

        // Credit detach bookkeeping (best-effort).
        if let Some(credit) = cfg.credit {
            let shared = router.as_ptr();
            let (eligible_mask, _no_credit_mask, _full_mask, depths) =
                compute_credit_masks(shared, credit.credit_max.max(1), &credit_stats.detached);
            credit_stats.depth = depths;

            if credit.policy == CreditPolicy::Detach {
                #[cfg(feature = "std")]
                {
                    let now = std::time::Instant::now();
                    let detach_after = credit
                        .detach_after_ms
                        .map(|ms| std::time::Duration::from_millis(ms as u64));
                    let credit_max = credit.credit_max.max(1) as u64;
                    let reattach_at = (credit_max / 2).min(credit_max.saturating_sub(1));

                    for i in 0..N {
                        if credit_stats.detached[i] {
                            if credit_stats.depth[i] <= reattach_at {
                                credit_stats.detached[i] = false;
                                credit_stats.exhausted_since[i] = None;
                            }
                            continue;
                        }

                        if credit_stats.depth[i] >= credit_max {
                            if credit_stats.exhausted_since[i].is_none() {
                                credit_stats.exhausted_since[i] = Some(now);
                            }

                            if let (Some(since), Some(threshold)) =
                                (credit_stats.exhausted_since[i], detach_after)
                            {
                                if now.duration_since(since) >= threshold {
                                    credit_stats.detached[i] = true;
                                    credit_stats.detach_count[i] =
                                        credit_stats.detach_count[i].wrapping_add(1);
                                    c.detaches += 1;
                                    credit_stats.exhausted_since[i] = None;
                                }
                            }
                        } else {
                            credit_stats.exhausted_since[i] = None;
                        }
                    }
                }
            }

            // If N<=64 and no consumers are eligible, keep eligible_mask computed for later.
            let _ = eligible_mask;
        }

        // Attempt at least one route.
        let first_routed = if let Some(credit) = cfg.credit {
            let shared = router.as_ptr();
            let (eligible, no_credit, full, depths) =
                compute_credit_masks(shared, credit.credit_max.max(1), &credit_stats.detached);
            credit_stats.depth = depths;

            let step = router
                .route_once_with_credit_masks(cfg.source, cfg.mode, eligible, no_credit, full);
            if !step.routed {
                false
            } else {
                c.routed += 1;
                c.delivered += step.delivered as u64;
                c.dropped += (step.dropped_full + step.dropped_no_credit) as u64;
                match cfg.mode {
                    RouterMode::Broadcast => c.dropped_queue_full += step.dropped_full as u64,
                    RouterMode::WorkQueue => c.dropped_all_full += step.dropped_full as u64,
                }
                c.dropped_no_credit += step.dropped_no_credit as u64;

                if cfg.mode == RouterMode::Broadcast {
                    apply_masks_to_stats(
                        credit_stats,
                        step.delivered_mask,
                        step.dropped_full_mask,
                        step.dropped_no_credit_mask,
                    );
                } else {
                    // Work-queue attribution is best-effort.
                    apply_masks_to_stats(credit_stats, step.delivered_mask, 0, 0);

                    // If we dropped due to no-credit/full, attribute using the precomputed masks.
                    if step.dropped_no_credit != 0 {
                        apply_masks_to_stats(credit_stats, 0, 0, no_credit);
                    }
                    if step.dropped_full != 0 {
                        apply_masks_to_stats(credit_stats, 0, full, 0);
                    }
                }

                true
            }
        } else {
            let first = router.route_once_with_stats(cfg.source, cfg.mode);
            if !first.routed {
                false
            } else {
                c.routed += 1;
                c.delivered += first.delivered as u64;
                c.dropped += first.dropped as u64;
                if first.dropped != 0 {
                    match cfg.mode {
                        RouterMode::Broadcast => c.dropped_queue_full += first.dropped as u64,
                        RouterMode::WorkQueue => c.dropped_all_full += first.dropped as u64,
                    }
                }
                true
            }
        };

        if !first_routed {
            c.idle_waits += 1;

            // Bounded idle spin window before invoking a heavier wait.
            if cfg.idle_spin_limit != 0 {
                let shared = router.as_ptr();
                for _ in 0..cfg.idle_spin_limit {
                    if should_stop() {
                        return c;
                    }
                    if has_pending_message(shared, cfg.source) {
                        break;
                    }
                    spin_loop();
                }
                if has_pending_message(shared, cfg.source) {
                    continue;
                }
            }

            if cfg.policy == BackpressurePolicy::Block {
                if let Some(br) = &blocking {
                    let expected = br.producer_seq();
                    c.wake_waits += 1;
                    match br.wait_producer_seq_ne(expected, Some(IDLE_WAIT_TIMEOUT)) {
                        Ok(true) => {}
                        Ok(false) => c.wake_timeouts += 1,
                        Err(_) => wait.wait(),
                    }
                } else {
                    wait.wait();
                }
            } else {
                wait.wait();
            }

            continue;
        }

        // Start a batch.
        let mut batch_size: u32 = 0;

        #[cfg(feature = "std")]
        let batch_start = std::time::Instant::now();

        batch_size += 1;

        while batch_size < batch_max {
            #[cfg(feature = "std")]
            if let Some(us) = cfg.batch_time_us {
                if batch_start.elapsed() >= Duration::from_micros(us as u64) {
                    break;
                }
            }

            // If we are about to hit pressure, end the batch and let the outer loop apply the
            // configured wait policy.
            if cfg.mode == RouterMode::WorkQueue
                && (cfg.policy == BackpressurePolicy::SpinThenBlock
                    || cfg.policy == BackpressurePolicy::Block)
            {
                let shared = router.as_ptr();
                let blocked = if let Some(credit) = cfg.credit {
                    has_pending_message(shared, cfg.source)
                        && !any_eligible_consumer_has_space_and_credit(
                            shared,
                            credit.credit_max.max(1),
                            &credit_stats.detached,
                        )
                } else {
                    has_pending_message(shared, cfg.source) && !any_consumer_has_space(shared)
                };

                if blocked {
                    break;
                }
            }

            if let Some(credit) = cfg.credit {
                let shared = router.as_ptr();
                let (eligible, no_credit, full, depths) =
                    compute_credit_masks(shared, credit.credit_max.max(1), &credit_stats.detached);
                credit_stats.depth = depths;

                let step = router
                    .route_once_with_credit_masks(cfg.source, cfg.mode, eligible, no_credit, full);
                if !step.routed {
                    break;
                }

                c.routed += 1;
                c.delivered += step.delivered as u64;
                c.dropped += (step.dropped_full + step.dropped_no_credit) as u64;
                match cfg.mode {
                    RouterMode::Broadcast => c.dropped_queue_full += step.dropped_full as u64,
                    RouterMode::WorkQueue => c.dropped_all_full += step.dropped_full as u64,
                }
                c.dropped_no_credit += step.dropped_no_credit as u64;

                if cfg.mode == RouterMode::Broadcast {
                    apply_masks_to_stats(
                        credit_stats,
                        step.delivered_mask,
                        step.dropped_full_mask,
                        step.dropped_no_credit_mask,
                    );
                } else {
                    apply_masks_to_stats(credit_stats, step.delivered_mask, 0, 0);
                    if step.dropped_no_credit != 0 {
                        apply_masks_to_stats(credit_stats, 0, 0, no_credit);
                    }
                    if step.dropped_full != 0 {
                        apply_masks_to_stats(credit_stats, 0, full, 0);
                    }
                }
            } else {
                let step = router.route_once_with_stats(cfg.source, cfg.mode);
                if !step.routed {
                    break;
                }
                c.routed += 1;
                c.delivered += step.delivered as u64;
                c.dropped += step.dropped as u64;
                if step.dropped != 0 {
                    match cfg.mode {
                        RouterMode::Broadcast => c.dropped_queue_full += step.dropped as u64,
                        RouterMode::WorkQueue => c.dropped_all_full += step.dropped as u64,
                    }
                }
            }
            batch_size += 1;
        }

        c.batches += 1;
        c.batch_sum += batch_size as u64;
        c.batch_max = c.batch_max.max(batch_size);
        wait.reset();
    }

    c
}

/// Run the router for a fixed duration.
///
/// Requires `std` because it uses `std::time::Instant`.
#[cfg(feature = "std")]
#[inline]
pub fn route_for<const N: usize, W: WaitStrategy>(
    router: &FanoutRouter<N>,
    cfg: RouterLoopConfig,
    wait: &mut W,
    dur: Duration,
) -> RouterCounters {
    let start = std::time::Instant::now();
    route_until(router, cfg, wait, || start.elapsed() >= dur)
}