Skip to main content

sozu_lib/
backends.rs

1use std::{
2    cell::{Cell, RefCell},
3    collections::HashMap,
4    net::SocketAddr,
5    rc::{Rc, Weak},
6    time::{Duration, Instant},
7};
8
9use mio::net::TcpStream;
10use rand::{
11    Rng, SeedableRng,
12    rngs::{StdRng, SysRng},
13};
14use sozu_command::{
15    proto::command::{
16        Event, EventKind, HealthCheckConfig, LoadBalancingAlgorithms, LoadBalancingParams,
17        LoadMetric, ShardMode,
18    },
19    secret::ProtectedHeaderSet,
20    state::ClusterId,
21};
22
23use crate::metrics::names;
24use crate::{
25    PeakEWMA,
26    load_balancing::{
27        Candidates, LeastLoaded, LoadBalancingAlgorithm, Maglev, PowerOfTwo, Random, Rendezvous,
28        RoundRobin, ShuffleSharding, count_sibling_step, hrw_score, outranks_at_address,
29    },
30    retry::{self, RetryPolicy},
31    server::{self, push_event},
32};
33
34#[derive(thiserror::Error, Debug)]
35pub enum BackendError {
36    #[error("No backend found for cluster {0}")]
37    NoBackendForCluster(String),
38    #[error("Failed to connect to socket with MIO: {0}")]
39    MioConnection(std::io::Error),
40    #[error("This backend is not in a normal status: status={0:?}")]
41    Status(BackendStatus),
42    #[error("could not connect {cluster_id} to {backend_address:?} ({failures} failures): {error}")]
43    ConnectionFailures {
44        cluster_id: String,
45        backend_address: SocketAddr,
46        failures: usize,
47        error: String,
48    },
49}
50
51#[derive(Debug, PartialEq, Eq, Clone)]
52pub enum BackendStatus {
53    Normal,
54    Closing,
55    Closed,
56}
57
58#[derive(Debug, PartialEq, Eq, Clone, Copy)]
59pub enum HealthStatus {
60    Healthy,
61    Unhealthy,
62}
63
64/// Per-cluster availability state, owned by `BackendList`. Flips between
65/// `Available` (≥1 backend can serve traffic) and `AllDown` (every backend
66/// fails the `health.is_healthy() && !retry_policy.is_down()` predicate)
67/// every time `BackendMap::record_cluster_availability` is invoked.
68/// Empty clusters never report `AllDown` to avoid log spam during cluster
69/// bootstrap.
70#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
71pub(crate) enum ClusterAvailability {
72    #[default]
73    Available,
74    AllDown,
75}
76
77#[derive(Debug, Clone, PartialEq)]
78pub struct HealthState {
79    pub status: HealthStatus,
80    pub consecutive_successes: u32,
81    pub consecutive_failures: u32,
82}
83
84impl Default for HealthState {
85    fn default() -> Self {
86        HealthState {
87            status: HealthStatus::Healthy,
88            consecutive_successes: 0,
89            consecutive_failures: 0,
90        }
91    }
92}
93
94impl HealthState {
95    /// Record a successful health check. Returns true if the backend transitioned to healthy.
96    pub fn record_success(&mut self, healthy_threshold: u32) -> bool {
97        let was_unhealthy = self.status == HealthStatus::Unhealthy;
98        let successes_before = self.consecutive_successes;
99        self.consecutive_failures = 0;
100        self.consecutive_successes += 1;
101
102        // A success resets the failure streak and advances the success streak by
103        // exactly one — the two counters are never both non-zero afterwards.
104        debug_assert_eq!(
105            self.consecutive_failures, 0,
106            "a success must clear the consecutive-failure streak"
107        );
108        debug_assert_eq!(
109            self.consecutive_successes,
110            successes_before + 1,
111            "a success must advance the success streak by exactly one"
112        );
113
114        if was_unhealthy && self.consecutive_successes >= healthy_threshold {
115            self.status = HealthStatus::Healthy;
116            // The transition is only reported when crossing the threshold from
117            // Unhealthy; a backend that was already Healthy never "transitions".
118            debug_assert!(
119                self.status == HealthStatus::Healthy,
120                "a reported recovery must leave the status Healthy"
121            );
122            return true;
123        }
124        false
125    }
126
127    /// Record a failed health check. Returns true if the backend transitioned to unhealthy.
128    pub fn record_failure(&mut self, unhealthy_threshold: u32) -> bool {
129        let was_healthy = self.status == HealthStatus::Healthy;
130        let failures_before = self.consecutive_failures;
131        self.consecutive_successes = 0;
132        self.consecutive_failures += 1;
133
134        // A failure resets the success streak and advances the failure streak by
135        // exactly one — symmetric with `record_success`.
136        debug_assert_eq!(
137            self.consecutive_successes, 0,
138            "a failure must clear the consecutive-success streak"
139        );
140        debug_assert_eq!(
141            self.consecutive_failures,
142            failures_before + 1,
143            "a failure must advance the failure streak by exactly one"
144        );
145
146        if was_healthy && self.consecutive_failures >= unhealthy_threshold {
147            self.status = HealthStatus::Unhealthy;
148            debug_assert!(
149                self.status == HealthStatus::Unhealthy,
150                "a reported drop must leave the status Unhealthy"
151            );
152            return true;
153        }
154        false
155    }
156
157    pub fn is_healthy(&self) -> bool {
158        self.status == HealthStatus::Healthy
159    }
160}
161
162#[derive(Debug, PartialEq, Clone)]
163pub struct Backend {
164    pub sticky_id: Option<String>,
165    pub backend_id: String,
166    pub address: SocketAddr,
167    pub status: BackendStatus,
168    pub retry_policy: retry::RetryPolicyWrapper,
169    pub active_connections: usize,
170    pub active_requests: usize,
171    pub failures: usize,
172    pub load_balancing_parameters: Option<LoadBalancingParams>,
173    pub backup: bool,
174    pub connection_time: PeakEWMA,
175    pub health: HealthState,
176    /// UNIX domain socket dialed instead of `address`, already checked by
177    /// `sozu_command::config::parse_backend_unix_path`. `address` still
178    /// identifies the backend.
179    pub unix_path: Option<String>,
180}
181
182/// Open a nonblocking stream connection to a backend: TCP to `address`, or a
183/// UNIX domain connect to `unix_path` when it is set (an absolute path, or
184/// `@name` for the Linux abstract namespace).
185///
186/// A UNIX stream comes back as the descriptor of a `mio::net::TcpStream`, so
187/// the H1, H2 and TCP pipelines take it unchanged. Everything they do on a
188/// backend socket is a socket call that answers the same way on both
189/// families: `read(2)`, `write(2)`, `writev(2)`, `shutdown(2)`, `SO_ERROR`,
190/// `getpeername(2)` through `socket2` (std's `peer_addr` rejects a non-INET
191/// family), and epoll registration. The TCP-only options are skipped by the
192/// callers (`TCP_NODELAY`) or already optional (`TCP_INFO` in
193/// `socket::stats::socket_rtt` answers `None`), and the peer the logs render
194/// is the identity `address` cached in `SessionTcpStream`.
195pub fn connect_backend(address: SocketAddr, unix_path: Option<&str>) -> std::io::Result<TcpStream> {
196    let Some(unix_path) = unix_path else {
197        return TcpStream::connect(address);
198    };
199    let unix_address = sozu_command::config::parse_backend_unix_path(unix_path)
200        .map_err(|reason| std::io::Error::new(std::io::ErrorKind::InvalidInput, reason))?;
201    // The error names the socket: `address` is all the failure logs carry.
202    let stream = mio::net::UnixStream::connect_addr(&unix_address)
203        .map_err(|error| std::io::Error::new(error.kind(), format!("{unix_path}: {error}")))?;
204    let stream = TcpStream::from_std(std::net::TcpStream::from(std::os::fd::OwnedFd::from(
205        stream,
206    )));
207    // POST: the descriptor is the UNIX socket just dialed, not a TCP one.
208    debug_assert_eq!(
209        socket2::SockRef::from(&stream).domain().ok(),
210        Some(socket2::Domain::UNIX),
211        "a unix_path backend dials a UNIX domain socket"
212    );
213    Ok(stream)
214}
215
216impl Backend {
217    /// A backend created now, on the wall clock.
218    ///
219    /// The production default: the one place backend construction reads the
220    /// clock, because the connection-time average decays from the instant the
221    /// backend was created. A simulator that must not read the wall clock
222    /// uses [`Self::new_at`].
223    pub fn new(
224        backend_id: &str,
225        address: SocketAddr,
226        sticky_id: Option<String>,
227        load_balancing_parameters: Option<LoadBalancingParams>,
228        backup: Option<bool>,
229    ) -> Backend {
230        Self::new_at(
231            backend_id,
232            address,
233            sticky_id,
234            load_balancing_parameters,
235            backup,
236            Instant::now(),
237        )
238    }
239
240    /// A backend created at `now` (#1684).
241    pub fn new_at(
242        backend_id: &str,
243        address: SocketAddr,
244        sticky_id: Option<String>,
245        load_balancing_parameters: Option<LoadBalancingParams>,
246        backup: Option<bool>,
247        now: Instant,
248    ) -> Backend {
249        let desired_policy = retry::ExponentialBackoffPolicy::new(6);
250        Backend {
251            sticky_id,
252            backend_id: backend_id.to_owned(),
253            address,
254            status: BackendStatus::Normal,
255            retry_policy: desired_policy.into(),
256            active_connections: 0,
257            active_requests: 0,
258            failures: 0,
259            load_balancing_parameters,
260            backup: backup.unwrap_or(false),
261            connection_time: PeakEWMA::new(now),
262            health: HealthState::default(),
263            unix_path: None,
264        }
265    }
266
267    /// This backend, dialed over the UNIX domain socket `unix_path` instead
268    /// of `address` when it is set.
269    pub fn with_unix_path(mut self, unix_path: Option<String>) -> Backend {
270        self.unix_path = unix_path;
271        self
272    }
273
274    /// Open a nonblocking connection to this backend, see [`connect_backend`].
275    pub fn connect_socket(&self) -> std::io::Result<TcpStream> {
276        connect_backend(self.address, self.unix_path.as_deref())
277    }
278
279    pub fn set_closing(&mut self) {
280        self.status = BackendStatus::Closing;
281    }
282
283    pub fn retry_policy(&mut self) -> &mut retry::RetryPolicyWrapper {
284        &mut self.retry_policy
285    }
286
287    /// Whether this backend can take a new connection at `now`.
288    pub fn can_open(&self, now: Instant) -> bool {
289        if !self.health.is_healthy() {
290            return false;
291        }
292        if let Some(action) = self.retry_policy.can_try(now) {
293            self.status == BackendStatus::Normal && action == retry::RetryAction::OKAY
294        } else {
295            false
296        }
297    }
298
299    /// Canonical "available" check used by per-backend metrics and cluster
300    /// availability accounting. Slightly more permissive than `can_open()`:
301    /// a backend currently in an exponential-backoff *wait window* still
302    /// counts as available because the next call after the window ends
303    /// will route to it without operator intervention. The dashboard
304    /// reading must reflect "operationally up, not exhausted" rather than
305    /// "ready to receive *this* request" — flicking the gauge to 0 on
306    /// every transient backoff would drown out genuine `is_down()`
307    /// transitions. Pairs with `BackendList::evaluate_availability`,
308    /// which applies the same predicate cluster-wide.
309    pub fn is_available(&self) -> bool {
310        self.health.is_healthy()
311            && self.status == BackendStatus::Normal
312            && !self.retry_policy.is_down()
313    }
314
315    pub fn inc_connections(&mut self) -> Option<usize> {
316        let before = self.active_connections;
317        if self.status == BackendStatus::Normal {
318            self.active_connections += 1;
319            // A `Normal` backend always increments by exactly one and reports the
320            // post-increment count back to the caller.
321            debug_assert_eq!(
322                self.active_connections,
323                before + 1,
324                "inc_connections must add exactly one active connection"
325            );
326            Some(self.active_connections)
327        } else {
328            // Non-`Normal` backends refuse new connections and leave the gauge
329            // untouched (no silent increment on a Closing/Closed backend).
330            debug_assert_eq!(
331                self.active_connections, before,
332                "inc_connections must not touch the count for a non-Normal backend"
333            );
334            None
335        }
336    }
337
338    /// TODO: normalize with saturating_sub()
339    pub fn dec_connections(&mut self) -> Option<usize> {
340        let before = self.active_connections;
341        match self.status {
342            BackendStatus::Normal => {
343                if self.active_connections > 0 {
344                    self.active_connections -= 1;
345                }
346                // The count drops by one when positive, otherwise saturates at
347                // zero — it must never wrap below zero (usize underflow).
348                debug_assert!(
349                    self.active_connections <= before,
350                    "dec_connections must never increase the active-connection count"
351                );
352                debug_assert_eq!(
353                    self.active_connections,
354                    before.saturating_sub(1),
355                    "dec_connections must drop by exactly one (saturating at zero)"
356                );
357                Some(self.active_connections)
358            }
359            BackendStatus::Closed => {
360                // A Closed backend has already been retired: nothing to decrement.
361                debug_assert_eq!(
362                    self.active_connections, before,
363                    "dec_connections on a Closed backend must not mutate the count"
364                );
365                None
366            }
367            BackendStatus::Closing => {
368                if self.active_connections > 0 {
369                    self.active_connections -= 1;
370                }
371                debug_assert_eq!(
372                    self.active_connections,
373                    before.saturating_sub(1),
374                    "dec_connections must drop by exactly one (saturating at zero)"
375                );
376                if self.active_connections == 0 {
377                    self.status = BackendStatus::Closed;
378                    // Draining a Closing backend to zero retires it: the
379                    // lifecycle advances to Closed and we stop reporting a count.
380                    debug_assert_eq!(
381                        self.status,
382                        BackendStatus::Closed,
383                        "a fully drained Closing backend must become Closed"
384                    );
385                    None
386                } else {
387                    Some(self.active_connections)
388                }
389            }
390        }
391    }
392
393    /// Reserve a connection on this backend for a dial about to start.
394    ///
395    /// `active_connections += 1`, exactly the count a successful
396    /// [`Self::try_connect`] takes, but before the connect: selection takes it
397    /// under the same borrow as the choice (`BackendMap::reserve_backend`).
398    /// Only a `Normal` backend can be chosen, and only a `Normal` backend
399    /// accepts the reservation.
400    pub fn reserve_connection(&mut self) -> Result<(), BackendError> {
401        match self.inc_connections() {
402            Some(_) => Ok(()),
403            None => Err(BackendError::Status(self.status.to_owned())),
404        }
405    }
406
407    /// Release the reservation of a dial whose connect failed at `now`, and
408    /// record the failure: `failures += 1` and the retry policy's backoff,
409    /// whose jitter is drawn from `rng`.
410    ///
411    /// Leaves the backend where a failed [`Self::try_connect`] leaves it: the
412    /// reservation it releases is the count that call never took.
413    pub fn release_failed_dial<R: Rng + ?Sized>(&mut self, now: Instant, rng: &mut R) {
414        let (connections_before, failures_before) = (self.active_connections, self.failures);
415        self.dec_connections();
416        self.failures += 1;
417        self.retry_policy.fail(now, rng);
418        debug_assert_eq!(
419            self.active_connections,
420            connections_before.saturating_sub(1),
421            "a failed dial must release exactly its one reservation (saturating at 0)"
422        );
423        debug_assert_eq!(
424            self.failures,
425            failures_before + 1,
426            "a failed dial must advance the failure counter by exactly one"
427        );
428    }
429
430    /// Record that a connection took `dur` to establish, observed at `now`.
431    pub fn set_connection_time(&mut self, dur: Duration, now: Instant) {
432        self.connection_time.observe(dur.as_nanos() as f64, now);
433    }
434
435    /// The connection-time cost of this backend, as seen at `now`.
436    pub fn peak_ewma_connection(&mut self, now: Instant) -> f64 {
437        self.connection_time.get(self.active_connections, now)
438    }
439
440    /// Open a connection to this backend. A failure at `now` arms the retry
441    /// policy, whose backoff jitter is drawn from `rng`.
442    pub fn try_connect<R: Rng + ?Sized>(
443        &mut self,
444        now: Instant,
445        rng: &mut R,
446    ) -> Result<mio::net::TcpStream, BackendError> {
447        if self.status != BackendStatus::Normal {
448            return Err(BackendError::Status(self.status.to_owned()));
449        }
450        // Reaching the connect attempt implies we passed the status gate; the
451        // failure counter is whatever prior attempts accumulated.
452        debug_assert_eq!(
453            self.status,
454            BackendStatus::Normal,
455            "try_connect only attempts a connection on a Normal backend"
456        );
457        let failures_before = self.failures;
458        let connections_before = self.active_connections;
459
460        match self.connect_socket() {
461            Ok(tcp_stream) => {
462                //self.retry_policy.succeed();
463                self.inc_connections();
464                // Success registers exactly one new active connection and never
465                // touches the failure counter.
466                debug_assert_eq!(
467                    self.active_connections,
468                    connections_before + 1,
469                    "a successful connect must register exactly one active connection"
470                );
471                debug_assert_eq!(
472                    self.failures, failures_before,
473                    "a successful connect must not bump the failure counter"
474                );
475                Ok(tcp_stream)
476            }
477            Err(io_error) => {
478                self.retry_policy.fail(now, rng);
479                self.failures += 1;
480                // A failed connect arms the retry policy and advances the
481                // failure counter by exactly one, leaving the connection gauge
482                // untouched (no connection was established).
483                debug_assert_eq!(
484                    self.failures,
485                    failures_before + 1,
486                    "a failed connect must advance the failure counter by exactly one"
487                );
488                debug_assert_eq!(
489                    self.active_connections, connections_before,
490                    "a failed connect must not register an active connection"
491                );
492                // TODO: handle EINPROGRESS. It is difficult. It is discussed here:
493                // https://docs.rs/mio/latest/mio/net/struct.TcpStream.html#method.connect
494                // with an example code here:
495                // https://github.com/Thomasdezeeuw/heph/blob/0c4f1ab3eaf08bea1d65776528bfd6114c9f8374/src/net/tcp/stream.rs#L560-L622
496                Err(BackendError::MioConnection(io_error))
497            }
498        }
499    }
500}
501
502// when a backend has been removed from configuration and the last connection to
503// it has stopped, it will be dropped, so we can notify that the backend server
504// can be safely stopped
505impl std::ops::Drop for Backend {
506    fn drop(&mut self) {
507        server::push_event(Event {
508            kind: EventKind::RemovedBackendHasNoConnections as i32,
509            backend_id: Some(self.backend_id.to_owned()),
510            address: Some(self.address.into()),
511            cluster_id: None,
512            metric_detail: None,
513        });
514    }
515}
516
517#[derive(Debug)]
518pub struct BackendMap {
519    pub backends: HashMap<ClusterId, BackendList>,
520    pub health_check_configs: HashMap<ClusterId, HealthCheckConfig>,
521    /// The values behind the headers of a health check, in memory only. An
522    /// entry is replaced together with its configuration and dropped with it:
523    /// a configuration that names headers without a set satisfying them sends
524    /// no probe.
525    pub protected_health_check_headers: HashMap<ClusterId, ProtectedHeaderSet>,
526    /// Whether the cluster's backends speak HTTP/2 (cluster.http2 = true).
527    /// Mirrors the same backend-capability hint the mux router reads at
528    /// `protocol/mux/router.rs::Router::plan_connect`. The health checker uses
529    /// it to switch the probe wire format from HTTP/1.1 to h2c so an
530    /// h2c-only backend is not probed with an HTTP/1.1 preface that
531    /// would always fail.
532    pub cluster_http2: HashMap<ClusterId, bool>,
533    /// The one source of randomness selection and backoff draw from: it seeds
534    /// every cluster's load-balancing policy and supplies the jitter of every
535    /// backoff window a failed connection arms.
536    ///
537    /// [`Self::new`] seeds it from the OS once, so each worker draws an
538    /// independent keystream, as the thread-local RNG it replaces did;
539    /// [`Self::with_seed`] makes the whole map reproducible for a simulator
540    /// (#1684). Taken at construction, never on the datapath.
541    rng: StdRng,
542}
543
544/// A generator seeded from the OS, for the production defaults of
545/// [`BackendMap::new`] and [`BackendList::new`].
546fn os_rng() -> StdRng {
547    StdRng::try_from_rng(&mut SysRng).expect("failed to seed random number generator from system")
548}
549
550impl Default for BackendMap {
551    fn default() -> Self {
552        Self::new()
553    }
554}
555
556impl BackendMap {
557    /// A backend map whose randomness is seeded from the OS.
558    pub fn new() -> BackendMap {
559        Self::with_rng(os_rng())
560    }
561
562    /// A backend map whose every policy seed and backoff jitter derives from
563    /// `seed`, for a deterministic simulator (#1684).
564    pub fn with_seed(seed: u64) -> BackendMap {
565        Self::with_rng(StdRng::seed_from_u64(seed))
566    }
567
568    fn with_rng(rng: StdRng) -> BackendMap {
569        BackendMap {
570            backends: HashMap::new(),
571            health_check_configs: HashMap::new(),
572            protected_health_check_headers: HashMap::new(),
573            cluster_http2: HashMap::new(),
574            rng,
575        }
576    }
577
578    /// The map's generator, for a caller that arms a backend's retry policy
579    /// outside [`Self::backend_from_cluster_id`] and needs its jitter.
580    pub fn rng(&mut self) -> &mut StdRng {
581        &mut self.rng
582    }
583
584    /// Re-evaluate the availability of `cluster_id`, publish the
585    /// `cluster.available_backends` / `cluster.total_backends` gauges,
586    /// and emit the transition log + counter + `Event` exactly when the
587    /// per-cluster state flips between `Available` and `AllDown`.
588    ///
589    /// Empty clusters (`total == 0`) never report `AllDown` — avoids log
590    /// spam during cluster bootstrap when backends are still being
591    /// registered. The (0, 0) gauges are still published so dashboards
592    /// see "cluster exists, zero backends configured" as a distinct
593    /// state from "cluster doesn't exist".
594    ///
595    /// Takes `&self` so callers that already hold `&mut BackendMap`
596    /// can drop their `&mut BackendList` borrow before invoking it
597    /// without re-borrowing.
598    pub(crate) fn record_cluster_availability(&self, cluster_id: &str) {
599        let Some(list) = self.backends.get(cluster_id) else {
600            return;
601        };
602
603        let (available, total) = list.evaluate_availability();
604        // A subset count can never exceed the whole, and it must match the
605        // live backend vector length the helper just walked.
606        debug_assert!(
607            available <= total,
608            "available backends ({available}) cannot exceed total ({total})"
609        );
610        debug_assert_eq!(
611            total,
612            list.backends.len(),
613            "total must equal the number of registered backends"
614        );
615        gauge!(
616            names::cluster::AVAILABLE_BACKENDS,
617            available,
618            Some(cluster_id),
619            None
620        );
621        gauge!(
622            names::cluster::TOTAL_BACKENDS,
623            total,
624            Some(cluster_id),
625            None
626        );
627
628        let new_state = if total > 0 && available == 0 {
629            ClusterAvailability::AllDown
630        } else {
631            ClusterAvailability::Available
632        };
633        // Empty clusters never report AllDown (avoids bootstrap log spam); a
634        // cluster with at least one available backend is always Available.
635        debug_assert!(
636            !(total == 0 && new_state == ClusterAvailability::AllDown),
637            "an empty cluster must never be reported AllDown"
638        );
639        debug_assert!(
640            !(available > 0 && new_state == ClusterAvailability::AllDown),
641            "a cluster with an available backend must not be AllDown"
642        );
643
644        let prev = list.availability.replace(new_state);
645        // The cell now holds exactly the freshly computed state.
646        debug_assert_eq!(
647            list.availability.get(),
648            new_state,
649            "the availability cell must latch the newly computed state"
650        );
651        if prev == new_state {
652            return;
653        }
654        match (prev, new_state) {
655            (ClusterAvailability::Available, ClusterAvailability::AllDown) => {
656                error!("cluster {}: all {} backends are down", cluster_id, total);
657                incr!(
658                    names::cluster::NO_AVAILABLE_BACKENDS,
659                    Some(cluster_id),
660                    None
661                );
662                push_event(Event {
663                    kind: EventKind::NoAvailableBackends as i32,
664                    cluster_id: Some(cluster_id.to_owned()),
665                    backend_id: None,
666                    address: None,
667                    metric_detail: None,
668                });
669            }
670            (ClusterAvailability::AllDown, ClusterAvailability::Available) => {
671                info!(
672                    "cluster {}: backends recovered ({}/{} available)",
673                    cluster_id, available, total
674                );
675                incr!(names::cluster::AVAILABLE_RECOVERED, Some(cluster_id), None);
676                push_event(Event {
677                    kind: EventKind::ClusterRecovered as i32,
678                    cluster_id: Some(cluster_id.to_owned()),
679                    backend_id: None,
680                    address: None,
681                    metric_detail: None,
682                });
683            }
684            _ => {}
685        }
686    }
687
688    /// Forget everything the map holds for `cluster_id`: its backend list,
689    /// with the load-balancing policy and shuffle sharding it carries, its
690    /// health-check configuration and its `http2` hint. A session that already
691    /// holds one of its backends keeps it until the session closes;
692    /// [`Self::close_backend_connection`] ignores a cluster that is gone.
693    /// Every backend it drops is marked `Closing`, as `BackendList::remove_backend`
694    /// marks the ones it drops, so a connection pooled on it drains the
695    /// requests it carries and takes no new one, even once a cluster of the
696    /// same id is added again.
697    pub fn remove_cluster(&mut self, cluster_id: &str) {
698        if let Some(list) = self.backends.remove(cluster_id) {
699            for backend in &list.backends {
700                backend.borrow_mut().set_closing();
701            }
702        }
703        self.health_check_configs.remove(cluster_id);
704        self.protected_health_check_headers.remove(cluster_id);
705        self.cluster_http2.remove(cluster_id);
706        debug_assert!(
707            !self.backends.contains_key(cluster_id)
708                && !self.health_check_configs.contains_key(cluster_id)
709                && !self.cluster_http2.contains_key(cluster_id),
710            "remove_cluster must leave nothing keyed by the cluster"
711        );
712    }
713
714    /// Record (or clear) the `cluster.http2` backend-capability hint for
715    /// `cluster_id`. The health checker reads the resulting map at probe
716    /// time so the wire format follows what the mux router will use to
717    /// connect to the same backends.
718    pub fn set_cluster_http2(&mut self, cluster_id: &str, http2: bool) {
719        if http2 {
720            self.cluster_http2.insert(cluster_id.into(), true);
721        } else {
722            self.cluster_http2.remove(cluster_id);
723        }
724    }
725
726    /// Replace the configuration and the values behind its headers as one
727    /// step: both change or neither does. A probe already in flight composed
728    /// its request from the previous set and is not affected.
729    pub fn set_protected_health_check(
730        &mut self,
731        cluster_id: &str,
732        config: HealthCheckConfig,
733        headers: ProtectedHeaderSet,
734    ) {
735        self.health_check_configs.insert(cluster_id.into(), config);
736        self.protected_health_check_headers
737            .insert(cluster_id.into(), headers);
738    }
739
740    /// Put back exactly what a protected health check replaced: the
741    /// configuration and the values behind its headers, each present or absent
742    /// as it was.
743    pub fn restore_health_check(
744        &mut self,
745        cluster_id: &str,
746        config: Option<HealthCheckConfig>,
747        headers: Option<ProtectedHeaderSet>,
748    ) {
749        match config {
750            Some(config) => {
751                self.health_check_configs.insert(cluster_id.into(), config);
752            }
753            None => {
754                self.health_check_configs.remove(cluster_id);
755            }
756        }
757        match headers {
758            Some(headers) => {
759                self.protected_health_check_headers
760                    .insert(cluster_id.into(), headers);
761            }
762            None => {
763                self.protected_health_check_headers.remove(cluster_id);
764            }
765        }
766    }
767
768    pub fn set_health_check_config(&mut self, cluster_id: &str, config: Option<HealthCheckConfig>) {
769        // The values kept for the previous configuration survive only when the
770        // new one names exactly the same headers: same count, order, names and
771        // full fingerprints (`ProtectedHeaderSet::satisfies`). A threshold or
772        // load-balancing update therefore keeps its probes; a cleared, reordered,
773        // foreign-epoch, malformed or merely same-named list drops the values, so
774        // a stale credential is never sent under new metadata.
775        let keeps_values = config.as_ref().is_some_and(|new| {
776            !new.headers.is_empty()
777                && self
778                    .protected_health_check_headers
779                    .get(cluster_id)
780                    .is_some_and(|held| held.satisfies(&new.headers))
781        });
782        if !keeps_values {
783            self.protected_health_check_headers.remove(cluster_id);
784        }
785        match config {
786            Some(c) => {
787                self.health_check_configs.insert(cluster_id.into(), c);
788            }
789            None => {
790                self.health_check_configs.remove(cluster_id);
791                // When the operator drops the health check, any
792                // previously-recorded `HealthState::Unhealthy` would
793                // otherwise stick — `next_available_backend` keeps
794                // skipping the backend even though we have stopped
795                // probing it. Reset every backend in the cluster to a
796                // pristine healthy state so the load balancer can
797                // route again.
798                if let Some(backend_list) = self.backends.get(cluster_id) {
799                    for backend in &backend_list.backends {
800                        backend.borrow_mut().health = HealthState::default();
801                    }
802                }
803                // Re-emit the rollup gauges so dashboards reflect the
804                // post-reset availability instead of holding the last
805                // health-check value indefinitely.
806                self.record_cluster_availability(cluster_id);
807            }
808        }
809    }
810
811    pub fn import_configuration_state(
812        &mut self,
813        backends: &HashMap<ClusterId, Vec<sozu_command::response::Backend>>,
814    ) {
815        // Seed the clusters in id order, not in the `HashMap`'s: its
816        // `RandomState` iterates differently in every process, so drawing in
817        // that order would hand a seeded map's seeds to different clusters on
818        // every run (#1684). Sorting the ids allocates once, on this
819        // control-plane replay path only.
820        let mut cluster_ids: Vec<&ClusterId> = backends.keys().collect();
821        cluster_ids.sort_unstable();
822        for cluster_id in cluster_ids {
823            let list =
824                BackendList::import_configuration_state(&backends[cluster_id], self.rng.next_u64());
825            self.backends.insert(cluster_id.clone(), list);
826        }
827        // Replay path inserts every cluster's backend list without
828        // touching the gauge emission sites used by add/remove/health.
829        // Latch `cluster.available_backends` and `.total_backends` here
830        // so a freshly-loaded worker reports correct values on the very
831        // first `QueryMetrics` instead of zero until something else
832        // mutates each cluster.
833        for cluster_id in backends.keys() {
834            self.record_cluster_availability(cluster_id);
835        }
836    }
837
838    pub fn add_backend(&mut self, cluster_id: &str, backend: Backend) {
839        let address = backend.address;
840        self.get_or_create_backend_list_for_cluster(cluster_id)
841            .add_backend(backend);
842        // Adding a backend must leave the cluster present and containing the
843        // just-added address (whether it created the entry or updated in place).
844        debug_assert!(
845            self.backends
846                .get(cluster_id)
847                .is_some_and(|list| list.has_backend(&address)),
848            "add_backend must leave the backend present in its cluster"
849        );
850        // Publish initial gauges and surface the corner case where a fresh
851        // cluster's first backend is already down (e.g. registered with a
852        // pre-existing failed retry policy). For an `Available` initial
853        // backend this is just a (1, 1) gauge emission with no transition.
854        self.record_cluster_availability(cluster_id);
855    }
856
857    // TODO: return <Result, BackendError>, log the error downstream
858    /// Remove the backend identified by `(backend_id, backend_address)` from
859    /// `cluster_id` and return whether it was present. A backend of another
860    /// id at the same address stays (#1821).
861    pub fn remove_backend(
862        &mut self,
863        cluster_id: &str,
864        backend_id: &str,
865        backend_address: &SocketAddr,
866    ) -> bool {
867        let removed = if let Some(backends) = self.backends.get_mut(cluster_id) {
868            let removed = backends.remove_backend(backend_id, backend_address);
869            if !removed {
870                warn!(
871                    "No backend matches id {} at address {:?} in cluster {}: nothing removed",
872                    backend_id, backend_address, cluster_id
873                );
874            }
875            removed
876        } else {
877            error!(
878                "Backend was already removed: cluster id {}, backend id {}, address {:?}",
879                cluster_id, backend_id, backend_address
880            );
881            return false;
882        };
883        debug_assert!(
884            self.backends.get(cluster_id).is_none_or(|list| list
885                .find_backend_by_identity(backend_id, backend_address)
886                .is_none()),
887            "remove_backend must evict the backend it names"
888        );
889        // Re-evaluate so removing the last backend logs an explicit
890        // `AllDown` transition (or, with `total == 0`, drops back to
891        // silent gauges).
892        self.record_cluster_availability(cluster_id);
893        removed
894    }
895
896    // TODO: return <Result, BackendError>, log the error downstream
897    pub fn close_backend_connection(&mut self, cluster_id: &str, addr: &SocketAddr) {
898        if let Some(cluster_backends) = self.backends.get_mut(cluster_id)
899            && let Some(ref mut backend) = cluster_backends.find_backend(addr)
900        {
901            backend.borrow_mut().dec_connections();
902        }
903    }
904
905    /// Whether `cluster_id` still holds a backend carrying `backend_id`, at
906    /// any address. Backend metrics are labelled by id alone, so every entry
907    /// sharing an id feeds one metrics row.
908    pub fn has_backend_id(&self, cluster_id: &str, backend_id: &str) -> bool {
909        self.backends.get(cluster_id).is_some_and(|backends| {
910            backends
911                .backends
912                .iter()
913                .any(|backend| backend.borrow().backend_id == backend_id)
914        })
915    }
916
917    pub fn has_backend(&self, cluster_id: &str, backend: &Backend) -> bool {
918        self.backends
919            .get(cluster_id)
920            .map(|backends| backends.has_backend(&backend.address))
921            .unwrap_or(false)
922    }
923
924    /// Select a backend of `cluster_id` at `now` and connect to it.
925    ///
926    /// The TCP proxy's entry point: it selects and connects in one call, and
927    /// `Backend::try_connect` counts the connection once it is established.
928    /// The HTTP mux reserves with [`Self::reserve_backend`] instead and
929    /// connects on its own side.
930    ///
931    /// `key` is the client's affinity key, which `HRW` and `MAGLEV` pin the
932    /// client on and every other policy ignores: the HTTP, HTTPS and TCP
933    /// datapaths pass the one they derived for a cluster using either
934    /// policy, and `None` otherwise (see
935    /// [`BackendList::next_available_backend_with_key`]).
936    pub fn backend_from_cluster_id(
937        &mut self,
938        cluster_id: &str,
939        key: Option<u64>,
940        now: Instant,
941    ) -> Result<(Rc<RefCell<Backend>>, TcpStream), BackendError> {
942        let next_backend = self.select_backend(cluster_id, key, now, &[])?;
943
944        let tcp_stream = {
945            let mut borrowed_backend = next_backend.borrow_mut();
946
947            debug!(
948                "Connecting {} -> {:?}",
949                cluster_id,
950                (
951                    borrowed_backend.address,
952                    borrowed_backend.active_connections,
953                    borrowed_backend.failures
954                )
955            );
956
957            borrowed_backend
958                .try_connect(now, &mut self.rng)
959                .map_err(|backend_error| BackendError::ConnectionFailures {
960                    cluster_id: cluster_id.to_owned(),
961                    backend_address: borrowed_backend.address,
962                    failures: borrowed_backend.failures,
963                    error: backend_error.to_string(),
964                })?
965        };
966
967        // Connection succeeded: re-evaluate so we capture an
968        // AllDown -> Available recovery transition the moment a request
969        // first hits a healthy backend after an outage. `next_backend` is
970        // not borrowed here (the inner block dropped `borrowed_backend`),
971        // so the helper's `BackendList::evaluate_availability` walk is
972        // free to call `borrow()` on every backend.
973        self.record_cluster_availability(cluster_id);
974
975        // The selected backend is a live member of the cluster it was drawn
976        // from — selection never fabricates or returns a stale backend.
977        debug_assert!(
978            self.backends.get(cluster_id).is_some_and(|list| {
979                let picked = next_backend.borrow().address;
980                list.has_backend(&picked)
981            }),
982            "the selected backend must belong to the cluster's live set"
983        );
984
985        Ok((next_backend.clone(), tcp_stream))
986    }
987
988    /// Select a backend of `cluster_id` at `now` and reserve a connection on
989    /// it, without connecting (#1684).
990    ///
991    /// The reservation is `active_connections += 1`, taken under the same
992    /// borrow as the choice, so the next selection weighs it exactly as it
993    /// weighed the connection `Backend::try_connect` used to count at the
994    /// dial. The caller connects. A `connect(2)` that fails releases the
995    /// reservation through [`Backend::release_failed_dial`], which also
996    /// records the failure. A dial that connects but is abandoned before its
997    /// connection is registered releases it with a plain
998    /// `Backend::dec_connections`, recording no failure (the mux's
999    /// `BackendChange::ConnectionClosed`, #1713). A connection that starts
1000    /// keeps it, and its close releases it like any other.
1001    ///
1002    /// `key` is the client's affinity key, as for
1003    /// [`Self::backend_from_cluster_id`].
1004    pub fn reserve_backend(
1005        &mut self,
1006        cluster_id: &str,
1007        key: Option<u64>,
1008        now: Instant,
1009    ) -> Result<Rc<RefCell<Backend>>, BackendError> {
1010        self.reserve_backend_excluding(cluster_id, key, now, &[])
1011    }
1012
1013    /// [`Self::reserve_backend`], preferring a backend whose address is not in
1014    /// `exclude`: the backends a request already failed to connect to
1015    /// (sozu-proxy/sozu#1800). See `BackendList::select_with_key_excluding`
1016    /// for what happens when every selectable backend is excluded.
1017    pub fn reserve_backend_excluding(
1018        &mut self,
1019        cluster_id: &str,
1020        key: Option<u64>,
1021        now: Instant,
1022        exclude: &[SocketAddr],
1023    ) -> Result<Rc<RefCell<Backend>>, BackendError> {
1024        let backend = self.select_backend(cluster_id, key, now, exclude)?;
1025        backend.borrow_mut().reserve_connection()?;
1026        // Re-evaluate on a successful selection, as the connecting path does
1027        // on a successful connect, so an AllDown -> Available recovery is
1028        // reported the moment a request reaches a backend again.
1029        self.record_cluster_availability(cluster_id);
1030        Ok(backend)
1031    }
1032
1033    /// Reserve a connection on the backend of `cluster_id` that
1034    /// `sticky_session` names, if it can take one at `now`, falling back to
1035    /// [`Self::reserve_backend`] under the client's affinity `key`.
1036    ///
1037    /// A cookie naming a live backend wins over the key: the key only decides
1038    /// where a client without a usable cookie lands.
1039    pub fn reserve_sticky_backend(
1040        &mut self,
1041        cluster_id: &str,
1042        sticky_session: &str,
1043        key: Option<u64>,
1044        now: Instant,
1045    ) -> Result<Rc<RefCell<Backend>>, BackendError> {
1046        self.reserve_sticky_backend_excluding(cluster_id, sticky_session, key, now, &[])
1047    }
1048
1049    /// [`Self::reserve_sticky_backend`], except that a cookie naming a backend
1050    /// in `exclude` is not followed: the request already failed to connect to
1051    /// it, so it goes to [`Self::reserve_backend_excluding`] instead
1052    /// (sozu-proxy/sozu#1800).
1053    pub fn reserve_sticky_backend_excluding(
1054        &mut self,
1055        cluster_id: &str,
1056        sticky_session: &str,
1057        key: Option<u64>,
1058        now: Instant,
1059        exclude: &[SocketAddr],
1060    ) -> Result<Rc<RefCell<Backend>>, BackendError> {
1061        let sticky = self
1062            .backends
1063            .get_mut(cluster_id)
1064            .and_then(|cluster_backends| cluster_backends.find_sticky(sticky_session, now))
1065            .filter(|backend| !exclude.contains(&backend.borrow().address))
1066            .cloned();
1067        match sticky {
1068            Some(backend) => {
1069                backend.borrow_mut().reserve_connection()?;
1070                Ok(backend)
1071            }
1072            None => {
1073                debug!(
1074                    "Couldn't find a backend corresponding to sticky_session {} for cluster {}",
1075                    sticky_session, cluster_id
1076                );
1077                self.reserve_backend_excluding(cluster_id, key, now, exclude)
1078            }
1079        }
1080    }
1081
1082    /// Choose a backend of `cluster_id` at `now`, publishing the cluster's
1083    /// availability when nothing can be chosen.
1084    fn select_backend(
1085        &mut self,
1086        cluster_id: &str,
1087        key: Option<u64>,
1088        now: Instant,
1089        exclude: &[SocketAddr],
1090    ) -> Result<Rc<RefCell<Backend>>, BackendError> {
1091        let cluster_backends = self
1092            .backends
1093            .get_mut(cluster_id)
1094            .ok_or(BackendError::NoBackendForCluster(cluster_id.to_owned()))?;
1095
1096        if cluster_backends.backends.is_empty() {
1097            // Drop the &mut BackendList borrow before the &self helper call.
1098            // `total == 0` falls into the "never report AllDown" branch in
1099            // record_cluster_availability, so this just publishes the (0, 0)
1100            // gauges.
1101            let _ = cluster_backends;
1102            self.record_cluster_availability(cluster_id);
1103            return Err(BackendError::NoBackendForCluster(cluster_id.to_owned()));
1104        }
1105        // Past the empty guard there is at least one backend to pick from.
1106        debug_assert!(
1107            !cluster_backends.backends.is_empty(),
1108            "selection runs only on a non-empty backend list"
1109        );
1110
1111        let (picked, outcome) = cluster_backends.select_with_key_excluding(key, now, exclude);
1112        record_shard_outcome(cluster_id, outcome, picked.is_some());
1113        match picked {
1114            Some(backend) => Ok(backend),
1115            None => {
1116                // Drop the &mut BackendList before the &self helper call.
1117                // The helper observes (available=0, total>0) and emits the
1118                // Available -> AllDown transition (log + counter + Event)
1119                // exactly once per regime entry. Subsequent calls in the
1120                // same AllDown regime are no-ops.
1121                let _ = cluster_backends;
1122                self.record_cluster_availability(cluster_id);
1123                Err(BackendError::NoBackendForCluster(cluster_id.to_owned()))
1124            }
1125        }
1126    }
1127
1128    /// Select a backend for `cluster_id`, optionally pinned by an affinity
1129    /// `key`, and return its `(backend_id, address)` **without** opening any
1130    /// connection.
1131    ///
1132    /// This is the UDP datapath's selection entry point. Unlike
1133    /// [`backend_from_cluster_id`](Self::backend_from_cluster_id) — which is
1134    /// TCP-specific because it calls `Backend::try_connect` and hands back a
1135    /// `TcpStream` — UDP owns its own per-flow connected `UdpSocket` (created in
1136    /// the shell via `socket::udp_connect`), so all the map needs to surface is
1137    /// the chosen endpoint identity. `key` is `Some(flow_hash)` so HRW/Maglev
1138    /// keep a client flow pinned to one backend; `None` behaves like the legacy
1139    /// round-robin selection. Fail-open (all-unhealthy ⇒ LB over the full set)
1140    /// is inherited from [`BackendList::next_available_backend_with_key`].
1141    pub fn backend_from_cluster_id_with_key(
1142        &mut self,
1143        cluster_id: &str,
1144        key: Option<u64>,
1145        now: Instant,
1146    ) -> Result<(String, SocketAddr), BackendError> {
1147        let cluster_backends = self
1148            .backends
1149            .get_mut(cluster_id)
1150            .ok_or(BackendError::NoBackendForCluster(cluster_id.to_owned()))?;
1151
1152        if cluster_backends.backends.is_empty() {
1153            let _ = cluster_backends;
1154            self.record_cluster_availability(cluster_id);
1155            return Err(BackendError::NoBackendForCluster(cluster_id.to_owned()));
1156        }
1157        debug_assert!(
1158            !cluster_backends.backends.is_empty(),
1159            "keyed selection runs only on a non-empty backend list"
1160        );
1161
1162        let (picked, outcome) = cluster_backends.select_with_key(key, now);
1163        record_shard_outcome(cluster_id, outcome, picked.is_some());
1164        let next_backend = match picked {
1165            Some(nb) => nb,
1166            None => {
1167                let _ = cluster_backends;
1168                self.record_cluster_availability(cluster_id);
1169                return Err(BackendError::NoBackendForCluster(cluster_id.to_owned()));
1170            }
1171        };
1172
1173        let (backend_id, address) = {
1174            let borrowed = next_backend.borrow();
1175            (borrowed.backend_id.to_owned(), borrowed.address)
1176        };
1177        // The keyed selection returns a live member of the cluster — the
1178        // surfaced identity (id, address) belongs to a registered backend.
1179        debug_assert!(
1180            cluster_backends.has_backend(&address),
1181            "keyed selection must return a backend in the cluster's live set"
1182        );
1183        Ok((backend_id, address))
1184    }
1185
1186    pub fn set_load_balancing_policy_for_cluster(
1187        &mut self,
1188        cluster_id: &str,
1189        lb_algo: LoadBalancingAlgorithms,
1190        metric: Option<LoadMetric>,
1191    ) {
1192        // The cluster can be created before the backends were registered because of the async config messages.
1193        // So when we set the load balancing policy, we have to create the backend list if if it doesn't exist yet.
1194        let seed = self.rng.next_u64();
1195        let cluster_backends = self.get_or_create_backend_list_for_cluster(cluster_id);
1196        cluster_backends.set_load_balancing_policy(lb_algo, metric, seed);
1197    }
1198
1199    /// Set or clear the shuffle sharding of `cluster_id`, creating its
1200    /// backend list if the cluster's backends have not arrived yet, as
1201    /// [`Self::set_load_balancing_policy_for_cluster`] does.
1202    pub fn set_shuffle_sharding_for_cluster(
1203        &mut self,
1204        cluster_id: &str,
1205        shuffle_sharding: Option<ShuffleSharding>,
1206    ) {
1207        self.get_or_create_backend_list_for_cluster(cluster_id)
1208            .set_shuffle_sharding(shuffle_sharding);
1209    }
1210
1211    pub fn get_or_create_backend_list_for_cluster(&mut self, cluster_id: &str) -> &mut BackendList {
1212        let rng = &mut self.rng;
1213        self.backends
1214            .entry(cluster_id.into())
1215            .or_insert_with(|| BackendList::with_seed(rng.next_u64()))
1216    }
1217}
1218
1219/// The worker's backend registry, seen by the UDP core as a selection view.
1220///
1221/// `UdpManager` selects inside the core now (#1340, Question 6) and reaches
1222/// the registry only through this, for the duration of one admission. It
1223/// delegates to [`BackendMap::backend_from_cluster_id_with_key`], the
1224/// connection-free selection entry point — the UDP datapath owns its own
1225/// per-flow `UdpSocket`, so all the map has to surface is the chosen
1226/// endpoint's identity.
1227impl crate::protocol::udp::BackendSource for BackendMap {
1228    fn select(
1229        &mut self,
1230        cluster: &str,
1231        key: Option<u64>,
1232        now: Instant,
1233    ) -> Option<(crate::protocol::udp::BackendId, SocketAddr)> {
1234        self.backend_from_cluster_id_with_key(cluster, key, now)
1235            .ok()
1236    }
1237}
1238
1239#[derive(Debug)]
1240pub struct BackendList {
1241    /// The cluster's backends, in registration order. Crate-private: the
1242    /// sibling state (`sibling_ring`, `shard_representative`,
1243    /// `shares_address`) and the policy table are derived from it and
1244    /// refreshed only by [`Self::add_backend`] and [`Self::remove_backend`],
1245    /// so a direct push or removal would leave them stale. Read it through
1246    /// [`Self::backends`].
1247    pub(crate) backends: Vec<Rc<RefCell<Backend>>>,
1248    pub next_id: u32,
1249    /// The cluster's selection policy. Crate-private: a Maglev policy keeps
1250    /// a lookup table built from `backends`, and only
1251    /// [`Self::set_load_balancing_policy`] seeds it when the policy is
1252    /// installed, so a direct assignment would leave the table empty or
1253    /// stale.
1254    pub(crate) load_balancing: Box<dyn LoadBalancingAlgorithm>,
1255    /// Latches the fail-open `warn!`. Set to `true` when fail-open routing
1256    /// emits its entry warning so subsequent routing decisions in the same
1257    /// regime stay quiet; reset to `false` when a healthy backend is
1258    /// available again, so the regime-exit transition is logged exactly once.
1259    /// Without this latch the warning fired per request, which under a
1260    /// universal outage (the exact scenario fail-open targets) is also the
1261    /// highest request-rate scenario — log volume would become catastrophic.
1262    fail_open_warned: bool,
1263    /// Per-cluster availability latched by `BackendMap::record_cluster_availability`.
1264    /// `Cell` (not `RefCell`) because the receiver is `&self` and the
1265    /// state is `Copy`. Worker runtime is single-threaded, so a `Cell` is
1266    /// sound — no synchronisation needed.
1267    pub(crate) availability: Cell<ClusterAvailability>,
1268    /// Positions in `backends` of the current selection's candidates, refilled
1269    /// by `collect_candidates` on every selection and lent to the policy as a
1270    /// [`Candidates`] view. Reused rather than rebuilt: `add_backend` reserves
1271    /// room for every backend, so a selection never allocates and clones no
1272    /// `Rc` but the one it returns.
1273    candidates: Vec<usize>,
1274    /// The cluster's shuffle sharding, `None` when it is off
1275    /// (sozu-proxy/sozu#524). Set by [`Self::set_shuffle_sharding`].
1276    shuffle_sharding: Option<ShuffleSharding>,
1277    /// Reused scratch for a keyed selection's shard: the HRW score of each
1278    /// primary backend and its position. Reserved in `add_backend` like
1279    /// `candidates`, so computing a shard never allocates.
1280    shard_scores: Vec<(f64, usize)>,
1281    /// The positions of the current selection's shard members, sorted.
1282    shard: Vec<usize>,
1283    /// Whether two backends of the list share an address. Recomputed by
1284    /// `refresh_siblings`, so a list without a shared address — the common
1285    /// case — skips `collapse_shared_addresses` entirely.
1286    shares_address: bool,
1287    /// Sibling ring: `sibling_ring[i]` is the next position whose backend
1288    /// shares the address of position `i`, cyclically through each address's
1289    /// ids in list order, and `i` itself for an address with one id. Lets a
1290    /// selection reach the siblings of an id without scanning the list.
1291    /// Recomputed by `refresh_siblings` on every mutation of the list or of
1292    /// a backend's configuration (`add_backend`, `remove_backend`).
1293    sibling_ring: Vec<usize>,
1294    /// Whether position `i` ranks for shuffle sharding: a primary backend no
1295    /// primary sibling outranks ([`outranks_at_address`]). A function of the
1296    /// configuration alone (address, weight, `backup`, list position), so it
1297    /// is precomputed by `refresh_siblings`.
1298    shard_representative: Vec<bool>,
1299    /// How many positions are `shard_representative`: the distinct primary
1300    /// addresses a shard size is computed from.
1301    shard_addresses: usize,
1302    /// Reused scratch `collapse_shared_addresses` filters the candidates
1303    /// into, reserved in `add_backend` like `candidates`.
1304    collapsed: Vec<usize>,
1305}
1306
1307/// How shuffle sharding shaped one selection, for the caller that knows the
1308/// cluster to count it under.
1309#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1310pub(crate) enum ShardOutcome {
1311    /// Sharding is off, inactive below `shard_min_backends`, or the request
1312    /// had no key: the whole cluster was eligible.
1313    Unsharded,
1314    /// The selection stayed inside the client's shard.
1315    InShard,
1316    /// No member of the shard could take a connection and `FALLBACK`
1317    /// selected over the rest of the cluster.
1318    SpilledOver,
1319    /// No member of the shard could take a connection and `STRICT` selected
1320    /// nothing.
1321    Exhausted,
1322}
1323
1324impl Default for BackendList {
1325    fn default() -> Self {
1326        Self::new()
1327    }
1328}
1329
1330impl BackendList {
1331    /// An empty list whose default `Random` policy is seeded from the OS.
1332    pub fn new() -> BackendList {
1333        Self::with_seed(os_rng().next_u64())
1334    }
1335
1336    /// An empty list whose default `Random` policy is seeded with `seed`.
1337    ///
1338    /// [`BackendMap`] builds every list it holds this way, from its own
1339    /// generator, so a seeded map seeds every cluster (#1684).
1340    pub fn with_seed(seed: u64) -> BackendList {
1341        BackendList {
1342            backends: Vec::new(),
1343            next_id: 0,
1344            load_balancing: Box::new(Random::with_seed(seed)),
1345            fail_open_warned: false,
1346            availability: Cell::new(ClusterAvailability::Available),
1347            candidates: Vec::new(),
1348            shuffle_sharding: None,
1349            shard_scores: Vec::new(),
1350            shard: Vec::new(),
1351            shares_address: false,
1352            sibling_ring: Vec::new(),
1353            shard_representative: Vec::new(),
1354            shard_addresses: 0,
1355            collapsed: Vec::new(),
1356        }
1357    }
1358
1359    /// The cluster's backends, in registration order.
1360    ///
1361    /// Read-only: the list is mutated only through [`Self::add_backend`] and
1362    /// [`Self::remove_backend`], which refresh the sibling state and the
1363    /// policy table derived from it. Code outside this crate cannot reach
1364    /// the field itself:
1365    ///
1366    /// ```
1367    /// use std::{cell::RefCell, rc::Rc};
1368    ///
1369    /// use sozu_lib::backends::{Backend, BackendList};
1370    ///
1371    /// let mut list = BackendList::with_seed(1);
1372    /// let backend = Backend::new("b1", "127.0.0.1:8080".parse().unwrap(), None, None, None);
1373    /// list.add_backend(backend);
1374    /// let backends: &[Rc<RefCell<Backend>>] = list.backends();
1375    /// assert_eq!(backends.len(), 1);
1376    /// ```
1377    ///
1378    /// ```compile_fail,E0616
1379    /// use std::{cell::RefCell, rc::Rc};
1380    ///
1381    /// use sozu_lib::backends::{Backend, BackendList};
1382    ///
1383    /// let mut list = BackendList::with_seed(1);
1384    /// let backend = Backend::new("b1", "127.0.0.1:8080".parse().unwrap(), None, None, None);
1385    /// list.backends.push(Rc::new(RefCell::new(backend)));
1386    /// ```
1387    pub fn backends(&self) -> &[Rc<RefCell<Backend>>] {
1388        &self.backends
1389    }
1390
1391    /// Count `(available, total)` for this cluster. Delegates to
1392    /// `Backend::is_available` so the per-cluster aggregate and the
1393    /// per-backend `backend.available` gauge stay in lock-step.
1394    pub(crate) fn evaluate_availability(&self) -> (usize, usize) {
1395        let total = self.backends.len();
1396        let available = self
1397            .backends
1398            .iter()
1399            .filter(|b| b.borrow().is_available())
1400            .count();
1401        // The available count is a filtered subset of the total, so it can
1402        // never exceed it, and `total` mirrors the backend vector length.
1403        debug_assert!(
1404            available <= total,
1405            "available ({available}) cannot exceed total ({total})"
1406        );
1407        debug_assert_eq!(total, self.backends.len(), "total must equal backend count");
1408        (available, total)
1409    }
1410
1411    /// Full invariant sweep over the backend list. Called as a `debug_assert!`
1412    /// postcondition by every mutating method so any cross-field corruption
1413    /// (duplicate addresses, a `next_id` that underflowed below the registered
1414    /// count) surfaces immediately under test/fuzz.
1415    #[cfg(debug_assertions)]
1416    fn check_invariants(&self) {
1417        // `next_id` is a monotonically incremented registration counter: it is
1418        // bumped once per *newly inserted* backend (never decremented, even on
1419        // removal), so it is always at least the current live count.
1420        debug_assert!(
1421            self.next_id as usize >= self.backends.len(),
1422            "next_id ({}) must be >= live backend count ({})",
1423            self.next_id,
1424            self.backends.len()
1425        );
1426        // The precomputed sibling state gates and drives every selection: a
1427        // stale value would silently hand a policy one candidate per id
1428        // again, or rank a wrong representative. Check it in `O(n)` plus the
1429        // rings' lengths, so the sweep stays affordable on large lists.
1430        let n = self.backends.len();
1431        debug_assert_eq!(self.sibling_ring.len(), n, "one ring slot per backend");
1432        debug_assert_eq!(self.shard_representative.len(), n);
1433        // A permutation whose links join ids of one address...
1434        let mut targeted = vec![false; n];
1435        for (index, &next) in self.sibling_ring.iter().enumerate() {
1436            debug_assert!(!targeted[next], "the sibling ring is a permutation");
1437            targeted[next] = true;
1438            debug_assert_eq!(
1439                self.backends[next].borrow().address,
1440                self.backends[index].borrow().address,
1441                "a sibling ring links only ids of one address"
1442            );
1443        }
1444        // ...with exactly one cycle per address: every id of an address
1445        // reaches the same cycle leader, the cycle's smallest position.
1446        let mut leaders: HashMap<SocketAddr, usize> = HashMap::new();
1447        for index in 0..n {
1448            let mut leader = index;
1449            let mut next = self.sibling_ring[index];
1450            while next != index {
1451                leader = leader.min(next);
1452                next = self.sibling_ring[next];
1453            }
1454            let address = self.backends[index].borrow().address;
1455            debug_assert_eq!(
1456                *leaders.entry(address).or_insert(leader),
1457                leader,
1458                "an address has one sibling ring"
1459            );
1460            let backend = self.backends[index].borrow();
1461            let mut represents = !backend.backup;
1462            let mut sibling = self.sibling_ring[index];
1463            while sibling != index {
1464                let other = self.backends[sibling].borrow();
1465                represents &=
1466                    other.backup || !outranks_at_address(&other, sibling, &backend, index);
1467                sibling = self.sibling_ring[sibling];
1468            }
1469            debug_assert_eq!(
1470                self.shard_representative[index], represents,
1471                "shard_representative must match the configuration"
1472            );
1473        }
1474        debug_assert_eq!(
1475            self.shares_address,
1476            leaders.len() < n,
1477            "shares_address must match the backend list"
1478        );
1479        debug_assert_eq!(
1480            self.shard_addresses,
1481            self.shard_representative.iter().filter(|r| **r).count()
1482        );
1483        // `(backend_id, address)` is the identity `remove_backend` keys on; two
1484        // live backends may legitimately share an address (A/B variant) but
1485        // must then differ by `backend_id`. The pair is therefore unique
1486        // across the live set.
1487        for (i, a) in self.backends.iter().enumerate() {
1488            let a = a.borrow();
1489            for b in self.backends.iter().skip(i + 1) {
1490                let b = b.borrow();
1491                debug_assert!(
1492                    a.address != b.address || a.backend_id != b.backend_id,
1493                    "duplicate (address, backend_id) in the live set: {:?} / {}",
1494                    a.address,
1495                    a.backend_id
1496                );
1497            }
1498        }
1499    }
1500
1501    /// A list holding `backend_vec`, its default policy seeded with `seed`.
1502    pub fn import_configuration_state(
1503        backend_vec: &[sozu_command_lib::response::Backend],
1504        seed: u64,
1505    ) -> BackendList {
1506        let mut list = BackendList::with_seed(seed);
1507        for backend in backend_vec {
1508            let backend = Backend::new(
1509                &backend.backend_id,
1510                backend.address,
1511                backend.sticky_id.clone(),
1512                backend.load_balancing_parameters,
1513                backend.backup,
1514            )
1515            .with_unix_path(backend.unix_path.clone());
1516            list.add_backend(backend);
1517        }
1518
1519        list
1520    }
1521
1522    pub fn add_backend(&mut self, backend: Backend) {
1523        let address = backend.address;
1524        let len_before = self.backends.len();
1525        let next_id_before = self.next_id;
1526        let existed = self.backends.iter().any(|b| {
1527            b.borrow().address == backend.address && b.borrow().backend_id == backend.backend_id
1528        });
1529        match self.backends.iter_mut().find(|b| {
1530            b.borrow().address == backend.address && b.borrow().backend_id == backend.backend_id
1531        }) {
1532            None => {
1533                let backend = Rc::new(RefCell::new(backend));
1534                self.backends.push(backend);
1535                self.next_id += 1;
1536            }
1537            // the backend already exists, update the configuration while
1538            // keeping connection retry state
1539            Some(old_backend) => {
1540                let mut b = old_backend.borrow_mut();
1541                b.sticky_id.clone_from(&backend.sticky_id);
1542                b.load_balancing_parameters
1543                    .clone_from(&backend.load_balancing_parameters);
1544                b.backup = backend.backup;
1545                b.unix_path.clone_from(&backend.unix_path);
1546            }
1547        }
1548        // Insert grows the list by exactly one and bumps `next_id`; an update
1549        // in place leaves both untouched. The address is present either way.
1550        debug_assert_eq!(
1551            self.backends.len(),
1552            len_before + (!existed) as usize,
1553            "add_backend grows the list by one only on a genuine insert"
1554        );
1555        debug_assert_eq!(
1556            self.next_id,
1557            next_id_before + (!existed) as u32,
1558            "next_id advances by one only on a genuine insert"
1559        );
1560        debug_assert!(
1561            self.has_backend(&address),
1562            "add_backend must leave the backend present in the list"
1563        );
1564        // Refresh table-based policies (Maglev) and the running weights of
1565        // round robin off the datapath whenever the full backend set or a
1566        // weight changes. This is the ONLY place the table is rebuilt on
1567        // mutation; selection never rebuilds. The default `rebuild` is a
1568        // no-op for the stateless policies.
1569        self.load_balancing.rebuild(&self.backends);
1570        // Size the candidate buffer for the whole list here, on the control
1571        // plane, so no selection ever grows it.
1572        self.candidates.clear();
1573        self.candidates.reserve(self.backends.len());
1574        self.shard_scores.clear();
1575        self.shard_scores.reserve(self.backends.len());
1576        self.shard.clear();
1577        self.shard.reserve(self.backends.len());
1578        self.collapsed.clear();
1579        self.collapsed.reserve(self.backends.len());
1580        self.refresh_siblings();
1581        #[cfg(debug_assertions)]
1582        self.check_invariants();
1583    }
1584
1585    /// Recompute the sibling state — [`Self::shares_address`],
1586    /// [`Self::sibling_ring`], [`Self::shard_representative`] and
1587    /// [`Self::shard_addresses`] — from the backend list and its
1588    /// configuration. Control plane only: it allocates a map of the
1589    /// addresses, linear in the list length, plus, for each shared address
1590    /// of `g` ids, up to `g²` comparisons to rank them.
1591    fn refresh_siblings(&mut self) {
1592        let mut groups: HashMap<SocketAddr, Vec<usize>> = HashMap::new();
1593        for (index, backend) in self.backends.iter().enumerate() {
1594            groups
1595                .entry(backend.borrow().address)
1596                .or_default()
1597                .push(index);
1598        }
1599        self.sibling_ring.clear();
1600        self.sibling_ring.extend(0..self.backends.len());
1601        self.shares_address = false;
1602        for group in groups.values() {
1603            if group.len() > 1 {
1604                self.shares_address = true;
1605                for (rank, &index) in group.iter().enumerate() {
1606                    self.sibling_ring[index] = group[(rank + 1) % group.len()];
1607                }
1608            }
1609        }
1610        self.shard_representative.clear();
1611        for (index, backend) in self.backends.iter().enumerate() {
1612            let backend = backend.borrow();
1613            let mut represents = !backend.backup;
1614            let mut sibling = self.sibling_ring[index];
1615            while represents && sibling != index {
1616                let other = self.backends[sibling].borrow();
1617                represents = other.backup || !outranks_at_address(&other, sibling, &backend, index);
1618                sibling = self.sibling_ring[sibling];
1619            }
1620            self.shard_representative.push(represents);
1621        }
1622        self.shard_addresses = self.shard_representative.iter().filter(|r| **r).count();
1623    }
1624
1625    /// Keep one candidate per address, its representative
1626    /// ([`outranks_at_address`]: the heaviest eligible id, the first in list
1627    /// order among equals), so every policy gives an address the share of
1628    /// one backend however many ids it carries (decided 2026-10-04). Runs on
1629    /// the candidates that passed the tier and shard filters, so a down or
1630    /// backing-off id never hides an eligible sibling. A filter: the
1631    /// candidates stay in list order, as `Candidates::new` requires.
1632    ///
1633    /// Free when no address is shared. Otherwise one pass over the
1634    /// candidates, where only an id with a sibling walks its address's ring
1635    /// (`sibling_ring`) and looks each sibling up among the candidates by
1636    /// binary search: linear in the `c` candidates plus, for each shared
1637    /// address of `g` ids, up to `g²` ring steps (about `g²/2` when the
1638    /// walks stop at the first outranking sibling), each with an
1639    /// `O(log c)` lookup — never a scan of the list per candidate.
1640    /// Allocates nothing (`collapsed` is reserved by `add_backend`).
1641    ///
1642    /// A list whose `backends` a caller grew or shrank directly, bypassing
1643    /// `add_backend`/`remove_backend`, has a ring of another length: the
1644    /// collapse is then skipped rather than indexing a stale ring.
1645    fn collapse_shared_addresses(&mut self) -> usize {
1646        if !self.shares_address || self.sibling_ring.len() != self.backends.len() {
1647            return self.candidates.len();
1648        }
1649        let backends = &self.backends;
1650        let candidates = &self.candidates;
1651        let ring = &self.sibling_ring;
1652        self.collapsed.clear();
1653        for &position in candidates {
1654            let mut outranked = false;
1655            let mut sibling = ring[position];
1656            if sibling != position {
1657                let backend = backends[position].borrow();
1658                while !outranked && sibling != position {
1659                    count_sibling_step();
1660                    outranked = candidates.binary_search(&sibling).is_ok()
1661                        && outranks_at_address(
1662                            &backends[sibling].borrow(),
1663                            sibling,
1664                            &backend,
1665                            position,
1666                        );
1667                    sibling = ring[sibling];
1668                }
1669            }
1670            if !outranked {
1671                self.collapsed.push(position);
1672            }
1673        }
1674        std::mem::swap(&mut self.candidates, &mut self.collapsed);
1675        // `collapsed` now holds the uncollapsed candidates: collapsing only
1676        // drops ids, and keeps one per address, so a non-empty set stays so.
1677        debug_assert!(
1678            self.candidates.len() <= self.collapsed.len(),
1679            "collapsing never adds a candidate"
1680        );
1681        debug_assert!(
1682            self.collapsed.is_empty() || !self.candidates.is_empty(),
1683            "every address keeps one candidate"
1684        );
1685        self.candidates.len()
1686    }
1687
1688    /// Set or clear the cluster's shuffle sharding. Takes effect from the
1689    /// next selection.
1690    pub fn set_shuffle_sharding(&mut self, shuffle_sharding: Option<ShuffleSharding>) {
1691        self.shuffle_sharding = shuffle_sharding;
1692    }
1693
1694    /// Fill `self.shard` with the positions of `key`'s shard, sorted, and
1695    /// say whether sharding applies to this selection at all.
1696    ///
1697    /// The shard is the top `k` of the HRW ranking of `key` over the
1698    /// **configured** primary addresses, healthy or not, with every primary
1699    /// id at those addresses (one address is one share, see
1700    /// `collapse_shared_addresses`): a backend going down
1701    /// must not pull another one into the shard, or a shard could never be
1702    /// exhausted and isolation would leak exactly when it matters. Ties are
1703    /// broken by list position so the shard is a function of the scores.
1704    /// Partitioning with `select_nth_unstable_by` is `O(N)`; each address's
1705    /// representative is precomputed (`shard_representative`), and a shared
1706    /// address brings its other ids in by walking its ring, so a shared
1707    /// address adds work only for its own ids. The buffers are reserved by
1708    /// `add_backend`, so this allocates nothing. If a caller grew or shrank
1709    /// the public `backends` directly, the precomputed state has another
1710    /// length and is ignored: every primary then ranks as its own address,
1711    /// as before shared addresses were collapsed, and nothing is indexed out
1712    /// of bounds.
1713    fn compute_shard(&mut self, key: Option<u64>) -> bool {
1714        let (Some(sharding), Some(key)) = (self.shuffle_sharding, key) else {
1715            return false;
1716        };
1717        // One address is one share: a shard counts and ranks primary
1718        // ADDRESSES, each scored through its representative primary id
1719        // (`outranks_at_address`), and takes in every primary id at the
1720        // addresses it selects, so the sibling of a down representative still
1721        // serves inside the shard. Without a shared address every primary is
1722        // its own representative.
1723        let backends = &self.backends;
1724        let shares_address = self.shares_address;
1725        let synced = self.shard_representative.len() == backends.len()
1726            && self.sibling_ring.len() == backends.len();
1727        let ring: &[usize] = if synced { &self.sibling_ring } else { &[] };
1728        // Size the shard before hashing anything, so a cluster below
1729        // `shard_min_backends` pays no walk and no hash.
1730        let primaries = if synced {
1731            self.shard_addresses
1732        } else {
1733            backends
1734                .iter()
1735                .filter(|backend| !backend.borrow().backup)
1736                .count()
1737        };
1738        let Some(k) = sharding.shard_size(primaries) else {
1739            return false;
1740        };
1741        self.shard_scores.clear();
1742        for (index, backend) in backends.iter().enumerate() {
1743            let represents = match self.shard_representative.get(index) {
1744                Some(&represents) if synced => represents,
1745                _ => !backend.borrow().backup,
1746            };
1747            if represents {
1748                self.shard_scores
1749                    .push((hrw_score(key, &backend.borrow()), index));
1750            }
1751        }
1752        debug_assert_eq!(
1753            self.shard_scores.len(),
1754            primaries,
1755            "one score per primary address"
1756        );
1757        let by_rank = |a: &(f64, usize), b: &(f64, usize)| b.0.total_cmp(&a.0).then(a.1.cmp(&b.1));
1758        if k < self.shard_scores.len() {
1759            self.shard_scores.select_nth_unstable_by(k - 1, by_rank);
1760        }
1761        self.shard.clear();
1762        for &(_, representative) in &self.shard_scores[..k] {
1763            self.shard.push(representative);
1764            let next = |position: usize| ring.get(position).copied().unwrap_or(representative);
1765            let mut sibling = next(representative);
1766            while sibling != representative {
1767                count_sibling_step();
1768                if !backends[sibling].borrow().backup {
1769                    self.shard.push(sibling);
1770                }
1771                sibling = next(sibling);
1772            }
1773        }
1774        self.shard.sort_unstable();
1775        debug_assert!(
1776            self.shard.len() >= k,
1777            "a shard holds every primary id of its k addresses"
1778        );
1779        debug_assert!(
1780            shares_address || self.shard.len() == k,
1781            "without a shared address a shard holds exactly k backends"
1782        );
1783        debug_assert!(
1784            self.shard
1785                .iter()
1786                .all(|&index| !self.backends[index].borrow().backup),
1787            "a shard holds primary backends only"
1788        );
1789        true
1790    }
1791
1792    /// Keep only the candidates that belong to the current shard; the
1793    /// candidates stay in list order, as `Candidates::new` requires.
1794    fn retain_shard_candidates(&mut self) -> usize {
1795        let shard = &self.shard;
1796        self.candidates
1797            .retain(|index| shard.binary_search(index).is_ok());
1798        self.candidates.len()
1799    }
1800
1801    /// Remove the backend identified by `(backend_id, backend_address)`, the
1802    /// identity `AddBackend` and `ConfigState` key backends on, and return
1803    /// whether it was present. Another id at the same address (A/B test,
1804    /// weighted variant) is a distinct backend and stays (#1821).
1805    pub fn remove_backend(&mut self, backend_id: &str, backend_address: &SocketAddr) -> bool {
1806        let len_before = self.backends.len();
1807        let mut removed = false;
1808        self.backends.retain(|backend| {
1809            let mut b = backend.borrow_mut();
1810            if b.backend_id == backend_id && &b.address == backend_address {
1811                removed = true;
1812                // A session may still hold this backend: retire it so none
1813                // of its pooled connections takes a new request.
1814                b.set_closing();
1815                false
1816            } else {
1817                true
1818            }
1819        });
1820        // The list holds each `(backend_id, address)` at most once, so it
1821        // shrinks by exactly one entry when the identity was present.
1822        debug_assert_eq!(
1823            self.backends.len(),
1824            len_before - removed as usize,
1825            "remove_backend must drop exactly the backend it reports"
1826        );
1827        debug_assert!(
1828            self.find_backend_by_identity(backend_id, backend_address)
1829                .is_none(),
1830            "remove_backend must evict the backend it names"
1831        );
1832        // Rebuild table-based policies (Maglev) and reset round robin's
1833        // running weights off the datapath after the set shrinks, only when
1834        // something was actually removed. No-op for the stateless policies.
1835        if removed {
1836            self.load_balancing.rebuild(&self.backends);
1837            self.refresh_siblings();
1838        }
1839        #[cfg(debug_assertions)]
1840        self.check_invariants();
1841        removed
1842    }
1843
1844    /// The live backend identified by `(backend_id, backend_address)`.
1845    pub fn find_backend_by_identity(
1846        &self,
1847        backend_id: &str,
1848        backend_address: &SocketAddr,
1849    ) -> Option<&Rc<RefCell<Backend>>> {
1850        self.backends.iter().find(|backend| {
1851            let b = backend.borrow();
1852            b.backend_id == backend_id && b.address == *backend_address
1853        })
1854    }
1855
1856    /// The live entry that is the `incarnation` a health probe captured when
1857    /// it launched, or `None` once that backend left the list: removed, or
1858    /// replaced by a backend re-added under the same id and address, which is
1859    /// a new incarnation. Comparing allocations is sound because the `Weak`
1860    /// keeps its allocation alive, so no later backend can reuse it (#1821).
1861    pub fn find_incarnation(
1862        &self,
1863        incarnation: &Weak<RefCell<Backend>>,
1864    ) -> Option<&Rc<RefCell<Backend>>> {
1865        self.backends
1866            .iter()
1867            .find(|backend| std::ptr::eq(Rc::as_ptr(backend), incarnation.as_ptr()))
1868    }
1869
1870    pub fn has_backend(&self, backend_address: &SocketAddr) -> bool {
1871        self.backends
1872            .iter()
1873            .any(|backend| backend.borrow().address == *backend_address)
1874    }
1875
1876    pub fn find_backend(
1877        &mut self,
1878        backend_address: &SocketAddr,
1879    ) -> Option<&mut Rc<RefCell<Backend>>> {
1880        self.backends
1881            .iter_mut()
1882            .find(|backend| backend.borrow().address == *backend_address)
1883    }
1884
1885    /// The backend `sticky_session` names, if it can take a connection at
1886    /// `now`.
1887    pub fn find_sticky(
1888        &mut self,
1889        sticky_session: &str,
1890        now: Instant,
1891    ) -> Option<&mut Rc<RefCell<Backend>>> {
1892        self.backends
1893            .iter_mut()
1894            .find(|b| b.borrow().sticky_id.as_deref() == Some(sticky_session))
1895            .filter(|b| b.borrow().can_open(now))
1896    }
1897
1898    /// The backends of one tier that can take a connection now, cloned into
1899    /// a fresh `Vec`. Selection does not call this: it walks the same
1900    /// predicate, `is_tier_candidate`, into a reused buffer instead.
1901    pub fn available_backends(&mut self, backup: bool, now: Instant) -> Vec<Rc<RefCell<Backend>>> {
1902        self.backends
1903            .iter()
1904            .filter(|backend| is_tier_candidate(&backend.borrow(), backup, now))
1905            .map(Clone::clone)
1906            .collect()
1907    }
1908
1909    /// Refill `candidates` with the position of every backend `keep` accepts,
1910    /// in list order, and return how many there are. Allocation-free once
1911    /// `add_backend` has reserved the buffer.
1912    fn collect_candidates(&mut self, keep: impl Fn(&Backend) -> bool) -> usize {
1913        self.candidates.clear();
1914        for (index, backend) in self.backends.iter().enumerate() {
1915            if keep(&backend.borrow()) {
1916                self.candidates.push(index);
1917            }
1918        }
1919        debug_assert!(
1920            self.candidates.len() <= self.backends.len(),
1921            "candidate set cannot be larger than the full backend list"
1922        );
1923        debug_assert!(
1924            self.candidates.windows(2).all(|pair| pair[0] < pair[1]),
1925            "candidate positions must follow list order"
1926        );
1927        self.candidates.len()
1928    }
1929
1930    /// Pick the next available backend at `now`.
1931    pub fn next_available_backend(&mut self, now: Instant) -> Option<Rc<RefCell<Backend>>> {
1932        self.next_available_backend_with_key(None, now)
1933    }
1934
1935    /// Pick the next available backend, optionally pinned by an affinity `key`.
1936    ///
1937    /// `key` is only consulted by consistent-hashing policies (HRW/Maglev);
1938    /// every other policy ignores it, so `next_available_backend_with_key(None)`
1939    /// is byte-for-byte the legacy behavior. The UDP datapath calls this with
1940    /// `Some(flow_hash)` to keep a client flow pinned to one backend; HTTP,
1941    /// HTTPS and TCP reach it through [`BackendMap::backend_from_cluster_id`]
1942    /// with the client affinity key of a cluster that uses HRW or Maglev.
1943    ///
1944    /// `now` is the instant the selection happens at: it decides which
1945    /// backends are out of their backoff window and how far each
1946    /// connection-time average has decayed, so the same `now` over the same
1947    /// state picks the same backend.
1948    pub fn next_available_backend_with_key(
1949        &mut self,
1950        key: Option<u64>,
1951        now: Instant,
1952    ) -> Option<Rc<RefCell<Backend>>> {
1953        self.select_with_key(key, now).0
1954    }
1955
1956    /// [`Self::next_available_backend_with_key`], also saying how shuffle
1957    /// sharding shaped the selection, for [`BackendMap`] to count under the
1958    /// cluster's id.
1959    ///
1960    /// Tiers, in order: the primary backends that can open, then the backup
1961    /// backends, then fail-open over every backend whose retry policy allows
1962    /// a try. When the cluster shards and `key` is known, the primary tier is
1963    /// first narrowed to the client's shard. An exhausted shard then either
1964    /// spills over to the unsharded tiers (`FALLBACK`) or ends the selection
1965    /// empty (`STRICT`), except in the fail-open regime — no backend of the
1966    /// whole cluster can open — where `STRICT` fails open inside its shard
1967    /// only.
1968    pub(crate) fn select_with_key(
1969        &mut self,
1970        key: Option<u64>,
1971        now: Instant,
1972    ) -> (Option<Rc<RefCell<Backend>>>, ShardOutcome) {
1973        self.select_tiers(key, now, &[])
1974    }
1975
1976    /// [`Self::select_with_key`] for a request that already failed to connect
1977    /// to the backends at the addresses in `exclude` (sozu-proxy/sozu#1800).
1978    ///
1979    /// Every policy picks from the candidate set this list builds, so leaving
1980    /// those backends out of it is what makes a retry skip them under round
1981    /// robin, random, least loaded, power of two, HRW and Maglev alike.
1982    ///
1983    /// The exclusion only narrows the healthy tiers, primary then backup. When
1984    /// it leaves both empty — every backend that can take a connection was
1985    /// already tried by this request — the selection runs again without it,
1986    /// exactly as [`Self::select_with_key`] would: the request retries a
1987    /// backend it already tried rather than failing, and the fail-open regime
1988    /// still decides when no backend can take a connection at all. A backend
1989    /// that failed is usually out of the healthy tiers already, its retry
1990    /// policy holding it in back-off; the exclusion matters when that back-off
1991    /// expires while the request is still retrying, which a connect timeout as
1992    /// long as the back-off makes routine.
1993    pub(crate) fn select_with_key_excluding(
1994        &mut self,
1995        key: Option<u64>,
1996        now: Instant,
1997        exclude: &[SocketAddr],
1998    ) -> (Option<Rc<RefCell<Backend>>>, ShardOutcome) {
1999        if !exclude.is_empty() {
2000            let selection = self.select_tiers(key, now, exclude);
2001            if selection.0.is_some() {
2002                debug_assert!(
2003                    selection
2004                        .0
2005                        .as_ref()
2006                        .is_some_and(|backend| !exclude.contains(&backend.borrow().address)),
2007                    "an excluding selection must not return an excluded backend"
2008                );
2009                return selection;
2010            }
2011        }
2012        self.select_tiers(key, now, &[])
2013    }
2014
2015    /// The selection itself. A non-empty `exclude` removes those addresses
2016    /// from the healthy tiers and skips the fail-open regime, which
2017    /// [`Self::select_with_key_excluding`] reaches through a second call
2018    /// without exclusion.
2019    fn select_tiers(
2020        &mut self,
2021        key: Option<u64>,
2022        now: Instant,
2023        exclude: &[SocketAddr],
2024    ) -> (Option<Rc<RefCell<Backend>>>, ShardOutcome) {
2025        let candidate = |backend: &Backend, backup: bool| {
2026            is_tier_candidate(backend, backup, now) && !exclude.contains(&backend.address)
2027        };
2028        let sharded = self.compute_shard(key);
2029        let strict = sharded
2030            && self
2031                .shuffle_sharding
2032                .is_some_and(|sharding| sharding.mode == ShardMode::Strict);
2033        let mut outcome = if sharded {
2034            ShardOutcome::InShard
2035        } else {
2036            ShardOutcome::Unsharded
2037        };
2038
2039        let mut available = self.collect_candidates(|backend| candidate(backend, false));
2040        if sharded {
2041            // Primary backends able to take a connection, inside the shard
2042            // or not: under `STRICT` they, and only they, decide between
2043            // refusing and failing open. Backups sit outside every shard, so
2044            // a healthy backup must not turn a strict shard's fail-open into
2045            // a refusal.
2046            let primaries_can_open = available != 0;
2047            let in_shard = self.retain_shard_candidates();
2048            if in_shard != 0 {
2049                available = in_shard;
2050            } else if strict {
2051                if primaries_can_open {
2052                    // Primary backends outside the shard could serve, but
2053                    // strict isolation forbids them.
2054                    return (None, ShardOutcome::Exhausted);
2055                }
2056                // Fail-open regime of the primary tier: fall through with an
2057                // empty primary tier; the backup tier is skipped below for a
2058                // strict shard, and fail-open is narrowed to the shard.
2059                available = 0;
2060            } else {
2061                outcome = ShardOutcome::SpilledOver;
2062                available = self.collect_candidates(|backend| candidate(backend, false));
2063            }
2064        }
2065
2066        if available == 0 && !strict {
2067            available = self.collect_candidates(|backend| candidate(backend, true));
2068        }
2069
2070        if available != 0 {
2071            // Healthy regime: log the fail-open exit transition exactly once.
2072            if self.fail_open_warned {
2073                info!(
2074                    "fail-open: cluster recovered, {} backends now healthy",
2075                    available
2076                );
2077                self.fail_open_warned = false;
2078            }
2079            self.collapse_shared_addresses();
2080            let picked = self.load_balancing.next_available_backend(
2081                key,
2082                Candidates::with_siblings(
2083                    &self.backends,
2084                    &self.candidates,
2085                    &self.sibling_ring,
2086                    now,
2087                ),
2088            );
2089            debug_assert!(
2090                picked.as_ref().is_none_or(|b| {
2091                    let addr = b.borrow().address;
2092                    self.backends.iter().any(|x| x.borrow().address == addr)
2093                }),
2094                "selection must return a backend present in the live list"
2095            );
2096            debug_assert!(
2097                outcome != ShardOutcome::InShard
2098                    || picked.as_ref().is_none_or(|b| {
2099                        let addr = b.borrow().address;
2100                        self.shard
2101                            .iter()
2102                            .any(|&index| self.backends[index].borrow().address == addr)
2103                    }),
2104                "an in-shard selection must return a member of the shard"
2105            );
2106            let outcome = self.settle_spill_outcome(outcome, picked.as_ref());
2107            return (picked, outcome);
2108        }
2109
2110        // Fail-open: when no backend passes the full `can_open()` gate,
2111        // route to backends that are administratively `Normal` AND whose
2112        // retry policy reports `OKAY` (i.e., not currently in
2113        // exponential-backoff). This prevents a shared dependency outage
2114        // (e.g., database) from making the entire cluster unavailable while
2115        // still respecting the per-backend back-off window — hammering a
2116        // backend at line rate during its back-off would defeat the back-off
2117        // itself. Ref: Amazon "Implementing Health Checks".
2118        //
2119        // An excluding selection stops before this regime: it is decided by
2120        // the second, unexcluding call `select_with_key_excluding` makes.
2121        if !exclude.is_empty() {
2122            return (None, outcome);
2123        }
2124        let mut available = self.collect_candidates(|backend| {
2125            backend.status == BackendStatus::Normal
2126                && matches!(
2127                    backend.retry_policy.can_try(now),
2128                    Some(retry::RetryAction::OKAY)
2129                )
2130        });
2131        if strict {
2132            // A strict shard fails open inside itself only.
2133            available = self.retain_shard_candidates();
2134            if available == 0 {
2135                return (None, ShardOutcome::Exhausted);
2136            }
2137        }
2138
2139        if available == 0 {
2140            return (None, outcome);
2141        }
2142
2143        // Latched warning + per-decision counter: the warn! fires once on
2144        // regime entry; the counter is the operator-visible per-request
2145        // signal that does not drown logs under universal outage.
2146        if !self.fail_open_warned {
2147            warn!(
2148                "fail-open: all backends unhealthy, routing to {} normal backends with retry-policy OKAY",
2149                available
2150            );
2151            self.fail_open_warned = true;
2152        }
2153        count!(names::backend::FAIL_OPEN, 1);
2154
2155        self.collapse_shared_addresses();
2156        let picked = self.load_balancing.next_available_backend(
2157            key,
2158            Candidates::with_siblings(&self.backends, &self.candidates, &self.sibling_ring, now),
2159        );
2160        let outcome = self.settle_spill_outcome(outcome, picked.as_ref());
2161        (picked, outcome)
2162    }
2163
2164    /// A `FALLBACK` selection that left its exhausted shard but whose pick is
2165    /// a shard member after all — fail-open over the whole cluster can land
2166    /// back in the shard — stayed in the shard, and is not a spill-over.
2167    fn settle_spill_outcome(
2168        &self,
2169        outcome: ShardOutcome,
2170        picked: Option<&Rc<RefCell<Backend>>>,
2171    ) -> ShardOutcome {
2172        match (outcome, picked) {
2173            (ShardOutcome::SpilledOver, Some(picked))
2174                if self
2175                    .shard
2176                    .iter()
2177                    .any(|&index| Rc::ptr_eq(&self.backends[index], picked)) =>
2178            {
2179                ShardOutcome::InShard
2180            }
2181            _ => outcome,
2182        }
2183    }
2184
2185    /// Replace the cluster's policy. `seed` seeds the policies that draw at
2186    /// random, `Random` and `PowerOfTwo`; the others ignore it.
2187    ///
2188    /// This is the only way to change the policy: it builds the Maglev
2189    /// lookup table from the current backends when it installs `Maglev`.
2190    /// Code outside this crate cannot assign the field directly:
2191    ///
2192    /// ```
2193    /// use sozu_command_lib::proto::command::LoadBalancingAlgorithms;
2194    /// use sozu_lib::backends::{Backend, BackendList};
2195    ///
2196    /// let mut list = BackendList::with_seed(1);
2197    /// list.add_backend(Backend::new("b1", "127.0.0.1:8080".parse().unwrap(), None, None, None));
2198    /// list.set_load_balancing_policy(LoadBalancingAlgorithms::Maglev, None, 1);
2199    /// ```
2200    ///
2201    /// ```compile_fail,E0616
2202    /// use sozu_lib::{backends::BackendList, load_balancing::Maglev};
2203    ///
2204    /// let mut list = BackendList::with_seed(1);
2205    /// list.load_balancing = Box::new(Maglev::new());
2206    /// ```
2207    pub fn set_load_balancing_policy(
2208        &mut self,
2209        load_balancing_policy: LoadBalancingAlgorithms,
2210        metric: Option<LoadMetric>,
2211        seed: u64,
2212    ) {
2213        match load_balancing_policy {
2214            LoadBalancingAlgorithms::RoundRobin => {
2215                let mut round_robin = RoundRobin::new();
2216                // Size the running weights now, off the datapath, so no
2217                // selection allocates them.
2218                round_robin.rebuild(&self.backends);
2219                self.load_balancing = Box::new(round_robin)
2220            }
2221            LoadBalancingAlgorithms::Random => {
2222                self.load_balancing = Box::new(Random::with_seed(seed))
2223            }
2224            LoadBalancingAlgorithms::LeastLoaded => {
2225                self.load_balancing = Box::new(LeastLoaded {
2226                    metric: metric.unwrap_or(LoadMetric::Connections),
2227                })
2228            }
2229            LoadBalancingAlgorithms::PowerOfTwo => {
2230                self.load_balancing = Box::new(PowerOfTwo::with_seed(
2231                    seed,
2232                    metric.unwrap_or(LoadMetric::Connections),
2233                ))
2234            }
2235            // Affinity policies. They consult the client key every datapath
2236            // derives (UDP affinity key; HTTP/HTTPS/TCP source IP, header or
2237            // cookie); with `None` they fall back to round-robin.
2238            LoadBalancingAlgorithms::Hrw => {
2239                let mut rendezvous = Rendezvous::new();
2240                // Size the round-robin fallback's running weights now.
2241                rendezvous.rebuild(&self.backends);
2242                self.load_balancing = Box::new(rendezvous)
2243            }
2244            LoadBalancingAlgorithms::Maglev => {
2245                let mut maglev = Maglev::new();
2246                // Seed the lookup table from the currently-known backends so
2247                // selection is correct before the first control-plane rebuild,
2248                // and size the round-robin fallback's running weights.
2249                LoadBalancingAlgorithm::rebuild(&mut maglev, &self.backends);
2250                self.load_balancing = Box::new(maglev);
2251            }
2252        }
2253    }
2254}
2255
2256/// Count what shuffle sharding did to one selection of `cluster_id`: a
2257/// spill-over that found a backend outside the shard, or a strict refusal.
2258/// A selection that stayed in its shard, or was not sharded, counts nothing.
2259fn record_shard_outcome(cluster_id: &str, outcome: ShardOutcome, picked: bool) {
2260    match outcome {
2261        ShardOutcome::SpilledOver if picked => {
2262            incr!(names::backend::SHARD_SPILLOVER, Some(cluster_id), None);
2263        }
2264        ShardOutcome::Exhausted => {
2265            debug_assert!(!picked, "a strict exhausted shard selects nothing");
2266            incr!(names::backend::SHARD_EXHAUSTED, Some(cluster_id), None);
2267        }
2268        ShardOutcome::Unsharded | ShardOutcome::InShard | ShardOutcome::SpilledOver => {}
2269    }
2270}
2271
2272/// Whether `backend` is a candidate of the primary (`backup == false`) or the
2273/// backup tier: it belongs to that tier and can take a connection at `now`.
2274fn is_tier_candidate(backend: &Backend, backup: bool, now: Instant) -> bool {
2275    backend.backup == backup && backend.can_open(now)
2276}
2277
2278#[cfg(test)]
2279mod backends_test {
2280
2281    use std::{net::TcpListener, sync::mpsc::*, thread};
2282
2283    use super::*;
2284
2285    /// Start a TCP server that accepts connections, and return its address.
2286    ///
2287    /// The kernel picks the port (`:0`): a fixed port collides with any other
2288    /// process bound to it, including a concurrent `cargo test` of this crate,
2289    /// and fails the test with `AddrInUse` before it asserts anything.
2290    fn run_mock_tcp_server(stopper: Receiver<()>) -> SocketAddr {
2291        let mut run = true;
2292        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
2293        let addr = listener.local_addr().unwrap();
2294
2295        thread::spawn(move || {
2296            while run {
2297                for _stream in listener.incoming() {
2298                    // accept connections
2299                    if let Ok(()) = stopper.try_recv() {
2300                        run = false;
2301                    }
2302                }
2303            }
2304        });
2305        addr
2306    }
2307
2308    #[test]
2309    fn it_should_retrieve_a_backend_from_cluster_id_when_backends_have_been_recorded() {
2310        let mut backend_map = BackendMap::new();
2311        let cluster_id = "mycluster";
2312
2313        let (sender, receiver) = channel();
2314        let backend_addr = run_mock_tcp_server(receiver);
2315
2316        backend_map.add_backend(
2317            cluster_id,
2318            Backend::new(&format!("{cluster_id}-1"), backend_addr, None, None, None),
2319        );
2320
2321        assert!(
2322            backend_map
2323                .backend_from_cluster_id(cluster_id, None, Instant::now())
2324                .is_ok()
2325        );
2326        sender.send(()).unwrap();
2327    }
2328
2329    /// A session keeps the `Rc` of a backend it dialled after the map drops
2330    /// it; the mux pool reads that handle's status to refuse reusing its
2331    /// connection, so removal must leave it `Closing`.
2332    #[test]
2333    fn removing_a_backend_or_its_cluster_marks_it_closing() {
2334        let mut backend_map = BackendMap::new();
2335        let first: SocketAddr = "127.0.0.1:9001".parse().unwrap();
2336        let second: SocketAddr = "127.0.0.1:9002".parse().unwrap();
2337        backend_map.add_backend("foo", Backend::new("foo-1", first, None, None, None));
2338        backend_map.add_backend("foo", Backend::new("foo-2", second, None, None, None));
2339        let held: Vec<Rc<RefCell<Backend>>> = backend_map.backends["foo"].backends.clone();
2340
2341        assert!(backend_map.remove_backend("foo", "foo-1", &first));
2342        assert_eq!(held[0].borrow().status, BackendStatus::Closing);
2343        assert_eq!(held[1].borrow().status, BackendStatus::Normal);
2344
2345        backend_map.remove_cluster("foo");
2346        assert_eq!(held[1].borrow().status, BackendStatus::Closing);
2347
2348        // The same address added again is a new backend, not the retired one.
2349        backend_map.add_backend("foo", Backend::new("foo-2", second, None, None, None));
2350        let readded = &backend_map.backends["foo"].backends[0];
2351        assert!(!Rc::ptr_eq(readded, &held[1]));
2352        assert_eq!(readded.borrow().status, BackendStatus::Normal);
2353    }
2354
2355    /// `AddBackend` admits two ids at one address; removing one of them must
2356    /// keep the other routable (#1821).
2357    #[test]
2358    fn removing_one_backend_id_keeps_its_same_address_sibling() {
2359        let mut backend_map = BackendMap::new();
2360        let shared: SocketAddr = "127.0.0.1:9003".parse().unwrap();
2361        backend_map.add_backend("foo", Backend::new("foo-a", shared, None, None, None));
2362        backend_map.add_backend("foo", Backend::new("foo-b", shared, None, None, None));
2363        let held: Vec<Rc<RefCell<Backend>>> = backend_map.backends["foo"].backends.clone();
2364
2365        assert!(backend_map.remove_backend("foo", "foo-a", &shared));
2366        assert_eq!(held[0].borrow().status, BackendStatus::Closing);
2367        assert_eq!(held[1].borrow().status, BackendStatus::Normal);
2368        let list = &backend_map.backends["foo"];
2369        assert_eq!(list.backends.len(), 1);
2370        assert!(Rc::ptr_eq(
2371            list.find_backend_by_identity("foo-b", &shared).unwrap(),
2372            &held[1]
2373        ));
2374        assert!(list.find_backend_by_identity("foo-a", &shared).is_none());
2375
2376        // Removing it again, or an id never added, removes nothing.
2377        assert!(!backend_map.remove_backend("foo", "foo-a", &shared));
2378        assert!(!backend_map.remove_backend("foo", "foo-c", &shared));
2379        assert_eq!(backend_map.backends["foo"].backends.len(), 1);
2380    }
2381
2382    #[test]
2383    fn it_should_not_retrieve_a_backend_from_cluster_id_when_backend_has_not_been_recorded() {
2384        let mut backend_map = BackendMap::new();
2385        let cluster_not_recorded = "not";
2386        backend_map.add_backend(
2387            "foo",
2388            Backend::new("foo-1", "127.0.0.1:9001".parse().unwrap(), None, None, None),
2389        );
2390
2391        assert!(
2392            backend_map
2393                .backend_from_cluster_id(cluster_not_recorded, None, Instant::now())
2394                .is_err()
2395        );
2396    }
2397
2398    #[test]
2399    fn it_should_not_retrieve_a_backend_from_cluster_id_when_backend_list_is_empty() {
2400        let mut backend_map = BackendMap::new();
2401
2402        assert!(
2403            backend_map
2404                .backend_from_cluster_id("dumb", None, Instant::now())
2405                .is_err()
2406        );
2407    }
2408
2409    #[test]
2410    fn it_should_retrieve_a_backend_from_sticky_session_when_the_backend_has_been_recorded() {
2411        let mut backend_map = BackendMap::new();
2412        let cluster_id = "mycluster";
2413        let sticky_session = "server-2";
2414
2415        for (index, port) in [(1u16, 9001u16), (2, 9000), (3, 9002)] {
2416            backend_map.add_backend(
2417                cluster_id,
2418                Backend::new(
2419                    &format!("{cluster_id}-{index}"),
2420                    SocketAddr::from(([127, 0, 0, 1], port)),
2421                    Some(format!("server-{index}")),
2422                    None,
2423                    None,
2424                ),
2425            );
2426        }
2427
2428        let backend = backend_map
2429            .reserve_sticky_backend(cluster_id, sticky_session, None, Instant::now())
2430            .expect("the backend the sticky session names can take a connection");
2431        let backend = backend.borrow();
2432        assert_eq!(
2433            backend.backend_id, "mycluster-2",
2434            "a sticky session must reach the backend it names"
2435        );
2436        assert_eq!(
2437            backend.active_connections, 1,
2438            "reaching it reserves a connection"
2439        );
2440    }
2441
2442    #[test]
2443    fn it_should_not_retrieve_a_backend_from_sticky_session_when_the_backend_has_not_been_recorded()
2444    {
2445        let mut backend_map = BackendMap::new();
2446        let cluster_id = "mycluster";
2447        let sticky_session = "test";
2448
2449        assert!(
2450            backend_map
2451                .reserve_sticky_backend(cluster_id, sticky_session, None, Instant::now())
2452                .is_err()
2453        );
2454    }
2455
2456    #[test]
2457    fn it_should_not_retrieve_a_backend_from_sticky_session_when_the_backend_list_is_empty() {
2458        let mut backend_map = BackendMap::new();
2459        let mycluster_not_recorded = "mycluster";
2460        let sticky_session = "test";
2461
2462        assert!(
2463            backend_map
2464                .reserve_sticky_backend(
2465                    mycluster_not_recorded,
2466                    sticky_session,
2467                    None,
2468                    Instant::now()
2469                )
2470                .is_err()
2471        );
2472    }
2473
2474    #[test]
2475    fn it_should_add_a_backend_when_he_doesnt_already_exist() {
2476        let backend_id = "myback";
2477        let mut backends_list = BackendList::new();
2478        backends_list.add_backend(Backend::new(
2479            backend_id,
2480            "127.0.0.1:80".parse().unwrap(),
2481            None,
2482            None,
2483            None,
2484        ));
2485
2486        assert_eq!(1, backends_list.backends.len());
2487    }
2488
2489    #[test]
2490    fn it_should_not_add_a_backend_when_he_already_exist() {
2491        let backend_id = "myback";
2492        let mut backends_list = BackendList::new();
2493        backends_list.add_backend(Backend::new(
2494            backend_id,
2495            "127.0.0.1:80".parse().unwrap(),
2496            None,
2497            None,
2498            None,
2499        ));
2500
2501        //same backend id
2502        backends_list.add_backend(Backend::new(
2503            backend_id,
2504            "127.0.0.1:80".parse().unwrap(),
2505            None,
2506            None,
2507            None,
2508        ));
2509
2510        assert_eq!(1, backends_list.backends.len());
2511    }
2512
2513    /// Build a backend addressed at 127.0.0.1:port and force it Unhealthy
2514    /// without going through the health-check loop.
2515    fn unhealthy_backend(id: &str, port: u16) -> Backend {
2516        let mut backend = Backend::new(
2517            id,
2518            format!("127.0.0.1:{port}").parse().unwrap(),
2519            None,
2520            None,
2521            None,
2522        );
2523        // Threshold = 1 transitions on the first failure.
2524        backend.health.record_failure(1);
2525        assert!(!backend.health.is_healthy());
2526        backend
2527    }
2528
2529    #[test]
2530    fn fail_open_picks_normal_backend_in_retry_policy_okay() {
2531        // All backends are unhealthy but their retry policy is fresh (OKAY),
2532        // so fail-open must select one. A fresh ExponentialBackoffPolicy
2533        // returns OKAY on `can_try()` until the first `fail()` arms a wait
2534        // window.
2535        let mut list = BackendList::new();
2536        list.add_backend(unhealthy_backend("b1", 9001));
2537        list.add_backend(unhealthy_backend("b2", 9002));
2538
2539        // Sanity: `available_backends` returns nothing (the regular path).
2540        assert!(list.available_backends(false, Instant::now()).is_empty());
2541        assert!(list.available_backends(true, Instant::now()).is_empty());
2542
2543        let picked = list.next_available_backend(Instant::now());
2544        assert!(
2545            picked.is_some(),
2546            "fail-open must pick a Normal+OKAY backend"
2547        );
2548        assert!(list.fail_open_warned, "regime entry must latch the warn!");
2549    }
2550
2551    #[test]
2552    fn fail_open_skips_backend_in_retry_backoff() {
2553        // Same shape as above, but each backend's retry policy is in the
2554        // WAIT window after a recorded failure. Fail-open must NOT pick any
2555        // of them — hammering a backend at line rate during its back-off
2556        // window is exactly what the back-off is protecting against — and
2557        // the regime-entry warn! must NOT latch (no log spam either).
2558        let mut list = BackendList::new();
2559        list.add_backend(unhealthy_backend("b1", 9011));
2560        list.add_backend(unhealthy_backend("b2", 9012));
2561        for backend_rc in &list.backends {
2562            backend_rc
2563                .borrow_mut()
2564                .retry_policy()
2565                .fail(Instant::now(), &mut rand::rng());
2566            assert_eq!(
2567                Some(retry::RetryAction::WAIT),
2568                backend_rc.borrow().retry_policy.can_try(Instant::now()),
2569                "test fixture must place retry policy in WAIT"
2570            );
2571        }
2572
2573        let picked = list.next_available_backend(Instant::now());
2574        assert!(
2575            picked.is_none(),
2576            "fail-open must skip backends whose retry policy is in WAIT"
2577        );
2578        assert!(
2579            !list.fail_open_warned,
2580            "no candidate backends, no regime entry"
2581        );
2582    }
2583
2584    #[test]
2585    fn fail_open_warn_latched() {
2586        // First call enters the regime → latch flips, warn! fires.
2587        // Second call stays in the regime → latch stays, no second warn!.
2588        // Recovering one backend → latch clears on the next routing call.
2589        let mut list = BackendList::new();
2590        list.add_backend(unhealthy_backend("b1", 9021));
2591        list.add_backend(unhealthy_backend("b2", 9022));
2592
2593        assert!(list.next_available_backend(Instant::now()).is_some());
2594        assert!(list.fail_open_warned, "first fail-open must latch");
2595
2596        assert!(list.next_available_backend(Instant::now()).is_some());
2597        assert!(
2598            list.fail_open_warned,
2599            "subsequent fail-open routing keeps the latch"
2600        );
2601
2602        // Heal one backend — the next routing call takes the healthy path
2603        // and must clear the latch (regime exit logged once).
2604        list.backends[0].borrow_mut().health.status = HealthStatus::Healthy;
2605        let picked = list.next_available_backend(Instant::now());
2606        assert!(
2607            picked.is_some(),
2608            "regular path must select the healed backend"
2609        );
2610        assert!(
2611            !list.fail_open_warned,
2612            "regime exit must clear the latch so the next entry is logged again"
2613        );
2614    }
2615
2616    // ── #892: per-cluster availability tracker ──────────────────────────
2617
2618    /// Build a backend that passes the `evaluate_availability` predicate
2619    /// (`status == Normal && health.is_healthy() && !retry_policy.is_down()`).
2620    /// A `Backend::new` returns Normal/Healthy with a fresh
2621    /// ExponentialBackoffPolicy that reports `is_down() == false` until the
2622    /// first `fail()`, so this is just `Backend::new` with a stable address.
2623    fn healthy_backend(id: &str, port: u16) -> Backend {
2624        Backend::new(
2625            id,
2626            format!("127.0.0.1:{port}").parse().unwrap(),
2627            None,
2628            None,
2629            None,
2630        )
2631    }
2632
2633    #[test]
2634    fn is_available_requires_health_status_and_retry_policy() {
2635        // Fresh backend: Healthy + Normal + retry_policy fresh (OKAY).
2636        let mut backend = Backend::new("b", "127.0.0.1:9050".parse().unwrap(), None, None, None);
2637        assert!(backend.is_available(), "fresh backend must be available");
2638
2639        // Unhealthy fails the predicate even with everything else OK.
2640        backend.health.record_failure(1);
2641        assert!(!backend.is_available(), "unhealthy must not be available");
2642
2643        // Restore health, then drive retry policy into the exhausted-budget
2644        // state via the test-only helper. Calling `fail()` in a tight loop
2645        // would early-return on the second invocation because the
2646        // exponential-backoff window has not elapsed yet, so the natural
2647        // path needs real-time sleeps the unit test cannot afford.
2648        backend.health.status = HealthStatus::Healthy;
2649        assert!(backend.is_available());
2650        backend.retry_policy.force_down();
2651        assert!(
2652            backend.retry_policy.is_down(),
2653            "test setup: retry policy budget must be exhausted",
2654        );
2655        assert!(
2656            !backend.is_available(),
2657            "retry-policy backoff must fail the predicate"
2658        );
2659
2660        // Reset retry, switch lifecycle to Closing.
2661        backend.retry_policy.succeed(Instant::now());
2662        backend.set_closing();
2663        assert!(
2664            !backend.is_available(),
2665            "Closing lifecycle status must fail the predicate"
2666        );
2667    }
2668
2669    #[test]
2670    fn evaluate_availability_empty_list_returns_zero_zero() {
2671        let list = BackendList::new();
2672        assert_eq!((0, 0), list.evaluate_availability());
2673    }
2674
2675    #[test]
2676    fn evaluate_availability_counts_only_healthy_normal_not_in_backoff() {
2677        let mut list = BackendList::new();
2678        list.add_backend(healthy_backend("b-ok-1", 9101));
2679        list.add_backend(healthy_backend("b-ok-2", 9102));
2680        list.add_backend(unhealthy_backend("b-bad", 9103));
2681        let (available, total) = list.evaluate_availability();
2682        assert_eq!(3, total, "every configured backend counts toward total");
2683        assert_eq!(
2684            2, available,
2685            "only the two healthy backends pass the predicate"
2686        );
2687    }
2688
2689    #[test]
2690    fn evaluate_availability_excludes_retry_policy_down() {
2691        let mut list = BackendList::new();
2692        list.add_backend(healthy_backend("b-fresh", 9111));
2693        list.add_backend(healthy_backend("b-fail", 9112));
2694        // Drive the second backend's retry policy into is_down() via the
2695        // test-only force_down() helper. A natural exhaustion would
2696        // require waiting through the exponential-backoff windows
2697        // (`fail()` early-returns when called inside one).
2698        list.backends[1].borrow_mut().retry_policy.force_down();
2699        let (available, total) = list.evaluate_availability();
2700        assert_eq!(2, total);
2701        assert_eq!(
2702            1, available,
2703            "retry-policy is_down() backend must be excluded even when health.is_healthy()"
2704        );
2705    }
2706
2707    #[test]
2708    fn record_cluster_availability_flips_to_alldown_then_idempotent() {
2709        let mut map = BackendMap::new();
2710        let cluster_id = "c-flap";
2711        map.add_backend(cluster_id, unhealthy_backend("b1", 9201));
2712        // After add_backend the helper has run; total=1, available=0,
2713        // so the cell must already be AllDown.
2714        let list = map.backends.get(cluster_id).expect("cluster present");
2715        assert_eq!(
2716            ClusterAvailability::AllDown,
2717            list.availability.get(),
2718            "single unhealthy backend must drive the cell to AllDown"
2719        );
2720        // Calling again in the same regime is a no-op (Cell already AllDown).
2721        map.record_cluster_availability(cluster_id);
2722        let list = map.backends.get(cluster_id).expect("cluster present");
2723        assert_eq!(
2724            ClusterAvailability::AllDown,
2725            list.availability.get(),
2726            "repeat call must keep the cell at AllDown without flipping"
2727        );
2728    }
2729
2730    #[test]
2731    fn record_cluster_availability_recovers_to_available() {
2732        let mut map = BackendMap::new();
2733        let cluster_id = "c-recover";
2734        map.add_backend(cluster_id, unhealthy_backend("b1", 9301));
2735        assert_eq!(
2736            ClusterAvailability::AllDown,
2737            map.backends.get(cluster_id).unwrap().availability.get()
2738        );
2739        // Heal the backend in place and re-evaluate. Without going through
2740        // a routing call the helper still fires from add_backend / HC, so
2741        // here we drive it manually.
2742        map.backends.get_mut(cluster_id).unwrap().backends[0]
2743            .borrow_mut()
2744            .health
2745            .status = HealthStatus::Healthy;
2746        map.record_cluster_availability(cluster_id);
2747        assert_eq!(
2748            ClusterAvailability::Available,
2749            map.backends.get(cluster_id).unwrap().availability.get(),
2750            "healed backend must flip the cell back to Available"
2751        );
2752    }
2753
2754    #[test]
2755    fn record_cluster_availability_empty_cluster_stays_available() {
2756        let mut map = BackendMap::new();
2757        let cluster_id = "c-empty";
2758        map.backends.insert(cluster_id.into(), BackendList::new());
2759        // total == 0 path: never report AllDown — avoids log spam during
2760        // cluster bootstrap when backends are still being registered.
2761        map.record_cluster_availability(cluster_id);
2762        assert_eq!(
2763            ClusterAvailability::Available,
2764            map.backends.get(cluster_id).unwrap().availability.get(),
2765            "empty cluster must keep the cell at the default Available"
2766        );
2767    }
2768
2769    #[test]
2770    fn record_cluster_availability_missing_cluster_is_noop() {
2771        let map = BackendMap::new();
2772        // No panic, no insert — just an early return.
2773        map.record_cluster_availability("c-absent");
2774        assert!(
2775            !map.backends.contains_key("c-absent"),
2776            "helper must not insert a BackendList for an unknown cluster_id"
2777        );
2778    }
2779
2780    #[test]
2781    fn import_configuration_state_latches_cluster_rollup_gauges() {
2782        use crate::metrics::METRICS;
2783        use sozu_command_lib::proto::command::QueryMetricsOptions;
2784        // Unique cluster id so the assertion is not perturbed by gauges
2785        // left in the thread-local METRICS aggregator by sibling tests.
2786        let cluster_id = "c-import-rollup-9701";
2787        let mut map = BackendMap::new();
2788        let mut input = HashMap::new();
2789        input.insert(
2790            cluster_id.into(),
2791            vec![sozu_command_lib::response::Backend {
2792                cluster_id: cluster_id.into(),
2793                backend_id: "b1".to_owned(),
2794                address: "127.0.0.1:9701".parse().unwrap(),
2795                sticky_id: None,
2796                load_balancing_parameters: None,
2797                backup: None,
2798                unix_path: None,
2799            }],
2800        );
2801        map.import_configuration_state(&input);
2802        let response = METRICS
2803            .with(|m| {
2804                m.borrow_mut().query(&QueryMetricsOptions {
2805                    metric_names: vec![
2806                        names::cluster::AVAILABLE_BACKENDS.to_owned(),
2807                        names::cluster::TOTAL_BACKENDS.to_owned(),
2808                    ],
2809                    cluster_ids: vec![cluster_id.to_owned()],
2810                    backend_ids: vec![],
2811                    list: false,
2812                    no_clusters: false,
2813                    workers: false,
2814                })
2815            })
2816            .expect("metrics query succeeds");
2817        let cluster_metrics = match response.content_type {
2818            Some(
2819                sozu_command_lib::proto::command::response_content::ContentType::WorkerMetrics(wm),
2820            ) => wm,
2821            Some(_) => panic!("expected WorkerMetrics, got another response variant"),
2822            None => panic!("expected WorkerMetrics, got no response content"),
2823        };
2824        let cm = cluster_metrics
2825            .clusters
2826            .get(cluster_id)
2827            .expect("imported cluster must have a ClusterMetrics entry");
2828        // Without the import-time `record_cluster_availability` call the
2829        // two rollup gauges would be absent here. The fix guarantees the
2830        // pair lands without waiting for any follow-up backend mutation.
2831        assert!(
2832            cm.cluster.contains_key(names::cluster::AVAILABLE_BACKENDS),
2833            "cluster.available_backends gauge must be latched at import time"
2834        );
2835        assert!(
2836            cm.cluster.contains_key(names::cluster::TOTAL_BACKENDS),
2837            "cluster.total_backends gauge must be latched at import time"
2838        );
2839    }
2840
2841    #[test]
2842    fn set_health_check_config_none_re_emits_rollup_after_reset() {
2843        let mut map = BackendMap::new();
2844        let cluster_id = "c-hc-reset";
2845        // Seed the cluster with an unhealthy backend so `add_backend`
2846        // drives the `availability` cell to AllDown.
2847        map.add_backend(cluster_id, unhealthy_backend("b1", 9801));
2848        assert_eq!(
2849            ClusterAvailability::AllDown,
2850            map.backends.get(cluster_id).unwrap().availability.get(),
2851            "test setup: unhealthy backend must register the cell at AllDown"
2852        );
2853        // Disabling the health check resets backend health to the default
2854        // pristine state AND must re-emit the rollup so the cell reflects
2855        // the post-reset availability instead of the stale AllDown.
2856        map.set_health_check_config(cluster_id, None);
2857        assert_eq!(
2858            ClusterAvailability::Available,
2859            map.backends.get(cluster_id).unwrap().availability.get(),
2860            "set_health_check_config(None) must re-emit the rollup after \
2861             resetting backend health, otherwise dashboards stay stuck at AllDown"
2862        );
2863    }
2864
2865    // ----- protected health-check headers survive only an exact metadata match -----
2866
2867    /// A configuration whose headers name `names` with the fingerprints of
2868    /// `held`, and the matching held set, all issued by one key.
2869    fn held_with_config(
2870        key: &sozu_command::secret::FingerprintKey,
2871        values: &[(&str, &[u8])],
2872    ) -> (HealthCheckConfig, ProtectedHeaderSet) {
2873        let headers: Vec<_> = values
2874            .iter()
2875            .map(|(name, value)| {
2876                let mut carrier = sozu_command::proto::command::HealthCheckHeaderValue {
2877                    name: (*name).to_owned(),
2878                    value: value.to_vec(),
2879                };
2880                let secret = sozu_command::secret::SecretValue::take_from_ingress(&mut carrier)
2881                    .expect("a valid value");
2882                sozu_command::secret::ProtectedHeader::new(key, (*name).to_owned(), secret)
2883            })
2884            .collect();
2885        let config = HealthCheckConfig {
2886            uri: "/".to_owned(),
2887            interval: 10,
2888            timeout: 5,
2889            healthy_threshold: 3,
2890            unhealthy_threshold: 3,
2891            headers: headers
2892                .iter()
2893                .map(|h| sozu_command::proto::command::HealthCheckHeader {
2894                    name: h.name().to_owned(),
2895                    fingerprint: h.fingerprint().to_string(),
2896                })
2897                .collect(),
2898            ..Default::default()
2899        };
2900        (config, ProtectedHeaderSet::new(headers))
2901    }
2902
2903    fn test_key(byte: u8, epoch: u8) -> sozu_command::secret::FingerprintKey {
2904        sozu_command::secret::FingerprintKey::from_parts(
2905            sozu_command::secret::Zeroizing::new([byte; 32]),
2906            [epoch; 16],
2907        )
2908    }
2909
2910    fn map_holding(config: &HealthCheckConfig, held: &ProtectedHeaderSet) -> BackendMap {
2911        let mut map = BackendMap::new();
2912        map.set_protected_health_check("app", config.to_owned(), held.to_owned());
2913        map
2914    }
2915
2916    #[test]
2917    fn a_config_update_with_identical_header_metadata_keeps_the_values() {
2918        let key = test_key(1, 2);
2919        let (config, held) =
2920            held_with_config(&key, &[("Authorization", b"s1"), ("X-Token", b"s2")]);
2921        let mut map = map_holding(&config, &held);
2922
2923        // Threshold and interval changed, headers metadata untouched.
2924        let mut updated = config.to_owned();
2925        updated.unhealthy_threshold = 7;
2926        updated.interval = 20;
2927        map.set_health_check_config("app", Some(updated.to_owned()));
2928
2929        assert_eq!(map.health_check_configs["app"], updated);
2930        assert!(
2931            map.protected_health_check_headers["app"].ptr_eq(&held),
2932            "the very same set stays, bytes untouched"
2933        );
2934    }
2935
2936    #[test]
2937    fn a_config_update_that_changes_header_metadata_drops_the_values() {
2938        let key = test_key(1, 2);
2939        let (config, held) =
2940            held_with_config(&key, &[("Authorization", b"s1"), ("X-Token", b"s2")]);
2941
2942        let mut cases: Vec<(&str, Vec<sozu_command::proto::command::HealthCheckHeader>)> =
2943            Vec::new();
2944        // The headers are cleared.
2945        cases.push(("clear", Vec::new()));
2946        // Same names, order swapped.
2947        let mut swapped = config.headers.to_owned();
2948        swapped.reverse();
2949        cases.push(("order", swapped));
2950        // Names alone match: one fingerprint belongs to a different value.
2951        let (other, _) = held_with_config(&key, &[("Authorization", b"other"), ("X-Token", b"s2")]);
2952        cases.push(("fingerprint", other.headers));
2953        // A fingerprint issued under a foreign key epoch.
2954        let (foreign, _) = held_with_config(
2955            &test_key(1, 9),
2956            &[("Authorization", b"s1"), ("X-Token", b"s2")],
2957        );
2958        cases.push(("foreign-epoch", foreign.headers));
2959        // A fingerprint that does not parse, an empty one, and a shorter list.
2960        let mut malformed = config.headers.to_owned();
2961        malformed[0].fingerprint = "not-a-fingerprint".to_owned();
2962        cases.push(("malformed", malformed));
2963        let mut empty = config.headers.to_owned();
2964        empty[1].fingerprint = String::new();
2965        cases.push(("empty-fingerprint", empty));
2966        cases.push(("count", config.headers[..1].to_vec()));
2967
2968        for (label, headers) in cases {
2969            let mut map = map_holding(&config, &held);
2970            let mut updated = config.to_owned();
2971            updated.headers = headers;
2972            map.set_health_check_config("app", Some(updated.to_owned()));
2973            assert_eq!(map.health_check_configs["app"], updated, "{label}");
2974            assert!(
2975                !map.protected_health_check_headers.contains_key("app"),
2976                "{label}: the values are dropped, never kept by names alone"
2977            );
2978        }
2979    }
2980
2981    #[test]
2982    fn a_config_update_without_a_held_set_stays_missing() {
2983        let key = test_key(1, 2);
2984        let (config, _) = held_with_config(&key, &[("Authorization", b"s1")]);
2985        let mut map = BackendMap::new();
2986        map.set_health_check_config("app", Some(config.to_owned()));
2987        assert_eq!(map.health_check_configs["app"], config);
2988        assert!(!map.protected_health_check_headers.contains_key("app"));
2989    }
2990
2991    #[test]
2992    fn removing_the_health_check_drops_the_values() {
2993        let key = test_key(1, 2);
2994        let (config, held) = held_with_config(&key, &[("Authorization", b"s1")]);
2995        let mut map = map_holding(&config, &held);
2996        map.set_health_check_config("app", None);
2997        assert!(!map.health_check_configs.contains_key("app"));
2998        assert!(!map.protected_health_check_headers.contains_key("app"));
2999    }
3000
3001    // ----- #1684: selection reads neither the wall clock nor an ambient RNG -----
3002
3003    /// Four backends of `cluster_id` under `policy`, created at `now` in a map
3004    /// that seeded the policy, and the addresses of `draws` selections at `now`.
3005    fn draws(
3006        map: &mut BackendMap,
3007        cluster_id: &str,
3008        policy: LoadBalancingAlgorithms,
3009        now: Instant,
3010        draws: usize,
3011    ) -> Vec<SocketAddr> {
3012        map.set_load_balancing_policy_for_cluster(
3013            cluster_id,
3014            policy,
3015            Some(LoadMetric::Connections),
3016        );
3017        for index in 0..4u16 {
3018            map.add_backend(
3019                cluster_id,
3020                Backend::new_at(
3021                    &format!("{cluster_id}-{index}"),
3022                    SocketAddr::from(([127, 0, 0, 1], 9100 + index)),
3023                    None,
3024                    None,
3025                    None,
3026                    now,
3027                ),
3028            );
3029        }
3030        let list = map
3031            .backends
3032            .get_mut(cluster_id)
3033            .expect("the policy call created the list");
3034        (0..draws)
3035            .map(|_| {
3036                list.next_available_backend(now)
3037                    .expect("four fresh backends can all open")
3038                    .borrow()
3039                    .address
3040            })
3041            .collect()
3042    }
3043
3044    /// #1684: a map built from a seed seeds every policy it creates from it,
3045    /// so one seed yields one sequence of picks.
3046    ///
3047    /// TO SEE THIS RED: build `BackendMap::with_seed` from `os_rng()` and
3048    /// ignore the seed; two maps of one seed then disagree.
3049    #[test]
3050    fn one_map_seed_yields_one_selection_sequence() {
3051        let now = Instant::now();
3052        for policy in [
3053            LoadBalancingAlgorithms::Random,
3054            LoadBalancingAlgorithms::PowerOfTwo,
3055        ] {
3056            let first = draws(&mut BackendMap::with_seed(7), "c", policy, now, 48);
3057            assert_eq!(
3058                first,
3059                draws(&mut BackendMap::with_seed(7), "c", policy, now, 48),
3060                "{policy:?}: one seed must yield one sequence of picks"
3061            );
3062            assert_ne!(
3063                first,
3064                draws(&mut BackendMap::with_seed(8), "c", policy, now, 48),
3065                "{policy:?}: the picks must come from the map's seed, so another seed differs"
3066            );
3067        }
3068    }
3069
3070    /// The addresses of `draws` selections at `now` from `cluster_id`.
3071    fn picks_of(
3072        map: &mut BackendMap,
3073        cluster_id: &str,
3074        now: Instant,
3075        draws: usize,
3076    ) -> Vec<SocketAddr> {
3077        let list = map
3078            .backends
3079            .get_mut(cluster_id)
3080            .expect("the cluster was created");
3081        (0..draws)
3082            .map(|_| {
3083                list.next_available_backend(now)
3084                    .expect("fresh backends can all open")
3085                    .borrow()
3086                    .address
3087            })
3088            .collect()
3089    }
3090
3091    /// #1684: a cluster the map creates while adding a backend, before any
3092    /// policy is set, is seeded from the map too, not from the OS.
3093    ///
3094    /// TO SEE THIS RED: create the list in `BackendMap::add_backend` with
3095    /// `or_default()`, which seeds its `Random` policy from the OS.
3096    #[test]
3097    fn a_cluster_created_by_adding_a_backend_is_seeded_by_the_map() {
3098        let now = Instant::now();
3099        let sequence = || {
3100            let mut map = BackendMap::with_seed(7);
3101            for index in 0..4u16 {
3102                map.add_backend(
3103                    "c",
3104                    Backend::new_at(
3105                        &format!("c-{index}"),
3106                        SocketAddr::from(([127, 0, 0, 1], 9300 + index)),
3107                        None,
3108                        None,
3109                        None,
3110                        now,
3111                    ),
3112                );
3113            }
3114            picks_of(&mut map, "c", now, 48)
3115        };
3116        assert_eq!(
3117            sequence(),
3118            sequence(),
3119            "a list created by add_backend must draw its seed from the map"
3120        );
3121    }
3122
3123    /// #1684: importing a configuration state seeds each cluster from the
3124    /// map in cluster-id order, so one map seed gives every cluster the same
3125    /// seed whatever order the state's `HashMap` iterates in.
3126    ///
3127    /// TO SEE THIS RED: draw the seeds while iterating `backends` directly in
3128    /// `BackendMap::import_configuration_state`; each `HashMap` has its own
3129    /// `RandomState`, so two imports hand the seeds out in different orders.
3130    #[test]
3131    fn importing_a_state_seeds_its_clusters_in_a_fixed_order() {
3132        const CLUSTERS: u16 = 16;
3133        let now = Instant::now();
3134        let state = |ids: &mut dyn Iterator<Item = u16>| {
3135            let mut state: HashMap<ClusterId, Vec<sozu_command_lib::response::Backend>> =
3136                HashMap::new();
3137            for id in ids {
3138                let cluster_id: ClusterId = format!("import-{id:02}").into();
3139                let backends = (0..4u16)
3140                    .map(|index| sozu_command_lib::response::Backend {
3141                        cluster_id: cluster_id.clone(),
3142                        backend_id: format!("{cluster_id}-{index}"),
3143                        address: SocketAddr::from(([127, 0, 0, 1], 9400 + 4 * id + index)),
3144                        sticky_id: None,
3145                        load_balancing_parameters: None,
3146                        backup: None,
3147                        unix_path: None,
3148                    })
3149                    .collect();
3150                state.insert(cluster_id, backends);
3151            }
3152            state
3153        };
3154        let forward = state(&mut (0..CLUSTERS));
3155        let backward = state(&mut (0..CLUSTERS).rev());
3156
3157        let mut first = BackendMap::with_seed(7);
3158        first.import_configuration_state(&forward);
3159        let mut second = BackendMap::with_seed(7);
3160        second.import_configuration_state(&backward);
3161        for id in 0..CLUSTERS {
3162            let cluster_id = format!("import-{id:02}");
3163            assert_eq!(
3164                picks_of(&mut first, &cluster_id, now, 32),
3165                picks_of(&mut second, &cluster_id, now, 32),
3166                "{cluster_id}: one map seed must give each imported cluster one seed"
3167            );
3168        }
3169    }
3170
3171    /// Regression guard moved from `load_balancing.rs`, where it held
3172    /// `Random::new()`: the production seed must never be a shared constant.
3173    /// If it were, every worker, on every cold start, would draw from a
3174    /// bit-for-bit identical keystream and pick the same backend at every
3175    /// step — the correlated-load event uniform selection exists to prevent,
3176    /// arriving at a synchronised redeploy. The OS seed now enters through
3177    /// `BackendMap::new`, which seeds every cluster's policy, so two maps (two
3178    /// workers) and two clusters of one map must each diverge.
3179    ///
3180    /// 48 draws over four backends collide with probability ~4^-48 if the
3181    /// keystreams are independent, so this cannot flake.
3182    #[test]
3183    fn random_policies_seeded_by_the_os_are_not_correlated() {
3184        let now = Instant::now();
3185        let policy = LoadBalancingAlgorithms::Random;
3186        let mut map = BackendMap::new();
3187        let first = draws(&mut map, "a", policy, now, 48);
3188        assert_ne!(
3189            first,
3190            draws(&mut BackendMap::new(), "a", policy, now, 48),
3191            "two BackendMap::new() must NOT seed Random from a shared/correlated keystream"
3192        );
3193        assert_ne!(
3194            first,
3195            draws(&mut map, "b", policy, now, 48),
3196            "two clusters of one map must NOT draw from a shared keystream"
3197        );
3198    }
3199
3200    /// The same guard for `PowerOfTwo`'s tie-break, moved from
3201    /// `load_balancing.rs`. Every backend starts at zero load, so a tie (and
3202    /// its coin flip) is the common case: a shared seed would resolve every
3203    /// worker's first ties identically after a synchronised redeploy,
3204    /// reintroducing the herding P2C exists to prevent.
3205    #[test]
3206    fn power_of_two_policies_seeded_by_the_os_are_not_correlated() {
3207        let now = Instant::now();
3208        let policy = LoadBalancingAlgorithms::PowerOfTwo;
3209        let mut map = BackendMap::new();
3210        let first = draws(&mut map, "a", policy, now, 48);
3211        assert_ne!(
3212            first,
3213            draws(&mut BackendMap::new(), "a", policy, now, 48),
3214            "two BackendMap::new() must NOT seed PowerOfTwo from a shared/correlated keystream"
3215        );
3216        assert_ne!(
3217            first,
3218            draws(&mut map, "b", policy, now, 48),
3219            "two clusters of one map must NOT draw from a shared keystream"
3220        );
3221    }
3222
3223    /// #1684: the connection-time average decays on the instants it is
3224    /// given. A backend created at `t0` and read at `t0 + 1s` has aged by
3225    /// exactly one decay constant, `e^-1`, whatever the wall clock did.
3226    ///
3227    /// TO SEE THIS RED: take `now` from `Instant::now()` inside
3228    /// `PeakEWMA::observe`; almost no real time passes, so the average does
3229    /// not decay and the cost stays at its 50 ms default.
3230    #[test]
3231    fn the_connection_time_average_decays_on_the_injected_clock() {
3232        let t0 = Instant::now();
3233        let mut backend = Backend::new_at(
3234            "ewma",
3235            SocketAddr::from(([127, 0, 0, 1], 9200)),
3236            None,
3237            None,
3238            None,
3239            t0,
3240        );
3241        let default_rtt = backend.connection_time.rtt;
3242        let cost = backend.peak_ewma_connection(t0 + Duration::from_secs(1));
3243        assert_eq!(
3244            cost,
3245            default_rtt * (-1f64).exp(),
3246            "one second after creation, the average must have decayed by e^-1"
3247        );
3248
3249        // One sequence of instants, one average, bit for bit.
3250        let replay = |backend: &mut Backend| {
3251            backend.set_connection_time(Duration::from_millis(80), t0 + Duration::from_millis(10));
3252            backend.set_connection_time(Duration::from_millis(20), t0 + Duration::from_millis(900));
3253            backend.peak_ewma_connection(t0 + Duration::from_millis(2_500))
3254        };
3255        let mut twin = Backend::new_at(
3256            "ewma",
3257            SocketAddr::from(([127, 0, 0, 1], 9200)),
3258            None,
3259            None,
3260            None,
3261            t0,
3262        );
3263        let mut again = Backend::new_at(
3264            "ewma",
3265            SocketAddr::from(([127, 0, 0, 1], 9200)),
3266            None,
3267            None,
3268            None,
3269            t0,
3270        );
3271        assert_eq!(
3272            replay(&mut twin).to_bits(),
3273            replay(&mut again).to_bits(),
3274            "one sequence of instants must yield one average, bit for bit"
3275        );
3276    }
3277
3278    // ----- #524: shuffle sharding over the HRW ranking -----
3279
3280    /// `n` primary backends on distinct loopback ports, under `policy`, sharded
3281    /// by `sharding`.
3282    fn sharded_list(
3283        n: u16,
3284        policy: LoadBalancingAlgorithms,
3285        sharding: Option<ShuffleSharding>,
3286    ) -> BackendList {
3287        let now = Instant::now();
3288        let mut list = BackendList::with_seed(524);
3289        list.set_load_balancing_policy(policy, Some(LoadMetric::Connections), 524);
3290        for index in 0..n {
3291            list.add_backend(Backend::new_at(
3292                &format!("shard-{index}"),
3293                SocketAddr::from(([127, 0, 0, 1], 20_000 + index)),
3294                None,
3295                None,
3296                None,
3297                now,
3298            ));
3299        }
3300        list.set_shuffle_sharding(sharding);
3301        list
3302    }
3303
3304    fn sharding(percent: u32, min_backends: u32, mode: ShardMode) -> Option<ShuffleSharding> {
3305        Some(ShuffleSharding {
3306            percent,
3307            min_backends,
3308            mode,
3309        })
3310    }
3311
3312    /// The addresses of `key`'s shard in `list`, and its members ranked best
3313    /// first.
3314    fn shard_of(list: &mut BackendList, key: u64) -> Vec<SocketAddr> {
3315        assert!(list.compute_shard(Some(key)), "the list must shard");
3316        let mut ranked: Vec<(f64, SocketAddr)> = list
3317            .shard
3318            .iter()
3319            .map(|&index| {
3320                let backend = list.backends[index].borrow();
3321                (hrw_score(key, &backend), backend.address)
3322            })
3323            .collect();
3324        ranked.sort_by(|a, b| b.0.total_cmp(&a.0));
3325        ranked.into_iter().map(|(_, address)| address).collect()
3326    }
3327
3328    /// #524: a shard is the top `k` of the key's HRW ranking over every
3329    /// primary backend — the same backends whatever their order in the list,
3330    /// on every call.
3331    #[test]
3332    fn a_shard_is_the_top_of_the_hrw_ranking() {
3333        let unweighted_list = || {
3334            sharded_list(
3335                12,
3336                LoadBalancingAlgorithms::RoundRobin,
3337                sharding(25, 8, ShardMode::Fallback),
3338            )
3339        };
3340        let mut unweighted = unweighted_list();
3341        // The same backends with non-uniform weights: the ranking is the
3342        // weighted HRW score, so heavier backends rank higher more often.
3343        let mut weighted = unweighted_list();
3344        for (index, backend) in weighted.backends.iter().enumerate() {
3345            backend.borrow_mut().load_balancing_parameters = Some(LoadBalancingParams {
3346                weight: 10 + 90 * (index as i32 % 4),
3347            });
3348        }
3349        let mut weights_moved_a_shard = false;
3350        for list in [&mut unweighted, &mut weighted] {
3351            for key in 0..64u64 {
3352                let shard = shard_of(list, key);
3353                assert_eq!(shard.len(), 3, "25% of 12 backends is a shard of 3");
3354                assert_eq!(shard, shard_of(list, key), "a shard is deterministic");
3355                let mut everyone: Vec<(f64, SocketAddr)> = list
3356                    .backends
3357                    .iter()
3358                    .map(|backend| {
3359                        let backend = backend.borrow();
3360                        (hrw_score(key, &backend), backend.address)
3361                    })
3362                    .collect();
3363                everyone.sort_by(|a, b| b.0.total_cmp(&a.0));
3364                let top: Vec<SocketAddr> = everyone.iter().take(3).map(|(_, a)| *a).collect();
3365                assert_eq!(shard, top, "key {key}: the shard is the HRW top 3");
3366            }
3367        }
3368        for key in 0..64u64 {
3369            weights_moved_a_shard |= shard_of(&mut unweighted, key) != shard_of(&mut weighted, key);
3370        }
3371        assert!(
3372            weights_moved_a_shard,
3373            "the weighted ranking must differ from the unweighted one for some key"
3374        );
3375    }
3376
3377    /// #524: a change of N moves a client only at the tail of its ranking.
3378    /// Adding a backend with k unchanged replaces at most the lowest-ranked
3379    /// member; adding one that makes k grow keeps every member; removing a
3380    /// non-member changes nothing, and removing a member keeps the others.
3381    ///
3382    /// TO SEE THIS RED: rank by list position instead of HRW score in
3383    /// `BackendList::compute_shard`; adding a backend then reshuffles shards
3384    /// that should not move.
3385    #[test]
3386    fn a_shard_changes_only_at_its_tail_when_backends_change() {
3387        let now = Instant::now();
3388        let extra = |port: u16| {
3389            Backend::new_at(
3390                &format!("extra-{port}"),
3391                SocketAddr::from(([127, 0, 0, 1], port)),
3392                None,
3393                None,
3394                None,
3395                now,
3396            )
3397        };
3398        for key in 0..128u64 {
3399            // 10% of 10 and of 11 both give k = max(2, ceil(1.0 | 1.1)) = 2,
3400            // so k stays 2 across the add.
3401            let mut list = sharded_list(
3402                10,
3403                LoadBalancingAlgorithms::RoundRobin,
3404                sharding(10, 8, ShardMode::Fallback),
3405            );
3406            let before = shard_of(&mut list, key);
3407            list.add_backend(extra(21_000));
3408            let after = shard_of(&mut list, key);
3409            let lost: Vec<_> = before.iter().filter(|a| !after.contains(a)).collect();
3410            assert!(lost.len() <= 1, "key {key}: at most one member displaced");
3411            if let Some(lost) = lost.first() {
3412                assert_eq!(
3413                    Some(*lost),
3414                    before.last(),
3415                    "key {key}: only the lowest-ranked member may be displaced"
3416                );
3417            }
3418
3419            // k grows with N: every member stays.
3420            let mut list = sharded_list(
3421                10,
3422                LoadBalancingAlgorithms::RoundRobin,
3423                sharding(20, 8, ShardMode::Fallback),
3424            );
3425            let before = shard_of(&mut list, key);
3426            list.add_backend(extra(21_001));
3427            let after = shard_of(&mut list, key);
3428            assert_eq!((before.len(), after.len()), (2, 3));
3429            assert!(
3430                before.iter().all(|a| after.contains(a)),
3431                "key {key}: a growing shard keeps its members"
3432            );
3433
3434            // Removing a non-member changes nothing; removing a member keeps
3435            // the others.
3436            let mut list = sharded_list(
3437                10,
3438                LoadBalancingAlgorithms::RoundRobin,
3439                sharding(20, 8, ShardMode::Fallback),
3440            );
3441            let before = shard_of(&mut list, key);
3442            let outsider = list
3443                .backends
3444                .iter()
3445                .map(|b| b.borrow().address)
3446                .find(|a| !before.contains(a))
3447                .expect("a shard of 2 among 10 leaves outsiders");
3448            // `sharded_list` names each backend after its port.
3449            let id_of = |address: &SocketAddr| format!("shard-{}", address.port() - 20_000);
3450            assert!(list.remove_backend(&id_of(&outsider), &outsider));
3451            assert_eq!(shard_of(&mut list, key), before, "key {key}");
3452            assert!(list.remove_backend(&id_of(&before[0]), &before[0]));
3453            assert!(
3454                shard_of(&mut list, key).contains(&before[1]),
3455                "key {key}: removing one member keeps the other"
3456            );
3457        }
3458    }
3459
3460    /// #524: every selection of a keyed client stays in its shard, under a
3461    /// policy that would otherwise visit every backend.
3462    #[test]
3463    fn a_sharded_selection_stays_in_the_shard() {
3464        let now = Instant::now();
3465        for policy in [
3466            LoadBalancingAlgorithms::RoundRobin,
3467            LoadBalancingAlgorithms::LeastLoaded,
3468            LoadBalancingAlgorithms::Random,
3469            LoadBalancingAlgorithms::Hrw,
3470        ] {
3471            let mut list = sharded_list(8, policy, sharding(25, 8, ShardMode::Fallback));
3472            for key in 0..32u64 {
3473                let shard = shard_of(&mut list, key);
3474                for _ in 0..8 {
3475                    let (picked, outcome) = list.select_with_key(Some(key), now);
3476                    let picked = picked.expect("a healthy shard selects").borrow().address;
3477                    assert_eq!(outcome, ShardOutcome::InShard);
3478                    assert!(
3479                        shard.contains(&picked),
3480                        "{policy:?} key {key}: {picked} is outside {shard:?}"
3481                    );
3482                }
3483            }
3484            // No key, or fewer primaries than the threshold: not sharded.
3485            assert_eq!(list.select_with_key(None, now).1, ShardOutcome::Unsharded);
3486            let mut small = sharded_list(7, policy, sharding(25, 8, ShardMode::Fallback));
3487            assert_eq!(
3488                small.select_with_key(Some(1), now).1,
3489                ShardOutcome::Unsharded
3490            );
3491        }
3492    }
3493
3494    /// Put every member of `key`'s shard out of selection, as a failed
3495    /// connection does: its retry policy enters its back-off window.
3496    fn fail_shard(list: &mut BackendList, key: u64) -> Vec<SocketAddr> {
3497        let shard = shard_of(list, key);
3498        let now = Instant::now();
3499        for backend in &list.backends {
3500            let mut backend = backend.borrow_mut();
3501            if shard.contains(&backend.address) {
3502                backend.retry_policy.fail(now, &mut rand::rng());
3503            }
3504        }
3505        shard
3506    }
3507
3508    /// #524: with its whole shard down, `STRICT` selects nothing although the
3509    /// rest of the cluster is healthy, and `FALLBACK` spills over to a backend
3510    /// outside the shard.
3511    ///
3512    /// TO SEE THIS RED: in `BackendList::select_with_key`, spill over in both
3513    /// modes (drop the `strict` early return); the strict selection then
3514    /// finds a backend outside the shard.
3515    #[test]
3516    fn an_exhausted_shard_refuses_in_strict_and_spills_in_fallback() {
3517        let now = Instant::now();
3518        for key in 0..32u64 {
3519            let mut strict = sharded_list(
3520                8,
3521                LoadBalancingAlgorithms::RoundRobin,
3522                sharding(25, 8, ShardMode::Strict),
3523            );
3524            fail_shard(&mut strict, key);
3525            let (picked, outcome) = strict.select_with_key(Some(key), now);
3526            assert!(picked.is_none(), "key {key}: strict never leaves the shard");
3527            assert_eq!(outcome, ShardOutcome::Exhausted);
3528
3529            let mut fallback = sharded_list(
3530                8,
3531                LoadBalancingAlgorithms::RoundRobin,
3532                sharding(25, 8, ShardMode::Fallback),
3533            );
3534            let shard = fail_shard(&mut fallback, key);
3535            let (picked, outcome) = fallback.select_with_key(Some(key), now);
3536            let picked = picked.expect("fallback spills over").borrow().address;
3537            assert_eq!(outcome, ShardOutcome::SpilledOver);
3538            assert!(
3539                !shard.contains(&picked),
3540                "key {key}: the spill leaves the shard"
3541            );
3542        }
3543    }
3544
3545    /// #524: a strict shard fails open inside itself only. When no backend of
3546    /// the whole cluster passes its health check (the fail-open regime), a
3547    /// strict selection still picks a shard member.
3548    #[test]
3549    fn a_strict_shard_fails_open_inside_itself() {
3550        let now = Instant::now();
3551        for key in 0..16u64 {
3552            let mut list = sharded_list(
3553                8,
3554                LoadBalancingAlgorithms::RoundRobin,
3555                sharding(25, 8, ShardMode::Strict),
3556            );
3557            let shard = shard_of(&mut list, key);
3558            for backend in &list.backends {
3559                backend.borrow_mut().health.status = HealthStatus::Unhealthy;
3560            }
3561            let (picked, outcome) = list.select_with_key(Some(key), now);
3562            let picked = picked.expect("fail-open still routes").borrow().address;
3563            assert!(
3564                shard.contains(&picked),
3565                "key {key}: fail-open stays in the shard"
3566            );
3567            assert_eq!(outcome, ShardOutcome::InShard);
3568        }
3569    }
3570
3571    /// Fail every member of `key`'s shard by health check, as a probe does:
3572    /// they stay `Normal` with an `OKAY` retry policy, so fail-open may still
3573    /// pick them.
3574    fn unhealthy_shard(list: &mut BackendList, key: u64) -> Vec<SocketAddr> {
3575        let shard = shard_of(list, key);
3576        for backend in &list.backends {
3577            let mut backend = backend.borrow_mut();
3578            if shard.contains(&backend.address) {
3579                backend.health.status = HealthStatus::Unhealthy;
3580            }
3581        }
3582        shard
3583    }
3584
3585    /// #524 review L3: `STRICT` refuses when its shard fails its health
3586    /// checks while primary backends outside it are healthy — the fail-open
3587    /// regime is not reached, because the cluster is not all down.
3588    ///
3589    /// TO SEE THIS RED: make the `primaries_can_open` guard of
3590    /// `BackendList::select_with_key` false; the strict shard then fails open
3591    /// and picks one of its unhealthy members.
3592    #[test]
3593    fn a_strict_shard_failing_health_checks_refuses_while_others_are_healthy() {
3594        let now = Instant::now();
3595        for key in 0..32u64 {
3596            let mut list = sharded_list(
3597                8,
3598                LoadBalancingAlgorithms::RoundRobin,
3599                sharding(25, 8, ShardMode::Strict),
3600            );
3601            unhealthy_shard(&mut list, key);
3602            let (picked, outcome) = list.select_with_key(Some(key), now);
3603            assert!(
3604                picked.is_none(),
3605                "key {key}: strict must refuse, not fail open"
3606            );
3607            assert_eq!(outcome, ShardOutcome::Exhausted);
3608        }
3609    }
3610
3611    /// #524 review M1: backups sit outside every shard, so a healthy backup
3612    /// must not decide a strict shard's fate. With every primary failing its
3613    /// health checks, a strict shard fails open inside itself whether or not
3614    /// a backup is healthy.
3615    ///
3616    /// TO SEE THIS RED: count the backup tier in `primaries_can_open`
3617    /// (`available != 0 || <any backup can open>`); the healthy backup then
3618    /// turns the fail-open into a refusal.
3619    #[test]
3620    fn a_healthy_backup_does_not_turn_a_strict_fail_open_into_a_refusal() {
3621        let now = Instant::now();
3622        for key in 0..32u64 {
3623            let mut list = sharded_list(
3624                8,
3625                LoadBalancingAlgorithms::RoundRobin,
3626                sharding(25, 8, ShardMode::Strict),
3627            );
3628            let shard = shard_of(&mut list, key);
3629            for backend in &list.backends {
3630                backend.borrow_mut().health.status = HealthStatus::Unhealthy;
3631            }
3632            list.add_backend(Backend::new_at(
3633                "healthy-backup",
3634                SocketAddr::from(([127, 0, 0, 1], 23_000)),
3635                None,
3636                None,
3637                Some(true),
3638                now,
3639            ));
3640            let (picked, outcome) = list.select_with_key(Some(key), now);
3641            let picked = picked
3642                .expect("a strict shard fails open inside itself")
3643                .borrow()
3644                .address;
3645            assert!(
3646                shard.contains(&picked),
3647                "key {key}: never the backup, never outside"
3648            );
3649            assert_eq!(outcome, ShardOutcome::InShard);
3650        }
3651    }
3652
3653    /// #524 review L2: in `FALLBACK`, fail-open over the whole cluster can
3654    /// pick a member of the exhausted shard; that pick stayed in the shard
3655    /// and must not count as a spill-over.
3656    ///
3657    /// TO SEE THIS RED: drop the `settle_spill_outcome` call after the
3658    /// fail-open pick in `BackendList::select_with_key`; an in-shard pick is
3659    /// then reported `SpilledOver`.
3660    #[test]
3661    fn a_fallback_fail_open_pick_inside_the_shard_is_not_a_spill_over() {
3662        let now = Instant::now();
3663        let key = 3;
3664        let mut list = sharded_list(
3665            8,
3666            LoadBalancingAlgorithms::RoundRobin,
3667            sharding(25, 8, ShardMode::Fallback),
3668        );
3669        let shard = shard_of(&mut list, key);
3670        for backend in &list.backends {
3671            backend.borrow_mut().health.status = HealthStatus::Unhealthy;
3672        }
3673        let (mut in_shard, mut outside) = (0, 0);
3674        // Round-robin over the fail-open set visits every backend in 8 picks.
3675        for _ in 0..8 {
3676            let (picked, outcome) = list.select_with_key(Some(key), now);
3677            let picked = picked.expect("fail-open routes").borrow().address;
3678            if shard.contains(&picked) {
3679                assert_eq!(outcome, ShardOutcome::InShard, "{picked} is a shard member");
3680                in_shard += 1;
3681            } else {
3682                assert_eq!(outcome, ShardOutcome::SpilledOver, "{picked} is outside");
3683                outside += 1;
3684            }
3685        }
3686        assert_eq!(
3687            (in_shard, outside),
3688            (2, 6),
3689            "round-robin visits every backend once"
3690        );
3691    }
3692
3693    /// #524: a sticky cookie naming a live backend wins, even outside the
3694    /// client's shard.
3695    #[test]
3696    fn a_sticky_backend_outside_the_shard_wins() {
3697        let now = Instant::now();
3698        let key = 7;
3699        let mut map = BackendMap::with_seed(524);
3700        map.set_load_balancing_policy_for_cluster(
3701            "sharded",
3702            LoadBalancingAlgorithms::RoundRobin,
3703            None,
3704        );
3705        for index in 0..8u16 {
3706            map.add_backend(
3707                "sharded",
3708                Backend::new_at(
3709                    &format!("shard-{index}"),
3710                    SocketAddr::from(([127, 0, 0, 1], 22_000 + index)),
3711                    Some(format!("sticky-{index}")),
3712                    None,
3713                    None,
3714                    now,
3715                ),
3716            );
3717        }
3718        map.set_shuffle_sharding_for_cluster("sharded", sharding(25, 8, ShardMode::Strict));
3719        let list = map.backends.get_mut("sharded").expect("the cluster exists");
3720        let shard = shard_of(list, key);
3721        let outsider = list
3722            .backends
3723            .iter()
3724            .find(|b| !shard.contains(&b.borrow().address))
3725            .map(|b| {
3726                b.borrow()
3727                    .sticky_id
3728                    .clone()
3729                    .expect("every backend has a sticky id")
3730            })
3731            .expect("a shard of 2 among 8 leaves outsiders");
3732        let reserved = map
3733            .reserve_sticky_backend("sharded", &outsider, Some(key), now)
3734            .expect("the sticky backend is live");
3735        assert_eq!(
3736            reserved.borrow().sticky_id.as_deref(),
3737            Some(outsider.as_str())
3738        );
3739        assert!(!shard.contains(&reserved.borrow().address));
3740    }
3741
3742    /// #524: a sharded selection allocates nothing: the scores, the shard and
3743    /// the candidates live in buffers `add_backend` reserved.
3744    ///
3745    /// TO SEE THIS RED: collect the shard into a fresh `Vec` in
3746    /// `BackendList::compute_shard`; each selection then allocates.
3747    #[test]
3748    fn a_sharded_selection_allocates_nothing() {
3749        use std::hint::black_box;
3750
3751        use crate::test_allocations::allocations;
3752
3753        let now = Instant::now();
3754        let mut list = sharded_list(
3755            16,
3756            LoadBalancingAlgorithms::LeastLoaded,
3757            sharding(25, 8, ShardMode::Fallback),
3758        );
3759        // Warm-up: the first selection may settle lazily built policy state.
3760        drop(list.select_with_key(Some(0), now));
3761        let mut allocated = 0;
3762        for key in 0..256u64 {
3763            let before = allocations();
3764            let picked = black_box(&mut list).select_with_key(black_box(Some(key)), now);
3765            allocated += allocations() - before;
3766            assert_eq!(picked.1, ShardOutcome::InShard);
3767        }
3768        assert_eq!(
3769            allocated, 0,
3770            "256 sharded selections made {allocated} allocations"
3771        );
3772    }
3773}
3774
3775/// A retry skips the backends its request already failed to connect to,
3776/// under every load-balancing policy (sozu-proxy/sozu#1800).
3777#[cfg(test)]
3778mod exclusion_tests {
3779    use super::*;
3780
3781    const POLICIES: [LoadBalancingAlgorithms; 6] = [
3782        LoadBalancingAlgorithms::RoundRobin,
3783        LoadBalancingAlgorithms::Random,
3784        LoadBalancingAlgorithms::LeastLoaded,
3785        LoadBalancingAlgorithms::PowerOfTwo,
3786        LoadBalancingAlgorithms::Hrw,
3787        LoadBalancingAlgorithms::Maglev,
3788    ];
3789
3790    fn address(port: u16) -> SocketAddr {
3791        SocketAddr::from(([127, 0, 0, 1], port))
3792    }
3793
3794    /// A list of `count` healthy backends on ports 9100.. under `policy`.
3795    fn list(policy: LoadBalancingAlgorithms, count: u16) -> BackendList {
3796        let mut list = BackendList::with_seed(7);
3797        for index in 0..count {
3798            list.add_backend(Backend::new(
3799                &format!("b{index}"),
3800                address(9100 + index),
3801                None,
3802                None,
3803                None,
3804            ));
3805        }
3806        list.set_load_balancing_policy(policy, None, 11);
3807        list
3808    }
3809
3810    #[test]
3811    fn a_selection_never_returns_an_excluded_backend_under_any_policy() {
3812        let now = Instant::now();
3813        let exclude = [address(9100), address(9101), address(9102)];
3814        for policy in POLICIES {
3815            let mut list = list(policy, 4);
3816            for key in 0..32 {
3817                // HRW and Maglev pin on the key; the others ignore it.
3818                let (picked, _) = list.select_with_key_excluding(Some(key), now, &exclude);
3819                let picked = picked
3820                    .expect("one backend is not excluded")
3821                    .borrow()
3822                    .address;
3823                assert_eq!(
3824                    picked,
3825                    address(9103),
3826                    "{policy:?} picked {picked}, which the request already tried"
3827                );
3828            }
3829        }
3830    }
3831
3832    #[test]
3833    fn an_empty_exclusion_selects_exactly_as_before() {
3834        let now = Instant::now();
3835        for policy in POLICIES {
3836            let mut excluding = list(policy, 4);
3837            let mut plain = list(policy, 4);
3838            for key in 0..16 {
3839                let left = excluding
3840                    .select_with_key_excluding(Some(key), now, &[])
3841                    .0
3842                    .map(|backend| backend.borrow().address);
3843                let right = plain
3844                    .select_with_key(Some(key), now)
3845                    .0
3846                    .map(|backend| backend.borrow().address);
3847                assert_eq!(left, right, "{policy:?} diverged on key {key}");
3848            }
3849        }
3850    }
3851
3852    #[test]
3853    fn excluding_every_backend_falls_back_to_the_plain_selection() {
3854        let now = Instant::now();
3855        let exclude = [address(9100), address(9101)];
3856        for policy in POLICIES {
3857            let mut list = list(policy, 2);
3858            let (picked, _) = list.select_with_key_excluding(Some(3), now, &exclude);
3859            assert!(
3860                picked.is_some(),
3861                "{policy:?}: a request that tried every backend retries one of them"
3862            );
3863        }
3864    }
3865
3866    #[test]
3867    fn an_excluded_sticky_backend_is_not_followed() {
3868        let now = Instant::now();
3869        let mut map = BackendMap::with_seed(5);
3870        for index in 0..2u16 {
3871            map.add_backend(
3872                "cluster",
3873                Backend::new(
3874                    &format!("b{index}"),
3875                    address(9200 + index),
3876                    Some(format!("sticky{index}")),
3877                    None,
3878                    None,
3879                ),
3880            );
3881        }
3882        let followed = map
3883            .reserve_sticky_backend_excluding("cluster", "sticky0", None, now, &[])
3884            .expect("a backend is selectable");
3885        assert_eq!(followed.borrow().address, address(9200));
3886        let skipped = map
3887            .reserve_sticky_backend_excluding("cluster", "sticky0", None, now, &[address(9200)])
3888            .expect("a backend is selectable");
3889        assert_eq!(
3890            skipped.borrow().address,
3891            address(9201),
3892            "the cookie names a backend the request already failed to reach"
3893        );
3894    }
3895}
3896
3897/// Two backend ids at one address share that address's load, under every
3898/// policy: the address is one candidate, whatever the number of ids
3899/// configured on it.
3900#[cfg(test)]
3901mod same_address_tests {
3902    use super::*;
3903    use crate::load_balancing::affinity_key_from_value;
3904
3905    const POLICIES: [LoadBalancingAlgorithms; 6] = [
3906        LoadBalancingAlgorithms::RoundRobin,
3907        LoadBalancingAlgorithms::Random,
3908        LoadBalancingAlgorithms::LeastLoaded,
3909        LoadBalancingAlgorithms::PowerOfTwo,
3910        LoadBalancingAlgorithms::Hrw,
3911        LoadBalancingAlgorithms::Maglev,
3912    ];
3913
3914    /// Picks per measured distribution.
3915    const PICKS: u32 = 20_000;
3916
3917    const SHARED: SocketAddr = SocketAddr::new(
3918        std::net::IpAddr::V4(std::net::Ipv4Addr::new(127, 0, 0, 1)),
3919        9300,
3920    );
3921    const OTHER: SocketAddr = SocketAddr::new(
3922        std::net::IpAddr::V4(std::net::Ipv4Addr::new(127, 0, 0, 1)),
3923        9301,
3924    );
3925
3926    fn weighted(id: &str, address: SocketAddr, weight: Option<i32>) -> Backend {
3927        Backend::new(
3928            id,
3929            address,
3930            None,
3931            weight.map(|weight| LoadBalancingParams { weight }),
3932            None,
3933        )
3934    }
3935
3936    /// `a` and `b` at [`SHARED`], `c` at [`OTHER`], in that list order.
3937    fn shared_list(policy: LoadBalancingAlgorithms, a: Backend, b: Backend) -> BackendList {
3938        let mut list = BackendList::with_seed(1839);
3939        list.add_backend(a);
3940        list.add_backend(b);
3941        list.add_backend(weighted("c", OTHER, None));
3942        list.set_load_balancing_policy(policy, Some(LoadMetric::Connections), 1839);
3943        list
3944    }
3945
3946    fn equal_list(policy: LoadBalancingAlgorithms) -> BackendList {
3947        shared_list(
3948            policy,
3949            weighted("a", SHARED, None),
3950            weighted("b", SHARED, None),
3951        )
3952    }
3953
3954    /// Run [`PICKS`] selections and return the picked backend ids. A
3955    /// load-aware policy sees each pick as one more open connection, and an
3956    /// affinity policy sees a fresh, well-spread client key per pick.
3957    fn picks(list: &mut BackendList) -> Vec<String> {
3958        let now = Instant::now();
3959        (0..PICKS)
3960            .map(|index| {
3961                let key = affinity_key_from_value(&index.to_le_bytes());
3962                let picked = list
3963                    .next_available_backend_with_key(Some(key), now)
3964                    .expect("three healthy backends");
3965                let mut picked = picked.borrow_mut();
3966                picked.active_connections += 1;
3967                picked.backend_id.to_owned()
3968            })
3969            .collect()
3970    }
3971
3972    fn share_of(ids: &[String], wanted: &[&str]) -> f64 {
3973        ids.iter()
3974            .filter(|id| wanted.contains(&id.as_str()))
3975            .count() as f64
3976            / ids.len() as f64
3977    }
3978
3979    /// The decision of 2026-10-04: one address, one share. Two ids at one
3980    /// address and one id at another, equal weights: each address takes
3981    /// half the selections, as HRW already gave it.
3982    ///
3983    /// TO SEE THIS RED: remove the `collapse_shared_addresses` calls from
3984    /// `BackendList::select_tiers`; round robin, random, least loaded and
3985    /// power of two then give the shared address two thirds.
3986    #[test]
3987    fn a_shared_address_takes_the_share_of_one_backend_under_every_policy() {
3988        let skewed: Vec<String> = POLICIES
3989            .into_iter()
3990            .filter_map(|policy| {
3991                let mut list = equal_list(policy);
3992                let shared = share_of(&picks(&mut list), &["a", "b"]);
3993                (!(0.47..=0.53).contains(&shared)).then(|| format!("{policy:?}: {shared:.3}"))
3994            })
3995            .collect();
3996        assert!(
3997            skewed.is_empty(),
3998            "the shared address did not take 1/2 of the selections under {skewed:?}"
3999        );
4000    }
4001
4002    /// The shared address is served by its first id in list order when the
4003    /// weights are equal: the second id is never picked while the first can
4004    /// take a connection.
4005    #[test]
4006    fn the_first_id_at_a_shared_address_represents_it() {
4007        for policy in POLICIES {
4008            let mut list = equal_list(policy);
4009            let ids = picks(&mut list);
4010            assert!(
4011                ids.iter().all(|id| id != "b"),
4012                "{policy:?} picked the second id of the shared address"
4013            );
4014            assert!(ids.iter().any(|id| id == "a"), "{policy:?} never picked a");
4015        }
4016    }
4017
4018    /// A down id never serves its address: its eligible sibling does, and
4019    /// the address keeps the share of one backend.
4020    #[test]
4021    fn a_down_id_yields_its_address_to_an_eligible_sibling() {
4022        for policy in POLICIES {
4023            let mut down = weighted("a", SHARED, None);
4024            down.health.record_failure(1);
4025            assert!(!down.health.is_healthy());
4026            let mut list = shared_list(policy, down, weighted("b", SHARED, None));
4027            let ids = picks(&mut list);
4028            assert!(
4029                ids.iter().all(|id| id != "a"),
4030                "{policy:?} picked the unhealthy id"
4031            );
4032            let shared = share_of(&ids, &["b"]);
4033            assert!(
4034                (0.47..=0.53).contains(&shared),
4035                "{policy:?} gave the shared address {shared:.3} of the selections, not 1/2"
4036            );
4037        }
4038    }
4039
4040    /// Unequal weights at one address: the heaviest id represents it, as the
4041    /// HRW score — monotonic in weight for one address — has always decided.
4042    #[test]
4043    fn the_heaviest_id_at_a_shared_address_represents_it() {
4044        for policy in POLICIES {
4045            let mut list = shared_list(
4046                policy,
4047                weighted("a", SHARED, Some(50)),
4048                weighted("b", SHARED, Some(200)),
4049            );
4050            let ids = picks(&mut list);
4051            assert!(
4052                ids.iter().all(|id| id != "a"),
4053                "{policy:?} picked the lighter id of the shared address"
4054            );
4055        }
4056    }
4057
4058    /// A shard counts addresses, not ids: eight addresses, one carrying two
4059    /// ids, at 25% make shards of two addresses, and a shard that holds the
4060    /// shared address holds both its ids, so the sibling of a down
4061    /// representative still serves inside the shard.
4062    #[test]
4063    fn a_shard_counts_a_shared_address_once_and_holds_all_its_ids() {
4064        let mut list = BackendList::with_seed(524);
4065        list.set_load_balancing_policy(LoadBalancingAlgorithms::Hrw, None, 524);
4066        for index in 0..8u16 {
4067            list.add_backend(Backend::new(
4068                &format!("shard-{index}"),
4069                SocketAddr::from(([127, 0, 0, 1], 20_000 + index)),
4070                None,
4071                None,
4072                None,
4073            ));
4074        }
4075        let shared = SocketAddr::from(([127, 0, 0, 1], 20_000));
4076        list.add_backend(Backend::new("shard-0-bis", shared, None, None, None));
4077        list.set_shuffle_sharding(Some(ShuffleSharding {
4078            percent: 25,
4079            min_backends: 4,
4080            mode: ShardMode::Fallback,
4081        }));
4082
4083        let mut holding_shared = 0;
4084        for key in 0..256u64 {
4085            assert!(list.compute_shard(Some(key)), "the list must shard");
4086            let mut addresses: Vec<SocketAddr> = list
4087                .shard
4088                .iter()
4089                .map(|&index| list.backends[index].borrow().address)
4090                .collect();
4091            let ids = addresses.len();
4092            addresses.sort_unstable();
4093            addresses.dedup();
4094            assert_eq!(addresses.len(), 2, "key {key}: a shard holds two addresses");
4095            if addresses.contains(&shared) {
4096                holding_shared += 1;
4097                assert_eq!(ids, 3, "key {key}: the shared address brings both ids");
4098            } else {
4099                assert_eq!(ids, 2, "key {key}: one id per unshared address");
4100            }
4101        }
4102        assert!(
4103            holding_shared > 0,
4104            "some shard must hold the shared address"
4105        );
4106    }
4107
4108    /// `LEAST_LOADED` and `POWER_OF_TWO` read an address's load as the sum
4109    /// over its ids: the representative `a` is idle, but its sibling `b`
4110    /// carries 5 connections, so the address weighs 5 against `c`'s 3 and
4111    /// `c` is picked.
4112    #[test]
4113    fn a_load_aware_policy_reads_the_load_of_the_whole_address() {
4114        let now = Instant::now();
4115        for policy in [
4116            LoadBalancingAlgorithms::LeastLoaded,
4117            LoadBalancingAlgorithms::PowerOfTwo,
4118        ] {
4119            let mut list = equal_list(policy);
4120            list.backends[1].borrow_mut().active_connections = 5;
4121            list.backends[2].borrow_mut().active_connections = 3;
4122            for key in 0..64 {
4123                let picked = list
4124                    .next_available_backend_with_key(Some(key), now)
4125                    .expect("three healthy backends");
4126                assert_eq!(
4127                    picked.borrow().backend_id,
4128                    "c",
4129                    "{policy:?} compared the representative's own load, not its address's"
4130                );
4131            }
4132        }
4133    }
4134
4135    /// The cost guard: one shared pair among 1000 backends costs a selection
4136    /// a few units of shared-address work (`SIBLING_STEPS`: one per
4137    /// representative comparison or sibling-ring step), not one per
4138    /// candidate. A scan of the candidates per candidate costs a million.
4139    ///
4140    /// TO SEE THIS RED: replace the ring walk in
4141    /// `BackendList::collapse_shared_addresses` with
4142    /// `candidates.iter().any(|&other| outranks_at_address(..))`.
4143    #[test]
4144    fn one_shared_pair_costs_a_selection_constant_work() {
4145        use crate::load_balancing::SIBLING_STEPS;
4146
4147        let now = Instant::now();
4148        // Built once: the debug-only invariant sweep after each insertion is
4149        // quadratic in the list, so a list per case would dominate the run.
4150        let mut list = BackendList::with_seed(1856);
4151        for index in 0..1000u16 {
4152            let address = SocketAddr::from(([10, (index >> 8) as u8, index as u8, 1], 8080));
4153            list.add_backend(Backend::new(
4154                &format!("b{index}"),
4155                address,
4156                None,
4157                None,
4158                None,
4159            ));
4160        }
4161        let shared = SocketAddr::from(([10, 0, 0, 1], 8080));
4162        list.add_backend(Backend::new("dup", shared, None, None, None));
4163        for sharded in [false, true] {
4164            list.set_shuffle_sharding(sharded.then_some(ShuffleSharding {
4165                percent: 25,
4166                min_backends: 8,
4167                mode: ShardMode::Fallback,
4168            }));
4169            for policy in POLICIES {
4170                list.set_load_balancing_policy(policy, Some(LoadMetric::Connections), 1856);
4171                let mut worst = 0;
4172                for key in 0..64u64 {
4173                    SIBLING_STEPS.with(|steps| steps.set(0));
4174                    assert!(list.select_with_key(Some(key), now).0.is_some());
4175                    worst = worst.max(SIBLING_STEPS.with(|steps| steps.get()));
4176                }
4177                assert!(
4178                    worst <= 8,
4179                    "{policy:?} (sharded: {sharded}) spent {worst} steps on one shared pair"
4180                );
4181            }
4182        }
4183    }
4184
4185    /// `backends` is public: a library user may push or pop a backend
4186    /// without `add_backend`/`remove_backend`, leaving the sibling state of
4187    /// another length. Selection must then neither panic nor index a stale
4188    /// ring, under every policy, sharded or not.
4189    #[test]
4190    fn a_list_edited_directly_still_selects_without_panicking() {
4191        let now = Instant::now();
4192        let sharding = Some(ShuffleSharding {
4193            percent: 50,
4194            min_backends: 2,
4195            mode: ShardMode::Fallback,
4196        });
4197        for policy in POLICIES {
4198            for sharded in [false, true] {
4199                // Grown: a pushed sibling of the shared address, and a
4200                // pushed backend at a new address.
4201                let mut grown = equal_list(policy);
4202                grown.set_shuffle_sharding(sharded.then_some(sharding).flatten());
4203                for (id, port) in [("pushed-sibling", 9300), ("pushed", 9302)] {
4204                    let address = SocketAddr::from(([127, 0, 0, 1], port));
4205                    grown
4206                        .backends
4207                        .push(Rc::new(RefCell::new(weighted(id, address, None))));
4208                }
4209                // Shrunk: the last backend popped, the ring still covers it.
4210                let mut shrunk = equal_list(policy);
4211                shrunk.set_shuffle_sharding(sharded.then_some(sharding).flatten());
4212                shrunk.backends.pop();
4213                for list in [&mut grown, &mut shrunk] {
4214                    for key in 0..64 {
4215                        assert!(
4216                            list.select_with_key(Some(key), now).0.is_some(),
4217                            "{policy:?} (sharded: {sharded}) selected nothing"
4218                        );
4219                    }
4220                }
4221            }
4222        }
4223    }
4224
4225    /// Collapsing allocates nothing: the scratch buffer is reserved on the
4226    /// control plane like the candidate buffer.
4227    #[test]
4228    fn a_selection_over_a_shared_address_allocates_nothing() {
4229        use std::hint::black_box;
4230
4231        use crate::test_allocations::allocations;
4232
4233        let now = Instant::now();
4234        for policy in POLICIES {
4235            let mut list = equal_list(policy);
4236            drop(list.select_with_key(Some(0), now));
4237            let mut allocated = 0;
4238            for key in 0..256u64 {
4239                let before = allocations();
4240                let picked = black_box(&mut list).select_with_key(black_box(Some(key)), now);
4241                allocated += allocations() - before;
4242                assert!(picked.0.is_some());
4243            }
4244            assert_eq!(
4245                allocated, 0,
4246                "{policy:?}: 256 selections made {allocated} allocations"
4247            );
4248        }
4249    }
4250}
4251
4252#[cfg(test)]
4253mod unix_path_tests {
4254    use std::{
4255        io::{Read, Write},
4256        os::unix::net::UnixListener,
4257        path::PathBuf,
4258        time::Instant,
4259    };
4260
4261    use super::*;
4262
4263    /// A filesystem socket path unique to this test process and `tag`.
4264    fn socket_path(tag: &str) -> PathBuf {
4265        let path = std::env::temp_dir().join(format!("sozu-lib-{tag}-{}.sock", std::process::id()));
4266        let _ = std::fs::remove_file(&path);
4267        path
4268    }
4269
4270    /// Wait for the nonblocking `stream` to finish connecting, then echo a
4271    /// payload through `listener` and back.
4272    fn assert_round_trip(stream: TcpStream, listener: &UnixListener) {
4273        let (mut accepted, _) = listener.accept().expect("the dial reaches the listener");
4274        let mut stream = std::net::TcpStream::from(std::os::fd::OwnedFd::from(stream));
4275        stream.set_nonblocking(false).unwrap();
4276        stream.write_all(b"ping").unwrap();
4277        let mut received = [0u8; 4];
4278        accepted.read_exact(&mut received).unwrap();
4279        assert_eq!(&received, b"ping");
4280        accepted.write_all(b"pong").unwrap();
4281        stream.read_exact(&mut received).unwrap();
4282        assert_eq!(&received, b"pong");
4283    }
4284
4285    #[test]
4286    fn a_unix_path_backend_dials_its_socket_not_its_address() {
4287        let path = socket_path("dial");
4288        let listener = UnixListener::bind(&path).unwrap();
4289        // Nothing listens on this port: a TCP dial would be refused.
4290        let unused: SocketAddr = "127.0.0.1:1".parse().unwrap();
4291        let stream = connect_backend(unused, path.to_str()).expect("dial the unix socket");
4292        assert_eq!(
4293            socket2::SockRef::from(&stream).domain().unwrap(),
4294            socket2::Domain::UNIX
4295        );
4296        assert_round_trip(stream, &listener);
4297        std::fs::remove_file(&path).unwrap();
4298    }
4299
4300    #[cfg(target_os = "linux")]
4301    #[test]
4302    fn an_abstract_unix_path_backend_dials_the_abstract_namespace() {
4303        use std::os::linux::net::SocketAddrExt;
4304        let name = format!("sozu-lib-abstract-{}", std::process::id());
4305        let address = std::os::unix::net::SocketAddr::from_abstract_name(name.as_bytes()).unwrap();
4306        let listener = UnixListener::bind_addr(&address).unwrap();
4307        let unused: SocketAddr = "127.0.0.1:1".parse().unwrap();
4308        let stream = connect_backend(unused, Some(&format!("@{name}"))).unwrap();
4309        assert_round_trip(stream, &listener);
4310    }
4311
4312    #[test]
4313    fn a_missing_unix_socket_is_a_failed_connect() {
4314        let mut backend = Backend::new("b", "127.0.0.1:1".parse().unwrap(), None, None, None)
4315            .with_unix_path(Some(socket_path("missing").to_str().unwrap().to_owned()));
4316        let mut rng = StdRng::seed_from_u64(0);
4317        let error = backend
4318            .try_connect(Instant::now(), &mut rng)
4319            .expect_err("no socket file means no connection");
4320        assert!(matches!(error, BackendError::MioConnection(_)));
4321        assert_eq!(backend.failures, 1);
4322        assert_eq!(backend.active_connections, 0);
4323    }
4324
4325    #[test]
4326    fn a_unix_backend_connects_through_try_connect() {
4327        let path = socket_path("try-connect");
4328        let listener = UnixListener::bind(&path).unwrap();
4329        let mut backend = Backend::new("b", "127.0.0.1:1".parse().unwrap(), None, None, None)
4330            .with_unix_path(Some(path.to_str().unwrap().to_owned()));
4331        let mut rng = StdRng::seed_from_u64(0);
4332        let stream = backend.try_connect(Instant::now(), &mut rng).unwrap();
4333        assert_eq!(backend.active_connections, 1);
4334        assert_eq!(backend.failures, 0);
4335        assert_round_trip(stream, &listener);
4336        std::fs::remove_file(&path).unwrap();
4337    }
4338
4339    #[test]
4340    fn re_adding_a_backend_updates_its_unix_path() {
4341        let address: SocketAddr = "127.0.0.1:1".parse().unwrap();
4342        let mut list = BackendList::new();
4343        list.add_backend(Backend::new("b", address, None, None, None));
4344        list.add_backend(
4345            Backend::new("b", address, None, None, None)
4346                .with_unix_path(Some("/run/b.sock".to_owned())),
4347        );
4348        assert_eq!(list.backends.len(), 1);
4349        assert_eq!(
4350            list.backends[0].borrow().unix_path.as_deref(),
4351            Some("/run/b.sock")
4352        );
4353    }
4354}