Skip to main content

ipfrs_network/
network_qos_manager.rs

1//! Network Quality-of-Service (QoS) Manager.
2//!
3//! Classifies traffic into four priority tiers, enforces SLA contracts via
4//! latency and throughput checks, and provides both strict-priority and
5//! weighted-round-robin scheduling with per-class bandwidth guarantees.
6
7use std::collections::{HashMap, VecDeque};
8
9// ────────────────────────────────────────────────────────────────────────────
10// Traffic classification
11// ────────────────────────────────────────────────────────────────────────────
12
13/// Four-tier traffic classification for QoS scheduling.
14#[derive(Clone, Debug, PartialEq, Eq, Hash)]
15pub enum TrafficClass {
16    /// Highest priority: real-time streams, consensus messages.
17    RealTime,
18    /// Second priority: interactive user sessions, RPC calls.
19    Interactive,
20    /// Third priority: bulk file transfers, sync operations.
21    BulkData,
22    /// Lowest priority: background maintenance, metrics gossip.
23    Background,
24}
25
26impl TrafficClass {
27    /// Numeric priority — higher value is serviced first (1..=4).
28    pub fn priority(&self) -> u8 {
29        match self {
30            TrafficClass::RealTime => 4,
31            TrafficClass::Interactive => 3,
32            TrafficClass::BulkData => 2,
33            TrafficClass::Background => 1,
34        }
35    }
36
37    /// Fraction of total bandwidth allocated to this class (sum = 1.0).
38    pub fn bandwidth_share(&self) -> f64 {
39        match self {
40            TrafficClass::RealTime => 0.4,
41            TrafficClass::Interactive => 0.3,
42            TrafficClass::BulkData => 0.2,
43            TrafficClass::Background => 0.1,
44        }
45    }
46
47    /// Construct from a raw priority byte; returns `None` for unknown values.
48    pub fn from_priority(p: u8) -> Option<Self> {
49        match p {
50            4 => Some(TrafficClass::RealTime),
51            3 => Some(TrafficClass::Interactive),
52            2 => Some(TrafficClass::BulkData),
53            1 => Some(TrafficClass::Background),
54            _ => None,
55        }
56    }
57}
58
59// ────────────────────────────────────────────────────────────────────────────
60// Core data types
61// ────────────────────────────────────────────────────────────────────────────
62
63/// A network packet with QoS metadata, ready for priority queuing.
64#[derive(Clone, Debug)]
65pub struct QoSPacket {
66    /// Unique packet identifier within this manager instance.
67    pub id: u64,
68    /// Traffic class that determines scheduling tier.
69    pub class: TrafficClass,
70    /// Payload size in bytes; used for bandwidth accounting.
71    pub size_bytes: usize,
72    /// Originating or destination peer identifier.
73    pub peer_id: String,
74    /// Monotonic timestamp (ms) at which the packet entered the queue.
75    pub enqueued_at: u64,
76}
77
78/// SLA contract for a single traffic class.
79#[derive(Clone, Debug)]
80pub struct SLASpec {
81    /// The class this spec governs.
82    pub class: TrafficClass,
83    /// Maximum acceptable average queueing latency in milliseconds.
84    pub max_latency_ms: u64,
85    /// Minimum guaranteed throughput in bits per second.
86    pub min_throughput_bps: u64,
87    /// Maximum acceptable jitter (latency variance) in milliseconds.
88    pub max_jitter_ms: u64,
89}
90
91/// A recorded SLA violation event.
92#[derive(Clone, Debug)]
93pub struct SLAViolation {
94    /// Traffic class that violated its SLA.
95    pub class: TrafficClass,
96    /// Human-readable metric name (e.g. `"avg_wait_ms"`).
97    pub metric: String,
98    /// Observed value.
99    pub actual: f64,
100    /// Allowed limit.
101    pub limit: f64,
102    /// Monotonic timestamp (ms) when the violation was detected.
103    pub timestamp: u64,
104}
105
106/// Snapshot of per-class queue statistics.
107#[derive(Clone, Debug)]
108pub struct QueueMetrics {
109    /// Traffic class these metrics belong to.
110    pub class: TrafficClass,
111    /// Number of packets currently in the queue.
112    pub queued_packets: usize,
113    /// Total byte count of all enqueued packets.
114    pub total_bytes: usize,
115    /// Exponentially-weighted moving average wait time (ms); α = 0.1.
116    pub avg_wait_ms: f64,
117    /// Total packets dropped from this queue since creation.
118    pub dropped_packets: u64,
119}
120
121/// Aggregate QoS statistics across all classes.
122#[derive(Clone, Debug)]
123pub struct QoSStats {
124    /// Lifetime enqueued packet count.
125    pub total_enqueued: u64,
126    /// Lifetime dequeued packet count.
127    pub total_dequeued: u64,
128    /// Lifetime dropped packet count.
129    pub total_dropped: u64,
130    /// Fraction of enqueued packets that were dropped (0.0..=1.0).
131    pub drop_rate: f64,
132    /// Number of priority queues that contain at least one packet.
133    pub active_queues: usize,
134    /// Total violations recorded in the violations log.
135    pub total_violations: usize,
136}
137
138// ────────────────────────────────────────────────────────────────────────────
139// Configuration
140// ────────────────────────────────────────────────────────────────────────────
141
142/// Configuration for the `NetworkQoSManager`.
143#[derive(Clone, Debug)]
144pub struct QoSConfig {
145    /// Maximum combined packet count across all queues before drops occur.
146    pub max_queue_size: usize,
147    /// Aggregate link capacity in bits per second.
148    pub total_bandwidth_bps: u64,
149    /// Per-class SLA specifications; can be empty (no SLA checking).
150    pub sla_specs: Vec<SLASpec>,
151    /// When `true`, always service the highest non-empty priority queue
152    /// (strict priority).  When `false`, use weighted round-robin.
153    pub enable_strict_priority: bool,
154}
155
156impl Default for QoSConfig {
157    fn default() -> Self {
158        Self {
159            max_queue_size: 10_000,
160            total_bandwidth_bps: 100_000_000, // 100 Mbps
161            sla_specs: Vec::new(),
162            enable_strict_priority: false,
163        }
164    }
165}
166
167// ────────────────────────────────────────────────────────────────────────────
168// Internal round-robin state
169// ────────────────────────────────────────────────────────────────────────────
170
171/// Tracks the weighted-round-robin deficit counter for one priority queue.
172#[derive(Debug, Default)]
173struct WrrState {
174    /// Accumulated "credits" not yet consumed by a dequeue.
175    deficit: f64,
176    /// Quantum of credits added each round (proportional to bandwidth_share).
177    quantum: f64,
178}
179
180// ────────────────────────────────────────────────────────────────────────────
181// Manager
182// ────────────────────────────────────────────────────────────────────────────
183
184/// Quality-of-Service manager: priority queuing, SLA enforcement, and
185/// bandwidth-share scheduling for the IPFRS network layer.
186pub struct NetworkQoSManager {
187    /// User-supplied configuration.
188    pub config: QoSConfig,
189    /// Priority queues keyed by `TrafficClass::priority()` (1..=4).
190    pub queues: HashMap<u8, VecDeque<QoSPacket>>,
191    /// Per-priority queue metrics keyed by priority (1..=4).
192    pub metrics: HashMap<u8, QueueMetrics>,
193    /// Bounded log of observed SLA violations (max 1 000 entries).
194    pub violations: VecDeque<SLAViolation>,
195    /// Lifetime counter of successfully enqueued packets.
196    pub total_enqueued: u64,
197    /// Lifetime counter of successfully dequeued packets.
198    pub total_dequeued: u64,
199    /// Lifetime counter of dropped packets.
200    pub total_dropped: u64,
201
202    // ── internal scheduling state ──────────────────────────────────────────
203    /// Weighted-round-robin deficit state keyed by priority (1..=4).
204    wrr: HashMap<u8, WrrState>,
205    /// Current round-robin position (priority value 1..=4).
206    wrr_cursor: u8,
207}
208
209/// Ordered priority levels from highest to lowest (used for iteration).
210const PRIORITIES: [u8; 4] = [4, 3, 2, 1];
211
212impl NetworkQoSManager {
213    // ── construction ────────────────────────────────────────────────────────
214
215    /// Create a new manager with the given configuration.
216    ///
217    /// Initialises four priority queues (keyed 1..=4) and pre-fills WRR
218    /// quanta proportional to each class's `bandwidth_share`.
219    pub fn new(config: QoSConfig) -> Self {
220        let mut queues = HashMap::with_capacity(4);
221        let mut metrics = HashMap::with_capacity(4);
222        let mut wrr = HashMap::with_capacity(4);
223
224        for &prio in &PRIORITIES {
225            let class = TrafficClass::from_priority(prio).unwrap_or(TrafficClass::Background);
226
227            queues.insert(prio, VecDeque::new());
228            metrics.insert(
229                prio,
230                QueueMetrics {
231                    class: class.clone(),
232                    queued_packets: 0,
233                    total_bytes: 0,
234                    avg_wait_ms: 0.0,
235                    dropped_packets: 0,
236                },
237            );
238
239            // WRR quantum = bandwidth_share * 1000 (normalised credits).
240            wrr.insert(
241                prio,
242                WrrState {
243                    deficit: 0.0,
244                    quantum: class.bandwidth_share() * 1000.0,
245                },
246            );
247        }
248
249        Self {
250            config,
251            queues,
252            metrics,
253            violations: VecDeque::new(),
254            total_enqueued: 0,
255            total_dequeued: 0,
256            total_dropped: 0,
257            wrr,
258            wrr_cursor: 4, // start at highest priority
259        }
260    }
261
262    // ── internal helpers ─────────────────────────────────────────────────────
263
264    /// Total number of packets across all queues.
265    fn total_queued_count(&self) -> usize {
266        self.queues.values().map(|q| q.len()).sum()
267    }
268
269    /// Update the `queued_packets` and `total_bytes` fields of the metrics
270    /// entry for `priority` to match the actual queue state.
271    fn sync_queue_metrics(&mut self, priority: u8) {
272        if let Some(q) = self.queues.get(&priority) {
273            let count = q.len();
274            let bytes: usize = q.iter().map(|p| p.size_bytes).sum();
275            if let Some(m) = self.metrics.get_mut(&priority) {
276                m.queued_packets = count;
277                m.total_bytes = bytes;
278            }
279        }
280    }
281
282    /// Apply an EWMA update (α = 0.1) to `avg_wait_ms` for the queue at
283    /// `priority` using the supplied observed wait duration.
284    fn update_avg_wait(&mut self, priority: u8, wait_ms: f64) {
285        const ALPHA: f64 = 0.1;
286        if let Some(m) = self.metrics.get_mut(&priority) {
287            if m.avg_wait_ms == 0.0 {
288                // Bootstrap: first sample sets the value directly.
289                m.avg_wait_ms = wait_ms;
290            } else {
291                m.avg_wait_ms = ALPHA * wait_ms + (1.0 - ALPHA) * m.avg_wait_ms;
292            }
293        }
294    }
295
296    // ── public API ───────────────────────────────────────────────────────────
297
298    /// Enqueue a packet into the priority queue determined by its class.
299    ///
300    /// If the total queue occupancy is at or above `max_queue_size`:
301    /// 1. Try to drop the **oldest** Background packet (priority 1).
302    /// 2. If Background is already empty, drop the **newest** Background
303    ///    packet (i.e. the one just being enqueued is discarded instead).
304    /// 3. If neither relieves enough space, return `false`.
305    ///
306    /// Returns `true` when the packet was successfully enqueued.
307    pub fn enqueue(&mut self, packet: QoSPacket, _now: u64) -> bool {
308        // Make room if over capacity.
309        if self.total_queued_count() >= self.config.max_queue_size {
310            // Drop oldest Background packet.
311            let dropped = self.drop_packet(1);
312            if !dropped {
313                // Background queue empty; if the incoming packet IS Background,
314                // discard it directly.
315                if packet.class.priority() == 1 {
316                    self.total_dropped += 1;
317                    if let Some(m) = self.metrics.get_mut(&1) {
318                        m.dropped_packets += 1;
319                    }
320                    return false;
321                }
322                // Otherwise we cannot make room.
323                return false;
324            }
325        }
326
327        let priority = packet.class.priority();
328        if let Some(q) = self.queues.get_mut(&priority) {
329            q.push_back(packet);
330            self.total_enqueued += 1;
331            self.sync_queue_metrics(priority);
332            true
333        } else {
334            false
335        }
336    }
337
338    /// Drop the **oldest** packet from the queue at `priority`.
339    ///
340    /// Returns `true` if a packet was removed, `false` if the queue was empty.
341    pub fn drop_packet(&mut self, priority: u8) -> bool {
342        if let Some(q) = self.queues.get_mut(&priority) {
343            if q.pop_front().is_some() {
344                self.total_dropped += 1;
345                if let Some(m) = self.metrics.get_mut(&priority) {
346                    m.dropped_packets += 1;
347                }
348                self.sync_queue_metrics(priority);
349                return true;
350            }
351        }
352        false
353    }
354
355    /// Dequeue the next packet according to the configured scheduling policy.
356    ///
357    /// * **Strict priority** (`enable_strict_priority = true`): always pick
358    ///   from the highest non-empty priority queue.
359    /// * **Weighted round-robin** (`enable_strict_priority = false`): iterate
360    ///   round-robin through classes; add `quantum` credits each visit and
361    ///   dequeue when credits ≥ packet size.  Falls back to the highest
362    ///   non-empty queue if no class accumulates enough credits.
363    ///
364    /// Returns `None` if all queues are empty.
365    pub fn dequeue(&mut self, now: u64) -> Option<QoSPacket> {
366        if self.config.enable_strict_priority {
367            self.dequeue_strict(now)
368        } else {
369            self.dequeue_wrr(now)
370        }
371    }
372
373    /// Strict-priority dequeue: highest non-empty queue wins.
374    fn dequeue_strict(&mut self, now: u64) -> Option<QoSPacket> {
375        for &prio in &PRIORITIES {
376            if let Some(q) = self.queues.get_mut(&prio) {
377                if let Some(pkt) = q.pop_front() {
378                    let wait_ms = now.saturating_sub(pkt.enqueued_at) as f64;
379                    self.total_dequeued += 1;
380                    self.sync_queue_metrics(prio);
381                    self.update_avg_wait(prio, wait_ms);
382                    return Some(pkt);
383                }
384            }
385        }
386        None
387    }
388
389    /// Weighted round-robin dequeue.
390    ///
391    /// Each class earns `quantum` credits per round and spends them on packet
392    /// sizes.  We iterate up to 4 full rounds to avoid infinite loops when all
393    /// queues are empty.
394    fn dequeue_wrr(&mut self, now: u64) -> Option<QoSPacket> {
395        // Quick exit when everything is empty.
396        if self.total_queued_count() == 0 {
397            return None;
398        }
399
400        // Try WRR: up to 4 * 4 = 16 cursor advances to find a serviced packet.
401        for _ in 0..16 {
402            let prio = self.wrr_cursor;
403
404            // Add quantum credits to this queue.
405            if let Some(state) = self.wrr.get_mut(&prio) {
406                state.deficit += state.quantum;
407            }
408
409            // Attempt to dequeue if we have enough credits.
410            let head_size = self
411                .queues
412                .get(&prio)
413                .and_then(|q| q.front())
414                .map(|p| p.size_bytes as f64);
415
416            if let Some(sz) = head_size {
417                let credits = self.wrr.get(&prio).map(|s| s.deficit).unwrap_or(0.0);
418                if credits >= sz {
419                    // Dequeue the packet.
420                    if let Some(q) = self.queues.get_mut(&prio) {
421                        if let Some(pkt) = q.pop_front() {
422                            if let Some(state) = self.wrr.get_mut(&prio) {
423                                state.deficit -= sz;
424                            }
425                            let wait_ms = now.saturating_sub(pkt.enqueued_at) as f64;
426                            self.total_dequeued += 1;
427                            self.sync_queue_metrics(prio);
428                            self.update_avg_wait(prio, wait_ms);
429                            // Advance cursor for next call.
430                            self.advance_cursor();
431                            return Some(pkt);
432                        }
433                    }
434                }
435            }
436
437            // Advance cursor regardless.
438            self.advance_cursor();
439        }
440
441        // WRR couldn't find a winner (all queues small enough credits);
442        // fall back to strict priority to prevent starvation.
443        self.dequeue_strict(now)
444    }
445
446    /// Advance the WRR cursor in decreasing priority order (4 → 3 → 2 → 1 → 4).
447    fn advance_cursor(&mut self) {
448        self.wrr_cursor = if self.wrr_cursor > 1 {
449            self.wrr_cursor - 1
450        } else {
451            4
452        };
453    }
454
455    /// Check every configured SLA spec against current queue metrics.
456    ///
457    /// Any violation is appended to the internal violations log (bounded to
458    /// 1 000 entries, oldest evicted on overflow).  The set of violations
459    /// discovered during **this call** is returned.
460    pub fn check_sla(&mut self, now: u64) -> Vec<SLAViolation> {
461        let specs = self.config.sla_specs.clone();
462        let mut found = Vec::new();
463
464        for spec in &specs {
465            let prio = spec.class.priority();
466            let avg_wait = self
467                .metrics
468                .get(&prio)
469                .map(|m| m.avg_wait_ms)
470                .unwrap_or(0.0);
471
472            if avg_wait > spec.max_latency_ms as f64 {
473                let v = SLAViolation {
474                    class: spec.class.clone(),
475                    metric: "avg_wait_ms".to_string(),
476                    actual: avg_wait,
477                    limit: spec.max_latency_ms as f64,
478                    timestamp: now,
479                };
480                found.push(v);
481            }
482        }
483
484        for v in &found {
485            if self.violations.len() >= 1_000 {
486                self.violations.pop_front();
487            }
488            self.violations.push_back(v.clone());
489        }
490
491        found
492    }
493
494    /// Return queue metrics for a specific traffic class, or `None` if the
495    /// class is not tracked (should not happen with a correctly initialised
496    /// manager).
497    pub fn queue_metrics(&self, class: &TrafficClass) -> Option<&QueueMetrics> {
498        self.metrics.get(&class.priority())
499    }
500
501    /// Return metrics for all four priority queues ordered highest-first.
502    pub fn all_metrics(&self) -> Vec<&QueueMetrics> {
503        PRIORITIES
504            .iter()
505            .filter_map(|p| self.metrics.get(p))
506            .collect()
507    }
508
509    /// Total byte count across all queued packets.
510    pub fn total_queued_bytes(&self) -> usize {
511        self.queues
512            .values()
513            .flat_map(|q| q.iter())
514            .map(|p| p.size_bytes)
515            .sum()
516    }
517
518    /// Immutable reference to the bounded violations log.
519    pub fn violations_log(&self) -> &VecDeque<SLAViolation> {
520        &self.violations
521    }
522
523    /// Aggregate statistics snapshot.
524    pub fn qos_stats(&self) -> QoSStats {
525        let active_queues = self.queues.values().filter(|q| !q.is_empty()).count();
526
527        let drop_rate = if self.total_enqueued > 0 {
528            self.total_dropped as f64 / self.total_enqueued as f64
529        } else {
530            0.0
531        };
532
533        QoSStats {
534            total_enqueued: self.total_enqueued,
535            total_dequeued: self.total_dequeued,
536            total_dropped: self.total_dropped,
537            drop_rate,
538            active_queues,
539            total_violations: self.violations.len(),
540        }
541    }
542
543    /// Reset all per-class EWMA wait metrics to zero (useful for test
544    /// isolation or after a reconfiguration event).
545    pub fn reset_metrics(&mut self) {
546        for m in self.metrics.values_mut() {
547            m.avg_wait_ms = 0.0;
548        }
549    }
550
551    /// Drain all packets from all queues and return the total count removed.
552    pub fn flush_all(&mut self) -> usize {
553        let mut count = 0;
554        for q in self.queues.values_mut() {
555            count += q.len();
556            q.clear();
557        }
558        for prio in &[1u8, 2, 3, 4] {
559            self.sync_queue_metrics(*prio);
560        }
561        count
562    }
563}
564
565// ────────────────────────────────────────────────────────────────────────────
566// Tests
567// ────────────────────────────────────────────────────────────────────────────
568
569#[cfg(test)]
570mod tests {
571    use super::{NetworkQoSManager, QoSConfig, QoSPacket, SLASpec, TrafficClass};
572
573    // ── helpers ──────────────────────────────────────────────────────────────
574
575    fn default_manager() -> NetworkQoSManager {
576        NetworkQoSManager::new(QoSConfig::default())
577    }
578
579    fn make_packet(id: u64, class: TrafficClass, size: usize, enqueued_at: u64) -> QoSPacket {
580        QoSPacket {
581            id,
582            class,
583            size_bytes: size,
584            peer_id: format!("peer-{id}"),
585            enqueued_at,
586        }
587    }
588
589    // ── TrafficClass unit tests ───────────────────────────────────────────────
590
591    #[test]
592    fn test_traffic_class_priority_values() {
593        assert_eq!(TrafficClass::RealTime.priority(), 4);
594        assert_eq!(TrafficClass::Interactive.priority(), 3);
595        assert_eq!(TrafficClass::BulkData.priority(), 2);
596        assert_eq!(TrafficClass::Background.priority(), 1);
597    }
598
599    #[test]
600    fn test_traffic_class_bandwidth_shares_sum_to_one() {
601        let sum = TrafficClass::RealTime.bandwidth_share()
602            + TrafficClass::Interactive.bandwidth_share()
603            + TrafficClass::BulkData.bandwidth_share()
604            + TrafficClass::Background.bandwidth_share();
605        assert!((sum - 1.0).abs() < 1e-9, "shares sum to {sum}");
606    }
607
608    #[test]
609    fn test_traffic_class_bandwidth_share_values() {
610        assert!((TrafficClass::RealTime.bandwidth_share() - 0.4).abs() < 1e-9);
611        assert!((TrafficClass::Interactive.bandwidth_share() - 0.3).abs() < 1e-9);
612        assert!((TrafficClass::BulkData.bandwidth_share() - 0.2).abs() < 1e-9);
613        assert!((TrafficClass::Background.bandwidth_share() - 0.1).abs() < 1e-9);
614    }
615
616    #[test]
617    fn test_from_priority_round_trip() {
618        for &p in &[1u8, 2, 3, 4] {
619            let cls = TrafficClass::from_priority(p).expect("known priority");
620            assert_eq!(cls.priority(), p);
621        }
622    }
623
624    #[test]
625    fn test_from_priority_unknown_returns_none() {
626        assert!(TrafficClass::from_priority(0).is_none());
627        assert!(TrafficClass::from_priority(5).is_none());
628        assert!(TrafficClass::from_priority(255).is_none());
629    }
630
631    // ── QoSConfig defaults ───────────────────────────────────────────────────
632
633    #[test]
634    fn test_config_defaults() {
635        let cfg = QoSConfig::default();
636        assert_eq!(cfg.max_queue_size, 10_000);
637        assert_eq!(cfg.total_bandwidth_bps, 100_000_000);
638        assert!(!cfg.enable_strict_priority);
639        assert!(cfg.sla_specs.is_empty());
640    }
641
642    // ── Construction ─────────────────────────────────────────────────────────
643
644    #[test]
645    fn test_new_initialises_four_queues() {
646        let mgr = default_manager();
647        assert_eq!(mgr.queues.len(), 4);
648        assert_eq!(mgr.metrics.len(), 4);
649    }
650
651    #[test]
652    fn test_new_all_queues_empty() {
653        let mgr = default_manager();
654        for q in mgr.queues.values() {
655            assert!(q.is_empty());
656        }
657    }
658
659    #[test]
660    fn test_new_counters_zero() {
661        let mgr = default_manager();
662        assert_eq!(mgr.total_enqueued, 0);
663        assert_eq!(mgr.total_dequeued, 0);
664        assert_eq!(mgr.total_dropped, 0);
665    }
666
667    // ── Enqueue ───────────────────────────────────────────────────────────────
668
669    #[test]
670    fn test_enqueue_basic_returns_true() {
671        let mut mgr = default_manager();
672        let pkt = make_packet(1, TrafficClass::RealTime, 100, 0);
673        assert!(mgr.enqueue(pkt, 0));
674        assert_eq!(mgr.total_enqueued, 1);
675    }
676
677    #[test]
678    fn test_enqueue_increments_queue_metrics() {
679        let mut mgr = default_manager();
680        mgr.enqueue(make_packet(1, TrafficClass::Interactive, 200, 0), 0);
681        let m = mgr
682            .queue_metrics(&TrafficClass::Interactive)
683            .expect("test: Interactive queue metrics must exist");
684        assert_eq!(m.queued_packets, 1);
685        assert_eq!(m.total_bytes, 200);
686    }
687
688    #[test]
689    fn test_enqueue_multiple_classes() {
690        let mut mgr = default_manager();
691        mgr.enqueue(make_packet(1, TrafficClass::RealTime, 50, 0), 0);
692        mgr.enqueue(make_packet(2, TrafficClass::BulkData, 150, 0), 0);
693        mgr.enqueue(make_packet(3, TrafficClass::Background, 10, 0), 0);
694        assert_eq!(mgr.total_enqueued, 3);
695    }
696
697    #[test]
698    fn test_enqueue_respects_fifo_within_class() {
699        let mut mgr = default_manager();
700        mgr.enqueue(make_packet(1, TrafficClass::BulkData, 100, 10), 10);
701        mgr.enqueue(make_packet(2, TrafficClass::BulkData, 200, 20), 20);
702        let q = mgr
703            .queues
704            .get(&2)
705            .expect("test: priority-2 queue must exist");
706        assert_eq!(
707            q.front().expect("test: queue must have a front element").id,
708            1
709        );
710        assert_eq!(
711            q.back().expect("test: queue must have a back element").id,
712            2
713        );
714    }
715
716    // ── Drop / overflow ───────────────────────────────────────────────────────
717
718    #[test]
719    fn test_drop_packet_returns_false_on_empty_queue() {
720        let mut mgr = default_manager();
721        assert!(!mgr.drop_packet(1));
722    }
723
724    #[test]
725    fn test_drop_packet_removes_oldest() {
726        let mut mgr = default_manager();
727        mgr.enqueue(make_packet(1, TrafficClass::Background, 10, 0), 0);
728        mgr.enqueue(make_packet(2, TrafficClass::Background, 20, 1), 1);
729        assert!(mgr.drop_packet(1));
730        // Oldest (id=1) should be gone; id=2 stays.
731        let q = mgr
732            .queues
733            .get(&1)
734            .expect("test: priority-1 queue must exist");
735        assert_eq!(
736            q.front()
737                .expect("test: queue must have a front element after drop")
738                .id,
739            2
740        );
741        assert_eq!(mgr.total_dropped, 1);
742    }
743
744    #[test]
745    fn test_enqueue_drops_background_when_at_capacity() {
746        let mut mgr = NetworkQoSManager::new(QoSConfig {
747            max_queue_size: 3,
748            ..Default::default()
749        });
750        // Fill with Background packets.
751        mgr.enqueue(make_packet(1, TrafficClass::Background, 10, 0), 0);
752        mgr.enqueue(make_packet(2, TrafficClass::Background, 10, 1), 1);
753        mgr.enqueue(make_packet(3, TrafficClass::Background, 10, 2), 2);
754
755        // Enqueueing a 4th (higher-priority) packet should drop oldest Background.
756        let result = mgr.enqueue(make_packet(4, TrafficClass::RealTime, 50, 3), 3);
757        assert!(result);
758        assert_eq!(mgr.total_dropped, 1);
759    }
760
761    #[test]
762    fn test_enqueue_drops_background_packet_itself_at_capacity() {
763        let mut mgr = NetworkQoSManager::new(QoSConfig {
764            max_queue_size: 2,
765            ..Default::default()
766        });
767        // Fill with non-Background packets.
768        mgr.enqueue(make_packet(1, TrafficClass::RealTime, 10, 0), 0);
769        mgr.enqueue(make_packet(2, TrafficClass::Interactive, 10, 1), 1);
770
771        // Trying to enqueue Background when Background queue is empty → dropped.
772        let result = mgr.enqueue(make_packet(3, TrafficClass::Background, 10, 2), 2);
773        assert!(!result);
774        assert_eq!(mgr.total_dropped, 1);
775    }
776
777    // ── Dequeue — strict priority ─────────────────────────────────────────────
778
779    #[test]
780    fn test_dequeue_strict_highest_priority_first() {
781        let mut mgr = NetworkQoSManager::new(QoSConfig {
782            enable_strict_priority: true,
783            ..Default::default()
784        });
785        mgr.enqueue(make_packet(1, TrafficClass::Background, 10, 0), 0);
786        mgr.enqueue(make_packet(2, TrafficClass::RealTime, 10, 0), 0);
787        mgr.enqueue(make_packet(3, TrafficClass::Interactive, 10, 0), 0);
788
789        let first = mgr
790            .dequeue(10)
791            .expect("test: dequeue must return a packet (RealTime enqueued)");
792        assert_eq!(first.id, 2); // RealTime → priority 4
793
794        let second = mgr
795            .dequeue(10)
796            .expect("test: dequeue must return a packet (Interactive enqueued)");
797        assert_eq!(second.id, 3); // Interactive → priority 3
798    }
799
800    #[test]
801    fn test_dequeue_strict_returns_none_on_empty() {
802        let mut mgr = NetworkQoSManager::new(QoSConfig {
803            enable_strict_priority: true,
804            ..Default::default()
805        });
806        assert!(mgr.dequeue(0).is_none());
807    }
808
809    #[test]
810    fn test_dequeue_strict_increments_total_dequeued() {
811        let mut mgr = NetworkQoSManager::new(QoSConfig {
812            enable_strict_priority: true,
813            ..Default::default()
814        });
815        mgr.enqueue(make_packet(1, TrafficClass::BulkData, 100, 0), 0);
816        mgr.dequeue(50);
817        assert_eq!(mgr.total_dequeued, 1);
818    }
819
820    #[test]
821    fn test_dequeue_strict_updates_avg_wait() {
822        let mut mgr = NetworkQoSManager::new(QoSConfig {
823            enable_strict_priority: true,
824            ..Default::default()
825        });
826        mgr.enqueue(make_packet(1, TrafficClass::RealTime, 100, 0), 0);
827        mgr.dequeue(100); // 100 ms wait
828
829        let m = mgr
830            .queue_metrics(&TrafficClass::RealTime)
831            .expect("test: RealTime queue metrics must exist");
832        // First sample bootstraps avg directly.
833        assert!((m.avg_wait_ms - 100.0).abs() < 1e-6);
834    }
835
836    // ── Dequeue — weighted round-robin ────────────────────────────────────────
837
838    #[test]
839    fn test_dequeue_wrr_returns_none_on_empty() {
840        let mut mgr = default_manager();
841        assert!(mgr.dequeue(0).is_none());
842    }
843
844    #[test]
845    fn test_dequeue_wrr_returns_packet_when_non_empty() {
846        let mut mgr = default_manager();
847        mgr.enqueue(make_packet(1, TrafficClass::Interactive, 200, 0), 0);
848        let pkt = mgr.dequeue(10);
849        assert!(pkt.is_some());
850        assert_eq!(mgr.total_dequeued, 1);
851    }
852
853    #[test]
854    fn test_dequeue_wrr_services_multiple_classes() {
855        let mut mgr = default_manager();
856        for i in 0..5 {
857            mgr.enqueue(make_packet(i, TrafficClass::RealTime, 50, 0), 0);
858            mgr.enqueue(make_packet(100 + i, TrafficClass::Background, 50, 0), 0);
859        }
860        // Drain everything.
861        let mut dequeued = 0;
862        while mgr.dequeue(10).is_some() {
863            dequeued += 1;
864        }
865        assert_eq!(dequeued, 10);
866    }
867
868    #[test]
869    fn test_dequeue_wrr_decrements_queue_metrics() {
870        let mut mgr = default_manager();
871        mgr.enqueue(make_packet(1, TrafficClass::BulkData, 512, 0), 0);
872        mgr.dequeue(10);
873        let m = mgr
874            .queue_metrics(&TrafficClass::BulkData)
875            .expect("test: BulkData queue metrics must exist");
876        assert_eq!(m.queued_packets, 0);
877        assert_eq!(m.total_bytes, 0);
878    }
879
880    // ── SLA checking ─────────────────────────────────────────────────────────
881
882    #[test]
883    fn test_check_sla_no_specs_returns_empty() {
884        let mut mgr = default_manager();
885        let violations = mgr.check_sla(100);
886        assert!(violations.is_empty());
887    }
888
889    #[test]
890    fn test_check_sla_no_violation_when_within_limit() {
891        let mut mgr = NetworkQoSManager::new(QoSConfig {
892            sla_specs: vec![SLASpec {
893                class: TrafficClass::RealTime,
894                max_latency_ms: 50,
895                min_throughput_bps: 0,
896                max_jitter_ms: 10,
897            }],
898            enable_strict_priority: true,
899            ..Default::default()
900        });
901        // avg_wait_ms starts at 0 → no violation.
902        let v = mgr.check_sla(0);
903        assert!(v.is_empty());
904    }
905
906    #[test]
907    fn test_check_sla_detects_latency_violation() {
908        let mut mgr = NetworkQoSManager::new(QoSConfig {
909            enable_strict_priority: true,
910            sla_specs: vec![SLASpec {
911                class: TrafficClass::RealTime,
912                max_latency_ms: 5,
913                min_throughput_bps: 0,
914                max_jitter_ms: 2,
915            }],
916            ..Default::default()
917        });
918        // Enqueue then dequeue with 100 ms wait — bootstraps avg_wait to 100.
919        mgr.enqueue(make_packet(1, TrafficClass::RealTime, 10, 0), 0);
920        mgr.dequeue(100);
921
922        let violations = mgr.check_sla(100);
923        assert_eq!(violations.len(), 1);
924        assert_eq!(violations[0].metric, "avg_wait_ms");
925        assert!(violations[0].actual > 5.0);
926    }
927
928    #[test]
929    fn test_check_sla_appends_to_violations_log() {
930        let mut mgr = NetworkQoSManager::new(QoSConfig {
931            enable_strict_priority: true,
932            sla_specs: vec![SLASpec {
933                class: TrafficClass::Interactive,
934                max_latency_ms: 1,
935                min_throughput_bps: 0,
936                max_jitter_ms: 1,
937            }],
938            ..Default::default()
939        });
940        mgr.enqueue(make_packet(1, TrafficClass::Interactive, 10, 0), 0);
941        mgr.dequeue(500); // 500 ms wait
942
943        mgr.check_sla(500);
944        assert!(!mgr.violations_log().is_empty());
945    }
946
947    #[test]
948    fn test_check_sla_violations_log_capped_at_1000() {
949        let mut mgr = NetworkQoSManager::new(QoSConfig {
950            enable_strict_priority: true,
951            sla_specs: vec![SLASpec {
952                class: TrafficClass::Background,
953                max_latency_ms: 1,
954                min_throughput_bps: 0,
955                max_jitter_ms: 1,
956            }],
957            ..Default::default()
958        });
959        // Manually set avg_wait high enough to always violate.
960        if let Some(m) = mgr.metrics.get_mut(&1) {
961            m.avg_wait_ms = 9_999.0;
962        }
963        for i in 0..1_200u64 {
964            mgr.check_sla(i);
965        }
966        assert!(mgr.violations_log().len() <= 1_000);
967    }
968
969    // ── Metrics helpers ───────────────────────────────────────────────────────
970
971    #[test]
972    fn test_queue_metrics_returns_correct_class() {
973        let mgr = default_manager();
974        let m = mgr
975            .queue_metrics(&TrafficClass::BulkData)
976            .expect("test: BulkData queue metrics must exist");
977        assert_eq!(m.class, TrafficClass::BulkData);
978    }
979
980    #[test]
981    fn test_all_metrics_returns_four_entries() {
982        let mgr = default_manager();
983        assert_eq!(mgr.all_metrics().len(), 4);
984    }
985
986    #[test]
987    fn test_all_metrics_ordered_highest_first() {
988        let mgr = default_manager();
989        let ms = mgr.all_metrics();
990        assert_eq!(ms[0].class, TrafficClass::RealTime);
991        assert_eq!(ms[1].class, TrafficClass::Interactive);
992        assert_eq!(ms[2].class, TrafficClass::BulkData);
993        assert_eq!(ms[3].class, TrafficClass::Background);
994    }
995
996    #[test]
997    fn test_total_queued_bytes_empty() {
998        let mgr = default_manager();
999        assert_eq!(mgr.total_queued_bytes(), 0);
1000    }
1001
1002    #[test]
1003    fn test_total_queued_bytes_after_enqueue() {
1004        let mut mgr = default_manager();
1005        mgr.enqueue(make_packet(1, TrafficClass::RealTime, 1000, 0), 0);
1006        mgr.enqueue(make_packet(2, TrafficClass::Background, 500, 0), 0);
1007        assert_eq!(mgr.total_queued_bytes(), 1500);
1008    }
1009
1010    #[test]
1011    fn test_total_queued_bytes_after_dequeue() {
1012        let mut mgr = NetworkQoSManager::new(QoSConfig {
1013            enable_strict_priority: true,
1014            ..Default::default()
1015        });
1016        mgr.enqueue(make_packet(1, TrafficClass::RealTime, 800, 0), 0);
1017        mgr.dequeue(10);
1018        assert_eq!(mgr.total_queued_bytes(), 0);
1019    }
1020
1021    // ── QoSStats ─────────────────────────────────────────────────────────────
1022
1023    #[test]
1024    fn test_qos_stats_initial_state() {
1025        let mgr = default_manager();
1026        let stats = mgr.qos_stats();
1027        assert_eq!(stats.total_enqueued, 0);
1028        assert_eq!(stats.total_dequeued, 0);
1029        assert_eq!(stats.total_dropped, 0);
1030        assert!((stats.drop_rate).abs() < 1e-9);
1031        assert_eq!(stats.active_queues, 0);
1032        assert_eq!(stats.total_violations, 0);
1033    }
1034
1035    #[test]
1036    fn test_qos_stats_active_queues_counted() {
1037        let mut mgr = default_manager();
1038        mgr.enqueue(make_packet(1, TrafficClass::RealTime, 10, 0), 0);
1039        mgr.enqueue(make_packet(2, TrafficClass::Background, 10, 0), 0);
1040        let stats = mgr.qos_stats();
1041        assert_eq!(stats.active_queues, 2);
1042    }
1043
1044    #[test]
1045    fn test_qos_stats_drop_rate_calculation() {
1046        let mut mgr = NetworkQoSManager::new(QoSConfig {
1047            max_queue_size: 1,
1048            ..Default::default()
1049        });
1050        // Enqueue one packet to fill capacity.
1051        mgr.enqueue(make_packet(1, TrafficClass::RealTime, 10, 0), 0);
1052        // Next Background packet cannot enter and is dropped.
1053        mgr.enqueue(make_packet(2, TrafficClass::Background, 10, 0), 0);
1054
1055        let stats = mgr.qos_stats();
1056        // 1 enqueued + 1 dropped; drop_rate = 1/2 = 0.5 (total_enqueued
1057        // counts only successful enqueues, so enqueued=1, dropped=1).
1058        assert!(stats.total_dropped >= 1);
1059    }
1060
1061    // ── Reset / flush ─────────────────────────────────────────────────────────
1062
1063    #[test]
1064    fn test_reset_metrics_clears_avg_wait() {
1065        let mut mgr = NetworkQoSManager::new(QoSConfig {
1066            enable_strict_priority: true,
1067            ..Default::default()
1068        });
1069        mgr.enqueue(make_packet(1, TrafficClass::BulkData, 10, 0), 0);
1070        mgr.dequeue(200);
1071        mgr.reset_metrics();
1072        let m = mgr
1073            .queue_metrics(&TrafficClass::BulkData)
1074            .expect("test: BulkData queue metrics must exist after reset");
1075        assert!((m.avg_wait_ms).abs() < 1e-9);
1076    }
1077
1078    #[test]
1079    fn test_flush_all_empties_queues() {
1080        let mut mgr = default_manager();
1081        mgr.enqueue(make_packet(1, TrafficClass::RealTime, 10, 0), 0);
1082        mgr.enqueue(make_packet(2, TrafficClass::BulkData, 20, 0), 0);
1083        let removed = mgr.flush_all();
1084        assert_eq!(removed, 2);
1085        assert_eq!(mgr.total_queued_bytes(), 0);
1086    }
1087
1088    #[test]
1089    fn test_flush_all_returns_zero_on_empty() {
1090        let mut mgr = default_manager();
1091        assert_eq!(mgr.flush_all(), 0);
1092    }
1093
1094    // ── EWMA wait time ────────────────────────────────────────────────────────
1095
1096    #[test]
1097    fn test_ewma_converges_towards_new_value() {
1098        let mut mgr = NetworkQoSManager::new(QoSConfig {
1099            enable_strict_priority: true,
1100            ..Default::default()
1101        });
1102        // Bootstrap with 100 ms wait.
1103        mgr.enqueue(make_packet(1, TrafficClass::Interactive, 10, 0), 0);
1104        mgr.dequeue(100);
1105
1106        // Feed several 10 ms waits; avg should converge below 100.
1107        for i in 0..20u64 {
1108            mgr.enqueue(make_packet(i + 10, TrafficClass::Interactive, 10, 0), 0);
1109            mgr.dequeue(10);
1110        }
1111        let m = mgr
1112            .queue_metrics(&TrafficClass::Interactive)
1113            .expect("test: Interactive queue metrics must exist");
1114        assert!(
1115            m.avg_wait_ms < 80.0,
1116            "EWMA should converge: {}",
1117            m.avg_wait_ms
1118        );
1119    }
1120
1121    // ── Drop metric consistency ───────────────────────────────────────────────
1122
1123    #[test]
1124    fn test_drop_increments_per_class_dropped_packets() {
1125        let mut mgr = default_manager();
1126        mgr.enqueue(make_packet(1, TrafficClass::Background, 10, 0), 0);
1127        mgr.drop_packet(1);
1128        let m = mgr
1129            .queue_metrics(&TrafficClass::Background)
1130            .expect("test: Background queue metrics must exist");
1131        assert_eq!(m.dropped_packets, 1);
1132    }
1133
1134    // ── Violations log ────────────────────────────────────────────────────────
1135
1136    #[test]
1137    fn test_violations_log_initially_empty() {
1138        let mgr = default_manager();
1139        assert!(mgr.violations_log().is_empty());
1140    }
1141
1142    #[test]
1143    fn test_violations_log_records_correct_class() {
1144        let mut mgr = NetworkQoSManager::new(QoSConfig {
1145            enable_strict_priority: true,
1146            sla_specs: vec![SLASpec {
1147                class: TrafficClass::RealTime,
1148                max_latency_ms: 1,
1149                min_throughput_bps: 0,
1150                max_jitter_ms: 1,
1151            }],
1152            ..Default::default()
1153        });
1154        mgr.enqueue(make_packet(1, TrafficClass::RealTime, 10, 0), 0);
1155        mgr.dequeue(999);
1156        mgr.check_sla(999);
1157
1158        let log = mgr.violations_log();
1159        assert!(!log.is_empty());
1160        assert_eq!(log[0].class, TrafficClass::RealTime);
1161    }
1162
1163    // ── Strict priority exhaustion ────────────────────────────────────────────
1164
1165    #[test]
1166    fn test_strict_priority_drains_all_queues() {
1167        let mut mgr = NetworkQoSManager::new(QoSConfig {
1168            enable_strict_priority: true,
1169            ..Default::default()
1170        });
1171        let classes = [
1172            TrafficClass::RealTime,
1173            TrafficClass::Interactive,
1174            TrafficClass::BulkData,
1175            TrafficClass::Background,
1176        ];
1177        for (i, cls) in classes.iter().enumerate() {
1178            mgr.enqueue(make_packet(i as u64, cls.clone(), 10, 0), 0);
1179        }
1180        let mut count = 0;
1181        while mgr.dequeue(10).is_some() {
1182            count += 1;
1183        }
1184        assert_eq!(count, 4);
1185        assert_eq!(mgr.total_dequeued, 4);
1186    }
1187}