use std::collections::{HashMap, VecDeque};
#[inline]
fn xorshift64(state: &mut u64) -> u64 {
let mut x = *state;
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
*state = x;
x
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum BandwidthClass {
High,
Normal,
Low,
Background,
}
impl BandwidthClass {
pub fn weight(self) -> f64 {
match self {
BandwidthClass::High => 4.0,
BandwidthClass::Normal => 2.0,
BandwidthClass::Low => 1.0,
BandwidthClass::Background => 0.5,
}
}
pub fn priority(self) -> u32 {
match self {
BandwidthClass::High => 4,
BandwidthClass::Normal => 3,
BandwidthClass::Low => 2,
BandwidthClass::Background => 1,
}
}
}
#[derive(Debug, Clone, PartialEq)]
pub enum AllocationPolicy {
EqualShare,
WeightedFair,
MinGuarantee(u64),
MaxCapacity(u64),
PriorityQueue,
}
#[derive(Debug, Clone)]
pub struct BandwidthWindow {
samples: VecDeque<u64>,
prng_state: u64,
capacity: usize,
}
impl BandwidthWindow {
const DEFAULT_CAPACITY: usize = 10;
pub fn new() -> Self {
Self {
samples: VecDeque::with_capacity(Self::DEFAULT_CAPACITY),
prng_state: 0xcafe_babe_dead_beef,
capacity: Self::DEFAULT_CAPACITY,
}
}
pub fn push(&mut self, raw_bps: u64) {
let jitter_range = (raw_bps / 200).max(1);
let jitter = xorshift64(&mut self.prng_state) % jitter_range;
let jittered = if xorshift64(&mut self.prng_state) & 1 == 0 {
raw_bps.saturating_add(jitter)
} else {
raw_bps.saturating_sub(jitter)
};
if self.samples.len() == self.capacity {
self.samples.pop_front();
}
self.samples.push_back(jittered);
}
pub fn average_bps(&self) -> u64 {
if self.samples.is_empty() {
return 0;
}
let sum: u64 = self.samples.iter().sum();
sum / self.samples.len() as u64
}
pub fn peak_bps(&self) -> u64 {
self.samples.iter().copied().max().unwrap_or(0)
}
pub fn len(&self) -> usize {
self.samples.len()
}
pub fn is_empty(&self) -> bool {
self.samples.is_empty()
}
}
impl Default for BandwidthWindow {
fn default() -> Self {
Self::new()
}
}
#[derive(Debug, Clone)]
pub struct PeerBandwidthProfile {
pub peer_id: String,
pub allocated_bps: u64,
pub used_bps: u64,
pub latency_ms: u64,
pub packet_loss_ppm: u32,
pub class: BandwidthClass,
pub last_updated: u64,
pub(crate) window: BandwidthWindow,
}
impl PeerBandwidthProfile {
fn new(peer_id: String, class: BandwidthClass, initial_bps: u64, now: u64) -> Self {
Self {
peer_id,
allocated_bps: initial_bps,
used_bps: 0,
latency_ms: 0,
packet_loss_ppm: 0,
class,
last_updated: now,
window: BandwidthWindow::new(),
}
}
}
#[derive(Debug, Clone)]
pub enum BandwidthEvent {
PeerAdded {
peer_id: String,
allocated_bps: u64,
timestamp: u64,
},
PeerRemoved { peer_id: String, timestamp: u64 },
BandwidthUpdated {
peer_id: String,
new_bps: u64,
timestamp: u64,
},
CongestionDetected {
peer_id: String,
loss_ppm: u32,
timestamp: u64,
},
AllocationRevised { timestamp: u64 },
}
impl BandwidthEvent {
pub fn timestamp(&self) -> u64 {
match self {
BandwidthEvent::PeerAdded { timestamp, .. } => *timestamp,
BandwidthEvent::PeerRemoved { timestamp, .. } => *timestamp,
BandwidthEvent::BandwidthUpdated { timestamp, .. } => *timestamp,
BandwidthEvent::CongestionDetected { timestamp, .. } => *timestamp,
BandwidthEvent::AllocationRevised { timestamp } => *timestamp,
}
}
}
#[derive(Debug, Clone)]
pub struct AllocatorConfig {
pub total_capacity_bps: u64,
pub min_allocation_bps: u64,
pub max_allocation_bps: u64,
pub policy: AllocationPolicy,
pub rebalance_interval_secs: u64,
pub congestion_threshold_ppm: u32,
}
impl AllocatorConfig {
pub fn validate(&self) -> Result<(), AllocatorError> {
if self.min_allocation_bps > self.max_allocation_bps {
return Err(AllocatorError::InvalidConfiguration(
"min_allocation_bps must be ≤ max_allocation_bps".to_string(),
));
}
if self.total_capacity_bps == 0 {
return Err(AllocatorError::InvalidConfiguration(
"total_capacity_bps must be > 0".to_string(),
));
}
if self.max_allocation_bps > self.total_capacity_bps {
return Err(AllocatorError::InvalidConfiguration(
"max_allocation_bps must be ≤ total_capacity_bps".to_string(),
));
}
Ok(())
}
}
#[derive(Debug, Clone)]
pub struct BandwidthStats {
pub total_allocated_bps: u64,
pub total_used_bps: u64,
pub utilization_pct: f64,
pub congested_peers: usize,
pub peer_count: usize,
pub fairness_index: f64,
}
#[derive(Debug, Clone, PartialEq, thiserror::Error)]
pub enum AllocatorError {
#[error("peer not found: {0}")]
PeerNotFound(String),
#[error("allocation of {0} bps exceeds remaining capacity")]
AllocationExceedsCapacity(u64),
#[error("invalid configuration: {0}")]
InvalidConfiguration(String),
#[error("congestion detected on peer {peer_id}: loss_ppm={loss_ppm}")]
CongestionDetected { peer_id: String, loss_ppm: u32 },
}
pub struct AdaptiveBandwidthAllocator {
config: AllocatorConfig,
peers: HashMap<String, PeerBandwidthProfile>,
events: VecDeque<BandwidthEvent>,
allocated_total: u64,
now_ms: u64,
}
impl AdaptiveBandwidthAllocator {
pub const MAX_EVENT_HISTORY: usize = 200;
pub fn new(config: AllocatorConfig) -> Self {
Self {
config,
peers: HashMap::new(),
events: VecDeque::with_capacity(Self::MAX_EVENT_HISTORY + 1),
allocated_total: 0,
now_ms: 1_000, }
}
pub fn advance_time(&mut self, delta_ms: u64) {
self.now_ms = self.now_ms.saturating_add(delta_ms);
}
pub fn add_peer(
&mut self,
peer_id: String,
class: BandwidthClass,
initial_bps: u64,
) -> Result<(), AllocatorError> {
let clamped = initial_bps.clamp(
self.config.min_allocation_bps,
self.config.max_allocation_bps,
);
let new_total = self.allocated_total.saturating_add(clamped);
if new_total > self.config.total_capacity_bps {
return Err(AllocatorError::AllocationExceedsCapacity(clamped));
}
let profile = PeerBandwidthProfile::new(peer_id.clone(), class, clamped, self.now_ms);
self.peers.insert(peer_id.clone(), profile);
self.allocated_total = new_total;
self.push_event(BandwidthEvent::PeerAdded {
peer_id,
allocated_bps: clamped,
timestamp: self.now_ms,
});
Ok(())
}
pub fn remove_peer(&mut self, peer_id: &str) -> Result<PeerBandwidthProfile, AllocatorError> {
let profile = self
.peers
.remove(peer_id)
.ok_or_else(|| AllocatorError::PeerNotFound(peer_id.to_string()))?;
self.allocated_total = self.allocated_total.saturating_sub(profile.allocated_bps);
self.push_event(BandwidthEvent::PeerRemoved {
peer_id: peer_id.to_string(),
timestamp: self.now_ms,
});
Ok(profile)
}
pub fn update_measurement(
&mut self,
peer_id: &str,
used_bps: u64,
latency_ms: u64,
packet_loss_ppm: u32,
) -> Result<BandwidthEvent, AllocatorError> {
let now = self.now_ms;
let profile = self
.peers
.get_mut(peer_id)
.ok_or_else(|| AllocatorError::PeerNotFound(peer_id.to_string()))?;
profile.used_bps = used_bps;
profile.latency_ms = latency_ms;
profile.packet_loss_ppm = packet_loss_ppm;
profile.last_updated = now;
profile.window.push(used_bps);
let event = if packet_loss_ppm > self.config.congestion_threshold_ppm {
BandwidthEvent::CongestionDetected {
peer_id: peer_id.to_string(),
loss_ppm: packet_loss_ppm,
timestamp: now,
}
} else {
BandwidthEvent::BandwidthUpdated {
peer_id: peer_id.to_string(),
new_bps: used_bps,
timestamp: now,
}
};
self.push_event(event.clone());
Ok(event)
}
pub fn reallocate(&mut self) -> Vec<BandwidthEvent> {
let mut out_events: Vec<BandwidthEvent> = Vec::new();
let now = self.now_ms;
if self.peers.is_empty() {
self.push_event(BandwidthEvent::AllocationRevised { timestamp: now });
out_events.push(BandwidthEvent::AllocationRevised { timestamp: now });
return out_events;
}
let new_allocs = self.compute_allocations();
let mut new_total: u64 = 0;
for (peer_id, alloc) in &new_allocs {
if let Some(profile) = self.peers.get_mut(peer_id) {
profile.allocated_bps = *alloc;
new_total = new_total.saturating_add(*alloc);
}
}
self.allocated_total = new_total;
let threshold = self.config.congestion_threshold_ppm;
let congested: Vec<(String, u32)> = self
.peers
.values()
.filter(|p| p.packet_loss_ppm > threshold)
.map(|p| (p.peer_id.clone(), p.packet_loss_ppm))
.collect();
for (peer_id, loss_ppm) in congested {
let ev = BandwidthEvent::CongestionDetected {
peer_id,
loss_ppm,
timestamp: now,
};
self.push_event(ev.clone());
out_events.push(ev);
}
let revised = BandwidthEvent::AllocationRevised { timestamp: now };
self.push_event(revised.clone());
out_events.push(revised);
out_events
}
pub fn get_allocation(&self, peer_id: &str) -> Result<u64, AllocatorError> {
self.peers
.get(peer_id)
.map(|p| p.allocated_bps)
.ok_or_else(|| AllocatorError::PeerNotFound(peer_id.to_string()))
}
pub fn stats(&self) -> BandwidthStats {
let peer_count = self.peers.len();
let total_allocated_bps: u64 = self.peers.values().map(|p| p.allocated_bps).sum();
let total_used_bps: u64 = self.peers.values().map(|p| p.used_bps).sum();
let utilization_pct = if total_allocated_bps == 0 {
0.0
} else {
(total_used_bps as f64 / total_allocated_bps as f64) * 100.0
};
let threshold = self.config.congestion_threshold_ppm;
let congested_peers = self
.peers
.values()
.filter(|p| p.packet_loss_ppm > threshold)
.count();
let fairness_index = Self::jain_fairness_index(
self.peers.values().map(|p| p.allocated_bps as f64),
peer_count,
);
BandwidthStats {
total_allocated_bps,
total_used_bps,
utilization_pct,
congested_peers,
peer_count,
fairness_index,
}
}
pub fn congested_peers(&self) -> Vec<String> {
let threshold = self.config.congestion_threshold_ppm;
self.peers
.values()
.filter(|p| p.packet_loss_ppm > threshold)
.map(|p| p.peer_id.clone())
.collect()
}
pub fn events_since(&self, since_ms: u64) -> Vec<BandwidthEvent> {
self.events
.iter()
.filter(|e| e.timestamp() >= since_ms)
.cloned()
.collect()
}
pub fn drain_events(&mut self) -> Vec<BandwidthEvent> {
self.events.drain(..).collect()
}
pub fn peer_profile(&self, peer_id: &str) -> Option<&PeerBandwidthProfile> {
self.peers.get(peer_id)
}
pub fn peer_count(&self) -> usize {
self.peers.len()
}
pub fn total_capacity_bps(&self) -> u64 {
self.config.total_capacity_bps
}
pub fn config(&self) -> &AllocatorConfig {
&self.config
}
fn push_event(&mut self, event: BandwidthEvent) {
if self.events.len() == Self::MAX_EVENT_HISTORY {
self.events.pop_front();
}
self.events.push_back(event);
}
fn compute_allocations(&self) -> HashMap<String, u64> {
match &self.config.policy {
AllocationPolicy::EqualShare => self.policy_equal_share(),
AllocationPolicy::WeightedFair => self.policy_weighted_fair(),
AllocationPolicy::MinGuarantee(min_n) => self.policy_min_guarantee(*min_n),
AllocationPolicy::MaxCapacity(max_n) => self.policy_max_capacity(*max_n),
AllocationPolicy::PriorityQueue => self.policy_priority_queue(),
}
}
fn policy_equal_share(&self) -> HashMap<String, u64> {
let n = self.peers.len() as u64;
let share = self.config.total_capacity_bps.checked_div(n).unwrap_or(0);
let clamped = share.clamp(
self.config.min_allocation_bps,
self.config.max_allocation_bps,
);
self.peers.keys().map(|id| (id.clone(), clamped)).collect()
}
fn policy_weighted_fair(&self) -> HashMap<String, u64> {
let total_weight: f64 = self.peers.values().map(|p| p.class.weight()).sum();
if total_weight == 0.0 {
return self.policy_equal_share();
}
self.peers
.values()
.map(|p| {
let raw = (self.config.total_capacity_bps as f64 * p.class.weight() / total_weight)
as u64;
let alloc = raw.clamp(
self.config.min_allocation_bps,
self.config.max_allocation_bps,
);
(p.peer_id.clone(), alloc)
})
.collect()
}
fn policy_min_guarantee(&self, min_n: u64) -> HashMap<String, u64> {
let peer_count = self.peers.len() as u64;
if peer_count == 0 {
return HashMap::new();
}
let effective_min = min_n
.max(self.config.min_allocation_bps)
.min(self.config.max_allocation_bps);
let guaranteed_total = effective_min.saturating_mul(peer_count);
let remainder = self
.config
.total_capacity_bps
.saturating_sub(guaranteed_total);
let bonus_per_peer = remainder / peer_count;
self.peers
.keys()
.map(|id| {
let alloc = (effective_min + bonus_per_peer).min(self.config.max_allocation_bps);
(id.clone(), alloc)
})
.collect()
}
fn policy_max_capacity(&self, max_n: u64) -> HashMap<String, u64> {
let peer_count = self.peers.len() as u64;
if peer_count == 0 {
return HashMap::new();
}
let effective_max = max_n.min(self.config.max_allocation_bps);
let equal = (self.config.total_capacity_bps / peer_count).min(effective_max);
let clamped = equal.clamp(self.config.min_allocation_bps, effective_max);
self.peers.keys().map(|id| (id.clone(), clamped)).collect()
}
fn policy_priority_queue(&self) -> HashMap<String, u64> {
let mut ordered: Vec<&PeerBandwidthProfile> = self.peers.values().collect();
ordered.sort_by(|a, b| {
b.class
.priority()
.cmp(&a.class.priority())
.then_with(|| a.peer_id.cmp(&b.peer_id))
});
let mut remaining = self.config.total_capacity_bps;
let mut allocs: HashMap<String, u64> = HashMap::new();
for profile in &ordered {
let grant = remaining.min(self.config.max_allocation_bps);
let grant = grant.max(self.config.min_allocation_bps);
let grant = if grant > remaining { remaining } else { grant };
allocs.insert(profile.peer_id.clone(), grant);
remaining = remaining.saturating_sub(grant);
}
for profile in &ordered {
let entry = allocs.entry(profile.peer_id.clone()).or_insert(0);
if *entry == 0 {
*entry = self.config.min_allocation_bps;
}
}
allocs
}
fn jain_fairness_index(allocations: impl Iterator<Item = f64>, n: usize) -> f64 {
if n == 0 {
return 1.0;
}
let (sum, sum_sq) = allocations.fold((0.0_f64, 0.0_f64), |(s, sq), x| (s + x, sq + x * x));
if sum_sq == 0.0 {
return 1.0;
}
(sum * sum) / (n as f64 * sum_sq)
}
}
#[cfg(test)]
mod tests {
use super::*;
fn default_config() -> AllocatorConfig {
AllocatorConfig {
total_capacity_bps: 100_000_000, min_allocation_bps: 1_000_000, max_allocation_bps: 50_000_000, policy: AllocationPolicy::EqualShare,
rebalance_interval_secs: 30,
congestion_threshold_ppm: 10_000, }
}
fn alloc_with_policy(policy: AllocationPolicy) -> AdaptiveBandwidthAllocator {
let mut cfg = default_config();
cfg.policy = policy;
AdaptiveBandwidthAllocator::new(cfg)
}
fn populated_allocator(n: usize, policy: AllocationPolicy) -> AdaptiveBandwidthAllocator {
let mut a = alloc_with_policy(policy);
for i in 0..n {
let class = match i % 4 {
0 => BandwidthClass::High,
1 => BandwidthClass::Normal,
2 => BandwidthClass::Low,
_ => BandwidthClass::Background,
};
a.add_peer(format!("peer-{i}"), class, 5_000_000)
.expect("test: failed to add peer in populated_allocator");
}
a
}
#[test]
fn test_xorshift64_non_zero_output() {
let mut state = 12345u64;
let v = xorshift64(&mut state);
assert_ne!(v, 0);
assert_ne!(state, 12345u64);
}
#[test]
fn test_xorshift64_deterministic() {
let mut s1 = 99999u64;
let mut s2 = 99999u64;
assert_eq!(xorshift64(&mut s1), xorshift64(&mut s2));
assert_eq!(s1, s2);
}
#[test]
fn test_xorshift64_different_states_differ() {
let mut s1 = 1u64;
let mut s2 = 2u64;
assert_ne!(xorshift64(&mut s1), xorshift64(&mut s2));
}
#[test]
fn test_bandwidth_class_weights() {
assert_eq!(BandwidthClass::High.weight(), 4.0);
assert_eq!(BandwidthClass::Normal.weight(), 2.0);
assert_eq!(BandwidthClass::Low.weight(), 1.0);
assert_eq!(BandwidthClass::Background.weight(), 0.5);
}
#[test]
fn test_bandwidth_class_priority_ordering() {
assert!(BandwidthClass::High.priority() > BandwidthClass::Normal.priority());
assert!(BandwidthClass::Normal.priority() > BandwidthClass::Low.priority());
assert!(BandwidthClass::Low.priority() > BandwidthClass::Background.priority());
}
#[test]
fn test_window_empty_returns_zero() {
let w = BandwidthWindow::new();
assert_eq!(w.average_bps(), 0);
assert_eq!(w.peak_bps(), 0);
assert!(w.is_empty());
}
#[test]
fn test_window_single_sample() {
let mut w = BandwidthWindow::new();
w.push(1_000_000);
assert!(!w.is_empty());
assert_eq!(w.len(), 1);
let avg = w.average_bps();
assert!((990_000..=1_010_000).contains(&avg));
}
#[test]
fn test_window_evicts_oldest_at_capacity() {
let mut w = BandwidthWindow::new();
for _ in 0..12 {
w.push(5_000_000);
}
assert_eq!(w.len(), 10);
}
#[test]
fn test_window_peak_is_max_sample() {
let mut w = BandwidthWindow::new();
for v in [1_000_000u64, 2_000_000, 3_000_000] {
w.push(v);
}
assert!(w.peak_bps() >= 2_900_000);
}
#[test]
fn test_window_average_convergence() {
let mut w = BandwidthWindow::new();
let target = 10_000_000u64;
for _ in 0..10 {
w.push(target);
}
let avg = w.average_bps();
assert!((avg as i64 - target as i64).abs() < 100_001);
}
#[test]
fn test_add_peer_success() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
assert!(a
.add_peer("p1".to_string(), BandwidthClass::Normal, 5_000_000)
.is_ok());
assert_eq!(a.peer_count(), 1);
}
#[test]
fn test_add_peer_clamps_to_max() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
a.add_peer("p1".to_string(), BandwidthClass::High, 999_000_000)
.expect("test: add high-bandwidth peer");
assert_eq!(
a.get_allocation("p1").expect("test: get allocation for p1"),
50_000_000
);
}
#[test]
fn test_add_peer_clamps_to_min() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
a.add_peer("p1".to_string(), BandwidthClass::Background, 1)
.expect("test: add background peer");
assert_eq!(
a.get_allocation("p1")
.expect("test: get allocation for background peer"),
1_000_000
);
}
#[test]
fn test_add_peer_exceeds_capacity() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
a.add_peer("p1".to_string(), BandwidthClass::High, 40_000_000)
.expect("test: add first high peer");
a.add_peer("p2".to_string(), BandwidthClass::High, 40_000_000)
.expect("test: add second high peer");
let err = a
.add_peer("p3".to_string(), BandwidthClass::Normal, 40_000_000)
.unwrap_err();
assert!(matches!(err, AllocatorError::AllocationExceedsCapacity(_)));
}
#[test]
fn test_add_peer_emits_event() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
a.add_peer("p1".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add normal peer for event test");
let events = a.drain_events();
assert_eq!(events.len(), 1);
assert!(matches!(events[0], BandwidthEvent::PeerAdded { .. }));
}
#[test]
fn test_remove_peer_success() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
a.add_peer("p1".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add peer before remove");
let profile = a.remove_peer("p1").expect("test: remove existing peer");
assert_eq!(profile.peer_id, "p1");
assert_eq!(a.peer_count(), 0);
}
#[test]
fn test_remove_peer_releases_capacity() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
a.add_peer("p1".to_string(), BandwidthClass::High, 40_000_000)
.expect("test: add p1 before release test");
a.add_peer("p2".to_string(), BandwidthClass::High, 40_000_000)
.expect("test: add p2 before release test");
a.remove_peer("p1")
.expect("test: remove p1 to release capacity");
assert!(a
.add_peer("p3".to_string(), BandwidthClass::High, 40_000_000)
.is_ok());
}
#[test]
fn test_remove_peer_not_found() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
let err = a.remove_peer("ghost").unwrap_err();
assert!(matches!(err, AllocatorError::PeerNotFound(_)));
}
#[test]
fn test_remove_peer_emits_event() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
a.add_peer("p1".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add peer before remove event test");
a.drain_events();
a.remove_peer("p1")
.expect("test: remove peer to trigger event");
let events = a.drain_events();
assert!(matches!(events[0], BandwidthEvent::PeerRemoved { .. }));
}
#[test]
fn test_update_measurement_normal() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
a.add_peer("p1".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add peer before measurement update");
let ev = a
.update_measurement("p1", 4_000_000, 20, 100)
.expect("test: update measurement for p1");
assert!(matches!(ev, BandwidthEvent::BandwidthUpdated { .. }));
}
#[test]
fn test_update_measurement_congestion_detected() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
a.add_peer("p1".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add peer before congestion test");
let ev = a
.update_measurement("p1", 2_000_000, 100, 50_000)
.expect("test: update measurement with high loss");
assert!(matches!(ev, BandwidthEvent::CongestionDetected { .. }));
}
#[test]
fn test_update_measurement_not_found() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
let err = a.update_measurement("ghost", 1_000, 5, 0).unwrap_err();
assert!(matches!(err, AllocatorError::PeerNotFound(_)));
}
#[test]
fn test_update_measurement_updates_window() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
a.add_peer("p1".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add peer before window test");
for _ in 0..5 {
a.update_measurement("p1", 4_000_000, 10, 0)
.expect("test: update measurement in loop");
}
let profile = a
.peer_profile("p1")
.expect("test: get peer profile after window updates");
assert_eq!(profile.window.len(), 5);
}
#[test]
fn test_update_measurement_updates_latency_and_loss() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
a.add_peer("p1".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add peer before latency/loss test");
a.update_measurement("p1", 3_000_000, 42, 500)
.expect("test: update measurement with latency and loss");
let p = a
.peer_profile("p1")
.expect("test: get peer profile for latency/loss check");
assert_eq!(p.latency_ms, 42);
assert_eq!(p.packet_loss_ppm, 500);
}
#[test]
fn test_reallocate_equal_share_basic() {
let mut a = alloc_with_policy(AllocationPolicy::EqualShare);
a.add_peer("p1".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add p1 for equal share test");
a.add_peer("p2".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add p2 for equal share test");
a.reallocate();
assert_eq!(
a.get_allocation("p1")
.expect("test: get p1 allocation after equal share reallocate"),
50_000_000
);
assert_eq!(
a.get_allocation("p2")
.expect("test: get p2 allocation after equal share reallocate"),
50_000_000
);
}
#[test]
fn test_reallocate_equal_share_zero_peers_is_safe() {
let mut a = alloc_with_policy(AllocationPolicy::EqualShare);
let events = a.reallocate();
assert!(!events.is_empty());
assert!(matches!(
events.last().expect("test: events list must not be empty"),
BandwidthEvent::AllocationRevised { .. }
));
}
#[test]
fn test_reallocate_equal_share_clamped_to_min() {
let mut cfg = default_config();
cfg.total_capacity_bps = 5_000_000; cfg.min_allocation_bps = 1_000_000;
cfg.max_allocation_bps = 5_000_000;
let mut a = AdaptiveBandwidthAllocator::new(cfg);
for i in 0..10 {
let _ = a.add_peer(format!("p{i}"), BandwidthClass::Low, 1_000_000);
}
a.reallocate();
for i in 0..a.peer_count() {
if let Ok(alloc) = a.get_allocation(&format!("p{i}")) {
assert!(alloc >= 1_000_000);
}
}
}
#[test]
fn test_reallocate_weighted_fair_respects_weight_order() {
let mut a = alloc_with_policy(AllocationPolicy::WeightedFair);
a.add_peer("high".to_string(), BandwidthClass::High, 5_000_000)
.expect("test: add high-priority peer for weight order test");
a.add_peer("low".to_string(), BandwidthClass::Low, 5_000_000)
.expect("test: add low-priority peer for weight order test");
a.reallocate();
let high_alloc = a
.get_allocation("high")
.expect("test: get allocation for high priority peer");
let low_alloc = a
.get_allocation("low")
.expect("test: get allocation for low priority peer");
assert!(
high_alloc > low_alloc,
"high priority must get more than low"
);
}
#[test]
fn test_reallocate_weighted_fair_sum_bounded_by_capacity() {
let mut a = populated_allocator(5, AllocationPolicy::WeightedFair);
a.reallocate();
let total: u64 = (0..5)
.map(|i| {
a.get_allocation(&format!("peer-{i}"))
.expect("test: get allocation for peer in weighted fair sum")
})
.sum();
assert!(total <= 100_000_000);
}
#[test]
fn test_reallocate_weighted_fair_single_peer() {
let mut a = alloc_with_policy(AllocationPolicy::WeightedFair);
a.add_peer("only".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add single peer for weighted fair test");
a.reallocate();
assert_eq!(
a.get_allocation("only")
.expect("test: get allocation for sole weighted fair peer"),
50_000_000
);
}
#[test]
fn test_reallocate_min_guarantee_basic() {
let mut a = alloc_with_policy(AllocationPolicy::MinGuarantee(5_000_000));
a.add_peer("p1".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add p1 for min guarantee basic test");
a.add_peer("p2".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add p2 for min guarantee basic test");
a.reallocate();
assert!(
a.get_allocation("p1")
.expect("test: get allocation for p1 in min guarantee basic")
>= 5_000_000
);
assert!(
a.get_allocation("p2")
.expect("test: get allocation for p2 in min guarantee basic")
>= 5_000_000
);
}
#[test]
fn test_reallocate_min_guarantee_distributes_surplus() {
let mut a = alloc_with_policy(AllocationPolicy::MinGuarantee(2_000_000));
a.add_peer("p1".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add p1 for min guarantee surplus test");
a.add_peer("p2".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add p2 for min guarantee surplus test");
a.reallocate();
assert_eq!(
a.get_allocation("p1")
.expect("test: get allocation for p1 after surplus distribution"),
50_000_000
);
}
#[test]
fn test_reallocate_min_guarantee_honors_config_min() {
let mut a = alloc_with_policy(AllocationPolicy::MinGuarantee(500_000));
a.add_peer("p1".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add p1 for min guarantee honors config min test");
a.reallocate();
assert!(
a.get_allocation("p1")
.expect("test: get allocation for p1 with config min enforcement")
>= 1_000_000
);
}
#[test]
fn test_reallocate_max_capacity_caps_peers() {
let mut a = alloc_with_policy(AllocationPolicy::MaxCapacity(10_000_000));
a.add_peer("p1".to_string(), BandwidthClass::High, 5_000_000)
.expect("test: add p1 for max capacity caps test");
a.add_peer("p2".to_string(), BandwidthClass::High, 5_000_000)
.expect("test: add p2 for max capacity caps test");
a.reallocate();
assert!(
a.get_allocation("p1")
.expect("test: get allocation for p1 under max capacity cap")
<= 10_000_000
);
assert!(
a.get_allocation("p2")
.expect("test: get allocation for p2 under max capacity cap")
<= 10_000_000
);
}
#[test]
fn test_reallocate_max_capacity_zero_peers_safe() {
let mut a = alloc_with_policy(AllocationPolicy::MaxCapacity(10_000_000));
a.reallocate();
assert_eq!(a.peer_count(), 0);
}
#[test]
fn test_reallocate_priority_queue_high_gets_more() {
let mut a = alloc_with_policy(AllocationPolicy::PriorityQueue);
a.add_peer("bg".to_string(), BandwidthClass::Background, 5_000_000)
.expect("test: add background peer for priority queue high-gets-more test");
a.add_peer("hi".to_string(), BandwidthClass::High, 5_000_000)
.expect("test: add high priority peer for priority queue high-gets-more test");
a.reallocate();
let hi = a
.get_allocation("hi")
.expect("test: get allocation for high priority peer in priority queue");
let bg = a
.get_allocation("bg")
.expect("test: get allocation for background peer in priority queue");
assert!(hi >= bg, "high priority should get >= background");
}
#[test]
fn test_reallocate_priority_queue_no_negative_allocations() {
let mut a = alloc_with_policy(AllocationPolicy::PriorityQueue);
for i in 0..8 {
a.add_peer(format!("p{i}"), BandwidthClass::High, 5_000_000)
.expect("test: add high-priority peer in no-negative-allocations test");
}
a.reallocate();
for i in 0..8 {
assert!(
a.get_allocation(&format!("p{i}"))
.expect("test: get allocation for peer in no-negative-allocations check")
>= 1_000_000
);
}
}
#[test]
fn test_reallocate_priority_queue_emits_revised() {
let mut a = populated_allocator(3, AllocationPolicy::PriorityQueue);
a.drain_events();
let events = a.reallocate();
assert!(events
.iter()
.any(|e| matches!(e, BandwidthEvent::AllocationRevised { .. })));
}
#[test]
fn test_get_allocation_not_found() {
let a = AdaptiveBandwidthAllocator::new(default_config());
let err = a.get_allocation("nobody").unwrap_err();
assert!(matches!(err, AllocatorError::PeerNotFound(_)));
}
#[test]
fn test_stats_empty_allocator() {
let a = AdaptiveBandwidthAllocator::new(default_config());
let s = a.stats();
assert_eq!(s.peer_count, 0);
assert_eq!(s.total_allocated_bps, 0);
assert_eq!(s.fairness_index, 1.0);
}
#[test]
fn test_stats_utilization_pct() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
a.add_peer("p1".to_string(), BandwidthClass::Normal, 10_000_000)
.expect("test: add p1 for stats utilization test");
a.update_measurement("p1", 5_000_000, 10, 0)
.expect("test: record measurement for p1 in utilization test");
let s = a.stats();
assert!((s.utilization_pct - 50.0).abs() < 1.0);
}
#[test]
fn test_stats_congested_count() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
a.add_peer("p1".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add p1 for stats congested count test");
a.add_peer("p2".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add p2 for stats congested count test");
a.update_measurement("p1", 3_000_000, 100, 50_000)
.expect("test: record congested measurement for p1"); a.update_measurement("p2", 4_000_000, 10, 0)
.expect("test: record healthy measurement for p2"); let s = a.stats();
assert_eq!(s.congested_peers, 1);
}
#[test]
fn test_stats_fairness_perfect() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
a.add_peer("p1".to_string(), BandwidthClass::Normal, 10_000_000)
.expect("test: add p1 for stats fairness perfect test");
a.add_peer("p2".to_string(), BandwidthClass::Normal, 10_000_000)
.expect("test: add p2 for stats fairness perfect test");
let s = a.stats();
assert!((s.fairness_index - 1.0).abs() < 1e-9);
}
#[test]
fn test_stats_fairness_unequal() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
a.add_peer("p1".to_string(), BandwidthClass::High, 10_000_000)
.expect("test: add p1 High class for stats fairness unequal test");
a.add_peer("p2".to_string(), BandwidthClass::High, 10_000_000)
.expect("test: add p2 High class for stats fairness unequal test");
a.reallocate();
let s = a.stats();
assert!(s.fairness_index > 0.0 && s.fairness_index <= 1.0);
}
#[test]
fn test_congested_peers_empty() {
let a = AdaptiveBandwidthAllocator::new(default_config());
assert!(a.congested_peers().is_empty());
}
#[test]
fn test_congested_peers_detection() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
a.add_peer("good".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add good peer for congested peers detection test");
a.add_peer("bad".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add bad peer for congested peers detection test");
a.update_measurement("bad", 1_000_000, 200, 100_000)
.expect("test: record high-latency measurement for bad peer");
let congested = a.congested_peers();
assert_eq!(congested.len(), 1);
assert_eq!(congested[0], "bad");
}
#[test]
fn test_congested_peers_at_threshold_not_flagged() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
a.add_peer("p1".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add p1 for congested peers at threshold test");
a.update_measurement("p1", 3_000_000, 50, 10_000)
.expect("test: record measurement at congestion threshold for p1");
assert!(a.congested_peers().is_empty());
}
#[test]
fn test_events_since_filters_by_timestamp() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
a.add_peer("p1".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add p1 for events since timestamp filter test");
a.advance_time(100);
let mid_ts = a.now_ms;
a.add_peer("p2".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add p2 after time advance for events since timestamp filter test");
let recent = a.events_since(mid_ts);
assert_eq!(recent.len(), 1);
}
#[test]
fn test_drain_events_clears_history() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
a.add_peer("p1".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add p1 for drain events clears history test");
assert!(!a.drain_events().is_empty());
assert!(a.drain_events().is_empty());
}
#[test]
fn test_event_history_bounded_at_200() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
a.add_peer("p1".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add p1 for event history bounded test");
for _ in 0..300 {
a.update_measurement("p1", 4_000_000, 10, 0)
.expect("test: record repeated measurement for event history bounded test");
}
let events = a.events_since(0);
assert!(events.len() <= AdaptiveBandwidthAllocator::MAX_EVENT_HISTORY);
}
#[test]
fn test_events_since_zero_returns_all() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
a.add_peer("p1".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add p1 for events since zero returns all test");
a.add_peer("p2".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add p2 for events since zero returns all test");
let all = a.events_since(0);
assert_eq!(all.len(), 2);
}
#[test]
fn test_config_validate_ok() {
assert!(default_config().validate().is_ok());
}
#[test]
fn test_config_validate_min_gt_max_fails() {
let mut cfg = default_config();
cfg.min_allocation_bps = 60_000_000;
cfg.max_allocation_bps = 10_000_000;
assert!(cfg.validate().is_err());
}
#[test]
fn test_config_validate_zero_capacity_fails() {
let mut cfg = default_config();
cfg.total_capacity_bps = 0;
assert!(cfg.validate().is_err());
}
#[test]
fn test_config_validate_max_gt_capacity_fails() {
let mut cfg = default_config();
cfg.max_allocation_bps = cfg.total_capacity_bps + 1;
assert!(cfg.validate().is_err());
}
#[test]
fn test_jain_fairness_all_equal() {
let fi =
AdaptiveBandwidthAllocator::jain_fairness_index([10.0, 10.0, 10.0].iter().copied(), 3);
assert!((fi - 1.0).abs() < 1e-9);
}
#[test]
fn test_jain_fairness_one_takes_all() {
let fi =
AdaptiveBandwidthAllocator::jain_fairness_index([100.0, 0.0, 0.0].iter().copied(), 3);
assert!((fi - 1.0 / 3.0).abs() < 1e-9);
}
#[test]
fn test_jain_fairness_zero_peers() {
let fi = AdaptiveBandwidthAllocator::jain_fairness_index(std::iter::empty(), 0);
assert_eq!(fi, 1.0);
}
#[test]
fn test_jain_fairness_single_peer() {
let fi = AdaptiveBandwidthAllocator::jain_fairness_index([42.0].iter().copied(), 1);
assert!((fi - 1.0).abs() < 1e-9);
}
#[test]
fn test_reallocate_emits_allocation_revised() {
let mut a = populated_allocator(3, AllocationPolicy::EqualShare);
a.drain_events();
let events = a.reallocate();
assert!(events
.iter()
.any(|e| matches!(e, BandwidthEvent::AllocationRevised { .. })));
}
#[test]
fn test_single_peer_equal_share() {
let mut a = alloc_with_policy(AllocationPolicy::EqualShare);
a.add_peer("only".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add single peer for equal share test");
a.reallocate();
assert_eq!(
a.get_allocation("only")
.expect("test: get allocation for single peer"),
50_000_000
);
}
#[test]
fn test_over_capacity_add_rejected() {
let mut cfg = default_config();
cfg.total_capacity_bps = 10_000_000; cfg.max_allocation_bps = 10_000_000;
let mut a = AdaptiveBandwidthAllocator::new(cfg);
a.add_peer("p1".to_string(), BandwidthClass::Normal, 10_000_000)
.expect("test: add p1 within capacity for over-capacity rejection test");
let err = a
.add_peer("p2".to_string(), BandwidthClass::Normal, 1_000_000)
.unwrap_err();
assert!(matches!(err, AllocatorError::AllocationExceedsCapacity(_)));
}
#[test]
fn test_reallocate_after_remove_rebalances_correctly() {
let mut a = alloc_with_policy(AllocationPolicy::EqualShare);
a.add_peer("p1".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add p1 for rebalance after remove test");
a.add_peer("p2".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add p2 for rebalance after remove test");
a.add_peer("p3".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add p3 to be removed for rebalance test");
a.remove_peer("p3")
.expect("test: remove p3 to trigger rebalance");
a.reallocate();
assert_eq!(
a.get_allocation("p1")
.expect("test: get allocation for p1 after rebalance"),
50_000_000
);
assert_eq!(
a.get_allocation("p2")
.expect("test: get allocation for p2 after rebalance"),
50_000_000
);
}
#[test]
fn test_congestion_clears_after_update() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
a.add_peer("p1".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add p1 for congestion clear test");
a.update_measurement("p1", 1_000_000, 200, 100_000)
.expect("test: update measurement to mark p1 as congested"); assert_eq!(a.congested_peers().len(), 1);
a.update_measurement("p1", 4_000_000, 10, 0)
.expect("test: update measurement to clear p1 congestion"); assert_eq!(a.congested_peers().len(), 0);
}
#[test]
fn test_advance_time_affects_events() {
let mut a = AdaptiveBandwidthAllocator::new(default_config());
a.add_peer("p1".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add p1 before time advance");
a.advance_time(500);
a.add_peer("p2".to_string(), BandwidthClass::Normal, 5_000_000)
.expect("test: add p2 after time advance");
let after_advance = a.events_since(a.now_ms);
assert_eq!(after_advance.len(), 1);
}
#[test]
fn test_multiple_reallocate_calls_stable() {
let mut a = populated_allocator(4, AllocationPolicy::WeightedFair);
for _ in 0..5 {
a.reallocate();
}
for i in 0..4 {
let alloc = a
.get_allocation(&format!("peer-{i}"))
.expect("test: get allocation for peer in stability test");
let cfg = a.config();
assert!(alloc >= cfg.min_allocation_bps);
assert!(alloc <= cfg.max_allocation_bps);
}
}
}