decdn-node 0.1.0

use super::*;
use std::net::Ipv4Addr;

fn metrics() -> Arc<Metrics> {
    Arc::new(Metrics::new())
}

fn peer(byte: u8) -> NodeId {
    NodeId::from_bytes([byte; 32])
}

fn ip(a: u8, b: u8, c: u8, d: u8) -> IpAddr {
    IpAddr::V4(Ipv4Addr::new(a, b, c, d))
}

fn strict_cfg() -> DhtRateLimitConfig {
    DhtRateLimitConfig {
        per_peer_rate_per_sec: 1.0,
        per_peer_burst: 1,
        per_ip_rate_per_sec: 1.0,
        per_ip_burst: 1,
        global_rate_per_sec: 1.0,
        global_burst: 1,
        max_tracked_per_ip: 4096,
        max_tracked_per_peer: 4096,
    }
}

#[test]
fn admits_first_request_under_default_caps() {
    let lim = DhtRateLimiter::new(&DhtRateLimitConfig::default(), metrics());
    assert_eq!(lim.check(&peer(1), Some(ip(10, 0, 0, 1))), Ok(()));
}

#[test]
fn per_peer_layer_rejects_after_burst() {
    let mut cfg = strict_cfg();
    // Loosen the global + per-IP layers so we're testing per-peer alone.
    cfg.global_burst = u32::MAX;
    cfg.global_rate_per_sec = 1e9;
    cfg.per_ip_burst = u32::MAX;
    cfg.per_ip_rate_per_sec = 1e9;
    let lim = DhtRateLimiter::new(&cfg, metrics());
    let p = peer(1);
    let i = Some(ip(10, 0, 0, 1));
    assert_eq!(lim.check(&p, i), Ok(()));
    assert_eq!(lim.check(&p, i), Err(DhtRejectLayer::PerPeer));
}

#[test]
fn per_ip_layer_rejects_after_burst() {
    let mut cfg = strict_cfg();
    cfg.global_burst = u32::MAX;
    cfg.global_rate_per_sec = 1e9;
    cfg.per_peer_burst = u32::MAX;
    cfg.per_peer_rate_per_sec = 1e9;
    let lim = DhtRateLimiter::new(&cfg, metrics());
    let i = Some(ip(10, 0, 0, 1));
    // Different peers from the same IP: first admits, second hits per-IP.
    assert_eq!(lim.check(&peer(1), i), Ok(()));
    assert_eq!(lim.check(&peer(2), i), Err(DhtRejectLayer::PerIp));
}

#[test]
fn global_layer_rejects_after_burst() {
    let mut cfg = strict_cfg();
    cfg.per_peer_burst = u32::MAX;
    cfg.per_peer_rate_per_sec = 1e9;
    cfg.per_ip_burst = u32::MAX;
    cfg.per_ip_rate_per_sec = 1e9;
    let lim = DhtRateLimiter::new(&cfg, metrics());
    // Different peer AND IP every time — only the global cap can fire.
    assert_eq!(lim.check(&peer(1), Some(ip(10, 0, 0, 1))), Ok(()));
    assert_eq!(
        lim.check(&peer(2), Some(ip(10, 0, 0, 2))),
        Err(DhtRejectLayer::Global)
    );
}

#[test]
fn cheapest_first_global_short_circuits_per_peer() {
    // burst=1 everywhere. The first request consumes the global bucket;
    // the second from the same peer+IP must be reported as global
    // (cheapest layer), not per-peer or per-ip.
    let lim = DhtRateLimiter::new(&strict_cfg(), metrics());
    assert_eq!(lim.check(&peer(1), Some(ip(10, 0, 0, 1))), Ok(()));
    assert_eq!(
        lim.check(&peer(1), Some(ip(10, 0, 0, 1))),
        Err(DhtRejectLayer::Global)
    );
}

#[test]
fn relay_only_connection_skips_per_ip_layer() {
    let mut cfg = strict_cfg();
    // Set per-IP very strict but everything else loose: a relay-only
    // peer (peer_ip = None) must still be admitted because per-IP is
    // skipped when no key is available.
    cfg.global_burst = u32::MAX;
    cfg.global_rate_per_sec = 1e9;
    cfg.per_peer_burst = u32::MAX;
    cfg.per_peer_rate_per_sec = 1e9;
    let lim = DhtRateLimiter::new(&cfg, metrics());
    for _ in 0..16 {
        assert_eq!(lim.check(&peer(1), None), Ok(()));
    }
}

// ---- #969: explicit evaluation-order coverage. ADR 005 §Probe rate
// limiting (and the mirroring ADR 022 §DHT Rate Limiting this limiter
// implements) specify cheapest-first ordering
// (global → per-IP → per-peer). ----

/// The global cap is the cheapest (first-checked) layer, so when both the
/// global bucket and the per-IP bucket are exhausted, the rejection must
/// be attributed to `Global` — never `PerIp`. This pins the
/// global-before-per-IP edge of the cheapest-first order specifically
/// (the existing `cheapest_first_global_short_circuits_per_peer` pins the
/// global-before-per-peer edge).
///
/// Construction: global burst=1 and per-IP burst=1 from the *same* IP but
/// *distinct* peers (so the per-peer layer never fires and can't be the
/// reported layer). The first request drains both global and per-IP; the
/// second would fail both, and global — checked first — must win.
#[test]
fn cheapest_first_global_fires_before_per_ip() {
    let mut cfg = strict_cfg();
    // global burst=1, per-IP burst=1 (both from strict_cfg); loosen
    // per-peer so it can never be the layer that fires. Pin the two
    // tested layers' refill to effectively zero (1e-9 req/sec) so a slow
    // CI scheduler can't refill a token mid-test and flake the second
    // `check`; burst=1 (from strict_cfg) is the only budget either gets.
    cfg.global_rate_per_sec = 1e-9;
    cfg.per_ip_rate_per_sec = 1e-9;
    cfg.per_peer_burst = u32::MAX;
    cfg.per_peer_rate_per_sec = 1e9;
    let lim = DhtRateLimiter::new(&cfg, metrics());
    let i = Some(ip(10, 0, 0, 1));
    // First request drains both the global and the per-IP bucket.
    assert_eq!(lim.check(&peer(1), i), Ok(()));
    // Second request from the SAME IP but a DIFFERENT peer would be
    // rejected by BOTH the global and the per-IP layer. Cheapest-first
    // ordering means global is consulted first and short-circuits, so
    // the reported layer must be `Global`, not `PerIp`.
    assert_eq!(lim.check(&peer(2), i), Err(DhtRejectLayer::Global));
}

/// The global rejection in `cheapest_first_global_fires_before_per_ip`
/// must increment the *global* counter only — a regression that consulted
/// per-IP first (or mis-attributed the counter) would bump
/// `decdn_dht_rate_limit_rejected_per_ip_total` instead. The
/// `Err(DhtRejectLayer::Global)` assertion alone can't catch a counter
/// mix-up, so confirm the attribution via the encoded scrape too.
#[test]
fn global_before_per_ip_attributes_rejection_to_global_counter() {
    let mut cfg = strict_cfg();
    // Pin the tested global + per-IP layers to no-refill (1e-9 req/sec) so
    // a slow CI scheduler can't refill a token between the two `check`
    // calls; burst=1 (from strict_cfg) is the entire budget.
    cfg.global_rate_per_sec = 1e-9;
    cfg.per_ip_rate_per_sec = 1e-9;
    cfg.per_peer_burst = u32::MAX;
    cfg.per_peer_rate_per_sec = 1e9;
    let metrics = metrics();
    let lim = DhtRateLimiter::new(&cfg, Arc::clone(&metrics));
    let i = Some(ip(10, 0, 0, 1));
    assert_eq!(lim.check(&peer(1), i), Ok(()));
    assert_eq!(lim.check(&peer(2), i), Err(DhtRejectLayer::Global));
    let text = metrics.encode().unwrap();
    assert!(
        text.contains("decdn_dht_rate_limit_rejected_global_total 1"),
        "global must be the counted layer when both global+per-IP would \
         reject; got:\n{text}"
    );
    assert!(
        text.contains("decdn_dht_rate_limit_rejected_per_ip_total 0"),
        "per-IP counter must NOT move — global short-circuits first; \
         got:\n{text}"
    );
}

fn ip6(segments: [u16; 8]) -> IpAddr {
    IpAddr::V6(std::net::Ipv6Addr::new(
        segments[0],
        segments[1],
        segments[2],
        segments[3],
        segments[4],
        segments[5],
        segments[6],
        segments[7],
    ))
}

#[test]
fn per_ip_layer_masks_ipv6_to_slash_64() {
    // #841: two distinct IPv6 addresses inside the same /64 must share one
    // per-IP bucket — otherwise an attacker rotating within a /64 mints a
    // fresh bucket per request and defeats the per-IP tier entirely.
    let mut cfg = strict_cfg();
    cfg.global_burst = u32::MAX;
    cfg.global_rate_per_sec = 1e9;
    cfg.per_peer_burst = u32::MAX;
    cfg.per_peer_rate_per_sec = 1e9;
    let lim = DhtRateLimiter::new(&cfg, metrics());
    let a = Some(ip6([0x2001, 0xdb8, 0, 0, 0, 0, 0, 1]));
    let b = Some(ip6([0x2001, 0xdb8, 0, 0, 0xffff, 0xffff, 0xffff, 0xfffe]));
    // Same /64 (only the host bits differ): second request hits per-IP.
    assert_eq!(lim.check(&peer(1), a), Ok(()));
    assert_eq!(lim.check(&peer(2), b), Err(DhtRejectLayer::PerIp));
    // A different /64 lands in a fresh bucket and is admitted.
    let c = Some(ip6([0x2001, 0xdb8, 0, 1, 0, 0, 0, 1]));
    assert_eq!(lim.check(&peer(3), c), Ok(()));
}

#[test]
fn rejection_increments_layer_metric_in_scrape() {
    // One unlabeled Counter per layer, matching the dispatch counters
    // (`dispatch_rejected_global` / `dispatch_rejected_per_source`) — the
    // sibling-counter convention settled in #1475, not a backend
    // limitation (`iroh_metrics` does support labels via `Family<L, M>`;
    // `probe_hold_unavailable` uses one). Operators recover the rolled-up
    // rate with
    // `sum(rate({__name__=~"decdn_dht_rate_limit_rejected_(per_peer|per_ip|global)_total"}[1m]))`;
    // per-layer alerts are unaffected. ADR 022 § Observability now
    // specifies this trio.
    let metrics = metrics();
    let lim = DhtRateLimiter::new(&strict_cfg(), Arc::clone(&metrics));
    lim.check(&peer(1), Some(ip(10, 0, 0, 1)))
        .expect("first ok");
    let _ = lim.check(&peer(1), Some(ip(10, 0, 0, 1)));
    let text = metrics.encode().unwrap();
    assert!(
        text.contains("decdn_dht_rate_limit_rejected_global_total 1"),
        "global rejection must appear in /metrics; got:\n{text}"
    );
}

#[test]
fn disabled_layer_admits_all() {
    let cfg = DhtRateLimitConfig {
        per_peer_rate_per_sec: 0.0,
        per_peer_burst: 0,
        per_ip_rate_per_sec: 0.0,
        per_ip_burst: 0,
        global_rate_per_sec: 0.0,
        global_burst: 0,
        max_tracked_per_ip: 0,
        max_tracked_per_peer: 0,
    };
    let lim = DhtRateLimiter::new(&cfg, metrics());
    for _ in 0..1000 {
        assert_eq!(lim.check(&peer(1), Some(ip(10, 0, 0, 1))), Ok(()));
    }
}

// ---- #648: two-stage batch token accounting (ADR 022 §Batch token
// accounting). `admit_batch_extra` is stage 2: after `check` has
// charged the stage-1 frame token, it consumes up to `extra` (= n-1)
// more units cheapest-first with short-circuit, exactly as `extra`
// separate `StoreRequest` admissions would. ----

#[test]
fn admit_batch_extra_grants_full_budget_when_permissive() {
    let lim = DhtRateLimiter::new(&DhtRateLimitConfig::default(), metrics());
    let p = peer(1);
    let i = Some(ip(10, 0, 0, 1));
    // Stage 1: the frame token (matches `serve`).
    assert_eq!(lim.check(&p, i), Ok(()));
    // Stage 2: a batch of n=10 wants n-1=9 extra; all granted under
    // the default burst caps (per-peer burst 40).
    assert_eq!(lim.admit_batch_extra(&p, i, 9), 9);
}

#[test]
fn admit_batch_extra_returns_zero_for_zero_extra() {
    // n=1 batch: extra = 0, nothing more to charge.
    let lim = DhtRateLimiter::new(&DhtRateLimitConfig::default(), metrics());
    assert_eq!(lim.admit_batch_extra(&peer(1), Some(ip(10, 0, 0, 1)), 0), 0);
}

#[test]
fn admit_batch_extra_bounded_by_per_peer_remaining() {
    // per-peer burst 5, everything else loose. Stage 1 burns 1
    // (4 left); a batch wanting 10 extra gets only the 4 remaining.
    let mut cfg = strict_cfg();
    cfg.per_peer_rate_per_sec = 1.0;
    cfg.per_peer_burst = 5;
    cfg.per_ip_burst = u32::MAX;
    cfg.per_ip_rate_per_sec = 1e9;
    cfg.global_burst = u32::MAX;
    cfg.global_rate_per_sec = 1e9;
    let lim = DhtRateLimiter::new(&cfg, metrics());
    let p = peer(1);
    let i = Some(ip(10, 0, 0, 1));
    assert_eq!(lim.check(&p, i), Ok(()));
    assert_eq!(lim.admit_batch_extra(&p, i, 10), 4);
    // Bucket now empty — a second batch gets nothing more.
    assert_eq!(lim.admit_batch_extra(&p, i, 10), 0);
}

#[test]
fn admit_batch_extra_bounded_by_min_across_layers() {
    // per-IP is the tightest layer (burst 3). Stage 1 burns 1 from
    // each (per-IP 2 left); the batch gets min remaining = 2.
    let mut cfg = strict_cfg();
    cfg.per_peer_rate_per_sec = 1e9;
    cfg.per_peer_burst = u32::MAX;
    cfg.per_ip_rate_per_sec = 1.0;
    cfg.per_ip_burst = 3;
    cfg.global_rate_per_sec = 1e9;
    cfg.global_burst = u32::MAX;
    let lim = DhtRateLimiter::new(&cfg, metrics());
    let p = peer(1);
    let i = Some(ip(10, 0, 0, 1));
    assert_eq!(lim.check(&p, i), Ok(()));
    assert_eq!(lim.admit_batch_extra(&p, i, 8), 2);
}

#[test]
fn admit_batch_extra_does_not_bump_rejection_counters() {
    // The partial-admit boundary is NOT a request rejection — the
    // frame was already admitted at stage 1. Draining the budget in
    // stage 2 must leave the rate-limit rejection counters at 0 so
    // operators don't see phantom rejections for a partially-admitted
    // batch.
    let mut cfg = strict_cfg();
    cfg.per_peer_rate_per_sec = 1.0;
    cfg.per_peer_burst = 2;
    cfg.per_ip_burst = u32::MAX;
    cfg.per_ip_rate_per_sec = 1e9;
    cfg.global_burst = u32::MAX;
    cfg.global_rate_per_sec = 1e9;
    let metrics = metrics();
    let lim = DhtRateLimiter::new(&cfg, Arc::clone(&metrics));
    let p = peer(1);
    let i = Some(ip(10, 0, 0, 1));
    assert_eq!(lim.check(&p, i), Ok(()));
    // extra=5 but only 1 token left after stage 1 → grants 1, then
    // the budget is exhausted for the rest (no counter bump).
    assert_eq!(lim.admit_batch_extra(&p, i, 5), 1);
    let text = metrics.encode().unwrap();
    for layer in ["per_peer", "per_ip", "global"] {
        assert!(
            text.contains(&format!("decdn_dht_rate_limit_rejected_{layer}_total 0")),
            "stage-2 partial admit must not bump the {layer} rejection counter:\n{text}"
        );
    }
}

#[test]
fn layer_label_strings_are_stable() {
    // Pinned because operators alert on these label values.
    assert_eq!(DhtRejectLayer::PerPeer.as_str(), "per_peer");
    assert_eq!(DhtRejectLayer::PerIp.as_str(), "per_ip");
    assert_eq!(DhtRejectLayer::Global.as_str(), "global");
}

// ---- #645: keyspace bound — periodic GC + lazy prune ----

#[test]
fn gc_per_ip_is_noop_when_layer_disabled() {
    // Disabled per-IP layer => no keyed map to prune. `gc_per_ip` must
    // return `None` (matches the dispatch-limiter
    // `gc_per_source_is_noop_when_layer_disabled` precedent).
    let mut cfg = strict_cfg();
    cfg.per_ip_rate_per_sec = 0.0;
    cfg.per_ip_burst = 0;
    let lim = DhtRateLimiter::new(&cfg, metrics());
    assert!(lim.gc_per_ip().is_none());
    assert_eq!(lim.per_ip_tracked(), 0);
}

#[test]
fn gc_per_peer_is_noop_when_layer_disabled() {
    let mut cfg = strict_cfg();
    cfg.per_peer_rate_per_sec = 0.0;
    cfg.per_peer_burst = 0;
    let lim = DhtRateLimiter::new(&cfg, metrics());
    assert!(lim.gc_per_peer().is_none());
    assert_eq!(lim.per_peer_tracked(), 0);
}

/// Fast refill (`rate=1000.0, burst=1`) so a 100ms sleep is enough
/// for `retain_recent` to drop every populated bucket. Mirrors
/// `dispatch.rs::gc_per_source_drops_refilled_buckets`.
#[tokio::test]
async fn gc_per_ip_drops_refilled_buckets() {
    let cfg = DhtRateLimitConfig {
        per_peer_rate_per_sec: 1e6,
        per_peer_burst: u32::MAX,
        per_ip_rate_per_sec: 1000.0,
        per_ip_burst: 1,
        global_rate_per_sec: 1e6,
        global_burst: u32::MAX,
        // 0 so lazy-prune in `check` is out of scope here — the GC
        // method is the only path that can drop refilled buckets.
        max_tracked_per_ip: 0,
        max_tracked_per_peer: 0,
    };
    let lim = DhtRateLimiter::new(&cfg, metrics());
    for i in 0..8 {
        let _ = lim.check(&peer(1), Some(ip(10, 0, 0, i)));
    }
    assert_eq!(lim.per_ip_tracked(), 8, "8 distinct IPs populated");
    tokio::time::sleep(Duration::from_millis(100)).await;
    let (before, after) = lim
        .gc_per_ip()
        .expect("per-IP layer enabled and CAS uncontended");
    assert_eq!(before, 8);
    assert_eq!(after, 0, "all buckets refilled and were dropped");
    assert_eq!(lim.per_ip_tracked(), 0);
}

#[tokio::test]
async fn gc_per_peer_drops_refilled_buckets() {
    let cfg = DhtRateLimitConfig {
        per_peer_rate_per_sec: 1000.0,
        per_peer_burst: 1,
        per_ip_rate_per_sec: 1e6,
        per_ip_burst: u32::MAX,
        global_rate_per_sec: 1e6,
        global_burst: u32::MAX,
        max_tracked_per_ip: 0,
        max_tracked_per_peer: 0,
    };
    let lim = DhtRateLimiter::new(&cfg, metrics());
    for byte in 0..8u8 {
        let _ = lim.check(&peer(byte), Some(ip(10, 0, 0, 1)));
    }
    assert_eq!(lim.per_peer_tracked(), 8);
    tokio::time::sleep(Duration::from_millis(100)).await;
    let (before, after) = lim
        .gc_per_peer()
        .expect("per-peer layer enabled and CAS uncontended");
    assert_eq!(before, 8);
    assert_eq!(after, 0);
    assert_eq!(lim.per_peer_tracked(), 0);
}

/// `cap_per_ip` bounds the per-IP map. 200 distinct IPs with a `cap`
/// of 16 should not exceed `cap + cap/10 + 1` (the `+1` covers
/// floor-division slack — the lazy prune fires at `len() > cap +
/// cap/10`, so the largest size we can transiently observe is `cap +
/// cap/10 + 1`). Fast refill so `retain_recent` can drop refilled
/// buckets between bursts.
#[tokio::test]
async fn lazy_prune_bounds_per_ip_under_flood() {
    let cap: usize = 16;
    let cfg = DhtRateLimitConfig {
        per_peer_rate_per_sec: 1e6,
        per_peer_burst: u32::MAX,
        per_ip_rate_per_sec: 1000.0,
        per_ip_burst: 1,
        global_rate_per_sec: 1e6,
        global_burst: u32::MAX,
        max_tracked_per_ip: cap,
        max_tracked_per_peer: 0,
    };
    let metrics_handle = metrics();
    let lim = DhtRateLimiter::new(&cfg, Arc::clone(&metrics_handle));
    for i in 0..200u16 {
        let a = u8::try_from((i >> 8) & 0xff).unwrap_or(0);
        let b = u8::try_from(i & 0xff).unwrap_or(0);
        let _ = lim.check(&peer(1), Some(ip(10, 0, a, b)));
        tokio::time::sleep(Duration::from_millis(2)).await;
    }
    let final_size = lim.per_ip_tracked();
    let bound = cap.saturating_add(cap / 10).saturating_add(1);
    assert!(
        final_size <= bound,
        "expected per_ip_tracked <= {bound}, got {final_size}"
    );
    // The lazy-prune path in `check` must also bump the prune-sweep
    // counter — a regression that dropped the metric call from the shared
    // engine's lazy-prune path (`ThreeLayerRateLimiter::maybe_prune_per_ip`,
    // but kept it on `gc_per_ip`) would not be caught by
    // `gauge_and_sweep_counter_emit_in_scrape` since that test only
    // exercises the GC path.
    let text = metrics_handle.encode().unwrap();
    assert!(
        text.contains("decdn_dht_rate_limit_prune_sweeps_per_ip_total"),
        "per-IP prune-sweep counter missing from scrape:\n{text}"
    );
    assert!(
        !text.contains("decdn_dht_rate_limit_prune_sweeps_per_ip_total 0"),
        "per-IP prune-sweep counter must have fired at least once during the flood:\n{text}"
    );
}

/// Pins the `0 = unbounded` contract: 1000 distinct keys with cap=0
/// stay in the keyed map (no prune fires). Operators opting in to
/// the unbounded mode see the `tracing::warn!` from the resolver
/// (covered by the `decdn-common` config tests, not here).
#[test]
fn unbounded_when_cap_is_zero_per_ip() {
    let cfg = DhtRateLimitConfig {
        per_peer_rate_per_sec: 1e6,
        per_peer_burst: u32::MAX,
        per_ip_rate_per_sec: 1e6,
        per_ip_burst: u32::MAX,
        global_rate_per_sec: 1e6,
        global_burst: u32::MAX,
        max_tracked_per_ip: 0,
        max_tracked_per_peer: 0,
    };
    let lim = DhtRateLimiter::new(&cfg, metrics());
    for i in 0..1000u16 {
        let a = u8::try_from((i >> 8) & 0xff).unwrap_or(0);
        let b = u8::try_from(i & 0xff).unwrap_or(0);
        assert!(lim.check(&peer(1), Some(ip(10, 0, a, b))).is_ok());
    }
    assert_eq!(lim.per_ip_tracked(), 1000);
}

#[test]
fn unbounded_when_cap_is_zero_per_peer() {
    let cfg = DhtRateLimitConfig {
        per_peer_rate_per_sec: 1e6,
        per_peer_burst: u32::MAX,
        per_ip_rate_per_sec: 1e6,
        per_ip_burst: u32::MAX,
        global_rate_per_sec: 1e6,
        global_burst: u32::MAX,
        max_tracked_per_ip: 0,
        max_tracked_per_peer: 0,
    };
    let lim = DhtRateLimiter::new(&cfg, metrics());
    for i in 0..1000u16 {
        let mut id = [0u8; 32];
        id[0] = u8::try_from((i >> 8) & 0xff).unwrap_or(0);
        id[1] = u8::try_from(i & 0xff).unwrap_or(0);
        assert!(
            lim.check(&NodeId::from_bytes(id), Some(ip(10, 0, 0, 1)))
                .is_ok()
        );
    }
    assert_eq!(lim.per_peer_tracked(), 1000);
}

#[tokio::test]
async fn lazy_prune_bounds_per_peer_under_flood() {
    let cap: usize = 16;
    let cfg = DhtRateLimitConfig {
        per_peer_rate_per_sec: 1000.0,
        per_peer_burst: 1,
        per_ip_rate_per_sec: 1e6,
        per_ip_burst: u32::MAX,
        global_rate_per_sec: 1e6,
        global_burst: u32::MAX,
        max_tracked_per_ip: 0,
        max_tracked_per_peer: cap,
    };
    let metrics_handle = metrics();
    let lim = DhtRateLimiter::new(&cfg, Arc::clone(&metrics_handle));
    for i in 0..200u16 {
        // Distinct NodeId per iteration — pack `i` into bytes 0..2.
        let mut id = [0u8; 32];
        id[0] = u8::try_from((i >> 8) & 0xff).unwrap_or(0);
        id[1] = u8::try_from(i & 0xff).unwrap_or(0);
        let _ = lim.check(&NodeId::from_bytes(id), Some(ip(10, 0, 0, 1)));
        tokio::time::sleep(Duration::from_millis(2)).await;
    }
    let final_size = lim.per_peer_tracked();
    let bound = cap.saturating_add(cap / 10).saturating_add(1);
    assert!(
        final_size <= bound,
        "expected per_peer_tracked <= {bound}, got {final_size}"
    );
    let text = metrics_handle.encode().unwrap();
    assert!(
        text.contains("decdn_dht_rate_limit_prune_sweeps_per_peer_total"),
        "per-peer prune-sweep counter missing from scrape:\n{text}"
    );
    assert!(
        !text.contains("decdn_dht_rate_limit_prune_sweeps_per_peer_total 0"),
        "per-peer prune-sweep counter must have fired at least once during the flood:\n{text}"
    );
}

/// Both sweep counters and gauges must appear in the encoded scrape
/// after `gc_per_ip` + `gc_per_peer` run on an enabled limiter. Uses
/// the same `encode().contains(...)` pattern as
/// `rejection_increments_layer_metric_in_scrape`.
#[tokio::test]
async fn gauge_and_sweep_counter_emit_in_scrape() {
    let metrics_handle = metrics();
    let cfg = DhtRateLimitConfig {
        per_peer_rate_per_sec: 1000.0,
        per_peer_burst: 1,
        per_ip_rate_per_sec: 1000.0,
        per_ip_burst: 1,
        global_rate_per_sec: 1e6,
        global_burst: u32::MAX,
        max_tracked_per_ip: 0,
        max_tracked_per_peer: 0,
    };
    let lim = DhtRateLimiter::new(&cfg, Arc::clone(&metrics_handle));
    for i in 0..4u8 {
        let mut id = [0u8; 32];
        id[0] = i;
        let _ = lim.check(&NodeId::from_bytes(id), Some(ip(10, 0, 0, i)));
    }
    tokio::time::sleep(Duration::from_millis(100)).await;
    assert!(lim.gc_per_ip().is_some());
    assert!(lim.gc_per_peer().is_some());
    let text = metrics_handle.encode().unwrap();
    assert!(
        text.contains("decdn_dht_rate_limit_prune_sweeps_per_ip_total 1"),
        "per-IP sweep counter missing from scrape:\n{text}"
    );
    assert!(
        text.contains("decdn_dht_rate_limit_prune_sweeps_per_peer_total 1"),
        "per-peer sweep counter missing from scrape:\n{text}"
    );
    assert!(
        text.contains("decdn_dht_rate_limit_tracked_per_ip 0"),
        "per-IP tracked gauge missing from scrape:\n{text}"
    );
    assert!(
        text.contains("decdn_dht_rate_limit_tracked_per_peer 0"),
        "per-peer tracked gauge missing from scrape:\n{text}"
    );
}