use std::{
cell::RefCell, fmt::Debug, hash::Hasher, net::SocketAddr, ops::Index, rc::Rc, time::Instant,
};
use rand::{
RngExt, SeedableRng,
distr::uniform::{UniformInt, UniformSampler},
prelude::IndexedRandom,
rngs::StdRng,
};
use sozu_command::config::DEFAULT_SHARD_MIN_BACKENDS;
use crate::{
backends::Backend,
sozu_command::proto::command::{LoadMetric, ShardMode},
};
const DEFAULT_WEIGHT: i32 = 100;
pub const DEFAULT_HASH_SEED: u64 = 0x9E37_79B9_7F4A_7C15;
fn backend_weight(backend: &Backend) -> u32 {
let weight = backend
.load_balancing_parameters
.as_ref()
.map(|p| p.weight)
.unwrap_or(DEFAULT_WEIGHT)
.max(1) as u32;
debug_assert!(weight >= 1, "backend weight must be clamped to at least 1");
weight
}
pub(crate) fn outranks_at_address(
other: &Backend,
other_position: usize,
backend: &Backend,
position: usize,
) -> bool {
count_sibling_step();
if other.address != backend.address || other_position == position {
return false;
}
let (other_weight, weight) = (backend_weight(other), backend_weight(backend));
other_weight > weight || (other_weight == weight && other_position < position)
}
#[cfg(test)]
thread_local! {
pub(crate) static SIBLING_STEPS: std::cell::Cell<u64> = const { std::cell::Cell::new(0) };
}
#[inline(always)]
pub(crate) fn count_sibling_step() {
#[cfg(test)]
SIBLING_STEPS.with(|steps| steps.set(steps.get() + 1));
}
fn hash_backend(seed: u64, key: u64, addr: &SocketAddr) -> u64 {
let mut h = FnvHasher::with_seed(seed);
h.write_u64(key);
match addr {
SocketAddr::V4(v4) => {
h.write_u8(4);
h.write(&v4.ip().octets());
h.write_u16(v4.port());
}
SocketAddr::V6(v6) => {
h.write_u8(6);
h.write(&v6.ip().octets());
h.write_u16(v6.port());
}
}
splitmix64_finalize(h.finish())
}
fn splitmix64_finalize(mut z: u64) -> u64 {
z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
z ^ (z >> 31)
}
const AFFINITY_DOMAIN_IP: u8 = 0x49; const AFFINITY_DOMAIN_VALUE: u8 = 0x56;
pub fn affinity_key_from_ip(ip: std::net::IpAddr) -> u64 {
let mut h = FnvHasher::with_seed(DEFAULT_HASH_SEED);
h.write_u8(AFFINITY_DOMAIN_IP);
match ip.to_canonical() {
std::net::IpAddr::V4(v4) => {
h.write_u8(4);
h.write(&v4.octets());
}
std::net::IpAddr::V6(v6) => {
h.write_u8(6);
h.write(&v6.octets());
}
}
splitmix64_finalize(h.finish())
}
pub fn affinity_key_from_value(value: &[u8]) -> u64 {
let mut h = FnvHasher::with_seed(DEFAULT_HASH_SEED);
h.write_u8(AFFINITY_DOMAIN_VALUE);
h.write(value);
splitmix64_finalize(h.finish())
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ShuffleSharding {
pub percent: u32,
pub min_backends: u32,
pub mode: ShardMode,
}
impl ShuffleSharding {
pub fn from_cluster(
shard_percent: Option<u32>,
shard_min_backends: Option<u32>,
shard_mode: Option<i32>,
) -> Option<Self> {
let percent = shard_percent?;
Some(Self {
percent,
min_backends: shard_min_backends.unwrap_or(DEFAULT_SHARD_MIN_BACKENDS),
mode: shard_mode
.and_then(|mode| ShardMode::try_from(mode).ok())
.unwrap_or(ShardMode::Fallback),
})
}
pub fn shard_size(&self, primaries: usize) -> Option<usize> {
if primaries < self.min_backends as usize || primaries == 0 {
return None;
}
let k = (self.percent as usize * primaries)
.div_ceil(100)
.max(2)
.min(primaries);
debug_assert!(
(1..=primaries).contains(&k),
"a shard holds at least one and at most every primary backend"
);
debug_assert!(
k >= 2 || primaries < 2,
"a shard keeps room for a retry whenever the cluster has two backends"
);
Some(k)
}
}
fn next_prime(n: usize) -> usize {
let mut candidate = n.max(2);
while !is_prime(candidate) {
candidate += 1;
}
debug_assert!(is_prime(candidate), "next_prime must return a prime");
debug_assert!(candidate >= 2, "next_prime must return at least 2");
debug_assert!(
candidate >= n,
"next_prime ({candidate}) must be >= the requested floor ({n})"
);
candidate
}
fn is_prime(x: usize) -> bool {
if x < 2 {
return false;
}
if x.is_multiple_of(2) {
return x == 2;
}
let mut d = 3;
while d * d <= x {
if x.is_multiple_of(d) {
return false;
}
d += 2;
}
true
}
struct FnvHasher {
state: u64,
}
impl FnvHasher {
const PRIME: u64 = 0x0000_0100_0000_01B3;
const OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
fn with_seed(seed: u64) -> Self {
Self {
state: Self::OFFSET ^ seed,
}
}
}
impl Hasher for FnvHasher {
fn finish(&self) -> u64 {
self.state
}
fn write(&mut self, bytes: &[u8]) {
for &b in bytes {
self.state ^= u64::from(b);
self.state = self.state.wrapping_mul(Self::PRIME);
}
}
}
#[derive(Clone, Copy, Debug)]
pub struct Candidates<'a> {
backends: &'a [Rc<RefCell<Backend>>],
indices: &'a [usize],
siblings: &'a [usize],
now: Instant,
}
impl<'a> Candidates<'a> {
pub fn new(backends: &'a [Rc<RefCell<Backend>>], indices: &'a [usize], now: Instant) -> Self {
debug_assert!(
indices.iter().all(|&index| index < backends.len()),
"a candidate index must address a slot of the backend list"
);
debug_assert!(
indices.windows(2).all(|pair| pair[0] < pair[1]),
"candidate indices must be strictly increasing"
);
Self {
backends,
indices,
siblings: &[],
now,
}
}
pub fn with_siblings(
backends: &'a [Rc<RefCell<Backend>>],
indices: &'a [usize],
siblings: &'a [usize],
now: Instant,
) -> Self {
let siblings = if siblings.len() == backends.len() {
siblings
} else {
&[]
};
debug_assert!(
siblings.iter().all(|&next| next < backends.len()),
"a sibling must address a slot of the backend list"
);
Self {
siblings,
..Self::new(backends, indices, now)
}
}
pub fn siblings(
&self,
index: usize,
) -> impl Iterator<Item = &'a Rc<RefCell<Backend>>> + use<'a> {
let start = self.indices[index];
let ring = self.siblings;
let backends = self.backends;
let next = move |position: usize| ring.get(position).copied().unwrap_or(start);
std::iter::successors(Some(next(start)), move |&position| Some(next(position)))
.take_while(move |&position| position != start)
.map(move |position| {
count_sibling_step();
&backends[position]
})
}
pub fn now(&self) -> Instant {
self.now
}
pub fn len(&self) -> usize {
self.indices.len()
}
pub fn is_empty(&self) -> bool {
self.indices.is_empty()
}
pub fn get(&self, index: usize) -> Option<&'a Rc<RefCell<Backend>>> {
self.indices.get(index).map(|&slot| &self.backends[slot])
}
pub fn iter(&self) -> impl ExactSizeIterator<Item = &'a Rc<RefCell<Backend>>> + use<'a> {
let backends = self.backends;
self.indices.iter().map(move |&slot| &backends[slot])
}
fn choose<R: rand::Rng + ?Sized>(&self, rng: &mut R) -> Option<&'a Rc<RefCell<Backend>>> {
self.indices.choose(rng).map(|&slot| &self.backends[slot])
}
}
impl Index<usize> for Candidates<'_> {
type Output = Rc<RefCell<Backend>>;
fn index(&self, index: usize) -> &Self::Output {
&self.backends[self.indices[index]]
}
}
pub trait LoadBalancingAlgorithm: Debug {
fn next_available_backend(
&mut self,
key: Option<u64>,
candidates: Candidates<'_>,
) -> Option<Rc<RefCell<Backend>>>;
fn rebuild(&mut self, _backends: &[Rc<RefCell<Backend>>]) {}
}
#[derive(Debug)]
pub struct RoundRobin {
pub next_backend: u32,
}
impl LoadBalancingAlgorithm for RoundRobin {
fn next_available_backend(
&mut self,
_key: Option<u64>,
backends: Candidates<'_>,
) -> Option<Rc<RefCell<Backend>>> {
if backends.is_empty() {
return None;
}
debug_assert!(
!backends.is_empty(),
"round-robin index math runs only on a non-empty set"
);
let index = self.next_backend as usize % backends.len();
debug_assert!(index < backends.len(), "round-robin index out of bounds");
let res = backends.get(index).cloned();
debug_assert!(
res.is_some(),
"round-robin must select a backend from a non-empty set"
);
self.next_backend = (self.next_backend + 1) % backends.len() as u32;
debug_assert!(
(self.next_backend as usize) < backends.len(),
"round-robin cursor must stay within bounds"
);
res
}
}
impl Default for RoundRobin {
fn default() -> Self {
Self::new()
}
}
impl RoundRobin {
pub fn new() -> Self {
Self { next_backend: 0 }
}
}
#[derive(Debug)]
pub struct Random {
rng: StdRng,
}
impl Random {
pub fn with_seed(seed: u64) -> Self {
Self {
rng: StdRng::seed_from_u64(seed),
}
}
}
impl Random {
fn weighted_index(&mut self, backends: Candidates<'_>) -> Option<usize> {
if backends.is_empty() {
return None;
}
let mut total: i32 = 0;
for backend in backends.iter() {
let weight = random_weight(&backend.borrow());
if weight < 0 {
return None;
}
total = total.checked_add(weight)?;
}
if total == 0 {
return None;
}
let drawn = UniformInt::<i32>::new(0, total)
.expect("a positive total is a non-empty range")
.sample(&mut self.rng);
debug_assert!(
(0..total).contains(&drawn),
"the weighted draw must land inside [0, total)"
);
let last = backends.len() - 1;
let mut prefix: i32 = 0;
for (index, backend) in backends.iter().take(last).enumerate() {
prefix += random_weight(&backend.borrow());
if prefix > drawn {
return Some(index);
}
}
Some(last)
}
}
fn random_weight(backend: &Backend) -> i32 {
backend
.load_balancing_parameters
.as_ref()
.map(|p| p.weight)
.unwrap_or(DEFAULT_WEIGHT)
}
impl LoadBalancingAlgorithm for Random {
fn next_available_backend(
&mut self,
_key: Option<u64>,
backends: Candidates<'_>,
) -> Option<Rc<RefCell<Backend>>> {
let len = backends.len();
if let Some(index) = self.weighted_index(backends) {
debug_assert!(index < len, "Random sampled an out-of-range index");
backends.get(index).cloned()
} else {
let chosen = backends.choose(&mut self.rng).cloned();
debug_assert_eq!(
chosen.is_some(),
len > 0,
"Random fallback selects iff the set is non-empty"
);
chosen
}
}
}
fn address_count(
candidates: Candidates<'_>,
index: usize,
count: impl Fn(&Backend) -> usize,
) -> usize {
let own = count(&candidates[index].borrow());
let total = candidates.siblings(index).fold(own, |sum, sibling| {
sum.saturating_add(count(&sibling.borrow()))
});
debug_assert!(
total >= own,
"an address carries at least its representative's load"
);
total
}
#[derive(Debug)]
pub struct LeastLoaded {
pub metric: LoadMetric,
}
impl LoadBalancingAlgorithm for LeastLoaded {
fn next_available_backend(
&mut self,
_key: Option<u64>,
backends: Candidates<'_>,
) -> Option<Rc<RefCell<Backend>>> {
let was_empty = backends.is_empty();
let opt_b = match self.metric {
LoadMetric::Connections => (0..backends.len())
.min_by_key(|&index| address_count(backends, index, |b| b.active_connections))
.and_then(|index| backends.get(index)),
LoadMetric::Requests => (0..backends.len())
.min_by_key(|&index| address_count(backends, index, |b| b.active_requests))
.and_then(|index| backends.get(index)),
LoadMetric::ConnectionTime => {
let mut b = None;
let now = backends.now();
for backend in backends.iter() {
let cost2 = backend.borrow_mut().peak_ewma_connection(now);
match b.take() {
None => b = Some((cost2, backend)),
Some((cost1, back1)) => {
if cost1 <= cost2 {
b = Some((cost1, back1));
} else {
b = Some((cost2, backend));
}
}
}
}
b.map(|(_cost, backend)| backend)
}
};
debug_assert_eq!(
opt_b.is_some(),
!was_empty,
"LeastLoaded selects iff the candidate set is non-empty"
);
opt_b.cloned()
}
}
#[derive(Debug)]
pub struct PowerOfTwo {
pub metric: LoadMetric,
rng: StdRng,
}
impl PowerOfTwo {
pub fn with_seed(seed: u64, metric: LoadMetric) -> Self {
Self {
metric,
rng: StdRng::seed_from_u64(seed),
}
}
fn measure(&self, backends: Candidates<'_>, index: usize) -> f64 {
match self.metric {
LoadMetric::Connections => {
address_count(backends, index, |b| b.active_connections) as f64
}
LoadMetric::Requests => address_count(backends, index, |b| b.active_requests) as f64,
LoadMetric::ConnectionTime => backends[index]
.borrow_mut()
.peak_ewma_connection(backends.now()),
}
}
}
impl LoadBalancingAlgorithm for PowerOfTwo {
fn next_available_backend(
&mut self,
_key: Option<u64>,
backends: Candidates<'_>,
) -> Option<Rc<RefCell<Backend>>> {
let len = backends.len();
match len {
0 => return None,
1 => return backends.get(0).cloned(),
_ => {}
}
let first = self.rng.random_range(0..len);
let mut second = self.rng.random_range(0..len - 1);
if second >= first {
second += 1;
}
debug_assert_ne!(
first, second,
"power-of-two must sample two distinct backends"
);
debug_assert!(
second < len,
"the shifted second index must stay inside the candidate set"
);
let first_measure = self.measure(backends, first);
let second_measure = self.measure(backends, second);
let chosen = if first_measure < second_measure {
first
} else if second_measure < first_measure {
second
} else if self.rng.random_bool(0.5) {
first
} else {
second
};
debug_assert!(
(chosen == first && first_measure <= second_measure)
|| (chosen == second && second_measure <= first_measure),
"power-of-two must never keep the strictly heavier of its two samples"
);
backends.get(chosen).cloned()
}
}
#[derive(Debug)]
pub struct Rendezvous {
seed: u64,
round_robin: RoundRobin,
}
impl Default for Rendezvous {
fn default() -> Self {
Self::new()
}
}
impl Rendezvous {
pub fn new() -> Self {
Self::with_seed(DEFAULT_HASH_SEED)
}
pub fn with_seed(seed: u64) -> Self {
Self {
seed,
round_robin: RoundRobin::new(),
}
}
fn score(&self, key: u64, backend: &Backend) -> f64 {
hrw_score_with_seed(self.seed, key, backend)
}
}
pub fn hrw_score(key: u64, backend: &Backend) -> f64 {
hrw_score_with_seed(DEFAULT_HASH_SEED, key, backend)
}
fn hrw_score_with_seed(seed: u64, key: u64, backend: &Backend) -> f64 {
let weight = backend_weight(backend) as f64;
let h = hash_backend(seed, key, &backend.address);
let unit = (h as f64 + 0.5) / (u64::MAX as f64 + 1.0);
debug_assert!(
unit > 0.0 && unit < 1.0,
"HRW unit must lie strictly inside (0, 1), got {unit}"
);
let score = -weight / unit.ln();
debug_assert!(
score.is_finite() && score > 0.0,
"HRW score must be finite and positive, got {score}"
);
score
}
impl LoadBalancingAlgorithm for Rendezvous {
fn next_available_backend(
&mut self,
key: Option<u64>,
backends: Candidates<'_>,
) -> Option<Rc<RefCell<Backend>>> {
let Some(key) = key else {
return self.round_robin.next_available_backend(None, backends);
};
if backends.is_empty() {
return None;
}
let mut best: Option<(f64, &Rc<RefCell<Backend>>)> = None;
for backend in backends.iter() {
let score = self.score(key, &backend.borrow());
match best {
Some((best_score, _)) if best_score >= score => {}
_ => best = Some((score, backend)),
}
}
debug_assert!(
best.is_some(),
"HRW must select a winner from a non-empty set"
);
#[cfg(debug_assertions)]
if let Some((best_score, _)) = best {
for backend in backends.iter() {
debug_assert!(
best_score >= self.score(key, &backend.borrow()),
"HRW winner does not maximize the score over the set"
);
}
}
best.map(|(_, backend)| backend.clone())
}
}
#[derive(Debug)]
pub struct Maglev {
seed: u64,
size: usize,
table: Vec<usize>,
backend_addrs: Vec<SocketAddr>,
round_robin: RoundRobin,
}
impl Default for Maglev {
fn default() -> Self {
Self::new()
}
}
impl Maglev {
pub const DEFAULT_TABLE_SIZE: usize = 65537;
pub fn new() -> Self {
Self::with_seed(DEFAULT_HASH_SEED)
}
pub fn with_seed(seed: u64) -> Self {
Self::with_seed_and_size(seed, Self::DEFAULT_TABLE_SIZE)
}
pub fn with_seed_and_size(seed: u64, size: usize) -> Self {
Self {
seed,
size: next_prime(size.max(2)),
table: Vec::new(),
backend_addrs: Vec::new(),
round_robin: RoundRobin::new(),
}
}
pub fn rebuild(&mut self, backends: &[Rc<RefCell<Backend>>]) {
self.rebuild_from(backends.iter());
}
fn rebuild_from<'a>(
&mut self,
backends: impl ExactSizeIterator<Item = &'a Rc<RefCell<Backend>>>,
) {
let ids = backends.len();
self.backend_addrs.clear();
self.table.clear();
if ids == 0 || self.size == 0 {
return;
}
let m = self.size;
let mut offsets = Vec::with_capacity(ids);
let mut skips = Vec::with_capacity(ids);
let mut weights: Vec<u64> = Vec::with_capacity(ids);
for backend in backends {
let b = backend.borrow();
let addr = b.address;
let w = backend_weight(&b) as u64;
if let Some(known) = self.backend_addrs.iter().position(|a| *a == addr) {
weights[known] = weights[known].max(w);
continue;
}
self.backend_addrs.push(addr);
let h1 = hash_backend(self.seed, 0x6F66_6673_6574, &addr); let h2 = hash_backend(self.seed, 0x736B_6970_5F5F, &addr); let offset = (h1 % m as u64) as usize;
let skip = (h2 % (m as u64 - 1)) as usize + 1;
debug_assert!(offset < m, "Maglev offset must be a valid slot");
debug_assert!(
(1..m).contains(&skip),
"Maglev skip must lie in [1, m-1] to stay coprime with the prime table"
);
offsets.push(offset);
skips.push(skip);
weights.push(w);
}
let n = self.backend_addrs.len();
let total_weight: u64 = weights.iter().sum();
debug_assert!(
(1..=ids).contains(&n),
"Maglev captures between one and every backend's address"
);
debug_assert!(
total_weight > 0,
"Maglev total weight must be positive for a non-empty backend set"
);
let mut targets = vec![0usize; n];
let mut assigned = 0usize;
for (i, &w) in weights.iter().enumerate() {
let t = ((w as u128 * m as u128) / total_weight as u128) as usize;
targets[i] = t;
assigned += t;
}
let mut i = 0;
while assigned < m {
targets[i % n] += 1;
assigned += 1;
i += 1;
}
debug_assert_eq!(
assigned, m,
"Maglev target budget must sum to the table size"
);
debug_assert_eq!(
targets.iter().sum::<usize>(),
m,
"Maglev per-backend targets must sum to the table size"
);
let mut table = vec![usize::MAX; m];
let mut next = vec![0usize; n];
let mut filled = vec![0usize; n];
let mut count = 0usize;
while count < m {
for b in 0..n {
if filled[b] >= targets[b] {
continue;
}
let mut c = (offsets[b] + next[b] * skips[b]) % m;
while table[c] != usize::MAX {
next[b] += 1;
c = (offsets[b] + next[b] * skips[b]) % m;
}
table[c] = b;
next[b] += 1;
filled[b] += 1;
count += 1;
if count >= m {
break;
}
}
}
debug_assert_eq!(count, m, "Maglev population loop must fill exactly m slots");
debug_assert!(
table.iter().all(|&slot| slot != usize::MAX),
"Maglev population loop left an unfilled slot"
);
debug_assert!(
filled == targets,
"Maglev filled counts must match the per-backend targets"
);
self.table = table;
#[cfg(debug_assertions)]
{
debug_assert!(
is_prime(self.size),
"Maglev table size {} is not prime",
self.size
);
if self.table.is_empty() {
debug_assert!(
self.backend_addrs.is_empty(),
"Maglev: empty table but non-empty backend_addrs"
);
} else {
debug_assert_eq!(
self.table.len(),
self.size,
"Maglev table must have exactly `size` slots"
);
debug_assert!(
!self.backend_addrs.is_empty(),
"Maglev: non-empty table but empty backend_addrs"
);
debug_assert!(
self.table.iter().all(|&idx| idx < self.backend_addrs.len()),
"Maglev table holds an index out of backend_addrs range"
);
}
}
}
}
impl LoadBalancingAlgorithm for Maglev {
fn next_available_backend(
&mut self,
key: Option<u64>,
backends: Candidates<'_>,
) -> Option<Rc<RefCell<Backend>>> {
let Some(key) = key else {
return self.round_robin.next_available_backend(None, backends);
};
if backends.is_empty() {
return None;
}
if self.table.is_empty() {
self.rebuild_from(backends.iter());
}
if self.table.is_empty() {
return None;
}
let start = (key % self.size as u64) as usize;
debug_assert!(start < self.size, "Maglev start slot out of range");
debug_assert_eq!(
self.table.len(),
self.size,
"Maglev lookup on a table whose length != size"
);
for i in 0..self.size {
let slot = (start + i) % self.size;
debug_assert!(slot < self.size, "Maglev probe slot out of range");
let idx = self.table[slot];
debug_assert!(
idx < self.backend_addrs.len(),
"Maglev table entry indexes outside the captured address set"
);
if let Some(addr) = self.backend_addrs.get(idx)
&& let Some(backend) = backends.iter().find(|b| b.borrow().address == *addr)
{
debug_assert!(
backends
.iter()
.any(|b| b.borrow().address == backend.borrow().address),
"Maglev must return a backend from the healthy subset"
);
return Some(backend.clone());
}
}
self.round_robin.next_available_backend(None, backends)
}
fn rebuild(&mut self, backends: &[Rc<RefCell<Backend>>]) {
Maglev::rebuild(self, backends);
}
}
#[cfg(test)]
mod test {
use std::{
net::{IpAddr, Ipv4Addr, SocketAddr},
time::Duration,
};
use super::*;
use crate::{
PeakEWMA,
backends::{BackendStatus, HealthState},
retry::{ExponentialBackoffPolicy, RetryPolicyWrapper},
sozu_command::proto::command::{LoadBalancingParams, LoadMetric},
};
fn pick(
policy: &mut (impl LoadBalancingAlgorithm + ?Sized),
key: Option<u64>,
backends: &[Rc<RefCell<Backend>>],
) -> Option<Rc<RefCell<Backend>>> {
pick_at(policy, key, backends, Instant::now())
}
fn pick_at(
policy: &mut (impl LoadBalancingAlgorithm + ?Sized),
key: Option<u64>,
backends: &[Rc<RefCell<Backend>>],
now: Instant,
) -> Option<Rc<RefCell<Backend>>> {
let indices: Vec<usize> = (0..backends.len()).collect();
policy.next_available_backend(key, Candidates::new(backends, &indices, now))
}
fn create_backend(id: String, connections: Option<usize>) -> Backend {
Backend {
sticky_id: None,
backend_id: id,
address: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 8080),
status: BackendStatus::Normal,
retry_policy: RetryPolicyWrapper::ExponentialBackoff(ExponentialBackoffPolicy::new(1)),
active_connections: connections.unwrap_or(0),
active_requests: 0,
failures: 0,
load_balancing_parameters: None,
backup: false,
connection_time: PeakEWMA::new(Instant::now()),
health: HealthState::default(),
}
}
#[test]
fn it_should_find_the_backend_with_least_connections() {
let backend_with_least_connection =
Rc::new(RefCell::new(create_backend("yolo".to_string(), Some(1))));
let backends = vec![
Rc::new(RefCell::new(create_backend("nolo".to_string(), Some(10)))),
Rc::new(RefCell::new(create_backend("philo".to_string(), Some(20)))),
backend_with_least_connection.clone(),
];
let mut least_connection_algorithm = LeastLoaded {
metric: LoadMetric::Connections,
};
let backend_res = pick(&mut least_connection_algorithm, None, &backends).unwrap();
let backend = backend_res.borrow();
assert!(*backend == *backend_with_least_connection.borrow());
}
#[test]
fn it_shouldnt_find_backend_with_least_connections_when_list_is_empty() {
let backends = vec![];
let mut least_connection_algorithm = LeastLoaded {
metric: LoadMetric::Connections,
};
let backend = pick(&mut least_connection_algorithm, None, &backends);
assert!(backend.is_none());
}
#[test]
fn it_should_find_backend_with_roundrobin_when_some_backends_were_removed() {
let mut backends = vec![
Rc::new(RefCell::new(create_backend("toto".to_string(), None))),
Rc::new(RefCell::new(create_backend("voto".to_string(), None))),
Rc::new(RefCell::new(create_backend("yoto".to_string(), None))),
];
let mut roundrobin = RoundRobin { next_backend: 1 };
let backend = pick(&mut roundrobin, None, &backends);
assert_eq!(backend.as_ref(), backends.get(1));
backends.remove(1);
let backend2 = pick(&mut roundrobin, None, &backends);
assert_eq!(backend2.as_ref(), backends.first());
}
fn addr_backend(id: &str, last_octet: u8, port: u16, weight: Option<i32>) -> Backend {
let mut b = create_backend(id.to_string(), None);
b.address = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, last_octet)), port);
b.load_balancing_parameters = weight.map(|weight| LoadBalancingParams { weight });
b
}
fn rc(b: Backend) -> Rc<RefCell<Backend>> {
Rc::new(RefCell::new(b))
}
fn make_backends(n: u8) -> Vec<Rc<RefCell<Backend>>> {
(0..n)
.map(|i| rc(addr_backend(&format!("b{i}"), i + 1, 8000 + i as u16, None)))
.collect()
}
fn chosen_addr(b: &Rc<RefCell<Backend>>) -> SocketAddr {
b.borrow().address
}
#[test]
fn hrw_is_deterministic_for_a_fixed_key() {
let backends = make_backends(5);
let mut hrw = Rendezvous::new();
let first = pick(&mut hrw, Some(42), &backends).map(|b| chosen_addr(&b));
for _ in 0..50 {
let again = pick(&mut hrw, Some(42), &backends).map(|b| chosen_addr(&b));
assert_eq!(first, again, "HRW must be deterministic for a fixed key");
}
}
#[test]
fn hrw_none_key_falls_back_to_round_robin() {
let backends = make_backends(3);
let mut hrw = Rendezvous::new();
let a = chosen_addr(&pick(&mut hrw, None, &backends).unwrap());
let b = chosen_addr(&pick(&mut hrw, None, &backends).unwrap());
let c = chosen_addr(&pick(&mut hrw, None, &backends).unwrap());
let d = chosen_addr(&pick(&mut hrw, None, &backends).unwrap());
assert_eq!(a, chosen_addr(&backends[0]));
assert_eq!(b, chosen_addr(&backends[1]));
assert_eq!(c, chosen_addr(&backends[2]));
assert_eq!(d, a, "round-robin should wrap around");
}
#[test]
fn hrw_minimal_disruption_when_removing_a_non_winner() {
let mut backends = make_backends(6);
let mut hrw = Rendezvous::new();
let removed_addr = chosen_addr(&backends[3]);
let mut before = std::collections::HashMap::new();
for key in 0..2000u64 {
let w = chosen_addr(&pick(&mut hrw, Some(key), &backends).unwrap());
before.insert(key, w);
}
backends.remove(3);
for key in 0..2000u64 {
let after = chosen_addr(&pick(&mut hrw, Some(key), &backends).unwrap());
let prev = before[&key];
if prev != removed_addr {
assert_eq!(
prev, after,
"removing a non-winner changed the choice for key {key}"
);
}
}
}
#[test]
fn hrw_distribution_is_roughly_even() {
let n = 5u8;
let backends = make_backends(n);
let mut hrw = Rendezvous::new();
let total = 20_000u64;
let mut counts: std::collections::HashMap<SocketAddr, u64> =
std::collections::HashMap::new();
for key in 0..total {
let w = chosen_addr(&pick(&mut hrw, Some(key), &backends).unwrap());
*counts.entry(w).or_default() += 1;
}
let expected = total / n as u64;
for b in &backends {
let c = counts.get(&chosen_addr(b)).copied().unwrap_or(0);
assert!(
c > expected * 65 / 100 && c < expected * 135 / 100,
"HRW distribution skewed: backend got {c}, expected ~{expected}"
);
}
}
#[test]
fn maglev_is_deterministic_for_a_fixed_key() {
let backends = make_backends(7);
let mut mag = Maglev::new();
mag.rebuild(&backends);
let sel = backends.clone();
let first = chosen_addr(&pick(&mut mag, Some(12345), &sel).unwrap());
for _ in 0..50 {
let again = chosen_addr(&pick(&mut mag, Some(12345), &sel).unwrap());
assert_eq!(first, again, "Maglev must be deterministic for a fixed key");
}
}
#[test]
fn maglev_none_key_falls_back_to_round_robin() {
let backends = make_backends(3);
let mut mag = Maglev::new();
mag.rebuild(&backends);
let a = chosen_addr(&pick(&mut mag, None, &backends).unwrap());
let b = chosen_addr(&pick(&mut mag, None, &backends).unwrap());
let c = chosen_addr(&pick(&mut mag, None, &backends).unwrap());
let d = chosen_addr(&pick(&mut mag, None, &backends).unwrap());
assert_eq!(a, chosen_addr(&backends[0]));
assert_eq!(b, chosen_addr(&backends[1]));
assert_eq!(c, chosen_addr(&backends[2]));
assert_eq!(d, a);
}
#[test]
fn maglev_distribution_is_roughly_even() {
let n = 5u8;
let backends = make_backends(n);
let mut mag = Maglev::with_seed_and_size(DEFAULT_HASH_SEED, 1009);
mag.rebuild(&backends);
let total = 50_000u64;
let mut counts: std::collections::HashMap<SocketAddr, u64> =
std::collections::HashMap::new();
let sel = backends.clone();
for key in 0..total {
let w = chosen_addr(&pick(&mut mag, Some(key), &sel).unwrap());
*counts.entry(w).or_default() += 1;
}
let expected = total / n as u64;
for b in &backends {
let c = counts.get(&chosen_addr(b)).copied().unwrap_or(0);
assert!(
c > expected * 85 / 100 && c < expected * 115 / 100,
"Maglev distribution skewed: backend got {c}, expected ~{expected}"
);
}
}
#[test]
fn maglev_table_rebuild_keeps_most_keys_stable() {
let backends5 = make_backends(5);
let mut mag = Maglev::with_seed_and_size(DEFAULT_HASH_SEED, 1009);
mag.rebuild(&backends5);
let total = 20_000u64;
let mut before = std::collections::HashMap::new();
let sel = backends5.clone();
for key in 0..total {
before.insert(key, chosen_addr(&pick(&mut mag, Some(key), &sel).unwrap()));
}
let mut backends6 = backends5.clone();
backends6.push(rc(addr_backend("b5", 6, 8005, None)));
mag.rebuild(&backends6);
let mut moved = 0u64;
let sel6 = backends6.clone();
for key in 0..total {
let after = chosen_addr(&pick(&mut mag, Some(key), &sel6).unwrap());
if after != before[&key] {
moved += 1;
}
}
assert!(
moved < total / 2,
"Maglev rebuild moved too many keys: {moved}/{total}"
);
}
#[test]
fn maglev_partial_outage_does_not_rebuild_and_stays_stable() {
let full = make_backends(5);
let mut mag = Maglev::with_seed_and_size(DEFAULT_HASH_SEED, 1009);
mag.rebuild(&full);
let table_before = mag.table.clone();
let addrs_before = mag.backend_addrs.clone();
assert_eq!(addrs_before.len(), 5, "table built from the full set");
let total = 4000u64;
let mut before = std::collections::HashMap::new();
let sel_full = full.clone();
for key in 0..total {
before.insert(
key,
chosen_addr(&pick(&mut mag, Some(key), &sel_full).unwrap()),
);
}
let unhealthy_addr = chosen_addr(&full[2]);
let subset: Vec<_> = full
.iter()
.filter(|b| chosen_addr(b) != unhealthy_addr)
.cloned()
.collect();
for key in 0..total {
let after = chosen_addr(&pick(&mut mag, Some(key), &subset).unwrap());
assert_ne!(
after, unhealthy_addr,
"selection returned the unhealthy backend for key {key}"
);
if before[&key] != unhealthy_addr {
assert_eq!(
before[&key], after,
"a healthy key moved during a partial outage (key {key})"
);
}
}
assert_eq!(
mag.table, table_before,
"partial outage must not rebuild the Maglev table"
);
assert_eq!(
mag.backend_addrs, addrs_before,
"partial outage must not change the captured backend set"
);
}
#[test]
fn maglev_all_table_backends_unhealthy_falls_back_to_round_robin() {
let table_set = make_backends(3);
let mut mag = Maglev::with_seed_and_size(DEFAULT_HASH_SEED, 1009);
mag.rebuild(&table_set);
let table_before = mag.table.clone();
let fresh = vec![
rc(addr_backend("n0", 50, 9000, None)),
rc(addr_backend("n1", 51, 9001, None)),
];
let fresh_addrs: Vec<_> = fresh.iter().map(chosen_addr).collect();
let picked = chosen_addr(&pick(&mut mag, Some(7), &fresh).unwrap());
assert!(
fresh_addrs.contains(&picked),
"fallback must route to a backend in the healthy subset"
);
assert_eq!(
mag.table, table_before,
"fallback must not rebuild the table"
);
}
#[test]
fn maglev_cold_start_builds_table_once() {
let backends = make_backends(4);
let mut mag = Maglev::with_seed_and_size(DEFAULT_HASH_SEED, 1009);
assert!(mag.table.is_empty(), "table starts empty (cold)");
let _ = pick(&mut mag, Some(99), &backends).unwrap();
assert_eq!(mag.table.len(), mag.size, "cold start populated the table");
let table_after_cold = mag.table.clone();
let _ = pick(&mut mag, Some(100), &backends).unwrap();
assert_eq!(
mag.table, table_after_cold,
"selection after cold start must not rebuild"
);
}
#[test]
fn round_robin_empty_set_returns_none_without_panic() {
let empty: Vec<Rc<RefCell<Backend>>> = vec![];
let mut rr = RoundRobin::new();
assert!(pick(&mut rr, None, &empty).is_none());
let mut hrw = Rendezvous::new();
assert!(pick(&mut hrw, None, &empty).is_none());
let mut mag = Maglev::new();
assert!(pick(&mut mag, None, &empty).is_none());
}
#[test]
fn maglev_table_size_one_is_clamped_to_a_prime() {
let mut mag = Maglev::with_seed_and_size(DEFAULT_HASH_SEED, 1);
assert!(
mag.size >= 2,
"size must be clamped to >= 2, got {}",
mag.size
);
let backends = make_backends(3);
mag.rebuild(&backends);
assert_eq!(mag.table.len(), mag.size);
let _ = pick(&mut mag, Some(1), &backends).unwrap();
}
#[test]
fn next_prime_picks_the_smallest_prime_at_least_n() {
assert_eq!(next_prime(0), 2);
assert_eq!(next_prime(1), 2);
assert_eq!(next_prime(2), 2);
assert_eq!(next_prime(3), 3);
assert_eq!(next_prime(4), 5);
assert_eq!(next_prime(1009), 1009); assert_eq!(next_prime(65537), 65537); }
#[test]
fn maglev_honors_weight() {
let backends = vec![
rc(addr_backend("light", 1, 8001, Some(100))),
rc(addr_backend("heavy", 2, 8002, Some(400))),
];
let mut mag = Maglev::with_seed_and_size(DEFAULT_HASH_SEED, 1009);
mag.rebuild(&backends);
let heavy_addr = chosen_addr(&backends[1]);
let total = 20_000u64;
let mut heavy = 0u64;
let sel = backends.clone();
for key in 0..total {
if chosen_addr(&pick(&mut mag, Some(key), &sel).unwrap()) == heavy_addr {
heavy += 1;
}
}
assert!(
heavy > total * 70 / 100,
"weighted Maglev did not favor the heavy backend: {heavy}/{total}"
);
}
#[test]
fn maglev_gives_a_shared_address_the_slots_of_one_backend() {
let count_shared = |backends: &[Rc<RefCell<Backend>>]| {
let mut mag = Maglev::with_seed_and_size(DEFAULT_HASH_SEED, 1009);
mag.rebuild(backends);
let shared = chosen_addr(&backends[0]);
mag.table
.iter()
.filter(|&&index| mag.backend_addrs[index] == shared)
.count()
};
let equal = vec![
rc(addr_backend("a", 1, 8001, None)),
rc(addr_backend("b", 1, 8001, None)),
rc(addr_backend("c", 2, 8002, None)),
];
assert_eq!(count_shared(&equal), 505);
let unequal = vec![
rc(addr_backend("a", 1, 8001, Some(100))),
rc(addr_backend("b", 1, 8001, Some(300))),
rc(addr_backend("c", 2, 8002, Some(100))),
];
assert_eq!(count_shared(&unequal), 1009 * 3 / 4 + 1);
}
fn addr_index(addr: SocketAddr) -> u8 {
match addr.ip() {
IpAddr::V4(v4) => v4.octets()[3] - 1,
IpAddr::V6(_) => unreachable!("test backends are always IPv4"),
}
}
#[test]
fn random_is_deterministic_and_matches_an_expected_sequence() {
let backends = make_backends(4);
let mut r1 = Random::with_seed(DEFAULT_HASH_SEED);
let mut r2 = Random::with_seed(DEFAULT_HASH_SEED);
let sel1 = backends.clone();
let sel2 = backends.clone();
let seq1: Vec<u8> = (0..20)
.map(|_| addr_index(chosen_addr(&pick(&mut r1, None, &sel1).unwrap())))
.collect();
let seq2: Vec<u8> = (0..20)
.map(|_| addr_index(chosen_addr(&pick(&mut r2, None, &sel2).unwrap())))
.collect();
assert_eq!(
seq1, seq2,
"Random::with_seed must reproduce the same sequence for the same seed"
);
let expected: Vec<u8> = vec![1, 2, 1, 1, 0, 2, 1, 3, 1, 0, 3, 1, 0, 0, 3, 2, 1, 3, 3, 1];
assert_eq!(
seq1, expected,
"Random sequence for DEFAULT_HASH_SEED regressed"
);
}
#[test]
fn random_distribution_does_not_collapse() {
let n = 5u8;
let backends = make_backends(n);
let mut r = Random::with_seed(DEFAULT_HASH_SEED);
let sel = backends.clone();
let total = 5_000u64;
let mut counts = [0u64; 5];
for _ in 0..total {
let idx = addr_index(chosen_addr(&pick(&mut r, None, &sel).unwrap()));
counts[idx as usize] += 1;
}
let expected = total / n as u64;
for (idx, &c) in counts.iter().enumerate() {
assert!(
c > expected * 65 / 100 && c < expected * 135 / 100,
"Random distribution skewed for backend {idx}: got {c}, expected ~{expected}"
);
}
}
#[test]
fn power_of_two_tie_break_is_deterministic_and_matches_an_expected_sequence() {
let backends = make_backends(4);
let mut p1 = PowerOfTwo::with_seed(DEFAULT_HASH_SEED, LoadMetric::Connections);
let mut p2 = PowerOfTwo::with_seed(DEFAULT_HASH_SEED, LoadMetric::Connections);
let sel1 = backends.clone();
let sel2 = backends.clone();
let seq1: Vec<u8> = (0..20)
.map(|_| addr_index(chosen_addr(&pick(&mut p1, None, &sel1).unwrap())))
.collect();
let seq2: Vec<u8> = (0..20)
.map(|_| addr_index(chosen_addr(&pick(&mut p2, None, &sel2).unwrap())))
.collect();
assert_eq!(
seq1, seq2,
"PowerOfTwo::with_seed must reproduce the same tie-break sequence for the same seed"
);
let expected: Vec<u8> = vec![1, 3, 1, 1, 1, 2, 1, 2, 0, 1, 2, 3, 1, 3, 3, 3, 0, 2, 0, 2];
assert_eq!(
seq1, expected,
"PowerOfTwo selection sequence for DEFAULT_HASH_SEED regressed"
);
}
#[test]
fn power_of_two_tie_break_distribution_does_not_collapse() {
let backends = make_backends(4);
let mut p = PowerOfTwo::with_seed(DEFAULT_HASH_SEED, LoadMetric::Connections);
let sel = backends.clone();
let total = 5_000u64;
let mut counts = [0u64; 4];
for _ in 0..total {
let idx = addr_index(chosen_addr(&pick(&mut p, None, &sel).unwrap()));
counts[idx as usize] += 1;
}
let expected = total / 4;
for (idx, &c) in counts.iter().enumerate() {
assert!(
c > expected * 65 / 100 && c < expected * 135 / 100,
"PowerOfTwo distribution skewed for backend {idx}: got {c}, expected ~{expected}"
);
}
}
#[test]
fn power_of_two_tie_break_is_decided_by_the_coin_flip_not_by_sample_position() {
const TOTAL: usize = 2_000;
const BACKENDS: u8 = 3;
let backends = make_backends(BACKENDS);
let sel = backends.clone();
let mut second_measured_won = 0usize;
for trial in 0..TOTAL as u64 {
let seed = DEFAULT_HASH_SEED ^ trial;
let mut p = PowerOfTwo::with_seed(seed, LoadMetric::Connections);
let mut replay = StdRng::seed_from_u64(seed);
let len = usize::from(BACKENDS);
let first = replay.random_range(0..len);
let mut second = replay.random_range(0..len - 1);
if second >= first {
second += 1;
}
let picked = usize::from(addr_index(chosen_addr(&pick(&mut p, None, &sel).unwrap())));
assert!(
picked == first || picked == second,
"trial {trial}: picked backend {picked}, outside the pair ({first}, {second}) \
the replay predicts; the sampler no longer draws the pair this test replays"
);
if picked == second {
second_measured_won += 1;
}
}
let low = TOTAL * 30 / 100;
let high = TOTAL * 70 / 100;
assert!(
second_measured_won > low && second_measured_won < high,
"power-of-two resolved {second_measured_won} of {TOTAL} exact ties in favour of \
the second-measured sample; a coin flip must land near {}, and a count at either \
end means the tie is being awarded by sample position instead",
TOTAL / 2
);
}
#[test]
fn power_of_two_touches_exactly_two_backends() {
const N: u8 = 64;
let backends = make_backends(N);
let before = Instant::now();
for backend in &backends {
backend.borrow_mut().connection_time.last_event = before;
}
let mut p = PowerOfTwo::with_seed(DEFAULT_HASH_SEED, LoadMetric::ConnectionTime);
assert!(pick_at(&mut p, None, &backends, before + Duration::from_nanos(1)).is_some());
let touched = backends
.iter()
.filter(|backend| backend.borrow().connection_time.last_event > before)
.count();
assert_eq!(
touched, 2,
"power-of-two must measure exactly 2 of the {N} backends; measuring {touched} means \
the policy reads every backend's load, which the name exists to rule out"
);
}
#[test]
fn power_of_two_always_returns_the_lighter_of_two_backends() {
let backends = make_backends(2);
backends[0].borrow_mut().active_connections = 7;
backends[1].borrow_mut().active_connections = 1;
let mut p = PowerOfTwo::with_seed(DEFAULT_HASH_SEED, LoadMetric::Connections);
let sel = backends.clone();
for call in 0..200 {
let picked = addr_index(chosen_addr(&pick(&mut p, None, &sel).unwrap()));
assert_eq!(
picked, 1,
"power-of-two must keep the lighter of the two sampled backends (call {call} \
picked backend {picked}, which carries 7 connections against 1)"
);
}
}
#[test]
fn power_of_two_handles_empty_and_singleton_sets_without_panic() {
let mut p = PowerOfTwo::with_seed(DEFAULT_HASH_SEED, LoadMetric::Connections);
let empty: Vec<Rc<RefCell<Backend>>> = vec![];
assert!(
pick(&mut p, None, &empty).is_none(),
"power-of-two selects nothing from an empty candidate set"
);
let single = make_backends(1);
for _ in 0..10 {
assert_eq!(
addr_index(chosen_addr(&pick(&mut p, None, &single).unwrap())),
0,
"power-of-two returns the only backend of a singleton set"
);
}
}
#[test]
fn power_of_two_never_returns_the_strictly_heaviest_backend() {
let backends = make_backends(3);
backends[0].borrow_mut().active_connections = 0;
backends[1].borrow_mut().active_connections = 1;
backends[2].borrow_mut().active_connections = 5;
let mut p = PowerOfTwo::with_seed(DEFAULT_HASH_SEED, LoadMetric::Connections);
let sel = backends.clone();
for call in 0..2_000 {
let picked = addr_index(chosen_addr(&pick(&mut p, None, &sel).unwrap()));
assert_ne!(
picked, 2,
"power-of-two returned the strictly heaviest backend on call {call}: it is \
heavier than either backend it can be paired against"
);
}
}
#[test]
fn power_of_two_sample_size_is_two_not_the_whole_set() {
const N: u8 = 100;
let backends = make_backends(N);
for backend in backends.iter().skip(1) {
backend.borrow_mut().active_connections = 1_000;
}
let mut p = PowerOfTwo::with_seed(DEFAULT_HASH_SEED, LoadMetric::Connections);
let sel = backends.clone();
let total = 100_000u32;
let mut hits = 0u32;
for _ in 0..total {
if addr_index(chosen_addr(&pick(&mut p, None, &sel).unwrap())) == 0 {
hits += 1;
}
}
let k_est = f64::from(hits) / f64::from(total) * f64::from(N);
assert!(
(1.5..=2.5).contains(&k_est),
"power-of-two consulted ~{k_est:.2} of {N} backends (unique minimum returned \
{hits}/{total} times); the algorithm must sample exactly 2"
);
}
fn weighted_index_reference(rng: &mut StdRng, weights: &[i32]) -> Option<usize> {
use rand::distr::{Distribution, weighted::WeightedIndex};
match WeightedIndex::new(weights.to_vec()) {
Ok(distribution) => Some(distribution.sample(rng)),
Err(_) => {
let positions: Vec<usize> = (0..weights.len()).collect();
positions.choose(rng).copied()
}
}
}
#[test]
fn random_weighted_pick_matches_weighted_index() {
let weight_sets: &[&[i32]] = &[
&[100, 100, 100],
&[1, 2, 3, 4, 5, 6, 7],
&[0, 5, 0, 0, 9, 0],
&[7, 0, 0],
&[0, 0, 0, 3],
&[0, 0, 0],
&[-1, 5, 5],
&[5, 5, -1, 5],
&[i32::MAX, 1],
&[i32::MAX / 2, i32::MAX / 2, 1],
&[i32::MAX / 2, i32::MAX / 2, 2],
&[42],
&[],
];
for (set, weights) in weight_sets.iter().enumerate() {
let backends: Vec<_> = weights
.iter()
.enumerate()
.map(|(index, &weight)| rc(addr_backend("w", index as u8 + 1, 80, Some(weight))))
.collect();
for seed in 0..64 {
let mut policy = Random::with_seed(seed);
let mut reference = StdRng::seed_from_u64(seed);
for draw in 0..32 {
let expected = weighted_index_reference(&mut reference, weights);
let got = pick(&mut policy, None, &backends).map(|backend| {
backends
.iter()
.position(|candidate| Rc::ptr_eq(candidate, &backend))
.expect("the pick is one of the candidates")
});
assert_eq!(
got, expected,
"weight set {set} {weights:?}, seed {seed}, draw {draw}"
);
}
}
}
}
#[test]
fn affinity_keys_are_pinned_across_workers_and_releases() {
let ip = IpAddr::V4(Ipv4Addr::new(192, 0, 2, 7));
assert_eq!(affinity_key_from_ip(ip), affinity_key_from_ip(ip));
assert_eq!(affinity_key_from_ip(ip), PINNED_IP_KEY);
assert_eq!(affinity_key_from_value(b"tenant-42"), PINNED_VALUE_KEY);
}
const PINNED_IP_KEY: u64 = 12_659_869_658_761_991_936;
const PINNED_VALUE_KEY: u64 = 4_927_821_811_038_511_670;
#[test]
fn an_ipv4_mapped_client_keys_as_its_ipv4_address() {
let v4 = Ipv4Addr::new(198, 51, 100, 9);
assert_eq!(
affinity_key_from_ip(IpAddr::V6(v4.to_ipv6_mapped())),
affinity_key_from_ip(IpAddr::V4(v4)),
"a mapped IPv4 client must key as its IPv4 address"
);
assert_ne!(
affinity_key_from_ip(IpAddr::V6(v4.to_ipv6_compatible())),
affinity_key_from_ip(IpAddr::V4(v4)),
"only the mapped form is the same client; a compatible address is a distinct IPv6 one"
);
}
#[test]
fn a_value_never_keys_as_the_address_it_spells() {
let v4 = Ipv4Addr::new(10, 1, 2, 3);
let mut spelled = vec![4u8];
spelled.extend_from_slice(&v4.octets());
assert_ne!(
affinity_key_from_value(&spelled),
affinity_key_from_ip(IpAddr::V4(v4))
);
assert_ne!(
affinity_key_from_value(b"tenant-a"),
affinity_key_from_value(b"tenant-b")
);
}
#[test]
fn a_shard_size_follows_the_decided_formula() {
let sharding = |percent, min_backends| ShuffleSharding {
percent,
min_backends,
mode: ShardMode::Fallback,
};
for (percent, min_backends, primaries, expected) in [
(1, 8, 8, Some(2)), (25, 8, 8, Some(2)), (30, 8, 10, Some(3)), (50, 8, 9, Some(5)), (34, 8, 100, Some(34)), (100, 8, 12, Some(12)), (25, 8, 7, None), (25, 2, 2, Some(2)), (1, 2, 3, Some(2)), ] {
assert_eq!(
sharding(percent, min_backends).shard_size(primaries),
expected,
"percent={percent} min={min_backends} N={primaries}"
);
}
assert_eq!(
ShuffleSharding::from_cluster(Some(10), None, None),
Some(sharding(10, DEFAULT_SHARD_MIN_BACKENDS)),
"min_backends defaults to 8 and the mode to FALLBACK"
);
assert_eq!(ShuffleSharding::from_cluster(None, Some(4), Some(1)), None);
}
}