1use std::collections::HashMap;
8
9#[derive(Clone, Copy, Debug, PartialEq, Eq)]
15pub enum LbStrategy {
16 RoundRobin,
18 LeastConnections,
20 WeightedRandom,
25 LowestLatency,
29}
30
31#[derive(Debug, Clone)]
37pub struct PeerLoad {
38 pub peer_id: String,
40 pub active_connections: u32,
42 pub weight: u32,
44 pub avg_latency_ms: f64,
46 pub total_requests: u64,
48 pub failed_requests: u64,
50}
51
52impl PeerLoad {
53 fn new(peer_id: impl Into<String>, weight: u32) -> Self {
55 Self {
56 peer_id: peer_id.into(),
57 active_connections: 0,
58 weight,
59 avg_latency_ms: 0.0,
60 total_requests: 0,
61 failed_requests: 0,
62 }
63 }
64
65 pub fn failure_rate(&self) -> f64 {
68 if self.total_requests == 0 {
69 0.0
70 } else {
71 self.failed_requests as f64 / self.total_requests as f64
72 }
73 }
74
75 pub fn is_healthy(&self) -> bool {
78 self.failure_rate() < 0.5 && self.active_connections < 1000
79 }
80}
81
82#[derive(Debug, Clone, PartialEq, Eq)]
88pub struct LbStats {
89 pub total_peers: usize,
91 pub healthy_peers: usize,
93 pub total_requests_routed: u64,
95 pub total_request_counter: u64,
97}
98
99pub struct PeerLoadBalancer {
105 pub peers: HashMap<String, PeerLoad>,
107 pub strategy: LbStrategy,
109 pub total_requests: u64,
112}
113
114impl PeerLoadBalancer {
115 pub fn new(strategy: LbStrategy) -> Self {
119 Self {
120 peers: HashMap::new(),
121 strategy,
122 total_requests: 0,
123 }
124 }
125
126 pub fn add_peer(&mut self, peer_id: &str, weight: u32) {
130 self.peers
131 .entry(peer_id.to_owned())
132 .and_modify(|p| p.weight = weight)
133 .or_insert_with(|| PeerLoad::new(peer_id, weight));
134 }
135
136 pub fn remove_peer(&mut self, peer_id: &str) -> bool {
138 self.peers.remove(peer_id).is_some()
139 }
140
141 pub fn select(&mut self) -> Option<String> {
149 let chosen = match self.strategy {
150 LbStrategy::RoundRobin => self.select_round_robin(),
151 LbStrategy::LeastConnections => self.select_least_connections(),
152 LbStrategy::WeightedRandom => self.select_weighted_random(),
153 LbStrategy::LowestLatency => self.select_lowest_latency(),
154 };
155
156 if chosen.is_some() {
157 self.total_requests += 1;
158 }
159
160 chosen
161 }
162
163 pub fn record_start(&mut self, peer_id: &str) -> bool {
170 match self.peers.get_mut(peer_id) {
171 Some(p) => {
172 p.active_connections = p.active_connections.saturating_add(1);
173 p.total_requests += 1;
174 true
175 }
176 None => false,
177 }
178 }
179
180 pub fn record_complete(&mut self, peer_id: &str, latency_ms: f64, success: bool) -> bool {
188 match self.peers.get_mut(peer_id) {
189 Some(p) => {
190 p.active_connections = p.active_connections.saturating_sub(1);
191 p.avg_latency_ms = 0.8 * p.avg_latency_ms + 0.2 * latency_ms;
192 if !success {
193 p.failed_requests += 1;
194 }
195 true
196 }
197 None => false,
198 }
199 }
200
201 pub fn stats(&self) -> LbStats {
205 let total_peers = self.peers.len();
206 let healthy_peers = self.peers.values().filter(|p| p.is_healthy()).count();
207 LbStats {
208 total_peers,
209 healthy_peers,
210 total_requests_routed: self.total_requests,
211 total_request_counter: self.total_requests,
212 }
213 }
214
215 fn healthy_sorted(&self) -> Vec<&PeerLoad> {
219 let mut peers: Vec<&PeerLoad> = self.peers.values().filter(|p| p.is_healthy()).collect();
220 peers.sort_by(|a, b| a.peer_id.cmp(&b.peer_id));
221 peers
222 }
223
224 fn select_round_robin(&self) -> Option<String> {
225 let peers = self.healthy_sorted();
226 if peers.is_empty() {
227 return None;
228 }
229 let idx = (self.total_requests as usize) % peers.len();
230 Some(peers[idx].peer_id.clone())
231 }
232
233 fn select_least_connections(&self) -> Option<String> {
234 self.peers
235 .values()
236 .filter(|p| p.is_healthy())
237 .min_by_key(|p| p.active_connections)
238 .map(|p| p.peer_id.clone())
239 }
240
241 fn select_weighted_random(&self) -> Option<String> {
242 let mut peers = self.healthy_sorted();
243 if peers.is_empty() {
244 return None;
245 }
246
247 let total_weight: u32 = peers.iter().map(|p| p.weight).sum();
248 if total_weight == 0 {
249 let idx = (self.total_requests as usize) % peers.len();
251 return Some(peers[idx].peer_id.clone());
252 }
253
254 let index = (self.total_requests % total_weight as u64) as u32;
255 let mut cumulative: u32 = 0;
256 for p in peers.drain(..) {
257 cumulative += p.weight;
258 if index < cumulative {
259 return Some(p.peer_id.clone());
260 }
261 }
262 None
264 }
265
266 fn select_lowest_latency(&self) -> Option<String> {
267 let peers: Vec<&PeerLoad> = self.peers.values().filter(|p| p.is_healthy()).collect();
268
269 if peers.is_empty() {
270 return None;
271 }
272
273 if peers.iter().all(|p| p.avg_latency_ms == 0.0) {
275 return self.select_round_robin();
276 }
277
278 peers
279 .into_iter()
280 .min_by(|a, b| {
281 a.avg_latency_ms
282 .partial_cmp(&b.avg_latency_ms)
283 .unwrap_or(std::cmp::Ordering::Equal)
284 })
285 .map(|p| p.peer_id.clone())
286 }
287}
288
289#[cfg(test)]
294mod tests {
295 use super::*;
296
297 #[test]
299 fn test_new_starts_empty() {
300 let lb = PeerLoadBalancer::new(LbStrategy::RoundRobin);
301 assert!(lb.peers.is_empty());
302 assert_eq!(lb.total_requests, 0);
303 }
304
305 #[test]
307 fn test_add_peer_creates_with_weight() {
308 let mut lb = PeerLoadBalancer::new(LbStrategy::RoundRobin);
309 lb.add_peer("alpha", 5);
310 let p = lb.peers.get("alpha").expect("peer must exist");
311 assert_eq!(p.weight, 5);
312 assert_eq!(p.active_connections, 0);
313 assert_eq!(p.total_requests, 0);
314 }
315
316 #[test]
318 fn test_add_peer_updates_weight() {
319 let mut lb = PeerLoadBalancer::new(LbStrategy::RoundRobin);
320 lb.add_peer("alpha", 1);
321 lb.add_peer("alpha", 10);
322 assert_eq!(
323 lb.peers
324 .get("alpha")
325 .expect("test: alpha peer must exist after duplicate add")
326 .weight,
327 10
328 );
329 assert_eq!(lb.peers.len(), 1);
330 }
331
332 #[test]
334 fn test_remove_peer_true() {
335 let mut lb = PeerLoadBalancer::new(LbStrategy::RoundRobin);
336 lb.add_peer("alpha", 1);
337 assert!(lb.remove_peer("alpha"));
338 }
339
340 #[test]
342 fn test_remove_peer_false() {
343 let mut lb = PeerLoadBalancer::new(LbStrategy::RoundRobin);
344 assert!(!lb.remove_peer("ghost"));
345 }
346
347 #[test]
349 fn test_select_none_when_no_peers() {
350 let mut lb = PeerLoadBalancer::new(LbStrategy::RoundRobin);
351 assert!(lb.select().is_none());
352 }
353
354 #[test]
356 fn test_round_robin_cycles_alpha_order() {
357 let mut lb = PeerLoadBalancer::new(LbStrategy::RoundRobin);
358 lb.add_peer("charlie", 1);
359 lb.add_peer("alpha", 1);
360 lb.add_peer("bravo", 1);
361
362 let first = lb
364 .select()
365 .expect("test: round-robin first select must succeed");
366 assert_eq!(first, "alpha");
367 let second = lb
368 .select()
369 .expect("test: round-robin second select must succeed");
370 assert_eq!(second, "bravo");
371 let third = lb
372 .select()
373 .expect("test: round-robin third select must succeed");
374 assert_eq!(third, "charlie");
375 let fourth = lb
377 .select()
378 .expect("test: round-robin wrap-around select must succeed");
379 assert_eq!(fourth, "alpha");
380 }
381
382 #[test]
384 fn test_least_connections() {
385 let mut lb = PeerLoadBalancer::new(LbStrategy::LeastConnections);
386 lb.add_peer("a", 1);
387 lb.add_peer("b", 1);
388 lb.peers
389 .get_mut("a")
390 .expect("test: peer 'a' must exist")
391 .active_connections = 5;
392 lb.peers
393 .get_mut("b")
394 .expect("test: peer 'b' must exist")
395 .active_connections = 2;
396
397 let chosen = lb
398 .select()
399 .expect("test: least-connections select must succeed");
400 assert_eq!(chosen, "b");
401 }
402
403 #[test]
405 fn test_weighted_random_proportional() {
406 let mut lb = PeerLoadBalancer::new(LbStrategy::WeightedRandom);
411 lb.add_peer("a", 1);
412 lb.add_peer("b", 2);
413
414 let r0 = lb
416 .select()
417 .expect("test: weighted-random select 0 must succeed"); assert_eq!(r0, "a");
420
421 let r1 = lb
423 .select()
424 .expect("test: weighted-random select 1 must succeed"); assert_eq!(r1, "b");
426
427 let r2 = lb
429 .select()
430 .expect("test: weighted-random select 2 must succeed"); assert_eq!(r2, "b");
432
433 let r3 = lb
435 .select()
436 .expect("test: weighted-random select 3 must succeed");
437 assert_eq!(r3, "a");
438 }
439
440 #[test]
442 fn test_lowest_latency_picks_lowest() {
443 let mut lb = PeerLoadBalancer::new(LbStrategy::LowestLatency);
444 lb.add_peer("fast", 1);
445 lb.add_peer("slow", 1);
446 lb.peers
447 .get_mut("fast")
448 .expect("test: peer 'fast' must exist")
449 .avg_latency_ms = 10.0;
450 lb.peers
451 .get_mut("slow")
452 .expect("test: peer 'slow' must exist")
453 .avg_latency_ms = 100.0;
454
455 let chosen = lb
456 .select()
457 .expect("test: lowest-latency select must succeed");
458 assert_eq!(chosen, "fast");
459 }
460
461 #[test]
463 fn test_lowest_latency_fallback_round_robin() {
464 let mut lb = PeerLoadBalancer::new(LbStrategy::LowestLatency);
465 lb.add_peer("beta", 1);
466 lb.add_peer("alpha", 1);
467 let first = lb
469 .select()
470 .expect("test: lowest-latency fallback first select must succeed");
471 assert_eq!(first, "alpha");
472 let second = lb
473 .select()
474 .expect("test: lowest-latency fallback second select must succeed");
475 assert_eq!(second, "beta");
476 }
477
478 #[test]
480 fn test_select_skips_unhealthy() {
481 let mut lb = PeerLoadBalancer::new(LbStrategy::RoundRobin);
482 lb.add_peer("sick", 1);
483 lb.add_peer("well", 1);
484 lb.peers
486 .get_mut("sick")
487 .expect("test: peer 'sick' must exist")
488 .total_requests = 2;
489 lb.peers
490 .get_mut("sick")
491 .expect("test: peer 'sick' must exist")
492 .failed_requests = 2;
493
494 let chosen = lb
496 .select()
497 .expect("test: select skipping unhealthy must return healthy peer");
498 assert_eq!(chosen, "well");
499 }
500
501 #[test]
503 fn test_record_start_increments_connections() {
504 let mut lb = PeerLoadBalancer::new(LbStrategy::RoundRobin);
505 lb.add_peer("x", 1);
506 assert!(lb.record_start("x"));
507 assert_eq!(lb.peers["x"].active_connections, 1);
508 assert_eq!(lb.peers["x"].total_requests, 1);
509 }
510
511 #[test]
513 fn test_record_start_false_unknown() {
514 let mut lb = PeerLoadBalancer::new(LbStrategy::RoundRobin);
515 assert!(!lb.record_start("unknown"));
516 }
517
518 #[test]
520 fn test_record_complete_decrements_connections() {
521 let mut lb = PeerLoadBalancer::new(LbStrategy::RoundRobin);
522 lb.add_peer("x", 1);
523 lb.record_start("x");
524 lb.record_complete("x", 50.0, true);
525 assert_eq!(lb.peers["x"].active_connections, 0);
526 }
527
528 #[test]
530 fn test_record_complete_ewma_latency() {
531 let mut lb = PeerLoadBalancer::new(LbStrategy::RoundRobin);
532 lb.add_peer("x", 1);
533 lb.record_complete("x", 100.0, true);
535 let lat = lb.peers["x"].avg_latency_ms;
536 assert!((lat - 20.0).abs() < f64::EPSILON);
537
538 lb.record_complete("x", 50.0, true);
540 let lat2 = lb.peers["x"].avg_latency_ms;
541 assert!((lat2 - 26.0).abs() < f64::EPSILON);
542 }
543
544 #[test]
546 fn test_record_complete_failure_increments_failed() {
547 let mut lb = PeerLoadBalancer::new(LbStrategy::RoundRobin);
548 lb.add_peer("x", 1);
549 lb.record_complete("x", 10.0, false);
550 assert_eq!(lb.peers["x"].failed_requests, 1);
551 }
552
553 #[test]
555 fn test_failure_rate() {
556 let mut p = PeerLoad::new("p", 1);
557 assert_eq!(p.failure_rate(), 0.0); p.total_requests = 4;
559 p.failed_requests = 1;
560 assert!((p.failure_rate() - 0.25).abs() < f64::EPSILON);
561 }
562
563 #[test]
565 fn test_is_healthy_false_high_failure_rate() {
566 let mut p = PeerLoad::new("p", 1);
567 p.total_requests = 2;
568 p.failed_requests = 1; assert!(!p.is_healthy());
570 }
571
572 #[test]
574 fn test_is_healthy_false_too_many_connections() {
575 let mut p = PeerLoad::new("p", 1);
576 p.active_connections = 1000;
577 assert!(!p.is_healthy());
578 }
579
580 #[test]
582 fn test_stats_peer_counts() {
583 let mut lb = PeerLoadBalancer::new(LbStrategy::RoundRobin);
584 lb.add_peer("a", 1);
585 lb.add_peer("b", 1);
586 lb.peers
587 .get_mut("b")
588 .expect("test: peer 'b' must exist")
589 .total_requests = 2;
590 lb.peers
591 .get_mut("b")
592 .expect("test: peer 'b' must exist")
593 .failed_requests = 2; let s = lb.stats();
596 assert_eq!(s.total_peers, 2);
597 assert_eq!(s.healthy_peers, 1);
598 }
599
600 #[test]
602 fn test_stats_requests_routed() {
603 let mut lb = PeerLoadBalancer::new(LbStrategy::RoundRobin);
604 lb.add_peer("a", 1);
605 lb.select();
606 lb.select();
607 assert_eq!(lb.stats().total_requests_routed, 2);
608 }
609
610 #[test]
612 fn test_saturating_decrement() {
613 let mut lb = PeerLoadBalancer::new(LbStrategy::RoundRobin);
614 lb.add_peer("x", 1);
615 lb.record_complete("x", 10.0, true);
617 assert_eq!(lb.peers["x"].active_connections, 0);
618 }
619
620 #[test]
622 fn test_select_none_all_unhealthy() {
623 let mut lb = PeerLoadBalancer::new(LbStrategy::LeastConnections);
624 lb.add_peer("a", 1);
625 lb.peers
626 .get_mut("a")
627 .expect("test: peer 'a' must exist")
628 .active_connections = 1000; assert!(lb.select().is_none());
630 }
631
632 #[test]
634 fn test_multiple_peers_round_robin() {
635 let mut lb = PeerLoadBalancer::new(LbStrategy::RoundRobin);
636 for name in &["peer-c", "peer-a", "peer-b"] {
637 lb.add_peer(name, 1);
638 }
639 let expected = ["peer-a", "peer-b", "peer-c", "peer-a", "peer-b", "peer-c"];
641 for &exp in &expected {
642 assert_eq!(
643 lb.select()
644 .expect("test: multi-peer round-robin select must succeed"),
645 exp
646 );
647 }
648 }
649
650 #[test]
652 fn test_record_complete_false_unknown() {
653 let mut lb = PeerLoadBalancer::new(LbStrategy::RoundRobin);
654 assert!(!lb.record_complete("ghost", 10.0, true));
655 }
656
657 #[test]
659 fn test_least_connections_with_two_zeros() {
660 let mut lb = PeerLoadBalancer::new(LbStrategy::LeastConnections);
661 lb.add_peer("x", 1);
662 lb.add_peer("y", 1);
663 assert!(lb.select().is_some());
665 }
666
667 #[test]
669 fn test_weighted_random_single_peer() {
670 let mut lb = PeerLoadBalancer::new(LbStrategy::WeightedRandom);
671 lb.add_peer("only", 7);
672 for _ in 0..10 {
673 assert_eq!(
674 lb.select()
675 .expect("test: weighted-random single peer must always select it"),
676 "only"
677 );
678 }
679 }
680}
681
682#[derive(Clone, Copy, Debug, PartialEq, Eq)]
688pub enum LbAlgorithm {
689 RoundRobin,
691 LeastConnections,
693 WeightedRoundRobin,
695 LeastLatency,
697 Random,
699 ConsistentHash,
701}
702
703#[derive(Clone, Debug)]
705pub struct LbPeer {
706 pub peer_id: String,
708 pub weight: u32,
710 pub active_connections: u32,
712 pub total_requests: u64,
714 pub total_errors: u64,
716 pub avg_latency_ms: f64,
718 pub healthy: bool,
720 pub last_used: u64,
722}
723
724impl LbPeer {
725 pub fn new(peer_id: impl Into<String>, weight: u32) -> Self {
727 Self {
728 peer_id: peer_id.into(),
729 weight,
730 active_connections: 0,
731 total_requests: 0,
732 total_errors: 0,
733 avg_latency_ms: 0.0,
734 healthy: true,
735 last_used: 0,
736 }
737 }
738
739 pub fn error_rate(&self) -> f64 {
741 if self.total_requests == 0 {
742 0.0
743 } else {
744 self.total_errors as f64 / self.total_requests as f64
745 }
746 }
747}
748
749#[derive(Clone, Debug)]
751pub struct LbRequest {
752 pub id: u64,
754 pub key: Option<String>,
756 pub size_bytes: usize,
758 pub timestamp: u64,
760}
761
762impl LbRequest {
763 pub fn new(id: u64, key: Option<String>, size_bytes: usize, timestamp: u64) -> Self {
765 Self {
766 id,
767 key,
768 size_bytes,
769 timestamp,
770 }
771 }
772}
773
774#[derive(Clone, Debug)]
776pub struct LbDecision {
777 pub peer_id: String,
779 pub algorithm_used: String,
781 pub request_id: u64,
783 pub decided_at: u64,
785}
786
787#[derive(Clone, Debug)]
789pub struct AdaptiveLbStats {
790 pub total_peers: usize,
792 pub healthy_peers: usize,
794 pub total_requests: u64,
796 pub total_errors: u64,
798 pub error_rate: f64,
800 pub avg_latency_ms: f64,
802}
803
804#[inline]
810fn fnv1a_64_bytes(data: &[u8]) -> u64 {
811 const OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
812 const PRIME: u64 = 0x0000_0100_0000_01b3;
813 let mut h = OFFSET;
814 for &b in data {
815 h ^= u64::from(b);
816 h = h.wrapping_mul(PRIME);
817 }
818 h
819}
820
821#[inline]
823fn fnv1a_64(s: &str) -> u64 {
824 fnv1a_64_bytes(s.as_bytes())
825}
826
827#[inline]
833fn xorshift64(state: &mut u64) -> u64 {
834 let mut x = *state;
835 if x == 0 {
837 x = 0x853c_49e6_748f_ea9b;
838 }
839 x ^= x << 13;
840 x ^= x >> 7;
841 x ^= x << 17;
842 *state = x;
843 x
844}
845
846pub struct AdaptiveLoadBalancer {
853 pub algorithm: LbAlgorithm,
855 pub peers: Vec<LbPeer>,
857 pub rr_index: usize,
859 pub virtual_nodes: usize,
861 pub total_requests: u64,
863 pub total_errors: u64,
865 ring: Vec<(u64, usize)>,
867 ring_dirty: bool,
869 prng_state: u64,
871}
872
873impl AdaptiveLoadBalancer {
874 pub fn new(algorithm: LbAlgorithm) -> Self {
879 Self {
880 algorithm,
881 peers: Vec::new(),
882 rr_index: 0,
883 virtual_nodes: 150,
884 total_requests: 0,
885 total_errors: 0,
886 ring: Vec::new(),
887 ring_dirty: true,
888 prng_state: 0,
889 }
890 }
891
892 pub fn add_peer(&mut self, peer: LbPeer) {
897 if let Some(existing) = self.peers.iter_mut().find(|p| p.peer_id == peer.peer_id) {
898 *existing = peer;
899 } else {
900 self.peers.push(peer);
901 }
902 self.ring_dirty = true;
903 }
904
905 pub fn remove_peer(&mut self, peer_id: &str) -> bool {
907 let before = self.peers.len();
908 self.peers.retain(|p| p.peer_id != peer_id);
909 let removed = self.peers.len() < before;
910 if removed {
911 self.ring_dirty = true;
912 if !self.peers.is_empty() {
914 self.rr_index %= self.peers.len();
915 } else {
916 self.rr_index = 0;
917 }
918 }
919 removed
920 }
921
922 pub fn select(&mut self, request: &LbRequest, now: u64) -> Option<LbDecision> {
928 let (peer_id, algo_name) = match self.algorithm {
929 LbAlgorithm::RoundRobin => self.select_round_robin()?,
930 LbAlgorithm::LeastConnections => self.select_least_connections()?,
931 LbAlgorithm::WeightedRoundRobin => self.select_weighted_round_robin()?,
932 LbAlgorithm::LeastLatency => self.select_least_latency()?,
933 LbAlgorithm::Random => self.select_random(request)?,
934 LbAlgorithm::ConsistentHash => self.select_consistent_hash(request)?,
935 };
936
937 self.total_requests += 1;
938
939 if let Some(p) = self.peers.iter_mut().find(|p| p.peer_id == peer_id) {
941 p.last_used = now;
942 }
943
944 Some(LbDecision {
945 peer_id,
946 algorithm_used: algo_name,
947 request_id: request.id,
948 decided_at: now,
949 })
950 }
951
952 pub fn record_connection_start(&mut self, peer_id: &str) -> bool {
956 match self.peers.iter_mut().find(|p| p.peer_id == peer_id) {
957 Some(p) => {
958 p.active_connections = p.active_connections.saturating_add(1);
959 true
960 }
961 None => false,
962 }
963 }
964
965 pub fn record_completion(
973 &mut self,
974 peer_id: &str,
975 latency_ms: f64,
976 success: bool,
977 _now: u64,
978 ) -> bool {
979 const ALPHA: f64 = 0.2;
980 match self.peers.iter_mut().find(|p| p.peer_id == peer_id) {
981 Some(p) => {
982 p.active_connections = p.active_connections.saturating_sub(1);
983 if p.avg_latency_ms == 0.0 {
985 p.avg_latency_ms = latency_ms;
986 } else {
987 p.avg_latency_ms = ALPHA * latency_ms + (1.0 - ALPHA) * p.avg_latency_ms;
988 }
989 if !success {
990 p.total_errors += 1;
991 self.total_errors += 1;
992 }
993 true
994 }
995 None => false,
996 }
997 }
998
999 pub fn mark_healthy(&mut self, peer_id: &str, healthy: bool) -> bool {
1001 match self.peers.iter_mut().find(|p| p.peer_id == peer_id) {
1002 Some(p) => {
1003 p.healthy = healthy;
1004 self.ring_dirty = true;
1005 true
1006 }
1007 None => false,
1008 }
1009 }
1010
1011 pub fn adjust_weights_by_latency(&mut self) {
1018 for p in self.peers.iter_mut() {
1019 if p.avg_latency_ms > 0.0 {
1020 p.weight = ((1000.0 / p.avg_latency_ms).round() as u32).max(1);
1021 }
1022 }
1023 }
1024
1025 pub fn healthy_peers(&self) -> Vec<&LbPeer> {
1029 self.peers.iter().filter(|p| p.healthy).collect()
1030 }
1031
1032 pub fn peer_count(&self) -> usize {
1034 self.peers.len()
1035 }
1036
1037 pub fn healthy_peer_count(&self) -> usize {
1039 self.peers.iter().filter(|p| p.healthy).count()
1040 }
1041
1042 pub fn stats(&self) -> AdaptiveLbStats {
1044 let total_peers = self.peers.len();
1045 let healthy_peers = self.healthy_peer_count();
1046
1047 let error_rate = if self.total_requests == 0 {
1048 0.0
1049 } else {
1050 self.total_errors as f64 / self.total_requests as f64
1051 };
1052
1053 let latency_peers: Vec<f64> = self
1055 .peers
1056 .iter()
1057 .filter(|p| p.avg_latency_ms > 0.0)
1058 .map(|p| p.avg_latency_ms)
1059 .collect();
1060 let avg_latency_ms = if latency_peers.is_empty() {
1061 0.0
1062 } else {
1063 latency_peers.iter().sum::<f64>() / latency_peers.len() as f64
1064 };
1065
1066 AdaptiveLbStats {
1067 total_peers,
1068 healthy_peers,
1069 total_requests: self.total_requests,
1070 total_errors: self.total_errors,
1071 error_rate,
1072 avg_latency_ms,
1073 }
1074 }
1075
1076 fn rebuild_ring_if_needed(&mut self) {
1079 if !self.ring_dirty {
1080 return;
1081 }
1082 self.ring.clear();
1083 for (idx, peer) in self.peers.iter().enumerate() {
1084 if peer.healthy {
1085 for i in 0..self.virtual_nodes {
1086 let key = format!("{}-{}", peer.peer_id, i);
1087 let hash = fnv1a_64(&key);
1088 self.ring.push((hash, idx));
1089 }
1090 }
1091 }
1092 self.ring.sort_unstable_by_key(|(h, _)| *h);
1093 self.ring_dirty = false;
1094 }
1095
1096 fn select_round_robin(&mut self) -> Option<(String, String)> {
1099 let healthy: Vec<usize> = self
1100 .peers
1101 .iter()
1102 .enumerate()
1103 .filter(|(_, p)| p.healthy)
1104 .map(|(i, _)| i)
1105 .collect();
1106 if healthy.is_empty() {
1107 return None;
1108 }
1109 let pos = self.rr_index % healthy.len();
1110 self.rr_index = pos + 1;
1111 Some((
1112 self.peers[healthy[pos]].peer_id.clone(),
1113 "RoundRobin".to_string(),
1114 ))
1115 }
1116
1117 fn select_least_connections(&self) -> Option<(String, String)> {
1118 self.peers
1119 .iter()
1120 .filter(|p| p.healthy)
1121 .min_by_key(|p| p.active_connections)
1122 .map(|p| (p.peer_id.clone(), "LeastConnections".to_string()))
1123 }
1124
1125 fn select_weighted_round_robin(&mut self) -> Option<(String, String)> {
1126 let healthy: Vec<&LbPeer> = self.peers.iter().filter(|p| p.healthy).collect();
1127 if healthy.is_empty() {
1128 return None;
1129 }
1130 let total_weight: u64 = healthy.iter().map(|p| u64::from(p.weight)).sum();
1131 if total_weight == 0 {
1132 let pos = self.rr_index % healthy.len();
1134 self.rr_index = pos + 1;
1135 return Some((
1136 healthy[pos].peer_id.clone(),
1137 "WeightedRoundRobin".to_string(),
1138 ));
1139 }
1140 let index = self.total_requests % total_weight;
1141 let mut cumulative: u64 = 0;
1142 for p in &healthy {
1143 cumulative += u64::from(p.weight);
1144 if index < cumulative {
1145 return Some((p.peer_id.clone(), "WeightedRoundRobin".to_string()));
1146 }
1147 }
1148 healthy
1150 .last()
1151 .map(|p| (p.peer_id.clone(), "WeightedRoundRobin".to_string()))
1152 }
1153
1154 fn select_least_latency(&self) -> Option<(String, String)> {
1155 let healthy: Vec<&LbPeer> = self.peers.iter().filter(|p| p.healthy).collect();
1156 if healthy.is_empty() {
1157 return None;
1158 }
1159 if healthy.iter().all(|p| p.avg_latency_ms == 0.0) {
1161 return healthy
1162 .first()
1163 .map(|p| (p.peer_id.clone(), "LeastLatency".to_string()));
1164 }
1165 healthy
1166 .iter()
1167 .filter(|p| p.avg_latency_ms > 0.0)
1168 .min_by(|a, b| {
1169 a.avg_latency_ms
1170 .partial_cmp(&b.avg_latency_ms)
1171 .unwrap_or(std::cmp::Ordering::Equal)
1172 })
1173 .or_else(|| healthy.first())
1174 .map(|p| (p.peer_id.clone(), "LeastLatency".to_string()))
1175 }
1176
1177 fn select_random(&mut self, request: &LbRequest) -> Option<(String, String)> {
1178 let healthy: Vec<&LbPeer> = self.peers.iter().filter(|p| p.healthy).collect();
1179 if healthy.is_empty() {
1180 return None;
1181 }
1182 if self.prng_state == 0 {
1184 self.prng_state = request.id.wrapping_add(request.timestamp).wrapping_add(1);
1185 }
1186 let r = xorshift64(&mut self.prng_state);
1187 let idx = (r as usize) % healthy.len();
1188 Some((healthy[idx].peer_id.clone(), "Random".to_string()))
1189 }
1190
1191 fn select_consistent_hash(&mut self, request: &LbRequest) -> Option<(String, String)> {
1192 self.rebuild_ring_if_needed();
1193 if self.ring.is_empty() {
1194 return None;
1195 }
1196 let key_hash = match &request.key {
1198 Some(k) => fnv1a_64(k),
1199 None => {
1200 let mut s = self.prng_state.wrapping_add(request.id).wrapping_add(1);
1202 xorshift64(&mut s)
1203 }
1204 };
1205
1206 let start = match self.ring.binary_search_by_key(&key_hash, |(h, _)| *h) {
1208 Ok(i) | Err(i) => i % self.ring.len(),
1209 };
1210
1211 let ring_len = self.ring.len();
1213 for offset in 0..ring_len {
1214 let (_, peer_idx) = self.ring[(start + offset) % ring_len];
1215 if peer_idx < self.peers.len() && self.peers[peer_idx].healthy {
1217 return Some((
1218 self.peers[peer_idx].peer_id.clone(),
1219 "ConsistentHash".to_string(),
1220 ));
1221 }
1222 }
1223 None
1224 }
1225}
1226
1227#[cfg(test)]
1232mod adaptive_tests {
1233 use crate::load_balancer::{AdaptiveLoadBalancer, LbAlgorithm, LbPeer, LbRequest};
1234
1235 fn req(id: u64, key: Option<&str>) -> LbRequest {
1237 LbRequest::new(id, key.map(|s| s.to_string()), 0, 1000)
1238 }
1239
1240 fn peer(id: &str, weight: u32) -> LbPeer {
1241 LbPeer::new(id, weight)
1242 }
1243
1244 #[test]
1246 fn test_new_empty() {
1247 let lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1248 assert_eq!(lb.peer_count(), 0);
1249 assert_eq!(lb.total_requests, 0);
1250 assert_eq!(lb.total_errors, 0);
1251 assert_eq!(lb.virtual_nodes, 150);
1252 }
1253
1254 #[test]
1256 fn test_add_peer_inserts() {
1257 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1258 lb.add_peer(peer("alice", 1));
1259 assert_eq!(lb.peer_count(), 1);
1260 }
1261
1262 #[test]
1264 fn test_add_peer_replaces_duplicate() {
1265 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1266 lb.add_peer(peer("alice", 1));
1267 let mut p2 = peer("alice", 10);
1268 p2.avg_latency_ms = 42.0;
1269 lb.add_peer(p2);
1270 assert_eq!(lb.peer_count(), 1);
1271 assert_eq!(lb.peers[0].weight, 10);
1272 assert!((lb.peers[0].avg_latency_ms - 42.0).abs() < f64::EPSILON);
1273 }
1274
1275 #[test]
1277 fn test_remove_peer_present() {
1278 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1279 lb.add_peer(peer("alice", 1));
1280 assert!(lb.remove_peer("alice"));
1281 assert_eq!(lb.peer_count(), 0);
1282 }
1283
1284 #[test]
1286 fn test_remove_peer_missing() {
1287 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1288 assert!(!lb.remove_peer("ghost"));
1289 }
1290
1291 #[test]
1293 fn test_select_none_no_peers() {
1294 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1295 assert!(lb.select(&req(1, None), 100).is_none());
1296 }
1297
1298 #[test]
1300 fn test_select_none_all_unhealthy() {
1301 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1302 let mut p = peer("a", 1);
1303 p.healthy = false;
1304 lb.add_peer(p);
1305 assert!(lb.select(&req(1, None), 100).is_none());
1306 }
1307
1308 #[test]
1310 fn test_round_robin_cycles() {
1311 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1312 lb.add_peer(peer("a", 1));
1313 lb.add_peer(peer("b", 1));
1314 lb.add_peer(peer("c", 1));
1315
1316 let d0 = lb.select(&req(0, None), 0).expect("must route");
1317 let d1 = lb.select(&req(1, None), 0).expect("must route");
1318 let d2 = lb.select(&req(2, None), 0).expect("must route");
1319 let d3 = lb.select(&req(3, None), 0).expect("must route");
1320
1321 assert_eq!(d0.peer_id, "a");
1322 assert_eq!(d1.peer_id, "b");
1323 assert_eq!(d2.peer_id, "c");
1324 assert_eq!(d3.peer_id, "a"); }
1326
1327 #[test]
1329 fn test_round_robin_skips_unhealthy() {
1330 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1331 lb.add_peer(peer("a", 1));
1332 let mut sick = peer("b", 1);
1333 sick.healthy = false;
1334 lb.add_peer(sick);
1335 lb.add_peer(peer("c", 1));
1336
1337 for _ in 0..6 {
1338 let d = lb.select(&req(0, None), 0).expect("must route");
1339 assert_ne!(d.peer_id, "b", "unhealthy peer must not be chosen");
1340 }
1341 }
1342
1343 #[test]
1345 fn test_least_connections() {
1346 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::LeastConnections);
1347 let mut a = peer("a", 1);
1348 a.active_connections = 5;
1349 lb.add_peer(a);
1350 let mut b = peer("b", 1);
1351 b.active_connections = 1;
1352 lb.add_peer(b);
1353 let d = lb.select(&req(1, None), 0).expect("must route");
1354 assert_eq!(d.peer_id, "b");
1355 }
1356
1357 #[test]
1359 fn test_weighted_round_robin_proportional() {
1360 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::WeightedRoundRobin);
1361 lb.add_peer(peer("a", 1)); lb.add_peer(peer("b", 3)); let d0 = lb.select(&req(0, None), 0).expect("d0");
1366 let d1 = lb.select(&req(1, None), 0).expect("d1");
1368 let d2 = lb.select(&req(2, None), 0).expect("d2");
1370 let d3 = lb.select(&req(3, None), 0).expect("d3");
1372 let d4 = lb.select(&req(4, None), 0).expect("d4");
1374
1375 assert_eq!(d0.peer_id, "a");
1376 assert_eq!(d1.peer_id, "b");
1377 assert_eq!(d2.peer_id, "b");
1378 assert_eq!(d3.peer_id, "b");
1379 assert_eq!(d4.peer_id, "a");
1380 }
1381
1382 #[test]
1384 fn test_least_latency_picks_lowest() {
1385 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::LeastLatency);
1386 let mut fast = peer("fast", 1);
1387 fast.avg_latency_ms = 5.0;
1388 lb.add_peer(fast);
1389 let mut slow = peer("slow", 1);
1390 slow.avg_latency_ms = 200.0;
1391 lb.add_peer(slow);
1392 let d = lb.select(&req(1, None), 0).expect("must route");
1393 assert_eq!(d.peer_id, "fast");
1394 }
1395
1396 #[test]
1398 fn test_least_latency_fallback_all_zero() {
1399 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::LeastLatency);
1400 lb.add_peer(peer("x", 1));
1401 lb.add_peer(peer("y", 1));
1402 assert!(lb.select(&req(1, None), 0).is_some());
1404 }
1405
1406 #[test]
1408 fn test_random_selects_healthy() {
1409 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::Random);
1410 lb.add_peer(peer("a", 1));
1411 lb.add_peer(peer("b", 1));
1412 let mut sick = peer("c", 1);
1413 sick.healthy = false;
1414 lb.add_peer(sick);
1415
1416 for i in 0..50u64 {
1417 let d = lb.select(&req(i, None), i).expect("must route");
1418 assert_ne!(d.peer_id, "c", "unhealthy peer must not be chosen");
1419 }
1420 }
1421
1422 #[test]
1424 fn test_consistent_hash_stable() {
1425 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::ConsistentHash);
1426 lb.add_peer(peer("n1", 1));
1427 lb.add_peer(peer("n2", 1));
1428 lb.add_peer(peer("n3", 1));
1429
1430 let first = lb.select(&req(1, Some("my-key")), 0).expect("must route");
1431 for _ in 0..10 {
1432 let repeated = lb.select(&req(1, Some("my-key")), 0).expect("must route");
1433 assert_eq!(first.peer_id, repeated.peer_id, "same key → same peer");
1434 }
1435 }
1436
1437 #[test]
1439 fn test_consistent_hash_distribution() {
1440 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::ConsistentHash);
1441 lb.add_peer(peer("n1", 1));
1442 lb.add_peer(peer("n2", 1));
1443 lb.add_peer(peer("n3", 1));
1444
1445 let mut seen = std::collections::HashSet::new();
1446 for i in 0..50u64 {
1447 let key = format!("key-{}", i);
1448 let d = lb.select(&req(i, Some(&key)), 0).expect("must route");
1449 seen.insert(d.peer_id);
1450 }
1451 assert!(
1453 seen.len() >= 2,
1454 "consistent hash should distribute across peers"
1455 );
1456 }
1457
1458 #[test]
1460 fn test_record_connection_start() {
1461 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1462 lb.add_peer(peer("x", 1));
1463 assert!(lb.record_connection_start("x"));
1464 assert_eq!(lb.peers[0].active_connections, 1);
1465 }
1466
1467 #[test]
1469 fn test_record_connection_start_unknown() {
1470 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1471 assert!(!lb.record_connection_start("ghost"));
1472 }
1473
1474 #[test]
1476 fn test_record_completion_decrements() {
1477 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1478 lb.add_peer(peer("x", 1));
1479 lb.record_connection_start("x");
1480 assert!(lb.record_completion("x", 50.0, true, 0));
1481 assert_eq!(lb.peers[0].active_connections, 0);
1482 }
1483
1484 #[test]
1486 fn test_record_completion_saturating() {
1487 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1488 lb.add_peer(peer("x", 1));
1489 lb.record_completion("x", 10.0, true, 0);
1491 assert_eq!(lb.peers[0].active_connections, 0);
1492 }
1493
1494 #[test]
1496 fn test_record_completion_first_latency() {
1497 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1498 lb.add_peer(peer("x", 1));
1499 lb.record_completion("x", 100.0, true, 0);
1500 assert!((lb.peers[0].avg_latency_ms - 100.0).abs() < f64::EPSILON);
1501 }
1502
1503 #[test]
1505 fn test_record_completion_ema_latency() {
1506 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1507 lb.add_peer(peer("x", 1));
1508 lb.record_completion("x", 100.0, true, 0); lb.record_completion("x", 200.0, true, 0); assert!((lb.peers[0].avg_latency_ms - 120.0).abs() < 1e-9);
1511 }
1512
1513 #[test]
1515 fn test_record_completion_errors() {
1516 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1517 lb.add_peer(peer("x", 1));
1518 lb.record_completion("x", 10.0, false, 0);
1519 lb.record_completion("x", 10.0, false, 0);
1520 assert_eq!(lb.peers[0].total_errors, 2);
1521 assert_eq!(lb.total_errors, 2);
1522 }
1523
1524 #[test]
1526 fn test_record_completion_unknown() {
1527 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1528 assert!(!lb.record_completion("ghost", 10.0, true, 0));
1529 }
1530
1531 #[test]
1533 fn test_mark_healthy() {
1534 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1535 lb.add_peer(peer("a", 1));
1536 assert!(lb.mark_healthy("a", false));
1537 assert!(!lb.peers[0].healthy);
1538 assert!(lb.mark_healthy("a", true));
1539 assert!(lb.peers[0].healthy);
1540 }
1541
1542 #[test]
1544 fn test_mark_healthy_unknown() {
1545 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1546 assert!(!lb.mark_healthy("ghost", true));
1547 }
1548
1549 #[test]
1551 fn test_adjust_weights_by_latency() {
1552 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::WeightedRoundRobin);
1553 lb.add_peer(peer("fast", 1));
1554 lb.add_peer(peer("slow", 1));
1555 lb.record_completion("fast", 10.0, true, 0); lb.record_completion("slow", 500.0, true, 0); lb.adjust_weights_by_latency();
1559
1560 let fast_w = lb
1561 .peers
1562 .iter()
1563 .find(|p| p.peer_id == "fast")
1564 .map(|p| p.weight)
1565 .unwrap_or(0);
1566 let slow_w = lb
1567 .peers
1568 .iter()
1569 .find(|p| p.peer_id == "slow")
1570 .map(|p| p.weight)
1571 .unwrap_or(0);
1572 assert_eq!(fast_w, 100, "fast peer: 1000/10 = 100");
1573 assert_eq!(slow_w, 2, "slow peer: 1000/500 = 2");
1574 }
1575
1576 #[test]
1578 fn test_adjust_weights_skips_zero_latency() {
1579 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::WeightedRoundRobin);
1580 let p = peer("x", 7);
1581 lb.add_peer(p);
1582 lb.adjust_weights_by_latency();
1583 assert_eq!(
1584 lb.peers[0].weight, 7,
1585 "weight must not change when latency is 0"
1586 );
1587 }
1588
1589 #[test]
1591 fn test_healthy_peers_accessor() {
1592 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1593 lb.add_peer(peer("a", 1));
1594 let mut b = peer("b", 1);
1595 b.healthy = false;
1596 lb.add_peer(b);
1597 lb.add_peer(peer("c", 1));
1598 let hp = lb.healthy_peers();
1599 assert_eq!(hp.len(), 2);
1600 assert!(hp.iter().all(|p| p.healthy));
1601 }
1602
1603 #[test]
1605 fn test_healthy_peer_count() {
1606 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1607 lb.add_peer(peer("a", 1));
1608 let mut b = peer("b", 1);
1609 b.healthy = false;
1610 lb.add_peer(b);
1611 assert_eq!(lb.healthy_peer_count(), 1);
1612 assert_eq!(lb.peer_count(), 2);
1613 }
1614
1615 #[test]
1617 fn test_stats_totals() {
1618 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1619 lb.add_peer(peer("a", 1));
1620 lb.select(&req(1, None), 0);
1621 lb.select(&req(2, None), 0);
1622 lb.record_completion("a", 50.0, false, 0); let s = lb.stats();
1625 assert_eq!(s.total_peers, 1);
1626 assert_eq!(s.healthy_peers, 1);
1627 assert_eq!(s.total_requests, 2);
1628 assert_eq!(s.total_errors, 1);
1629 assert!((s.error_rate - 0.5).abs() < 1e-9);
1630 }
1631
1632 #[test]
1634 fn test_stats_avg_latency() {
1635 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::LeastLatency);
1636 lb.add_peer(peer("a", 1));
1637 lb.add_peer(peer("b", 1));
1638 lb.record_completion("a", 100.0, true, 0); lb.record_completion("b", 200.0, true, 0); let s = lb.stats();
1641 assert!((s.avg_latency_ms - 150.0).abs() < 1e-9);
1643 }
1644
1645 #[test]
1647 fn test_decision_algorithm_name() {
1648 let algos: &[(LbAlgorithm, &str)] = &[
1649 (LbAlgorithm::RoundRobin, "RoundRobin"),
1650 (LbAlgorithm::LeastConnections, "LeastConnections"),
1651 (LbAlgorithm::WeightedRoundRobin, "WeightedRoundRobin"),
1652 (LbAlgorithm::LeastLatency, "LeastLatency"),
1653 (LbAlgorithm::Random, "Random"),
1654 (LbAlgorithm::ConsistentHash, "ConsistentHash"),
1655 ];
1656 for &(algo, name) in algos {
1657 let mut lb = AdaptiveLoadBalancer::new(algo);
1658 lb.add_peer(peer("x", 1));
1659 let d = lb.select(&req(1, Some("k")), 0).expect("must route");
1660 assert_eq!(d.algorithm_used, name, "algorithm mismatch for {:?}", algo);
1661 }
1662 }
1663
1664 #[test]
1666 fn test_decision_metadata() {
1667 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1668 lb.add_peer(peer("a", 1));
1669 let d = lb.select(&req(42, None), 999).expect("must route");
1670 assert_eq!(d.request_id, 42);
1671 assert_eq!(d.decided_at, 999);
1672 }
1673
1674 #[test]
1676 fn test_consistent_hash_no_healthy() {
1677 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::ConsistentHash);
1678 let mut p = peer("a", 1);
1679 p.healthy = false;
1680 lb.add_peer(p);
1681 assert!(lb.select(&req(1, Some("key")), 0).is_none());
1682 }
1683
1684 #[test]
1686 fn test_consistent_hash_reroute_after_unhealthy() {
1687 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::ConsistentHash);
1688 lb.add_peer(peer("n1", 1));
1689 lb.add_peer(peer("n2", 1));
1690 lb.add_peer(peer("n3", 1));
1691
1692 let first_peer = lb
1694 .select(&req(1, Some("testkey")), 0)
1695 .expect("must route")
1696 .peer_id;
1697
1698 lb.mark_healthy(&first_peer, false);
1700 let second_peer = lb
1701 .select(&req(1, Some("testkey")), 0)
1702 .expect("must route after reroute")
1703 .peer_id;
1704 assert_ne!(
1705 second_peer, first_peer,
1706 "should reroute away from unhealthy peer"
1707 );
1708 }
1709
1710 #[test]
1712 fn test_total_requests_increments() {
1713 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1714 lb.add_peer(peer("a", 1));
1715 for i in 0..10u64 {
1716 lb.select(&req(i, None), 0).expect("must route");
1717 }
1718 assert_eq!(lb.total_requests, 10);
1719 }
1720
1721 #[test]
1723 fn test_lb_peer_error_rate_zero() {
1724 let p = LbPeer::new("x", 1);
1725 assert_eq!(p.error_rate(), 0.0);
1726 }
1727
1728 #[test]
1730 fn test_lb_peer_error_rate_computed() {
1731 let mut p = LbPeer::new("x", 1);
1732 p.total_requests = 10;
1733 p.total_errors = 3;
1734 assert!((p.error_rate() - 0.3).abs() < 1e-9);
1735 }
1736
1737 #[test]
1739 fn test_select_sets_last_used() {
1740 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1741 lb.add_peer(peer("a", 1));
1742 lb.select(&req(1, None), 12345).expect("must route");
1743 assert_eq!(lb.peers[0].last_used, 12345);
1744 }
1745
1746 #[test]
1748 fn test_weighted_round_robin_zero_weights() {
1749 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::WeightedRoundRobin);
1750 lb.add_peer(LbPeer {
1751 weight: 0,
1752 ..LbPeer::new("a", 0)
1753 });
1754 lb.add_peer(LbPeer {
1755 weight: 0,
1756 ..LbPeer::new("b", 0)
1757 });
1758 assert!(lb.select(&req(1, None), 0).is_some());
1760 }
1761
1762 #[test]
1764 fn test_remove_peer_clamps_rr_index() {
1765 let mut lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1766 lb.add_peer(peer("a", 1));
1767 lb.add_peer(peer("b", 1));
1768 lb.add_peer(peer("c", 1));
1769 lb.select(&req(0, None), 0);
1771 lb.select(&req(1, None), 0);
1772 lb.remove_peer("c");
1774 assert!(lb.select(&req(2, None), 0).is_some());
1776 }
1777
1778 #[test]
1780 fn test_stats_error_rate_zero_no_requests() {
1781 let lb = AdaptiveLoadBalancer::new(LbAlgorithm::RoundRobin);
1782 assert_eq!(lb.stats().error_rate, 0.0);
1783 }
1784
1785 #[test]
1787 fn test_virtual_nodes_default() {
1788 let lb = AdaptiveLoadBalancer::new(LbAlgorithm::ConsistentHash);
1789 assert_eq!(lb.virtual_nodes, 150);
1790 }
1791
1792 #[test]
1794 fn test_all_algorithms_route() {
1795 let algos = [
1796 LbAlgorithm::RoundRobin,
1797 LbAlgorithm::LeastConnections,
1798 LbAlgorithm::WeightedRoundRobin,
1799 LbAlgorithm::LeastLatency,
1800 LbAlgorithm::Random,
1801 LbAlgorithm::ConsistentHash,
1802 ];
1803 for algo in algos {
1804 let mut lb = AdaptiveLoadBalancer::new(algo);
1805 lb.add_peer(peer("p1", 2));
1806 lb.add_peer(peer("p2", 3));
1807 let d = lb
1808 .select(&req(1, Some("routing-key")), 0)
1809 .unwrap_or_else(|| panic!("algo {:?} returned None", algo));
1810 assert!(!d.peer_id.is_empty());
1811 }
1812 }
1813}