dynamo-kv-router 1.3.0

// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

use std::cmp::Ordering;
use std::collections::BinaryHeap;

use ordered_float::OrderedFloat;
use serde::{Deserialize, Serialize};

use super::config::RouterQueuePolicy;
use super::policy_config::{PolicyClassConfig, PolicyProfile};

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct QueueSnapshot {
    pub raw_isl_tokens: usize,
    pub cached_tokens: usize,
    pub uncached_tokens: usize,
    pub scheduling_cost_tokens: usize,
}

impl QueueSnapshot {
    /// Keeps exact uncached tokens for classification while clamping only the
    /// scheduling cost so zero-work requests still participate in DRR/WSPT.
    pub fn new(raw_isl_tokens: usize, cached_tokens: usize) -> Self {
        let cached_tokens = cached_tokens.min(raw_isl_tokens);
        Self {
            raw_isl_tokens,
            cached_tokens,
            uncached_tokens: raw_isl_tokens.saturating_sub(cached_tokens),
            scheduling_cost_tokens: raw_isl_tokens.saturating_sub(cached_tokens).max(1),
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum QueueLimitKind {
    Requests,
    RawIslTokens,
    CachedTokens,
}

impl std::fmt::Display for QueueLimitKind {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Requests => formatter.write_str("requests"),
            Self::RawIslTokens => formatter.write_str("raw_isl_tokens"),
            Self::CachedTokens => formatter.write_str("cached_tokens"),
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
#[error(
    "router policy class {policy_class:?} queue {limit_kind} limit reached \
     (current={current}, limit={limit})"
)]
pub struct QueueRejection {
    pub policy_class: String,
    pub limit_kind: QueueLimitKind,
    pub current: usize,
    pub limit: usize,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct PolicyQueueStats {
    pub requests: usize,
    pub raw_isl_tokens: usize,
    pub cached_tokens: usize,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct QueuePriority {
    strict_priority: u32,
    policy_score: OrderedFloat<f64>,
}

pub struct PolicyQueueEntry<T> {
    class_index: usize,
    priority: QueuePriority,
    enqueue_seq: u64,
    snapshot: QueueSnapshot,
    payload: T,
}

impl<T> PolicyQueueEntry<T> {
    pub fn class_index(&self) -> usize {
        self.class_index
    }

    pub fn snapshot(&self) -> QueueSnapshot {
        self.snapshot
    }

    pub fn payload(&self) -> &T {
        &self.payload
    }

    pub fn payload_mut(&mut self) -> &mut T {
        &mut self.payload
    }

    pub fn into_payload(self) -> T {
        self.payload
    }
}

impl<T> Eq for PolicyQueueEntry<T> {}

impl<T> PartialEq for PolicyQueueEntry<T> {
    fn eq(&self, other: &Self) -> bool {
        self.priority == other.priority && self.enqueue_seq == other.enqueue_seq
    }
}

impl<T> Ord for PolicyQueueEntry<T> {
    fn cmp(&self, other: &Self) -> Ordering {
        self.priority
            .strict_priority
            .cmp(&other.priority.strict_priority)
            .then_with(|| self.priority.policy_score.cmp(&other.priority.policy_score))
            .then_with(|| other.enqueue_seq.cmp(&self.enqueue_seq))
    }
}

impl<T> PartialOrd for PolicyQueueEntry<T> {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

struct PolicyClassQueue<T> {
    config: PolicyClassConfig,
    pending: BinaryHeap<PolicyQueueEntry<T>>,
    stats: PolicyQueueStats,
    deficit: usize,
}

pub struct PolicyQueue<T> {
    classes: Vec<PolicyClassQueue<T>>,
    next_class: usize,
    next_enqueue_seq: u64,
    pending_count: usize,
    dispatchable: Vec<bool>,
}

impl<T> PolicyQueue<T> {
    pub fn new(profile: PolicyProfile) -> Self {
        let class_count = profile.classes().len();
        Self {
            classes: profile
                .classes()
                .iter()
                .cloned()
                .map(|config| PolicyClassQueue {
                    config,
                    pending: BinaryHeap::new(),
                    stats: PolicyQueueStats::default(),
                    deficit: 0,
                })
                .collect(),
            next_class: 0,
            next_enqueue_seq: 0,
            pending_count: 0,
            dispatchable: vec![false; class_count],
        }
    }

    pub fn pending_count(&self) -> usize {
        self.pending_count
    }

    pub fn class_count(&self) -> usize {
        self.classes.len()
    }

    pub fn class_config(&self, class_index: usize) -> &PolicyClassConfig {
        &self.classes[class_index].config
    }

    pub fn class_stats(&self, class_index: usize) -> PolicyQueueStats {
        self.classes[class_index].stats
    }

    pub fn has_backlog(&self, class_index: usize) -> bool {
        !self.classes[class_index].pending.is_empty()
    }

    pub fn entries(&self) -> impl Iterator<Item = &PolicyQueueEntry<T>> {
        self.classes.iter().flat_map(|class| class.pending.iter())
    }

    /// Remove queued entries that no longer satisfy `keep`, rebuilding queue
    /// accounting while preserving each retained entry's scheduling key.
    pub fn retain(&mut self, mut keep: impl FnMut(&T) -> bool) {
        self.pending_count = 0;
        for class in &mut self.classes {
            let pending = std::mem::take(&mut class.pending);
            class.stats = PolicyQueueStats::default();
            for entry in pending {
                if !keep(entry.payload()) {
                    continue;
                }
                add_stats(&mut class.stats, entry.snapshot);
                class.pending.push(entry);
                self.pending_count += 1;
            }
            if class.pending.is_empty() {
                class.deficit = 0;
            }
        }
    }

    #[allow(clippy::too_many_arguments)]
    /// Applies class-local, worker-scaled limits against pre-add counters, then
    /// captures the immutable scheduling key and accounting snapshot.
    pub fn enqueue(
        &mut self,
        class_index: usize,
        worker_count: usize,
        snapshot: QueueSnapshot,
        arrival_offset_secs: f64,
        priority_jump: f64,
        strict_priority: u32,
        payload: T,
    ) -> Result<(), (QueueRejection, T)> {
        let class = &mut self.classes[class_index];
        if let Some(rejection) = queue_rejection(class, worker_count) {
            return Err((rejection, payload));
        }

        let policy_score = match class.config.queue_policy {
            RouterQueuePolicy::Fcfs => priority_jump.max(0.0) - arrival_offset_secs.max(0.0),
            RouterQueuePolicy::Wspt => {
                (1.0 + priority_jump.max(0.0)) / snapshot.scheduling_cost_tokens as f64
            }
            RouterQueuePolicy::Lcfs => priority_jump.max(0.0) + arrival_offset_secs.max(0.0),
        };
        let entry = PolicyQueueEntry {
            class_index,
            priority: QueuePriority {
                strict_priority,
                policy_score: OrderedFloat(policy_score),
            },
            enqueue_seq: self.next_enqueue_seq,
            snapshot,
            payload,
        };
        self.next_enqueue_seq = self.next_enqueue_seq.wrapping_add(1);
        add_stats(&mut class.stats, snapshot);
        class.pending.push(entry);
        self.pending_count += 1;
        Ok(())
    }

    /// Runs one DRR ring pass over dispatchable class heads. If no head has
    /// enough credit, bulk-adds the minimum complete rounds needed for progress.
    pub fn pop_next(
        &mut self,
        mut is_dispatchable: impl FnMut(usize, &PolicyClassConfig, &T) -> bool,
    ) -> Option<PolicyQueueEntry<T>> {
        if self.pending_count == 0 {
            return None;
        }

        let class_count = self.classes.len();
        self.dispatchable.fill(false);
        for offset in 0..class_count {
            // Rotate the starting point across calls so class vector order
            // cannot become a permanent scheduling preference.
            let class_index = (self.next_class + offset) % class_count;
            let class = &mut self.classes[class_index];
            let Some(front) = class.pending.peek() else {
                class.deficit = 0;
                continue;
            };
            if !is_dispatchable(class_index, &class.config, front.payload()) {
                // A blocked class retains earned credit but does not accrue
                // more until its head becomes dispatchable.
                continue;
            }

            self.dispatchable[class_index] = true;
            if front.snapshot.scheduling_cost_tokens <= class.deficit {
                // Quantum is granted per ring round, not per request. Spend
                // carried credit before granting this class another quantum.
                return Some(self.pop_class(class_index));
            }
            class.deficit = class.deficit.saturating_add(class.config.quantum);
            if front.snapshot.scheduling_cost_tokens <= class.deficit {
                // The normal single-round visit made this head affordable.
                return Some(self.pop_class(class_index));
            }
        }

        // Fast-forward the minimum number of complete virtual rounds needed
        // for any dispatchable head to progress, avoiding repeated ring scans
        // for requests much larger than their class quantum. If every head was
        // blocked, `min()` returns `None` without changing any deficit.
        let rounds = self
            .dispatchable
            .iter()
            .enumerate()
            .filter_map(|(class_index, dispatchable)| {
                if !dispatchable {
                    return None;
                }
                let class = &self.classes[class_index];
                let cost = class.pending.peek()?.snapshot.scheduling_cost_tokens;
                let missing = cost.saturating_sub(class.deficit);
                Some(missing.div_ceil(class.config.quantum))
            })
            .min()?;

        for (class_index, dispatchable) in self.dispatchable.iter().copied().enumerate() {
            if !dispatchable {
                continue;
            }
            let class = &mut self.classes[class_index];
            // Applying the same virtual round count preserves weighting
            // because each class scales the credit by its own quantum.
            class.deficit = class
                .deficit
                .saturating_add(class.config.quantum.saturating_mul(rounds));
        }

        for offset in 0..class_count {
            let class_index = (self.next_class + offset) % class_count;
            let class = &self.classes[class_index];
            if self.dispatchable[class_index]
                && class
                    .pending
                    .peek()
                    .is_some_and(|entry| entry.snapshot.scheduling_cost_tokens <= class.deficit)
            {
                return Some(self.pop_class(class_index));
            }
        }

        None
    }

    pub fn drain(self) -> impl Iterator<Item = PolicyQueueEntry<T>> {
        self.classes
            .into_iter()
            .flat_map(|class| class.pending.into_iter())
    }

    fn pop_class(&mut self, class_index: usize) -> PolicyQueueEntry<T> {
        let class = &mut self.classes[class_index];
        let entry = class.pending.pop().expect("policy class front vanished");
        class.deficit = class
            .deficit
            .saturating_sub(entry.snapshot.scheduling_cost_tokens);
        subtract_stats(&mut class.stats, entry.snapshot);
        self.pending_count -= 1;
        // Empty classes discard stale credit. A class that can already afford
        // its next head keeps the cursor and spends its weighted burst;
        // otherwise the next call starts at the following class.
        if class.pending.is_empty() {
            class.deficit = 0;
            self.next_class = (class_index + 1) % self.classes.len();
        } else if class
            .pending
            .peek()
            .is_some_and(|next| next.snapshot.scheduling_cost_tokens <= class.deficit)
        {
            self.next_class = class_index;
        } else {
            self.next_class = (class_index + 1) % self.classes.len();
        }
        entry
    }
}

fn queue_rejection<T>(class: &PolicyClassQueue<T>, worker_count: usize) -> Option<QueueRejection> {
    // Limits scale from the current discovered endpoint count and intentionally
    // compare only existing usage; the request that crosses a cap is accepted.
    for (limit_kind, current, limit_per_worker) in [
        (
            QueueLimitKind::Requests,
            class.stats.requests,
            class.config.request_queue_limit_per_worker,
        ),
        (
            QueueLimitKind::RawIslTokens,
            class.stats.raw_isl_tokens,
            class.config.raw_isl_token_queue_limit_per_worker,
        ),
        (
            QueueLimitKind::CachedTokens,
            class.stats.cached_tokens,
            class.config.cached_token_queue_limit_per_worker,
        ),
    ] {
        let limit = limit_per_worker.map(|limit| limit.saturating_mul(worker_count));
        if limit.is_some_and(|limit| current >= limit) {
            return Some(QueueRejection {
                policy_class: class.config.name.clone(),
                limit_kind,
                current,
                limit: limit.expect("checked as some"),
            });
        }
    }

    None
}

fn add_stats(stats: &mut PolicyQueueStats, snapshot: QueueSnapshot) {
    stats.requests += 1;
    stats.raw_isl_tokens = stats.raw_isl_tokens.saturating_add(snapshot.raw_isl_tokens);
    stats.cached_tokens = stats.cached_tokens.saturating_add(snapshot.cached_tokens);
}

fn subtract_stats(stats: &mut PolicyQueueStats, snapshot: QueueSnapshot) {
    stats.requests = stats.requests.saturating_sub(1);
    stats.raw_isl_tokens = stats.raw_isl_tokens.saturating_sub(snapshot.raw_isl_tokens);
    stats.cached_tokens = stats.cached_tokens.saturating_sub(snapshot.cached_tokens);
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::scheduling::RouterPolicyConfig;

    fn profile(yaml: &str) -> PolicyProfile {
        RouterPolicyConfig::from_yaml(yaml)
            .unwrap()
            .resolve_profile(None, None, RouterQueuePolicy::Fcfs)
    }

    #[test]
    fn per_worker_caps_scale_and_remain_pre_add() {
        let mut queue = PolicyQueue::new(profile(
            r#"
default_policy_family: capped
uncached_isl_buckets:
  - min_tokens: 0
    bucket: all
policy_classes:
  - name: capped
    policy_family: capped
    cache_bucket: all
    quantum: 10
    request_queue_limit_per_worker: 1
    raw_isl_token_queue_limit_per_worker: 5
    cached_token_queue_limit_per_worker: 3
"#,
        ));
        queue
            .enqueue(0, 2, QueueSnapshot::new(8, 4), 0.0, 0.0, 0, "first")
            .unwrap();
        queue
            .enqueue(0, 2, QueueSnapshot::new(100, 100), 1.0, 0.0, 0, "overshoot")
            .unwrap();
        let (rejection, payload) = queue
            .enqueue(0, 2, QueueSnapshot::new(1, 0), 2.0, 0.0, 0, "rejected")
            .unwrap_err();
        assert_eq!(payload, "rejected");
        assert_eq!(rejection.limit_kind, QueueLimitKind::Requests);
        assert_eq!(rejection.current, 2);
        assert_eq!(rejection.limit, 2);
        assert_eq!(queue.class_stats(0).raw_isl_tokens, 108);
        assert_eq!(queue.class_stats(0).cached_tokens, 104);
    }

    #[test]
    fn retain_removes_payload_and_rebuilds_queue_accounting() {
        let mut queue = PolicyQueue::new(profile(
            r#"
default_policy_family: default
uncached_isl_buckets:
  - min_tokens: 0
    bucket: all
policy_classes:
  - name: default
    policy_family: default
    cache_bucket: all
    quantum: 10
"#,
        ));
        queue
            .enqueue(0, 1, QueueSnapshot::new(8, 4), 0.0, 0.0, 0, "keep")
            .unwrap();
        queue
            .enqueue(0, 1, QueueSnapshot::new(16, 6), 1.0, 0.0, 0, "remove")
            .unwrap();

        queue.retain(|payload| *payload != "remove");

        assert_eq!(queue.pending_count(), 1);
        assert_eq!(queue.class_stats(0).requests, 1);
        assert_eq!(queue.class_stats(0).raw_isl_tokens, 8);
        assert_eq!(queue.class_stats(0).cached_tokens, 4);
        assert_eq!(
            queue.pop_next(|_, _, _| true).unwrap().into_payload(),
            "keep"
        );
    }

    #[test]
    fn per_worker_token_caps_follow_capacity_without_evicting() {
        let mut queue = PolicyQueue::new(profile(
            r#"
default_policy_family: raw
uncached_isl_buckets:
  - min_tokens: 0
    bucket: all
policy_classes:
  - name: raw
    policy_family: raw
    cache_bucket: all
    quantum: 1
    raw_isl_token_queue_limit_per_worker: 10
  - name: cached
    policy_family: cached
    cache_bucket: all
    quantum: 1
    cached_token_queue_limit_per_worker: 5
  - name: zero
    policy_family: zero
    cache_bucket: all
    quantum: 1
    request_queue_limit_per_worker: 0
  - name: no-workers
    policy_family: no-workers
    cache_bucket: all
    quantum: 1
    request_queue_limit_per_worker: 1
"#,
        ));
        queue
            .enqueue(0, 2, QueueSnapshot::new(11, 0), 0.0, 0.0, 0, "raw-queued")
            .unwrap();
        let (raw_rejection, _) = queue
            .enqueue(0, 1, QueueSnapshot::new(1, 0), 1.0, 0.0, 0, "raw-rejected")
            .unwrap_err();
        assert_eq!(raw_rejection.limit_kind, QueueLimitKind::RawIslTokens);
        assert_eq!(raw_rejection.current, 11);
        assert_eq!(raw_rejection.limit, 10);
        assert_eq!(queue.class_stats(0).raw_isl_tokens, 11);

        queue
            .enqueue(
                0,
                2,
                QueueSnapshot::new(10, 0),
                2.0,
                0.0,
                0,
                "raw-after-growth",
            )
            .unwrap();
        let (grown_rejection, _) = queue
            .enqueue(
                0,
                2,
                QueueSnapshot::new(1, 0),
                3.0,
                0.0,
                0,
                "raw-at-grown-cap",
            )
            .unwrap_err();
        assert_eq!(grown_rejection.current, 21);
        assert_eq!(grown_rejection.limit, 20);

        queue
            .enqueue(1, 2, QueueSnapshot::new(8, 6), 0.0, 0.0, 0, "cached-queued")
            .unwrap();
        let (cached_rejection, _) = queue
            .enqueue(
                1,
                1,
                QueueSnapshot::new(1, 1),
                1.0,
                0.0,
                0,
                "cached-rejected",
            )
            .unwrap_err();
        assert_eq!(cached_rejection.limit_kind, QueueLimitKind::CachedTokens);
        assert_eq!(cached_rejection.current, 6);
        assert_eq!(cached_rejection.limit, 5);

        let (zero_rejection, _) = queue
            .enqueue(2, 4, QueueSnapshot::new(1, 0), 0.0, 0.0, 0, "zero")
            .unwrap_err();
        assert_eq!(zero_rejection.limit_kind, QueueLimitKind::Requests);
        assert_eq!(zero_rejection.limit, 0);

        let (no_workers_rejection, _) = queue
            .enqueue(3, 0, QueueSnapshot::new(1, 0), 0.0, 0.0, 0, "no-workers")
            .unwrap_err();
        assert_eq!(no_workers_rejection.current, 0);
        assert_eq!(no_workers_rejection.limit, 0);
    }

    #[test]
    fn per_worker_limit_multiplication_saturates() {
        let mut queue = PolicyQueue::new(profile(&format!(
            r#"
default_policy_family: capped
uncached_isl_buckets:
  - min_tokens: 0
    bucket: all
policy_classes:
  - name: capped
    policy_family: capped
    cache_bucket: all
    quantum: 1
    request_queue_limit_per_worker: {}
"#,
            usize::MAX
        )));
        queue
            .enqueue(0, 2, QueueSnapshot::new(1, 0), 0.0, 0.0, 0, "queued")
            .unwrap();
    }

    #[test]
    fn fcfs_and_wspt_order_only_within_each_class() {
        let mut queue = PolicyQueue::new(profile(
            r#"
default_policy_family: fcfs
uncached_isl_buckets:
  - min_tokens: 0
    bucket: all
policy_classes:
  - name: fcfs
    policy_family: fcfs
    cache_bucket: all
    queue_policy: fcfs
    quantum: 50
  - name: wspt
    policy_family: wspt
    cache_bucket: all
    queue_policy: wspt
    quantum: 50
"#,
        ));
        queue
            .enqueue(0, 1, QueueSnapshot::new(50, 0), 0.0, 0.0, 0, "fcfs-long")
            .unwrap();
        queue
            .enqueue(0, 1, QueueSnapshot::new(1, 0), 1.0, 0.0, 0, "fcfs-short")
            .unwrap();
        queue
            .enqueue(1, 1, QueueSnapshot::new(50, 0), 0.0, 0.0, 0, "wspt-long")
            .unwrap();
        queue
            .enqueue(1, 1, QueueSnapshot::new(1, 0), 1.0, 0.0, 0, "wspt-short")
            .unwrap();

        let first = queue.pop_next(|_, _, _| true).unwrap();
        let second = queue.pop_next(|_, _, _| true).unwrap();
        assert_eq!(first.into_payload(), "fcfs-long");
        assert_eq!(second.into_payload(), "wspt-short");
    }

    #[test]
    fn drr_weights_progress_and_skips_blocked_classes_without_credit() {
        let mut queue = PolicyQueue::new(profile(
            r#"
default_policy_family: slow
uncached_isl_buckets:
  - min_tokens: 0
    bucket: all
policy_classes:
  - name: slow
    policy_family: slow
    cache_bucket: all
    quantum: 1
  - name: fast
    policy_family: fast
    cache_bucket: all
    quantum: 3
"#,
        ));
        for index in 0..6 {
            queue
                .enqueue(0, 1, QueueSnapshot::new(1, 0), index as f64, 0.0, 0, "slow")
                .unwrap();
            queue
                .enqueue(1, 1, QueueSnapshot::new(1, 0), index as f64, 0.0, 0, "fast")
                .unwrap();
        }

        let mut first_six = Vec::new();
        for _ in 0..6 {
            first_six.push(queue.pop_next(|_, _, _| true).unwrap().into_payload());
        }
        assert!(first_six.iter().filter(|value| **value == "fast").count() >= 3);

        let blocked_deficit = queue.classes[1].deficit;
        let slow = queue.pop_next(|class, _, _| class == 0).unwrap();
        assert_eq!(slow.into_payload(), "slow");
        assert_eq!(queue.classes[1].deficit, blocked_deficit);
    }

    #[test]
    fn drr_serves_exact_quantum_ratio_for_equal_cost_backlogs() {
        let mut queue = PolicyQueue::new(profile(
            r#"
default_policy_family: one
uncached_isl_buckets:
  - min_tokens: 0
    bucket: all
policy_classes:
  - name: one
    policy_family: one
    cache_bucket: all
    quantum: 1
  - name: three
    policy_family: three
    cache_bucket: all
    quantum: 3
"#,
        ));
        for index in 0..20 {
            queue
                .enqueue(0, 1, QueueSnapshot::new(1, 0), index as f64, 0.0, 0, "one")
                .unwrap();
        }
        for index in 0..60 {
            queue
                .enqueue(
                    1,
                    1,
                    QueueSnapshot::new(1, 0),
                    index as f64,
                    0.0,
                    0,
                    "three",
                )
                .unwrap();
        }

        let dispatches = (0..80)
            .map(|_| queue.pop_next(|_, _, _| true).unwrap().into_payload())
            .collect::<Vec<_>>();
        for round in dispatches.chunks_exact(4) {
            assert_eq!(round, ["one", "three", "three", "three"]);
        }
    }

    #[test]
    fn fully_blocked_ring_returns_without_accruing_deficit() {
        let mut queue = PolicyQueue::new(profile(
            r#"
default_policy_family: first
uncached_isl_buckets:
  - min_tokens: 0
    bucket: all
policy_classes:
  - name: first
    policy_family: first
    cache_bucket: all
    quantum: 7
  - name: second
    policy_family: second
    cache_bucket: all
    quantum: 11
"#,
        ));
        queue
            .enqueue(0, 1, QueueSnapshot::new(100, 0), 0.0, 0.0, 0, "first")
            .unwrap();
        queue
            .enqueue(1, 1, QueueSnapshot::new(100, 0), 0.0, 0.0, 0, "second")
            .unwrap();

        for _ in 0..10_000 {
            assert!(queue.pop_next(|_, _, _| false).is_none());
        }
        assert_eq!(queue.classes[0].deficit, 0);
        assert_eq!(queue.classes[1].deficit, 0);
    }

    #[test]
    fn oversized_heads_bulk_add_deficit_and_make_progress() {
        let mut queue = PolicyQueue::new(profile(
            r#"
default_policy_family: large
uncached_isl_buckets:
  - min_tokens: 0
    bucket: all
policy_classes:
  - name: large
    policy_family: large
    cache_bucket: all
    quantum: 4
  - name: blocked
    policy_family: blocked
    cache_bucket: all
    quantum: 100
"#,
        ));
        queue
            .enqueue(0, 1, QueueSnapshot::new(101, 0), 0.0, 0.0, 0, "large")
            .unwrap();
        queue
            .enqueue(1, 1, QueueSnapshot::new(1, 0), 0.0, 0.0, 0, "blocked")
            .unwrap();

        let popped = queue
            .pop_next(|class, _, _| class == 0)
            .expect("oversized request should make bounded progress");
        assert_eq!(popped.into_payload(), "large");
        assert_eq!(queue.pending_count(), 1);
    }
}