1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
//! PeerMessagePrioritizer — Multi-level priority queue with aging for outbound messages.
//!
//! Each peer gets its own sorted queue (highest effective priority first, FIFO within same
//! priority). An aging mechanism prevents message starvation by promoting lower-priority
//! messages after they have waited for a configurable number of ticks.
//!
//! ## Design highlights
//!
//! - **Four priority levels**: Background → Normal → High → Urgent
//! - **Aging / anti-starvation**: `advance_tick` promotes stale messages one level at a time
//! - **Per-peer capacity cap**: when full, the lowest-priority message is evicted to make room
//! - **Rich statistics**: enqueued, dequeued, promoted, dropped counters with a throughput ratio
//! - **No `unwrap()`**: every fallible operation is expressed through `Option` / explicit checks
use std::collections::HashMap;
// ─── MessagePriority ──────────────────────────────────────────────────────────
/// Four-level priority for outbound peer messages.
///
/// The integer discriminants are intentional: `Ord` derived from them means
/// `Background < Normal < High < Urgent` so arithmetic promotion (`as u8 + 1`)
/// produces the next level.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub enum MessagePriority {
/// Best-effort background traffic (gossip, keep-alives, …).
Background = 0,
/// Default priority for ordinary messages.
Normal = 1,
/// Important messages that should be delivered before normal ones.
High = 2,
/// Time-sensitive messages that preempt all others.
Urgent = 3,
}
impl MessagePriority {
/// Promote this priority one level upward, capped at `Urgent`.
fn promote(self) -> Self {
match self {
Self::Background => Self::Normal,
Self::Normal => Self::High,
Self::High => Self::Urgent,
Self::Urgent => Self::Urgent,
}
}
}
// ─── PrioritizedMessage ───────────────────────────────────────────────────────
/// A single outbound message sitting in a peer's priority queue.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct PrioritizedMessage {
/// Monotonically increasing identifier assigned at enqueue time.
pub msg_id: u64,
/// Destination peer identifier.
pub peer_id: String,
/// Original priority assigned by the caller.
pub priority: MessagePriority,
/// Size of the message payload in bytes.
pub payload_bytes: u64,
/// Value of `current_tick` when the message was enqueued.
pub enqueued_tick: u64,
/// Effective priority after aging promotions; starts equal to `priority`.
pub effective_priority: MessagePriority,
}
// ─── AgingConfig ─────────────────────────────────────────────────────────────
/// Tunable parameters for the aging / anti-starvation mechanism.
#[derive(Clone, Debug)]
pub struct AgingConfig {
/// Number of ticks a message must wait before its effective priority is
/// promoted by one level. Default: 50.
pub promote_after_ticks: u64,
/// Maximum number of messages kept per peer queue. Default: 256.
pub max_queue_size: usize,
}
impl Default for AgingConfig {
fn default() -> Self {
Self {
promote_after_ticks: 50,
max_queue_size: 256,
}
}
}
// ─── PrioritizerStats ────────────────────────────────────────────────────────
/// Aggregate statistics for a `PeerMessagePrioritizer` instance.
#[derive(Clone, Debug, Default)]
pub struct PrioritizerStats {
/// Total messages successfully enqueued (including those that displaced a
/// dropped message).
pub total_enqueued: u64,
/// Total messages removed via `dequeue`.
pub total_dequeued: u64,
/// Total aging promotions applied across all messages and all ticks.
pub total_promoted: u64,
/// Total messages evicted because the per-peer queue was at capacity.
pub total_dropped: u64,
}
impl PrioritizerStats {
/// Fraction of enqueued messages that have been dequeued.
///
/// Returns `0.0` when nothing has been enqueued yet.
pub fn throughput_ratio(&self) -> f64 {
self.total_dequeued as f64 / self.total_enqueued.max(1) as f64
}
}
// ─── PeerMessagePrioritizer ───────────────────────────────────────────────────
/// Multi-level outbound message prioritizer with per-peer queues and aging.
///
/// # Ordering guarantee
///
/// Within a peer's queue messages are sorted so that:
/// 1. Higher `effective_priority` comes first.
/// 2. Among messages with the same `effective_priority`, the one enqueued
/// earliest (lowest `enqueued_tick`) comes first (FIFO).
pub struct PeerMessagePrioritizer {
/// Per-peer queues; each `Vec` is kept sorted (front = highest priority).
pub queues: HashMap<String, Vec<PrioritizedMessage>>,
/// Aging and capacity configuration.
pub config: AgingConfig,
/// Running statistics.
pub stats: PrioritizerStats,
/// Logical clock advanced via `advance_tick`.
pub current_tick: u64,
/// Counter used to assign unique message identifiers.
pub next_id: u64,
}
impl PeerMessagePrioritizer {
/// Create a new prioritizer with the supplied configuration.
pub fn new(config: AgingConfig) -> Self {
Self {
queues: HashMap::new(),
config,
stats: PrioritizerStats::default(),
current_tick: 0,
next_id: 1,
}
}
// ── Internal helpers ──────────────────────────────────────────────────────
/// Insert `msg` into `queue` maintaining the sorted invariant.
///
/// Sort key: descending `effective_priority`, then ascending `enqueued_tick`
/// (FIFO within the same effective priority).
fn insert_sorted(queue: &mut Vec<PrioritizedMessage>, msg: PrioritizedMessage) {
// Binary-search for the correct insertion point.
let pos = queue.partition_point(|m| {
// Items that should come *before* `msg` satisfy this predicate.
m.effective_priority > msg.effective_priority
|| (m.effective_priority == msg.effective_priority
&& m.enqueued_tick <= msg.enqueued_tick)
});
queue.insert(pos, msg);
}
/// Drop the message with the lowest priority (and, within that, the most
/// recently enqueued one) from `queue`. Returns `true` if a message was
/// removed.
fn drop_lowest(queue: &mut Vec<PrioritizedMessage>) -> bool {
if queue.is_empty() {
return false;
}
// The queue is sorted highest-first so the lowest-priority messages are
// at the tail. Among equal-priority tail messages prefer removing the
// one with the highest `enqueued_tick` (most recent, least-waited).
let lowest_prio = queue.last().map(|m| m.effective_priority);
if let Some(lp) = lowest_prio {
// Find the index of the most-recently-enqueued message among those
// sharing the lowest effective priority.
let victim = queue
.iter()
.enumerate()
.rev() // iterate from tail toward head
.take_while(|(_, m)| m.effective_priority == lp)
.max_by_key(|(_, m)| m.enqueued_tick)
.map(|(i, _)| i);
if let Some(idx) = victim {
queue.remove(idx);
return true;
}
}
false
}
// ── Public API ────────────────────────────────────────────────────────────
/// Enqueue a new message for `peer_id`.
///
/// If the peer's queue is already at `max_queue_size`:
/// 1. Attempt to drop the lowest-priority message (`Background` first, then
/// `Normal`).
/// 2. If the queue is still full after the drop attempt, return `None`.
///
/// On success, returns `Some(msg_id)`.
pub fn enqueue(
&mut self,
peer_id: String,
priority: MessagePriority,
payload_bytes: u64,
) -> Option<u64> {
let queue = self.queues.entry(peer_id.clone()).or_default();
if queue.len() >= self.config.max_queue_size {
// Try to make room by evicting the lowest-priority message.
if Self::drop_lowest(queue) {
self.stats.total_dropped += 1;
}
// Re-check — if still full we cannot accept the message.
if queue.len() >= self.config.max_queue_size {
return None;
}
}
let msg_id = self.next_id;
self.next_id += 1;
let msg = PrioritizedMessage {
msg_id,
peer_id,
priority,
payload_bytes,
enqueued_tick: self.current_tick,
effective_priority: priority,
};
Self::insert_sorted(queue, msg);
self.stats.total_enqueued += 1;
Some(msg_id)
}
/// Remove and return the highest-priority message for `peer_id`, or `None`
/// if the peer has no queued messages.
pub fn dequeue(&mut self, peer_id: &str) -> Option<PrioritizedMessage> {
let queue = self.queues.get_mut(peer_id)?;
if queue.is_empty() {
return None;
}
let msg = queue.remove(0);
self.stats.total_dequeued += 1;
Some(msg)
}
/// Advance the logical clock by one tick and apply aging promotions.
///
/// Any message whose waiting time (`current_tick - enqueued_tick`) meets or
/// exceeds `promote_after_ticks` and whose `effective_priority` is below
/// `Urgent` is promoted by one level. Queues that had at least one
/// promotion are re-sorted to preserve the ordering invariant.
pub fn advance_tick(&mut self) {
self.current_tick += 1;
let promote_after = self.config.promote_after_ticks;
let current_tick = self.current_tick;
for queue in self.queues.values_mut() {
let mut any_promoted = false;
for msg in queue.iter_mut() {
if msg.effective_priority < MessagePriority::Urgent
&& current_tick.saturating_sub(msg.enqueued_tick) >= promote_after
{
msg.effective_priority = msg.effective_priority.promote();
self.stats.total_promoted += 1;
any_promoted = true;
}
}
if any_promoted {
// Re-sort: descending effective_priority, then ascending enqueued_tick.
queue.sort_by(|a, b| {
b.effective_priority
.cmp(&a.effective_priority)
.then_with(|| a.enqueued_tick.cmp(&b.enqueued_tick))
});
}
}
}
/// Return a reference to the accumulated statistics.
pub fn stats(&self) -> &PrioritizerStats {
&self.stats
}
/// Return the number of messages currently queued for `peer_id`.
pub fn queue_depth(&self, peer_id: &str) -> usize {
self.queues.get(peer_id).map_or(0, |q| q.len())
}
}
// ─── Tests ────────────────────────────────────────────────────────────────────
#[cfg(test)]
mod tests {
use super::*;
fn default_prioritizer() -> PeerMessagePrioritizer {
PeerMessagePrioritizer::new(AgingConfig::default())
}
// ── 1. enqueue returns a unique msg_id ────────────────────────────────────
#[test]
fn enqueue_returns_msg_id() {
let mut p = default_prioritizer();
let id = p.enqueue("peer-a".to_string(), MessagePriority::Normal, 128);
assert!(id.is_some(), "enqueue should return Some(id)");
assert_eq!(id.expect("test: enqueue should return msg_id"), 1u64);
}
// ── 2. msg_ids are monotonically increasing ───────────────────────────────
#[test]
fn msg_ids_are_monotonically_increasing() {
let mut p = default_prioritizer();
let id1 = p
.enqueue("peer-a".to_string(), MessagePriority::Normal, 64)
.expect("test: first enqueue should return msg_id");
let id2 = p
.enqueue("peer-a".to_string(), MessagePriority::Normal, 64)
.expect("test: second enqueue should return msg_id");
let id3 = p
.enqueue("peer-b".to_string(), MessagePriority::High, 64)
.expect("test: third enqueue should return msg_id");
assert!(id1 < id2);
assert!(id2 < id3);
}
// ── 3. dequeue returns highest effective priority first ───────────────────
#[test]
fn dequeue_highest_priority_first() {
let mut p = default_prioritizer();
p.enqueue("peer-a".to_string(), MessagePriority::Background, 10)
.expect("test: enqueue Background should succeed");
p.enqueue("peer-a".to_string(), MessagePriority::Urgent, 10)
.expect("test: enqueue Urgent should succeed");
p.enqueue("peer-a".to_string(), MessagePriority::Normal, 10)
.expect("test: enqueue Normal should succeed");
let first = p
.dequeue("peer-a")
.expect("test: first dequeue should return Urgent message");
assert_eq!(first.effective_priority, MessagePriority::Urgent);
let second = p
.dequeue("peer-a")
.expect("test: second dequeue should return Normal message");
assert_eq!(second.effective_priority, MessagePriority::Normal);
let third = p
.dequeue("peer-a")
.expect("test: third dequeue should return Background message");
assert_eq!(third.effective_priority, MessagePriority::Background);
}
// ── 4. FIFO within the same priority ─────────────────────────────────────
#[test]
fn fifo_within_same_priority() {
let mut p = default_prioritizer();
let id1 = p
.enqueue("peer-a".to_string(), MessagePriority::Normal, 10)
.expect("test: first enqueue should succeed");
let id2 = p
.enqueue("peer-a".to_string(), MessagePriority::Normal, 20)
.expect("test: second enqueue should succeed");
let id3 = p
.enqueue("peer-a".to_string(), MessagePriority::Normal, 30)
.expect("test: third enqueue should succeed");
assert_eq!(
p.dequeue("peer-a")
.expect("test: first dequeue should return message")
.msg_id,
id1
);
assert_eq!(
p.dequeue("peer-a")
.expect("test: second dequeue should return message")
.msg_id,
id2
);
assert_eq!(
p.dequeue("peer-a")
.expect("test: third dequeue should return message")
.msg_id,
id3
);
}
// ── 5. FIFO is stable across ticks ───────────────────────────────────────
#[test]
fn fifo_stable_across_ticks() {
let config = AgingConfig {
promote_after_ticks: 1000, // never promote during this test
max_queue_size: 256,
};
let mut p = PeerMessagePrioritizer::new(config);
let id1 = p
.enqueue("peer-a".to_string(), MessagePriority::High, 10)
.expect("test: first enqueue at tick 0 should succeed");
p.advance_tick();
let id2 = p
.enqueue("peer-a".to_string(), MessagePriority::High, 10)
.expect("test: second enqueue at tick 1 should succeed");
p.advance_tick();
let id3 = p
.enqueue("peer-a".to_string(), MessagePriority::High, 10)
.expect("test: third enqueue at tick 2 should succeed");
assert_eq!(
p.dequeue("peer-a")
.expect("test: first dequeue should return message")
.msg_id,
id1
);
assert_eq!(
p.dequeue("peer-a")
.expect("test: second dequeue should return message")
.msg_id,
id2
);
assert_eq!(
p.dequeue("peer-a")
.expect("test: third dequeue should return message")
.msg_id,
id3
);
}
// ── 6. advance_tick promotes after threshold ──────────────────────────────
#[test]
fn advance_tick_promotes_after_threshold() {
let config = AgingConfig {
promote_after_ticks: 3,
max_queue_size: 256,
};
let mut p = PeerMessagePrioritizer::new(config);
p.enqueue("peer-a".to_string(), MessagePriority::Background, 10)
.expect("test: enqueue Background should succeed");
// Advance 2 ticks — not enough.
p.advance_tick();
p.advance_tick();
assert_eq!(p.stats().total_promoted, 0);
// Third tick crosses the threshold.
p.advance_tick();
assert_eq!(p.stats().total_promoted, 1);
let msg = p
.dequeue("peer-a")
.expect("test: dequeue should return promoted message");
assert_eq!(msg.effective_priority, MessagePriority::Normal);
assert_eq!(msg.priority, MessagePriority::Background); // original unchanged
}
// ── 7. Promotion is capped at Urgent ──────────────────────────────────────
#[test]
fn promotion_capped_at_urgent() {
let config = AgingConfig {
promote_after_ticks: 1,
max_queue_size: 256,
};
let mut p = PeerMessagePrioritizer::new(config);
p.enqueue("peer-a".to_string(), MessagePriority::High, 10)
.expect("test: enqueue High should succeed");
// Advance enough ticks to try to promote beyond Urgent.
for _ in 0..10 {
p.advance_tick();
}
let msg = p
.dequeue("peer-a")
.expect("test: dequeue should return Urgent message after promotion");
assert_eq!(msg.effective_priority, MessagePriority::Urgent);
}
// ── 8. total_promoted counter ─────────────────────────────────────────────
#[test]
fn total_promoted_counter() {
let config = AgingConfig {
promote_after_ticks: 2,
max_queue_size: 256,
};
let mut p = PeerMessagePrioritizer::new(config);
// Two Background messages enqueued at tick 0.
p.enqueue("peer-a".to_string(), MessagePriority::Background, 10)
.expect("test: first enqueue Background should succeed");
p.enqueue("peer-a".to_string(), MessagePriority::Background, 10)
.expect("test: second enqueue Background should succeed");
p.advance_tick(); // tick 1 — waited 1 tick, not yet promoted
assert_eq!(p.stats().total_promoted, 0);
p.advance_tick(); // tick 2 — waited 2 ticks, both promoted
assert_eq!(p.stats().total_promoted, 2);
}
// ── 9. drop lowest priority when full ────────────────────────────────────
#[test]
fn drop_lowest_priority_when_full() {
let config = AgingConfig {
promote_after_ticks: 9999,
max_queue_size: 3,
};
let mut p = PeerMessagePrioritizer::new(config);
p.enqueue("peer-a".to_string(), MessagePriority::Normal, 10)
.expect("test: enqueue Normal should succeed");
p.enqueue("peer-a".to_string(), MessagePriority::Background, 10)
.expect("test: enqueue Background should succeed");
p.enqueue("peer-a".to_string(), MessagePriority::High, 10)
.expect("test: enqueue High should succeed");
// Queue is now full (3 messages).
// Enqueue Urgent — should evict Background, succeed.
let result = p.enqueue("peer-a".to_string(), MessagePriority::Urgent, 10);
assert!(
result.is_some(),
"should succeed by evicting lowest-priority message"
);
assert_eq!(p.stats().total_dropped, 1);
assert_eq!(p.queue_depth("peer-a"), 3);
// All remaining messages should have priority >= Normal.
let priorities: Vec<_> = p.queues["peer-a"]
.iter()
.map(|m| m.effective_priority)
.collect();
for prio in &priorities {
assert!(
*prio >= MessagePriority::Normal,
"Background should have been evicted"
);
}
}
// ── 10. total_dropped counter ─────────────────────────────────────────────
#[test]
fn total_dropped_counter() {
let config = AgingConfig {
promote_after_ticks: 9999,
max_queue_size: 2,
};
let mut p = PeerMessagePrioritizer::new(config);
p.enqueue("peer-a".to_string(), MessagePriority::Background, 10)
.expect("test: first enqueue Background should succeed");
p.enqueue("peer-a".to_string(), MessagePriority::Background, 10)
.expect("test: second enqueue Background should succeed");
// Evict 1 Background, enqueue Normal.
p.enqueue("peer-a".to_string(), MessagePriority::Normal, 10)
.expect("test: enqueue Normal should succeed");
assert_eq!(p.stats().total_dropped, 1);
}
// ── 11. returns None when queue still full after drop attempt ─────────────
#[test]
fn returns_none_when_still_full_after_drop() {
let config = AgingConfig {
promote_after_ticks: 9999,
max_queue_size: 2,
};
let mut p = PeerMessagePrioritizer::new(config);
p.enqueue("peer-a".to_string(), MessagePriority::Urgent, 10)
.expect("test: first enqueue Urgent should succeed");
p.enqueue("peer-a".to_string(), MessagePriority::Urgent, 10)
.expect("test: second enqueue Urgent should succeed");
// Queue full with Urgents; nothing lower to drop — new message rejected.
// But drop_lowest will remove one (same priority, most recent).
// Let's fill with two Urgents then try a third Urgent.
let third = p.enqueue("peer-a".to_string(), MessagePriority::Urgent, 10);
// After dropping one Urgent there is space, so this succeeds.
assert!(third.is_some());
assert_eq!(p.stats().total_dropped, 1);
assert_eq!(p.queue_depth("peer-a"), 2);
}
// ── 12. empty dequeue returns None ───────────────────────────────────────
#[test]
fn empty_dequeue_returns_none() {
let mut p = default_prioritizer();
assert!(p.dequeue("nonexistent-peer").is_none());
}
// ── 13. dequeue from empty queue returns None ─────────────────────────────
#[test]
fn dequeue_exhausted_queue_returns_none() {
let mut p = default_prioritizer();
p.enqueue("peer-a".to_string(), MessagePriority::Normal, 10)
.expect("test: enqueue Normal should succeed");
p.dequeue("peer-a");
assert!(p.dequeue("peer-a").is_none());
}
// ── 14. queue_depth reflects current state ────────────────────────────────
#[test]
fn queue_depth_reflects_state() {
let mut p = default_prioritizer();
assert_eq!(p.queue_depth("peer-a"), 0);
p.enqueue("peer-a".to_string(), MessagePriority::Normal, 10)
.expect("test: first enqueue should succeed");
assert_eq!(p.queue_depth("peer-a"), 1);
p.enqueue("peer-a".to_string(), MessagePriority::High, 10)
.expect("test: second enqueue should succeed");
assert_eq!(p.queue_depth("peer-a"), 2);
p.dequeue("peer-a");
assert_eq!(p.queue_depth("peer-a"), 1);
}
// ── 15. throughput_ratio ──────────────────────────────────────────────────
#[test]
fn throughput_ratio_correct() {
let mut p = default_prioritizer();
// Zero enqueued → ratio = 0.0
assert!((p.stats().throughput_ratio() - 0.0).abs() < f64::EPSILON);
p.enqueue("peer-a".to_string(), MessagePriority::Normal, 10)
.expect("test: first enqueue should succeed");
p.enqueue("peer-a".to_string(), MessagePriority::Normal, 10)
.expect("test: second enqueue should succeed");
p.enqueue("peer-a".to_string(), MessagePriority::Normal, 10)
.expect("test: third enqueue should succeed");
p.dequeue("peer-a");
// 1 dequeued / 3 enqueued = 0.333…
let ratio = p.stats().throughput_ratio();
assert!((ratio - 1.0 / 3.0).abs() < 1e-9, "ratio = {ratio}");
}
// ── 16. total_enqueued and total_dequeued counters ────────────────────────
#[test]
fn enqueued_and_dequeued_counters() {
let mut p = default_prioritizer();
for _ in 0..5 {
p.enqueue("peer-a".to_string(), MessagePriority::Normal, 10)
.expect("test: enqueue in loop should succeed");
}
assert_eq!(p.stats().total_enqueued, 5);
for _ in 0..3 {
p.dequeue("peer-a");
}
assert_eq!(p.stats().total_dequeued, 3);
}
// ── 17. multi-peer isolation ──────────────────────────────────────────────
#[test]
fn multi_peer_isolation() {
let mut p = default_prioritizer();
p.enqueue("peer-a".to_string(), MessagePriority::Urgent, 10)
.expect("test: enqueue Urgent for peer-a should succeed");
p.enqueue("peer-b".to_string(), MessagePriority::Background, 10)
.expect("test: enqueue Background for peer-b should succeed");
let msg_a = p
.dequeue("peer-a")
.expect("test: dequeue peer-a should return Urgent message");
assert_eq!(msg_a.effective_priority, MessagePriority::Urgent);
assert_eq!(msg_a.peer_id, "peer-a");
let msg_b = p
.dequeue("peer-b")
.expect("test: dequeue peer-b should return Background message");
assert_eq!(msg_b.effective_priority, MessagePriority::Background);
assert_eq!(msg_b.peer_id, "peer-b");
}
// ── 18. advance_tick does not promote Urgent messages ────────────────────
#[test]
fn advance_tick_does_not_promote_urgent() {
let config = AgingConfig {
promote_after_ticks: 1,
max_queue_size: 256,
};
let mut p = PeerMessagePrioritizer::new(config);
p.enqueue("peer-a".to_string(), MessagePriority::Urgent, 10)
.expect("test: enqueue Urgent should succeed");
p.advance_tick();
assert_eq!(p.stats().total_promoted, 0);
let msg = p
.dequeue("peer-a")
.expect("test: dequeue should return Urgent message");
assert_eq!(msg.effective_priority, MessagePriority::Urgent);
}
// ── 19. payload_bytes is preserved ───────────────────────────────────────
#[test]
fn payload_bytes_preserved() {
let mut p = default_prioritizer();
p.enqueue("peer-a".to_string(), MessagePriority::Normal, 4096)
.expect("test: enqueue 4096 bytes should succeed");
let msg = p
.dequeue("peer-a")
.expect("test: dequeue should return message with payload");
assert_eq!(msg.payload_bytes, 4096);
}
// ── 20. promote_after_ticks boundary (strictly >=) ───────────────────────
#[test]
fn promote_at_exact_threshold() {
let config = AgingConfig {
promote_after_ticks: 5,
max_queue_size: 256,
};
let mut p = PeerMessagePrioritizer::new(config);
p.enqueue("peer-a".to_string(), MessagePriority::Background, 10)
.expect("test: enqueue Background should succeed");
for _ in 0..4 {
p.advance_tick();
}
assert_eq!(
p.stats().total_promoted,
0,
"should not promote before threshold"
);
p.advance_tick(); // tick 5 — exactly at threshold
assert_eq!(
p.stats().total_promoted,
1,
"should promote at exact threshold"
);
}
}