Skip to main content

sozu_command_lib/
state.rs

1use std::{
2    collections::{
3        BTreeMap, BTreeSet, HashMap, HashSet, btree_map::Entry as BTreeMapEntry,
4        hash_map::DefaultHasher,
5    },
6    fmt,
7    fs::File,
8    hash::{Hash, Hasher},
9    io::Write,
10    iter::repeat,
11    net::SocketAddr,
12};
13
14use prost::{Message, UnknownEnumValue};
15
16use crate::{
17    ObjectKind,
18    certificate::{CertificateError, Fingerprint, calculate_fingerprint},
19    config::validate_sni_pattern,
20    proto::{
21        command::{
22            ActivateListener, AddBackend, AddCertificate, CertificateAndKey, Cluster,
23            ClusterInformation, CustomHttpAnswers, DeactivateListener, FrontendFilters,
24            HealthChecksList, HttpListenerConfig, HttpsListenerConfig, InitialState,
25            ListedFrontends, ListenerType, ListenersList, PathRule, QueryCertificatesFilters,
26            RemoveBackend, RemoveCertificate, RemoveListener, ReplaceCertificate, Request,
27            RequestCounts, RequestHttpFrontend, RequestTcpFrontend, RequestUdpFrontend,
28            SetHealthCheck, SocketAddress, TcpListenerConfig, UdpListenerConfig,
29            UpdateHttpListenerConfig, UpdateHttpsListenerConfig, UpdateTcpListenerConfig,
30            UpdateUdpListenerConfig, WorkerRequest, request::RequestType,
31        },
32        display::format_request_type,
33    },
34    response::{Backend, HttpFrontend, TcpFrontend, UdpFrontend},
35};
36
37/// To use throughout Sōzu
38pub type ClusterId = String;
39
40#[derive(thiserror::Error)]
41pub enum StateError {
42    #[error("Request came in empty")]
43    EmptyRequest,
44    #[error("dispatching this request did not bring any change to the state")]
45    NoChange,
46    #[error("State can not handle this request")]
47    UndispatchableRequest,
48    #[error("Did not find {kind:?} with address or id_bytes={}", .id.len())]
49    NotFound { kind: ObjectKind, id: String },
50    #[error(
51        "{kind:?} with id_bytes={} already exists; remove it first, or apply the corresponding update if the object supports one, instead of re-adding it",
52        .id.len()
53    )]
54    Exists { kind: ObjectKind, id: String },
55    #[error("Wrong field value: {0}")]
56    WrongFieldValue(UnknownEnumValue),
57    #[error("Could not add certificate: error_kind={}", certificate_error_kind(.0))]
58    AddCertificate(CertificateError),
59    #[error("Could not remove certificate: fingerprint_bytes={}", .0.len())]
60    RemoveCertificate(String),
61    #[error("Could not replace certificate: error_bytes={}", .0.len())]
62    ReplaceCertificate(String),
63    #[error("Could not convert frontend: frontend_bytes={} error_bytes={}", .frontend.len(), .error.len())]
64    FrontendConversion { frontend: String, error: String },
65    #[error("Could not write state to file: io_kind={:?} os_code={:?}", .0.kind(), .0.raw_os_error())]
66    FileError(std::io::Error),
67    #[error("Invalid value for field '{field}': {reason}")]
68    InvalidValue {
69        field: &'static str,
70        reason: &'static str,
71    },
72    /// Mirrors the worker's `ProxyError::InvalidTcpFrontend`
73    /// (`lib/src/tcp.rs`): the master rejects an `AddTcpFrontend` for the
74    /// same reasons `validate_new_tcp_front` would, so a route never lands
75    /// in `ConfigState` only to be NACKed by every worker on the next
76    /// fan-out (sozu-proxy/sozu#1290).
77    #[error("invalid TCP frontend for address {address}: reason_bytes={}", .reason.len())]
78    InvalidTcpFrontend { address: SocketAddr, reason: String },
79}
80
81impl fmt::Debug for StateError {
82    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
83        write!(f, "StateError({self})")
84    }
85}
86
87fn certificate_error_kind(error: &CertificateError) -> &'static str {
88    match error {
89        CertificateError::ParsePEMCertificate(_) => "parse_pem_certificate",
90        CertificateError::ParseX509Certificate(_) => "parse_x509_certificate",
91        CertificateError::InvalidTlsVersion(_) => "invalid_tls_version",
92        CertificateError::InvalidFingerprint(_) => "invalid_fingerprint",
93        CertificateError::LoadFile { .. } => "load_file",
94        CertificateError::DecodeError(_) => "decode_error",
95    }
96}
97
98/// The `ConfigState` represents the state of Sōzu's business, which is to forward traffic
99/// from frontends to backends. Hence, it contains all details about:
100///
101/// - listeners (socket addresses, for TCP and HTTP connections)
102/// - frontends (bind to a listener)
103/// - backends (to forward connections to)
104/// - clusters (routing rules from frontends to backends)
105/// - TLS certificates
106#[derive(Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
107pub struct ConfigState {
108    pub clusters: BTreeMap<ClusterId, Cluster>,
109    pub backends: BTreeMap<ClusterId, Vec<Backend>>,
110    /// socket address -> HTTP listener
111    pub http_listeners: BTreeMap<SocketAddr, HttpListenerConfig>,
112    /// socket address -> HTTPS listener
113    pub https_listeners: BTreeMap<SocketAddr, HttpsListenerConfig>,
114    /// socket address -> TCP listener
115    pub tcp_listeners: BTreeMap<SocketAddr, TcpListenerConfig>,
116    /// socket address -> UDP listener
117    pub udp_listeners: BTreeMap<SocketAddr, UdpListenerConfig>,
118    /// HTTP frontends, indexed by a summary of each front's address;hostname;path, for uniqueness.
119    /// For example: `"0.0.0.0:8080;lolcatho.st;P/api"`
120    pub http_fronts: BTreeMap<String, HttpFrontend>,
121    /// indexed by (address, hostname, path)
122    pub https_fronts: BTreeMap<String, HttpFrontend>,
123    pub tcp_fronts: HashMap<ClusterId, Vec<TcpFrontend>>,
124    pub udp_fronts: HashMap<ClusterId, Vec<UdpFrontend>>,
125    pub certificates: HashMap<SocketAddr, HashMap<Fingerprint, CertificateAndKey>>,
126    /// A census of requests that were received. Name of the request -> number of occurences
127    pub request_counts: BTreeMap<String, i32>,
128}
129
130impl std::fmt::Debug for ConfigState {
131    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
132        let backends_count = self
133            .backends
134            .values()
135            .map(Vec::len)
136            .fold(0usize, usize::saturating_add);
137        let tcp_frontends_count = self
138            .tcp_fronts
139            .values()
140            .map(Vec::len)
141            .fold(0usize, usize::saturating_add);
142        let udp_frontends_count = self
143            .udp_fronts
144            .values()
145            .map(Vec::len)
146            .fold(0usize, usize::saturating_add);
147        let certificates_count = self
148            .certificates
149            .values()
150            .map(HashMap::len)
151            .fold(0usize, usize::saturating_add);
152
153        f.debug_struct("ConfigState")
154            .field("clusters_count", &self.clusters.len())
155            .field("backend_clusters_count", &self.backends.len())
156            .field("backends_count", &backends_count)
157            .field("http_listeners_count", &self.http_listeners.len())
158            .field("https_listeners_count", &self.https_listeners.len())
159            .field("tcp_listeners_count", &self.tcp_listeners.len())
160            .field("udp_listeners_count", &self.udp_listeners.len())
161            .field("http_frontends_count", &self.http_fronts.len())
162            .field("https_frontends_count", &self.https_fronts.len())
163            .field("tcp_frontend_clusters_count", &self.tcp_fronts.len())
164            .field("tcp_frontends_count", &tcp_frontends_count)
165            .field("udp_frontend_clusters_count", &self.udp_fronts.len())
166            .field("udp_frontends_count", &udp_frontends_count)
167            .field("certificate_addresses_count", &self.certificates.len())
168            .field("certificates_count", &certificates_count)
169            .field("request_counts_count", &self.request_counts.len())
170            .finish_non_exhaustive()
171    }
172}
173
174impl ConfigState {
175    pub fn new() -> Self {
176        Self::default()
177    }
178
179    pub fn dispatch(&mut self, request: &Request) -> Result<(), StateError> {
180        let request_type = match &request.request_type {
181            Some(t) => t,
182            None => return Err(StateError::EmptyRequest),
183        };
184
185        self.increment_request_count(request);
186
187        let result = match request_type {
188            RequestType::AddCluster(cluster) => self.add_cluster(cluster),
189            RequestType::RemoveCluster(cluster_id) => self.remove_cluster(cluster_id),
190            RequestType::AddHttpListener(listener) => self.add_http_listener(listener),
191            RequestType::AddHttpsListener(listener) => self.add_https_listener(listener),
192            RequestType::AddTcpListener(listener) => self.add_tcp_listener(listener),
193            RequestType::AddUdpListener(listener) => self.add_udp_listener(listener),
194            RequestType::RemoveListener(remove) => self.remove_listener(remove),
195            RequestType::ActivateListener(activate) => self.activate_listener(activate),
196            RequestType::DeactivateListener(deactivate) => self.deactivate_listener(deactivate),
197            RequestType::AddHttpFrontend(front) => self.add_http_frontend(front),
198            RequestType::RemoveHttpFrontend(front) => self.remove_http_frontend(front),
199            RequestType::AddCertificate(add) => self.add_certificate(add),
200            RequestType::RemoveCertificate(remove) => self.remove_certificate(remove),
201            RequestType::ReplaceCertificate(replace) => self.replace_certificate(replace),
202            RequestType::AddHttpsFrontend(front) => self.add_https_frontend(front),
203            RequestType::RemoveHttpsFrontend(front) => self.remove_https_frontend(front),
204            RequestType::AddTcpFrontend(front) => self.add_tcp_frontend(front),
205            RequestType::RemoveTcpFrontend(front) => self.remove_tcp_frontend(front),
206            RequestType::AddUdpFrontend(front) => self.add_udp_frontend(front),
207            RequestType::RemoveUdpFrontend(front) => self.remove_udp_frontend(front),
208            RequestType::AddBackend(add_backend) => self.add_backend(add_backend),
209            RequestType::RemoveBackend(backend) => self.remove_backend(backend),
210            RequestType::UpdateHttpListener(patch) => self.update_http_listener(patch),
211            RequestType::UpdateHttpsListener(patch) => self.update_https_listener(patch),
212            RequestType::UpdateTcpListener(patch) => self.update_tcp_listener(patch),
213            RequestType::UpdateUdpListener(patch) => self.update_udp_listener(patch),
214            RequestType::SetHealthCheck(set) => self.set_health_check(set),
215            RequestType::RemoveHealthCheck(cluster_id) => self.remove_health_check(cluster_id),
216
217            // This is to avoid the error message. These request types are
218            // worker-only / runtime-only and do not affect the persisted
219            // ConfigState (e.g., a worker-side global limit set via
220            // SetMaxConnectionsPerIp does NOT survive a worker restart;
221            // operators must mirror the change in the TOML to make it
222            // sticky).
223            RequestType::Logging(_)
224            | RequestType::CountRequests(_)
225            | RequestType::Status(_)
226            | RequestType::SoftStop(_)
227            | RequestType::QueryCertificatesFromWorkers(_)
228            | RequestType::QueryClusterById(_)
229            | RequestType::QueryClustersByDomain(_)
230            | RequestType::QueryMetrics(_)
231            | RequestType::QueryClustersHashes(_)
232            | RequestType::ConfigureMetrics(_)
233            | RequestType::SetMetricDetail(_)
234            | RequestType::ReturnListenSockets(_)
235            | RequestType::SetMaxConnectionsPerIp(_)
236            | RequestType::QueryMaxConnectionsPerIp(_)
237            | RequestType::HardStop(_) => Ok(()),
238
239            _other_request => Err(StateError::UndispatchableRequest),
240        };
241
242        // Run-to-completion postcondition: whatever path `dispatch` took, the
243        // cross-map invariants of the model must hold once it returns. We run
244        // the full sweep on both success and error: a failed mutating handler
245        // (e.g. a duplicate `add_*` or an absent `remove_*`) is required to be
246        // a no-op, so the invariants must be intact regardless of the result.
247        #[cfg(debug_assertions)]
248        self.check_invariants();
249
250        result
251    }
252
253    /// Full cross-map invariant sweep for the control-plane state model.
254    ///
255    /// This is the run-to-completion postcondition called via a
256    /// `#[cfg(debug_assertions)]` guard at the end of [`Self::dispatch`]. It
257    /// encodes the coherence invariants the diff/replay machinery relies on:
258    /// every map entry is self-consistent (a value's stored key/cluster_id
259    /// matches the key it is filed under), and the public accounting helpers
260    /// (`count_frontends`/`count_backends`) agree with the raw map contents.
261    ///
262    /// Compiled out entirely in release builds (no body, no callers).
263    #[cfg(debug_assertions)]
264    fn check_invariants(&self) {
265        // Listener maps: the value's `address` field must match the SocketAddr
266        // key it is filed under, or a hot-upgrade replay (which re-derives the
267        // key from the value) would land the entry under a different key.
268        for (addr, listener) in &self.http_listeners {
269            debug_assert_eq!(
270                SocketAddr::from(listener.address),
271                *addr,
272                "http_listener value address must match its map key"
273            );
274        }
275        for (addr, listener) in &self.https_listeners {
276            debug_assert_eq!(
277                SocketAddr::from(listener.address),
278                *addr,
279                "https_listener value address must match its map key"
280            );
281        }
282        for (addr, listener) in &self.tcp_listeners {
283            debug_assert_eq!(
284                SocketAddr::from(listener.address),
285                *addr,
286                "tcp_listener value address must match its map key"
287            );
288        }
289
290        // Clusters: the value's `cluster_id` must match the key it is filed
291        // under (replay re-keys on `cluster.cluster_id`).
292        for (cluster_id, cluster) in &self.clusters {
293            debug_assert_eq!(
294                &cluster.cluster_id, cluster_id,
295                "cluster value cluster_id must match its map key"
296            );
297        }
298
299        // Backends: grouped by cluster_id. Every backend in a bucket must carry
300        // that bucket's cluster_id (replay groups on `backend.cluster_id`), and
301        // the per-cluster Vec must stay deduplicated on (backend_id, address) —
302        // `add_backend` upserts, so duplicates would mean lost state.
303        for (cluster_id, backends) in &self.backends {
304            for backend in backends {
305                debug_assert_eq!(
306                    &backend.cluster_id, cluster_id,
307                    "backend cluster_id must match its bucket key"
308                );
309            }
310            let unique: HashSet<(&String, &SocketAddr)> = backends
311                .iter()
312                .map(|b| (&b.backend_id, &b.address))
313                .collect();
314            debug_assert_eq!(
315                unique.len(),
316                backends.len(),
317                "backends within a cluster must be unique on (backend_id, address)"
318            );
319        }
320
321        // TCP frontends: grouped by cluster_id. Every frontend in a bucket must
322        // carry that bucket's cluster_id, and `add_tcp_frontend` rejects exact
323        // duplicates so each bucket stays a set.
324        for (cluster_id, fronts) in &self.tcp_fronts {
325            for front in fronts {
326                debug_assert_eq!(
327                    &front.cluster_id, cluster_id,
328                    "tcp_frontend cluster_id must match its bucket key"
329                );
330            }
331            let unique: HashSet<&TcpFrontend> = fronts.iter().collect();
332            debug_assert_eq!(
333                unique.len(),
334                fronts.len(),
335                "tcp frontends within a cluster must be unique"
336            );
337        }
338
339        // Certificates: nested map keyed by (address, fingerprint). The inner
340        // map's fingerprint key is the addressing identity used by diff; we do
341        // not recompute it here (expensive), but the outer/inner structure must
342        // not hold an empty inner map silently produced outside the API — an
343        // empty bucket is a benign no-op for diff/replay, so we only assert the
344        // address-key relationship is preserved by construction (trivially true
345        // for a BTree/HashMap), leaving the costly fingerprint recompute out.
346
347        // Public accounting helpers must agree with the raw maps. These are the
348        // numbers the CLI and metrics surface; a drift here is a real bug.
349        let raw_frontends = self.http_fronts.len()
350            + self.https_fronts.len()
351            + self.count_tcp_frontends_raw()
352            + self.udp_fronts.values().map(|v| v.len()).sum::<usize>();
353        debug_assert_eq!(
354            self.count_frontends(),
355            raw_frontends,
356            "count_frontends must equal the sum of all frontend map entries"
357        );
358        let raw_backends: usize = self.backends.values().map(|v| v.len()).sum();
359        debug_assert_eq!(
360            self.count_backends(),
361            raw_backends,
362            "count_backends must equal the sum of all backend Vec lengths"
363        );
364    }
365
366    /// Raw count of TCP frontends across all clusters. Used by the debug-only
367    /// [`Self::check_invariants`] to cross-check `count_frontends`, and by
368    /// `add_tcp_frontend`'s no-mutation-before-admission accounting.
369    /// Compiled unconditionally (unlike `check_invariants`): it is a trivial
370    /// sum of bucket lengths, and `debug_assert!` arguments still typecheck
371    /// in release builds even though they compile out — a debug-gated helper
372    /// referenced there breaks the release/bench build.
373    fn count_tcp_frontends_raw(&self) -> usize {
374        self.tcp_fronts.values().map(|v| v.len()).sum()
375    }
376
377    /// Increments the count for this request type
378    fn increment_request_count(&mut self, request: &Request) {
379        if let Some(request_type) = &request.request_type {
380            let count = self
381                .request_counts
382                .entry(format_request_type(request_type).to_owned())
383                .or_insert(1);
384            *count += 1;
385        }
386    }
387
388    pub fn get_request_counts(&self) -> RequestCounts {
389        RequestCounts {
390            map: self.request_counts.clone(),
391        }
392    }
393
394    fn add_cluster(&mut self, cluster: &Cluster) -> Result<(), StateError> {
395        // Validate any inline `cluster.health_check` before mutating state so
396        // an invalid config (zero thresholds, missing leading `/`, CR/LF/NUL/C0
397        // in URI) cannot ride in via the AddCluster path. Without this, TOML
398        // reload, SaveState/LoadState, and direct API AddCluster requests
399        // bypass the SetHealthCheck-side check and let an attacker-controlled
400        // health-check URI smuggle CRLF into outbound HTTP/1.1 probes.
401        if let Some(hc) = cluster.health_check.as_ref()
402            && let Err(reason) = crate::config::validate_health_check_config(hc)
403        {
404            return Err(StateError::InvalidValue {
405                field: "health_check",
406                reason,
407            });
408        }
409        let cluster = cluster.clone();
410        // AddCluster is an upsert (replacing an existing cluster_id keeps the
411        // entry count flat), so we assert on presence/key-coherence rather than
412        // a strict +1 on len.
413        let cluster_id = cluster.cluster_id.clone();
414        self.clusters.insert(cluster_id.clone(), cluster);
415        debug_assert!(
416            self.clusters.contains_key(&cluster_id),
417            "add_cluster must leave the cluster present in the map"
418        );
419        debug_assert_eq!(
420            self.clusters.get(&cluster_id).map(|c| &c.cluster_id),
421            Some(&cluster_id),
422            "stored cluster must be keyed by its own cluster_id"
423        );
424        Ok(())
425    }
426
427    fn remove_cluster(&mut self, cluster_id: &str) -> Result<(), StateError> {
428        let before = self.clusters.len();
429        match self.clusters.remove(cluster_id) {
430            Some(_) => {
431                debug_assert!(
432                    !self.clusters.contains_key(cluster_id),
433                    "remove_cluster must evict the cluster"
434                );
435                debug_assert_eq!(
436                    self.clusters.len(),
437                    before - 1,
438                    "remove_cluster must drop exactly one entry"
439                );
440                Ok(())
441            }
442            None => {
443                debug_assert_eq!(
444                    self.clusters.len(),
445                    before,
446                    "a failed remove_cluster must not mutate the map"
447                );
448                Err(StateError::NotFound {
449                    kind: ObjectKind::Cluster,
450                    id: cluster_id.to_owned(),
451                })
452            }
453        }
454    }
455
456    fn set_health_check(&mut self, set: &SetHealthCheck) -> Result<(), StateError> {
457        // Validate before mutating state so an invalid config (zero
458        // thresholds, missing leading `/`, CR/LF/NUL/C0 in URI) cannot
459        // round-trip through SaveState/LoadState. The worker also
460        // validates at the SetHealthCheck handler — this is the
461        // master-side mirror so off-channel TOML reload paths don't
462        // bypass the policy.
463        if let Err(reason) = crate::config::validate_health_check_config(&set.config) {
464            return Err(StateError::InvalidValue {
465                field: "health_check",
466                reason,
467            });
468        }
469        match self.clusters.get_mut(&set.cluster_id) {
470            Some(cluster) => {
471                cluster.health_check = Some(set.config.to_owned());
472                Ok(())
473            }
474            None => Err(StateError::NotFound {
475                kind: ObjectKind::Cluster,
476                id: set.cluster_id.to_owned(),
477            }),
478        }
479    }
480
481    fn remove_health_check(&mut self, cluster_id: &str) -> Result<(), StateError> {
482        match self.clusters.get_mut(cluster_id) {
483            Some(cluster) => {
484                cluster.health_check = None;
485                Ok(())
486            }
487            None => Err(StateError::NotFound {
488                kind: ObjectKind::Cluster,
489                id: cluster_id.to_owned(),
490            }),
491        }
492    }
493
494    pub fn list_health_checks(&self, cluster_id: Option<&str>) -> HealthChecksList {
495        let map = self
496            .clusters
497            .iter()
498            .filter(|(id, _)| cluster_id.is_none_or(|filter| filter == id.as_str()))
499            .filter_map(|(id, cluster)| {
500                cluster
501                    .health_check
502                    .as_ref()
503                    .map(|hc| (id.to_owned(), hc.to_owned()))
504            })
505            .collect();
506        HealthChecksList { map }
507    }
508
509    fn add_http_listener(&mut self, listener: &HttpListenerConfig) -> Result<(), StateError> {
510        let address: SocketAddr = listener.address.into();
511        let before = self.http_listeners.len();
512        match self.http_listeners.entry(address) {
513            BTreeMapEntry::Vacant(vacant_entry) => vacant_entry.insert(listener.clone()),
514            BTreeMapEntry::Occupied(_) => {
515                debug_assert_eq!(
516                    self.http_listeners.len(),
517                    before,
518                    "a rejected duplicate add_http_listener must not mutate the map"
519                );
520                return Err(StateError::Exists {
521                    kind: ObjectKind::HttpListener,
522                    id: address.to_string(),
523                });
524            }
525        };
526        debug_assert!(
527            self.http_listeners.contains_key(&address),
528            "add_http_listener must insert the listener under its address"
529        );
530        debug_assert_eq!(
531            self.http_listeners.len(),
532            before + 1,
533            "add_http_listener inserts exactly one entry on the vacant path"
534        );
535        Ok(())
536    }
537
538    fn add_https_listener(&mut self, listener: &HttpsListenerConfig) -> Result<(), StateError> {
539        let address: SocketAddr = listener.address.into();
540        let before = self.https_listeners.len();
541        match self.https_listeners.entry(address) {
542            BTreeMapEntry::Vacant(vacant_entry) => vacant_entry.insert(listener.clone()),
543            BTreeMapEntry::Occupied(_) => {
544                debug_assert_eq!(
545                    self.https_listeners.len(),
546                    before,
547                    "a rejected duplicate add_https_listener must not mutate the map"
548                );
549                return Err(StateError::Exists {
550                    kind: ObjectKind::HttpsListener,
551                    id: address.to_string(),
552                });
553            }
554        };
555        debug_assert!(
556            self.https_listeners.contains_key(&address),
557            "add_https_listener must insert the listener under its address"
558        );
559        debug_assert_eq!(
560            self.https_listeners.len(),
561            before + 1,
562            "add_https_listener inserts exactly one entry on the vacant path"
563        );
564        Ok(())
565    }
566
567    fn add_tcp_listener(&mut self, listener: &TcpListenerConfig) -> Result<(), StateError> {
568        let address: SocketAddr = listener.address.into();
569        let before = self.tcp_listeners.len();
570        match self.tcp_listeners.entry(address) {
571            BTreeMapEntry::Vacant(vacant_entry) => vacant_entry.insert(*listener),
572            BTreeMapEntry::Occupied(_) => {
573                debug_assert_eq!(
574                    self.tcp_listeners.len(),
575                    before,
576                    "a rejected duplicate add_tcp_listener must not mutate the map"
577                );
578                return Err(StateError::Exists {
579                    kind: ObjectKind::TcpListener,
580                    id: address.to_string(),
581                });
582            }
583        };
584        debug_assert!(
585            self.tcp_listeners.contains_key(&address),
586            "add_tcp_listener must insert the listener under its address"
587        );
588        debug_assert_eq!(
589            self.tcp_listeners.len(),
590            before + 1,
591            "add_tcp_listener inserts exactly one entry on the vacant path"
592        );
593        Ok(())
594    }
595
596    fn add_udp_listener(&mut self, listener: &UdpListenerConfig) -> Result<(), StateError> {
597        let address: SocketAddr = listener.address.into();
598        match self.udp_listeners.entry(address) {
599            BTreeMapEntry::Vacant(vacant_entry) => vacant_entry.insert(*listener),
600            BTreeMapEntry::Occupied(_) => {
601                return Err(StateError::Exists {
602                    kind: ObjectKind::UdpListener,
603                    id: address.to_string(),
604                });
605            }
606        };
607        Ok(())
608    }
609
610    fn remove_listener(&mut self, remove: &RemoveListener) -> Result<(), StateError> {
611        match ListenerType::try_from(remove.proxy).map_err(StateError::WrongFieldValue)? {
612            ListenerType::Http => self.remove_http_listener(&remove.address.into()),
613            ListenerType::Https => self.remove_https_listener(&remove.address.into()),
614            ListenerType::Tcp => self.remove_tcp_listener(&remove.address.into()),
615            ListenerType::Udp => self.remove_udp_listener(&remove.address.into()),
616        }
617    }
618
619    fn remove_http_listener(&mut self, address: &SocketAddr) -> Result<(), StateError> {
620        let before = self.http_listeners.len();
621        if self.http_listeners.remove(address).is_none() {
622            debug_assert_eq!(
623                self.http_listeners.len(),
624                before,
625                "a failed remove_http_listener must not mutate the map"
626            );
627            return Err(StateError::NoChange);
628        }
629        debug_assert!(
630            !self.http_listeners.contains_key(address),
631            "remove_http_listener must evict the address"
632        );
633        debug_assert_eq!(
634            self.http_listeners.len(),
635            before - 1,
636            "remove_http_listener drops exactly one entry"
637        );
638        Ok(())
639    }
640
641    fn remove_https_listener(&mut self, address: &SocketAddr) -> Result<(), StateError> {
642        let before = self.https_listeners.len();
643        if self.https_listeners.remove(address).is_none() {
644            debug_assert_eq!(
645                self.https_listeners.len(),
646                before,
647                "a failed remove_https_listener must not mutate the map"
648            );
649            return Err(StateError::NoChange);
650        }
651        debug_assert!(
652            !self.https_listeners.contains_key(address),
653            "remove_https_listener must evict the address"
654        );
655        debug_assert_eq!(
656            self.https_listeners.len(),
657            before - 1,
658            "remove_https_listener drops exactly one entry"
659        );
660        Ok(())
661    }
662
663    fn remove_tcp_listener(&mut self, address: &SocketAddr) -> Result<(), StateError> {
664        let before = self.tcp_listeners.len();
665        if self.tcp_listeners.remove(address).is_none() {
666            debug_assert_eq!(
667                self.tcp_listeners.len(),
668                before,
669                "a failed remove_tcp_listener must not mutate the map"
670            );
671            return Err(StateError::NoChange);
672        }
673        debug_assert!(
674            !self.tcp_listeners.contains_key(address),
675            "remove_tcp_listener must evict the address"
676        );
677        debug_assert_eq!(
678            self.tcp_listeners.len(),
679            before - 1,
680            "remove_tcp_listener drops exactly one entry"
681        );
682        Ok(())
683    }
684
685    fn remove_udp_listener(&mut self, address: &SocketAddr) -> Result<(), StateError> {
686        if self.udp_listeners.remove(address).is_none() {
687            return Err(StateError::NoChange);
688        }
689        Ok(())
690    }
691
692    /// Validate and apply a partial patch to an existing HTTP listener.
693    ///
694    /// Only `Some` fields in the patch are written; `None` fields preserve the
695    /// current value. Returns `StateError::NotFound` if the address is unknown,
696    /// `StateError::InvalidValue` if a flood-knob value is below the required
697    /// minimum.
698    fn update_http_listener(&mut self, patch: &UpdateHttpListenerConfig) -> Result<(), StateError> {
699        validate_h2_flood_knobs_http(patch)?;
700
701        let address: SocketAddr = patch.address.into();
702        let listener =
703            self.http_listeners
704                .get_mut(&address)
705                .ok_or_else(|| StateError::NotFound {
706                    kind: ObjectKind::HttpListener,
707                    id: address.to_string(),
708                })?;
709
710        // Shared session-at-accept / per-connection knobs
711        if let Some(v) = patch.public_address {
712            listener.public_address = Some(v);
713        }
714        if let Some(v) = patch.expect_proxy {
715            listener.expect_proxy = v;
716        }
717        if let Some(ref v) = patch.sticky_name {
718            listener.sticky_name = v.to_owned();
719        }
720        if let Some(v) = patch.front_timeout {
721            listener.front_timeout = v;
722        }
723        if let Some(v) = patch.back_timeout {
724            listener.back_timeout = v;
725        }
726        if let Some(v) = patch.connect_timeout {
727            listener.connect_timeout = v;
728        }
729        if let Some(v) = patch.request_timeout {
730            listener.request_timeout = v;
731        }
732        if let Some(patch_answers) = patch.http_answers.as_ref() {
733            merge_custom_http_answers(&mut listener.http_answers, patch_answers);
734        }
735        // H2 flood knobs
736        if let Some(v) = patch.h2_max_rst_stream_per_window {
737            listener.h2_max_rst_stream_per_window = Some(v);
738        }
739        if let Some(v) = patch.h2_max_ping_per_window {
740            listener.h2_max_ping_per_window = Some(v);
741        }
742        if let Some(v) = patch.h2_max_settings_per_window {
743            listener.h2_max_settings_per_window = Some(v);
744        }
745        if let Some(v) = patch.h2_max_empty_data_per_window {
746            listener.h2_max_empty_data_per_window = Some(v);
747        }
748        if let Some(v) = patch.h2_max_continuation_frames {
749            listener.h2_max_continuation_frames = Some(v);
750        }
751        if let Some(v) = patch.h2_max_glitch_count {
752            listener.h2_max_glitch_count = Some(v);
753        }
754        if let Some(v) = patch.h2_initial_connection_window {
755            listener.h2_initial_connection_window = Some(v);
756        }
757        if let Some(v) = patch.h2_max_concurrent_streams {
758            listener.h2_max_concurrent_streams = Some(v);
759        }
760        if let Some(v) = patch.h2_stream_shrink_ratio {
761            listener.h2_stream_shrink_ratio = Some(v);
762        }
763        if let Some(v) = patch.h2_max_rst_stream_lifetime {
764            listener.h2_max_rst_stream_lifetime = Some(v);
765        }
766        if let Some(v) = patch.h2_max_rst_stream_abusive_lifetime {
767            listener.h2_max_rst_stream_abusive_lifetime = Some(v);
768        }
769        if let Some(v) = patch.h2_max_rst_stream_emitted_lifetime {
770            listener.h2_max_rst_stream_emitted_lifetime = Some(v);
771        }
772        if let Some(v) = patch.h2_max_header_list_size {
773            listener.h2_max_header_list_size = Some(v);
774        }
775        if let Some(v) = patch.h2_max_header_table_size {
776            listener.h2_max_header_table_size = Some(v);
777        }
778        if let Some(v) = patch.h2_max_header_fields {
779            listener.h2_max_header_fields = Some(v);
780        }
781        if let Some(v) = patch.h2_stream_idle_timeout_seconds {
782            listener.h2_stream_idle_timeout_seconds = Some(v);
783        }
784        // 0 is valid for graceful_shutdown_deadline (means "wait forever")
785        if let Some(v) = patch.h2_graceful_shutdown_deadline_seconds {
786            listener.h2_graceful_shutdown_deadline_seconds = Some(v);
787        }
788        if let Some(v) = patch.h2_max_window_update_stream0_per_window {
789            listener.h2_max_window_update_stream0_per_window = Some(v);
790        }
791        if let Some(ref v) = patch.sozu_id_header {
792            validate_sozu_id_header(v)?;
793            listener.sozu_id_header = Some(v.to_owned());
794        }
795        Ok(())
796    }
797
798    /// Validate and apply a partial patch to an existing HTTPS listener.
799    ///
800    /// Only `Some` fields in the patch are written; `None` fields preserve the
801    /// current value. Returns `StateError::NotFound` if the address is unknown,
802    /// `StateError::InvalidValue` if a flood-knob value is below the required
803    /// minimum or an ALPN value is unknown.
804    fn update_https_listener(
805        &mut self,
806        patch: &UpdateHttpsListenerConfig,
807    ) -> Result<(), StateError> {
808        validate_h2_flood_knobs_https(patch)?;
809
810        let address: SocketAddr = patch.address.into();
811        let listener =
812            self.https_listeners
813                .get_mut(&address)
814                .ok_or_else(|| StateError::NotFound {
815                    kind: ObjectKind::HttpsListener,
816                    id: address.to_string(),
817                })?;
818
819        // Shared session-at-accept / per-connection knobs
820        if let Some(v) = patch.public_address {
821            listener.public_address = Some(v);
822        }
823        if let Some(v) = patch.expect_proxy {
824            listener.expect_proxy = v;
825        }
826        if let Some(ref v) = patch.sticky_name {
827            listener.sticky_name = v.to_owned();
828        }
829        if let Some(v) = patch.front_timeout {
830            listener.front_timeout = v;
831        }
832        if let Some(v) = patch.back_timeout {
833            listener.back_timeout = v;
834        }
835        if let Some(v) = patch.connect_timeout {
836            listener.connect_timeout = v;
837        }
838        if let Some(v) = patch.request_timeout {
839            listener.request_timeout = v;
840        }
841        if let Some(patch_answers) = patch.http_answers.as_ref() {
842            merge_custom_http_answers(&mut listener.http_answers, patch_answers);
843        }
844        // HTTPS-only knobs
845        if let Some(ref alpn_wrapper) = patch.alpn_protocols {
846            validate_alpn_protocols(&alpn_wrapper.values)?;
847            // Empty values vec = reset to default (runtime treats empty as default)
848            listener.alpn_protocols = alpn_wrapper.values.clone();
849        }
850        if let Some(v) = patch.strict_sni_binding {
851            listener.strict_sni_binding = Some(v);
852        }
853        if let Some(v) = patch.disable_http11 {
854            listener.disable_http11 = Some(v);
855        }
856        // H2 flood knobs
857        if let Some(v) = patch.h2_max_rst_stream_per_window {
858            listener.h2_max_rst_stream_per_window = Some(v);
859        }
860        if let Some(v) = patch.h2_max_ping_per_window {
861            listener.h2_max_ping_per_window = Some(v);
862        }
863        if let Some(v) = patch.h2_max_settings_per_window {
864            listener.h2_max_settings_per_window = Some(v);
865        }
866        if let Some(v) = patch.h2_max_empty_data_per_window {
867            listener.h2_max_empty_data_per_window = Some(v);
868        }
869        if let Some(v) = patch.h2_max_continuation_frames {
870            listener.h2_max_continuation_frames = Some(v);
871        }
872        if let Some(v) = patch.h2_max_glitch_count {
873            listener.h2_max_glitch_count = Some(v);
874        }
875        if let Some(v) = patch.h2_initial_connection_window {
876            listener.h2_initial_connection_window = Some(v);
877        }
878        if let Some(v) = patch.h2_max_concurrent_streams {
879            listener.h2_max_concurrent_streams = Some(v);
880        }
881        if let Some(v) = patch.h2_stream_shrink_ratio {
882            listener.h2_stream_shrink_ratio = Some(v);
883        }
884        if let Some(v) = patch.h2_max_rst_stream_lifetime {
885            listener.h2_max_rst_stream_lifetime = Some(v);
886        }
887        if let Some(v) = patch.h2_max_rst_stream_abusive_lifetime {
888            listener.h2_max_rst_stream_abusive_lifetime = Some(v);
889        }
890        if let Some(v) = patch.h2_max_rst_stream_emitted_lifetime {
891            listener.h2_max_rst_stream_emitted_lifetime = Some(v);
892        }
893        if let Some(v) = patch.h2_max_header_list_size {
894            listener.h2_max_header_list_size = Some(v);
895        }
896        if let Some(v) = patch.h2_max_header_table_size {
897            listener.h2_max_header_table_size = Some(v);
898        }
899        if let Some(v) = patch.h2_max_header_fields {
900            listener.h2_max_header_fields = Some(v);
901        }
902        if let Some(v) = patch.h2_stream_idle_timeout_seconds {
903            listener.h2_stream_idle_timeout_seconds = Some(v);
904        }
905        // 0 is valid for graceful_shutdown_deadline (means "wait forever")
906        if let Some(v) = patch.h2_graceful_shutdown_deadline_seconds {
907            listener.h2_graceful_shutdown_deadline_seconds = Some(v);
908        }
909        if let Some(v) = patch.h2_max_window_update_stream0_per_window {
910            listener.h2_max_window_update_stream0_per_window = Some(v);
911        }
912        if let Some(ref v) = patch.sozu_id_header {
913            validate_sozu_id_header(v)?;
914            listener.sozu_id_header = Some(v.to_owned());
915        }
916        Ok(())
917    }
918
919    /// Validate and apply a partial patch to an existing TCP listener.
920    ///
921    /// Only `Some` fields in the patch are written; `None` fields preserve the
922    /// current value. Returns `StateError::NotFound` if the address is unknown.
923    fn update_tcp_listener(&mut self, patch: &UpdateTcpListenerConfig) -> Result<(), StateError> {
924        let address: SocketAddr = patch.address.into();
925        let listener =
926            self.tcp_listeners
927                .get_mut(&address)
928                .ok_or_else(|| StateError::NotFound {
929                    kind: ObjectKind::TcpListener,
930                    id: address.to_string(),
931                })?;
932
933        if let Some(v) = patch.public_address {
934            listener.public_address = Some(v);
935        }
936        if let Some(v) = patch.expect_proxy {
937            listener.expect_proxy = v;
938        }
939        if let Some(v) = patch.front_timeout {
940            listener.front_timeout = v;
941        }
942        if let Some(v) = patch.back_timeout {
943            listener.back_timeout = v;
944        }
945        if let Some(v) = patch.connect_timeout {
946            listener.connect_timeout = v;
947        }
948        Ok(())
949    }
950
951    /// Validate and apply a partial patch to an existing UDP listener.
952    ///
953    /// Only `Some` fields in the patch are written; `None` fields preserve the
954    /// current value. Returns `StateError::NotFound` if the address is unknown.
955    fn update_udp_listener(&mut self, patch: &UpdateUdpListenerConfig) -> Result<(), StateError> {
956        let address: SocketAddr = patch.address.into();
957        let listener =
958            self.udp_listeners
959                .get_mut(&address)
960                .ok_or_else(|| StateError::NotFound {
961                    kind: ObjectKind::UdpListener,
962                    id: address.to_string(),
963                })?;
964
965        if let Some(v) = patch.public_address {
966            listener.public_address = Some(v);
967        }
968        if let Some(v) = patch.front_timeout {
969            listener.front_timeout = v;
970        }
971        if let Some(v) = patch.back_timeout {
972            listener.back_timeout = v;
973        }
974        if let Some(v) = patch.max_rx_datagram_size {
975            listener.max_rx_datagram_size = v;
976        }
977        if let Some(v) = patch.max_flows {
978            listener.max_flows = v;
979        }
980        Ok(())
981    }
982
983    fn activate_listener(&mut self, activate: &ActivateListener) -> Result<(), StateError> {
984        match ListenerType::try_from(activate.proxy).map_err(StateError::WrongFieldValue)? {
985            ListenerType::Http => self
986                .http_listeners
987                .get_mut(&activate.address.into())
988                .map(|listener| listener.active = true)
989                .ok_or(StateError::NotFound {
990                    kind: ObjectKind::HttpListener,
991                    id: activate.address.to_string(),
992                }),
993            ListenerType::Https => self
994                .https_listeners
995                .get_mut(&activate.address.into())
996                .map(|listener| listener.active = true)
997                .ok_or(StateError::NotFound {
998                    kind: ObjectKind::HttpsListener,
999                    id: activate.address.to_string(),
1000                }),
1001            ListenerType::Tcp => self
1002                .tcp_listeners
1003                .get_mut(&activate.address.into())
1004                .map(|listener| listener.active = true)
1005                .ok_or(StateError::NotFound {
1006                    kind: ObjectKind::TcpListener,
1007                    id: activate.address.to_string(),
1008                }),
1009            ListenerType::Udp => self
1010                .udp_listeners
1011                .get_mut(&activate.address.into())
1012                .map(|listener| listener.active = true)
1013                .ok_or(StateError::NotFound {
1014                    kind: ObjectKind::UdpListener,
1015                    id: activate.address.to_string(),
1016                }),
1017        }
1018    }
1019
1020    fn deactivate_listener(&mut self, deactivate: &DeactivateListener) -> Result<(), StateError> {
1021        match ListenerType::try_from(deactivate.proxy).map_err(StateError::WrongFieldValue)? {
1022            ListenerType::Http => self
1023                .http_listeners
1024                .get_mut(&deactivate.address.into())
1025                .map(|listener| listener.active = false)
1026                .ok_or(StateError::NotFound {
1027                    kind: ObjectKind::HttpListener,
1028                    id: deactivate.address.to_string(),
1029                }),
1030            ListenerType::Https => self
1031                .https_listeners
1032                .get_mut(&deactivate.address.into())
1033                .map(|listener| listener.active = false)
1034                .ok_or(StateError::NotFound {
1035                    kind: ObjectKind::HttpsListener,
1036                    id: deactivate.address.to_string(),
1037                }),
1038            ListenerType::Tcp => self
1039                .tcp_listeners
1040                .get_mut(&deactivate.address.into())
1041                .map(|listener| listener.active = false)
1042                .ok_or(StateError::NotFound {
1043                    kind: ObjectKind::TcpListener,
1044                    id: deactivate.address.to_string(),
1045                }),
1046            ListenerType::Udp => self
1047                .udp_listeners
1048                .get_mut(&deactivate.address.into())
1049                .map(|listener| listener.active = false)
1050                .ok_or(StateError::NotFound {
1051                    kind: ObjectKind::UdpListener,
1052                    id: deactivate.address.to_string(),
1053                }),
1054        }
1055    }
1056
1057    fn add_http_frontend(&mut self, front: &RequestHttpFrontend) -> Result<(), StateError> {
1058        let front_as_key = front.to_string();
1059        let before = self.http_fronts.len();
1060
1061        match self.http_fronts.entry(front.to_string()) {
1062            BTreeMapEntry::Vacant(e) => {
1063                e.insert(front.clone().to_frontend().map_err(|into_error| {
1064                    StateError::FrontendConversion {
1065                        frontend: front_as_key,
1066                        error: into_error.to_string(),
1067                    }
1068                })?)
1069            }
1070            BTreeMapEntry::Occupied(_) => {
1071                debug_assert_eq!(
1072                    self.http_fronts.len(),
1073                    before,
1074                    "a rejected duplicate add_http_frontend must not mutate the map"
1075                );
1076                return Err(StateError::Exists {
1077                    kind: ObjectKind::HttpFrontend,
1078                    id: front.to_string(),
1079                });
1080            }
1081        };
1082        // On the conversion-error path the `?` already returned, so reaching
1083        // here means exactly one entry was inserted under the route key.
1084        debug_assert!(
1085            self.http_fronts.contains_key(&front.to_string()),
1086            "add_http_frontend must insert the route key on success"
1087        );
1088        debug_assert_eq!(
1089            self.http_fronts.len(),
1090            before + 1,
1091            "add_http_frontend inserts exactly one entry on success"
1092        );
1093        Ok(())
1094    }
1095
1096    fn add_https_frontend(&mut self, front: &RequestHttpFrontend) -> Result<(), StateError> {
1097        let front_as_key = front.to_string();
1098        let before = self.https_fronts.len();
1099
1100        match self.https_fronts.entry(front.to_string()) {
1101            BTreeMapEntry::Vacant(e) => {
1102                e.insert(front.clone().to_frontend().map_err(|into_error| {
1103                    StateError::FrontendConversion {
1104                        frontend: front_as_key,
1105                        error: into_error.to_string(),
1106                    }
1107                })?)
1108            }
1109            BTreeMapEntry::Occupied(_) => {
1110                debug_assert_eq!(
1111                    self.https_fronts.len(),
1112                    before,
1113                    "a rejected duplicate add_https_frontend must not mutate the map"
1114                );
1115                return Err(StateError::Exists {
1116                    kind: ObjectKind::HttpsFrontend,
1117                    id: front.to_string(),
1118                });
1119            }
1120        };
1121        debug_assert!(
1122            self.https_fronts.contains_key(&front.to_string()),
1123            "add_https_frontend must insert the route key on success"
1124        );
1125        debug_assert_eq!(
1126            self.https_fronts.len(),
1127            before + 1,
1128            "add_https_frontend inserts exactly one entry on success"
1129        );
1130        Ok(())
1131    }
1132
1133    fn remove_http_frontend(&mut self, front: &RequestHttpFrontend) -> Result<(), StateError> {
1134        let key = front.to_string();
1135        let before = self.http_fronts.len();
1136        self.http_fronts.remove(&key).ok_or(StateError::NotFound {
1137            kind: ObjectKind::HttpFrontend,
1138            id: front.to_string(),
1139        })?;
1140        debug_assert!(
1141            !self.http_fronts.contains_key(&key),
1142            "remove_http_frontend must evict the route key"
1143        );
1144        debug_assert_eq!(
1145            self.http_fronts.len(),
1146            before - 1,
1147            "remove_http_frontend drops exactly one entry"
1148        );
1149        Ok(())
1150    }
1151
1152    fn remove_https_frontend(&mut self, front: &RequestHttpFrontend) -> Result<(), StateError> {
1153        let key = front.to_string();
1154        let before = self.https_fronts.len();
1155        self.https_fronts.remove(&key).ok_or(StateError::NotFound {
1156            kind: ObjectKind::HttpsFrontend,
1157            id: front.to_string(),
1158        })?;
1159        debug_assert!(
1160            !self.https_fronts.contains_key(&key),
1161            "remove_https_frontend must evict the route key"
1162        );
1163        debug_assert_eq!(
1164            self.https_fronts.len(),
1165            before - 1,
1166            "remove_https_frontend drops exactly one entry"
1167        );
1168        Ok(())
1169    }
1170
1171    fn add_certificate(&mut self, add: &AddCertificate) -> Result<(), StateError> {
1172        let fingerprint = add
1173            .certificate
1174            .fingerprint()
1175            .map_err(StateError::AddCertificate)?;
1176
1177        let entry = self.certificates.entry(add.address.into()).or_default();
1178
1179        let mut add = add.clone();
1180        add.certificate
1181            .apply_overriding_names()
1182            .map_err(StateError::AddCertificate)?;
1183
1184        if entry.contains_key(&fingerprint) {
1185            let names_bytes = add
1186                .certificate
1187                .names
1188                .iter()
1189                .fold(0usize, |total, name| total.saturating_add(name.len()));
1190            info!(
1191                "Skip loading certificate with fingerprint_bytes={} names_count={} names_bytes={} on listener '{}', the certificate is already present.",
1192                fingerprint.0.len(),
1193                add.certificate.names.len(),
1194                names_bytes,
1195                add.address
1196            );
1197            return Ok(());
1198        }
1199
1200        let before = entry.len();
1201        entry.insert(fingerprint.clone(), add.certificate);
1202        debug_assert!(
1203            entry.contains_key(&fingerprint),
1204            "add_certificate must insert the fingerprint under its address"
1205        );
1206        debug_assert_eq!(
1207            entry.len(),
1208            before + 1,
1209            "add_certificate inserts exactly one fingerprint on the new path"
1210        );
1211        Ok(())
1212    }
1213
1214    fn remove_certificate(&mut self, remove: &RemoveCertificate) -> Result<(), StateError> {
1215        let fingerprint = Fingerprint(
1216            hex::decode(&remove.fingerprint)
1217                .map_err(|decode_error| StateError::RemoveCertificate(decode_error.to_string()))?,
1218        );
1219
1220        if let Some(index) = self.certificates.get_mut(&remove.address.into()) {
1221            index.remove(&fingerprint);
1222            debug_assert!(
1223                !index.contains_key(&fingerprint),
1224                "remove_certificate must evict the fingerprint when the address is known"
1225            );
1226        }
1227
1228        Ok(())
1229    }
1230
1231    /// - Remove old certificate from certificates, using the old fingerprint
1232    /// - calculate the new fingerprint
1233    /// - insert the new certificate with the new fingerprint as key
1234    /// - check that the new entry is present in the certificates hashmap
1235    fn replace_certificate(&mut self, replace: &ReplaceCertificate) -> Result<(), StateError> {
1236        let replace_address = replace.address.into();
1237        let old_fingerprint = Fingerprint(
1238            hex::decode(&replace.old_fingerprint)
1239                .map_err(|decode_error| StateError::RemoveCertificate(decode_error.to_string()))?,
1240        );
1241
1242        self.certificates
1243            .get_mut(&replace_address)
1244            .ok_or(StateError::NotFound {
1245                kind: ObjectKind::Certificate,
1246                id: replace.address.to_string(),
1247            })?
1248            .remove(&old_fingerprint);
1249
1250        let new_fingerprint = Fingerprint(
1251            calculate_fingerprint(replace.new_certificate.certificate.as_bytes()).map_err(
1252                |fingerprint_err| StateError::ReplaceCertificate(fingerprint_err.to_string()),
1253            )?,
1254        );
1255
1256        self.certificates
1257            .get_mut(&replace_address)
1258            .map(|certs| certs.insert(new_fingerprint.clone(), replace.new_certificate.clone()));
1259
1260        if !self
1261            .certificates
1262            .get(&replace_address)
1263            .ok_or(StateError::ReplaceCertificate(
1264                "Unlikely error. This entry in the certificate hashmap should be present"
1265                    .to_string(),
1266            ))?
1267            .contains_key(&new_fingerprint)
1268        {
1269            return Err(StateError::ReplaceCertificate(format!(
1270                "Failed to insert the new certificate for address {}",
1271                replace.address
1272            )));
1273        }
1274        // Postcondition: the new fingerprint is keyed under the address, and
1275        // (unless old and new collide, e.g. a self-replace) the old one is gone.
1276        debug_assert!(
1277            self.certificates
1278                .get(&replace_address)
1279                .is_some_and(|certs| certs.contains_key(&new_fingerprint)),
1280            "replace_certificate must leave the new fingerprint present"
1281        );
1282        debug_assert!(
1283            new_fingerprint == old_fingerprint
1284                || self
1285                    .certificates
1286                    .get(&replace_address)
1287                    .is_none_or(|certs| !certs.contains_key(&old_fingerprint)),
1288            "replace_certificate must evict the old fingerprint unless it equals the new one"
1289        );
1290        Ok(())
1291    }
1292
1293    /// Admission control mirrors the worker's `validate_new_tcp_front`
1294    /// (`lib/src/tcp.rs`) so the master never persists a route its own
1295    /// workers reject on the next fan-out (sozu-proxy/sozu#1290): the
1296    /// command server mutates master state before dispatching to workers,
1297    /// and a worker NACK never rolls the master back. Every check below
1298    /// runs, and every canonicalization is computed, BEFORE `self.tcp_fronts`
1299    /// is touched -- an early `Err` return is a true no-op on `self`.
1300    fn add_tcp_frontend(&mut self, front: &RequestTcpFrontend) -> Result<(), StateError> {
1301        let address: SocketAddr = front.address.into();
1302
1303        // Canonicalize first: shape-validate and lowercase `sni` through the
1304        // SAME `validate_sni_pattern` config-load and the worker use, and
1305        // reduce `alpn` to its canonical sorted+deduped set. Both canonical
1306        // forms are what gets stored (so the state's own hashing/diffing/
1307        // `generate_requests` replay sees one identity per route,
1308        // independent of how the caller cased/ordered its request) AND what
1309        // every admission check below compares against.
1310        let normalized_sni = match &front.sni {
1311            Some(sni) => Some(validate_sni_pattern(sni).map_err(|config_error| {
1312                StateError::InvalidTcpFrontend {
1313                    address,
1314                    reason: format!("sni {sni:?} failed SNI shape validation: {config_error}"),
1315                }
1316            })?),
1317            None => None,
1318        };
1319        let alpn = canonical_tcp_alpn(&front.alpn);
1320
1321        // Worker parity (`validate_new_tcp_front`'s first no-SNI check):
1322        // alpn only matches within an SNI-scoped preread, so a frontend
1323        // without sni would silently ignore its alpn list. Config-load's
1324        // `to_tcp_front` already rejects this shape; a raw `AddTcpFrontend`
1325        // over the command socket must not slip past the master either.
1326        if normalized_sni.is_none() && !alpn.is_empty() {
1327            return Err(StateError::InvalidTcpFrontend {
1328                address,
1329                reason: format!(
1330                    "alpn = {alpn:?} set without sni: alpn only matches within an SNI-scoped \
1331                     preread, so a frontend without sni would silently ignore its alpn list"
1332                ),
1333            });
1334        }
1335
1336        let total_before = self.count_tcp_frontends_raw();
1337
1338        // Listener-wide scan across ALL clusters' buckets at this address --
1339        // the previous check only walked `front.cluster_id`'s own bucket, so
1340        // a duplicate identity, a no-SNI/SNI mix, a second catch-all, or an
1341        // overlapping ALPN set under a DIFFERENT cluster_id all used to pass
1342        // here and then fail on every worker.
1343        let mut listener_has_no_sni_front = false;
1344        let mut listener_has_sni_front = false;
1345        for existing in self
1346            .tcp_fronts
1347            .values()
1348            .flatten()
1349            .filter(|existing| existing.address == address)
1350        {
1351            if tcp_frontend_matches(existing, address, &normalized_sni, &alpn) {
1352                return Err(StateError::Exists {
1353                    kind: ObjectKind::TcpFrontend,
1354                    id: format!(
1355                        "TcpFrontend {{ address: {address}, sni: {normalized_sni:?}, \
1356                         alpn: {alpn:?} }}"
1357                    ),
1358                });
1359            }
1360            match &existing.sni {
1361                None => listener_has_no_sni_front = true,
1362                Some(existing_sni) => {
1363                    listener_has_sni_front = true;
1364                    // Catch-all / ALPN-overlap only matter between two
1365                    // SNI-scoped fronts sharing the exact same normalized
1366                    // sni -- exact and wildcard patterns are DISTINCT keys
1367                    // here, never matched against each other (no trie
1368                    // matching in `ConfigState`), mirroring the worker's own
1369                    // `accept_wildcard: false` self-lookup.
1370                    if normalized_sni.as_deref() != Some(existing_sni.as_str()) {
1371                        continue;
1372                    }
1373                    let existing_is_catch_all = existing.alpn.is_empty();
1374                    let new_is_catch_all = alpn.is_empty();
1375                    if existing_is_catch_all && new_is_catch_all {
1376                        return Err(StateError::InvalidTcpFrontend {
1377                            address,
1378                            reason: format!(
1379                                "sni {existing_sni:?} already has a catch-all (empty alpn) \
1380                                 frontend: at most one frontend per (address, sni) may omit \
1381                                 alpn"
1382                            ),
1383                        });
1384                    }
1385                    // A catch-all coexisting with an ALPN-scoped sibling on
1386                    // the same sni is legal (worker parity: `AlpnMatcher::Any`
1387                    // only conflicts with another `Any`, never with a
1388                    // `OneOf`) -- only two ALPN-scoped fronts can overlap.
1389                    if !existing_is_catch_all
1390                        && !new_is_catch_all
1391                        && let Some(overlap) = alpn
1392                            .iter()
1393                            .find(|protocol| existing.alpn.contains(protocol))
1394                    {
1395                        return Err(StateError::InvalidTcpFrontend {
1396                            address,
1397                            reason: format!(
1398                                "sni {existing_sni:?} already has a frontend matching ALPN \
1399                                 protocol {overlap:?}: ALPN matchers for the same (address, \
1400                                 sni) must not overlap"
1401                            ),
1402                        });
1403                    }
1404                }
1405            }
1406        }
1407        match &normalized_sni {
1408            None if listener_has_sni_front => {
1409                return Err(StateError::InvalidTcpFrontend {
1410                    address,
1411                    reason: "a no-SNI frontend cannot be added to a listener that already has \
1412                             SNI-scoped routes"
1413                        .to_string(),
1414                });
1415            }
1416            Some(_) if listener_has_no_sni_front => {
1417                return Err(StateError::InvalidTcpFrontend {
1418                    address,
1419                    reason: "an SNI-scoped frontend cannot be added to a listener that already \
1420                             has a no-SNI catch-all cluster"
1421                        .to_string(),
1422                });
1423            }
1424            _ => {}
1425        }
1426        // POST: every admission check above either returned `Err` or fell
1427        // through -- none of them may mutate `self`, so the total frontend
1428        // count observed before the scan must still hold at this point.
1429        debug_assert_eq!(
1430            self.count_tcp_frontends_raw(),
1431            total_before,
1432            "add_tcp_frontend must not mutate tcp_fronts before every admission check has \
1433             passed"
1434        );
1435
1436        let tcp_frontend = TcpFrontend {
1437            cluster_id: front.cluster_id.clone(),
1438            address,
1439            tags: front.tags.clone(),
1440            sni: normalized_sni,
1441            alpn,
1442        };
1443        let tcp_frontends = self.tcp_fronts.entry(front.cluster_id.clone()).or_default();
1444        let before = tcp_frontends.len();
1445        debug_assert_eq!(
1446            tcp_frontend.cluster_id, front.cluster_id,
1447            "the built frontend must carry its bucket's cluster_id"
1448        );
1449        tcp_frontends.push(tcp_frontend);
1450        debug_assert_eq!(
1451            tcp_frontends.len(),
1452            before + 1,
1453            "add_tcp_frontend appends exactly one entry"
1454        );
1455        Ok(())
1456    }
1457
1458    fn remove_tcp_frontend(
1459        &mut self,
1460        front_to_remove: &RequestTcpFrontend,
1461    ) -> Result<(), StateError> {
1462        let tcp_frontends =
1463            self.tcp_fronts
1464                .get_mut(&front_to_remove.cluster_id)
1465                .ok_or(StateError::NotFound {
1466                    kind: ObjectKind::TcpFrontend,
1467                    id: format!(
1468                        "RequestTcpFrontend {{ cluster_id: {:?}, address: {:?}, tags: {:?}, sni: {:?}, alpn: {:?} }}",
1469                        front_to_remove.cluster_id,
1470                        front_to_remove.address,
1471                        front_to_remove.tags,
1472                        front_to_remove.sni,
1473                        front_to_remove.alpn,
1474                    ),
1475                })?;
1476
1477        let len = tcp_frontends.len();
1478        let remove_address: SocketAddr = front_to_remove.address.into();
1479        // Canonicalize for matching, mirroring add_tcp_frontend and the
1480        // worker: lowercase `sni` (the SNI-preread core normalizes to
1481        // lowercase before ever looking a route up, and `TcpFrontend.sni` is
1482        // stored in that same normalized form) and reduce `alpn` to its
1483        // canonical sorted+deduped set, so a request differing only in sni
1484        // case or alpn order still matches the stored frontend instead of
1485        // falling through to `NoChange`. Shape is intentionally NOT
1486        // re-validated here: a malformed pattern could never have been
1487        // stored post-fix, so it simply matches nothing below.
1488        let remove_sni = front_to_remove.sni.as_deref().map(str::to_ascii_lowercase);
1489        let remove_alpn = canonical_tcp_alpn(&front_to_remove.alpn);
1490        // INV: removal identity mirrors add_tcp_frontend's (address, sni,
1491        // alpn) key — removing one SNI-scoped frontend on a listener must
1492        // not also evict a sibling frontend at the same address with a
1493        // different sni/alpn.
1494        tcp_frontends.retain(|front| {
1495            !tcp_frontend_matches(front, remove_address, &remove_sni, &remove_alpn)
1496        });
1497        let after = tcp_frontends.len();
1498        if after == len {
1499            return Err(StateError::NoChange);
1500        }
1501        // `retain` may drop more than one entry only if duplicates on the same
1502        // (address, sni, alpn) ever existed; `add_tcp_frontend` forbids that,
1503        // so a successful removal must drop exactly one and leave none
1504        // matching.
1505        debug_assert_eq!(
1506            after,
1507            len - 1,
1508            "remove_tcp_frontend drops exactly one entry"
1509        );
1510        debug_assert!(
1511            !tcp_frontends.iter().any(|f| tcp_frontend_matches(
1512                f,
1513                remove_address,
1514                &remove_sni,
1515                &remove_alpn
1516            )),
1517            "remove_tcp_frontend must leave no frontend matching the removed (address, sni, alpn)"
1518        );
1519        Ok(())
1520    }
1521
1522    fn add_udp_frontend(&mut self, front: &RequestUdpFrontend) -> Result<(), StateError> {
1523        let udp_frontends = self.udp_fronts.entry(front.cluster_id.clone()).or_default();
1524
1525        let udp_frontend = UdpFrontend {
1526            cluster_id: front.cluster_id.clone(),
1527            address: front.address.into(),
1528            tags: front.tags.clone(),
1529        };
1530        if udp_frontends.contains(&udp_frontend) {
1531            return Err(StateError::Exists {
1532                kind: ObjectKind::UdpFrontend,
1533                id: format!("{udp_frontend:?}"),
1534            });
1535        }
1536
1537        udp_frontends.push(udp_frontend);
1538        Ok(())
1539    }
1540
1541    fn remove_udp_frontend(
1542        &mut self,
1543        front_to_remove: &RequestUdpFrontend,
1544    ) -> Result<(), StateError> {
1545        let udp_frontends =
1546            self.udp_fronts
1547                .get_mut(&front_to_remove.cluster_id)
1548                .ok_or(StateError::NotFound {
1549                    kind: ObjectKind::UdpFrontend,
1550                    id: format!("{front_to_remove:?}"),
1551                })?;
1552
1553        let len = udp_frontends.len();
1554        udp_frontends.retain(|front| front.address != front_to_remove.address.into());
1555        if udp_frontends.len() == len {
1556            return Err(StateError::NoChange);
1557        }
1558        Ok(())
1559    }
1560
1561    fn add_backend(&mut self, add_backend: &AddBackend) -> Result<(), StateError> {
1562        let backend = Backend {
1563            address: add_backend.address.into(),
1564            cluster_id: add_backend.cluster_id.clone(),
1565            backend_id: add_backend.backend_id.clone(),
1566            sticky_id: add_backend.sticky_id.clone(),
1567            load_balancing_parameters: add_backend.load_balancing_parameters,
1568            backup: add_backend.backup,
1569        };
1570        let backends = self.backends.entry(backend.cluster_id.clone()).or_default();
1571        let backend_id = backend.backend_id.clone();
1572        let backend_address = backend.address;
1573        let before = backends.len();
1574
1575        // we might be modifying the sticky id or load balancing parameters:
1576        // the retain drops at most one prior copy (the map stays deduplicated
1577        // on (backend_id, address)), then we re-push the new version. So the
1578        // net length grows by exactly one iff this was a brand-new backend.
1579        let was_present = backends
1580            .iter()
1581            .any(|b| b.backend_id == backend_id && b.address == backend_address);
1582        backends.retain(|b| b.backend_id != backend.backend_id || b.address != backend.address);
1583        debug_assert_eq!(
1584            backends.len(),
1585            before - was_present as usize,
1586            "the upsert retain must drop exactly the prior copy iff it existed"
1587        );
1588        backends.push(backend);
1589        backends.sort();
1590
1591        debug_assert_eq!(
1592            backends.len(),
1593            before + (!was_present) as usize,
1594            "add_backend grows the bucket by one iff the backend was new"
1595        );
1596        debug_assert_eq!(
1597            backends
1598                .iter()
1599                .filter(|b| b.backend_id == backend_id && b.address == backend_address)
1600                .count(),
1601            1,
1602            "exactly one copy of the upserted backend must remain"
1603        );
1604        Ok(())
1605    }
1606
1607    fn remove_backend(&mut self, backend: &RemoveBackend) -> Result<(), StateError> {
1608        let backend_list =
1609            self.backends
1610                .get_mut(&backend.cluster_id)
1611                .ok_or(StateError::NotFound {
1612                    kind: ObjectKind::Backend,
1613                    id: backend.backend_id.to_owned(),
1614                })?;
1615
1616        let len = backend_list.len();
1617        let remove_address: SocketAddr = backend.address.into();
1618        backend_list.retain(|b| b.backend_id != backend.backend_id || b.address != remove_address);
1619        backend_list.sort();
1620        let after = backend_list.len();
1621        if after == len {
1622            return Err(StateError::NoChange);
1623        }
1624        // The list is deduplicated on (backend_id, address), so a matching
1625        // removal drops exactly one entry and leaves nothing matching.
1626        debug_assert_eq!(after, len - 1, "remove_backend drops exactly one entry");
1627        debug_assert!(
1628            !backend_list
1629                .iter()
1630                .any(|b| b.backend_id == backend.backend_id && b.address == remove_address),
1631            "remove_backend must leave no backend matching (backend_id, address)"
1632        );
1633        Ok(())
1634    }
1635
1636    /// creates all requests needed to bootstrap the state
1637    fn generate_requests(&self) -> Vec<Request> {
1638        let mut v: Vec<Request> = Vec::new();
1639
1640        for listener in self.http_listeners.values() {
1641            v.push(RequestType::AddHttpListener(listener.clone()).into());
1642            if listener.active {
1643                v.push(
1644                    RequestType::ActivateListener(ActivateListener {
1645                        address: listener.address,
1646                        proxy: ListenerType::Http.into(),
1647                        from_scm: false,
1648                    })
1649                    .into(),
1650                );
1651            }
1652        }
1653
1654        for listener in self.https_listeners.values() {
1655            v.push(RequestType::AddHttpsListener(listener.clone()).into());
1656            if listener.active {
1657                v.push(
1658                    RequestType::ActivateListener(ActivateListener {
1659                        address: listener.address,
1660                        proxy: ListenerType::Https.into(),
1661                        from_scm: false,
1662                    })
1663                    .into(),
1664                );
1665            }
1666        }
1667
1668        for listener in self.tcp_listeners.values() {
1669            v.push(RequestType::AddTcpListener(*listener).into());
1670            if listener.active {
1671                v.push(
1672                    RequestType::ActivateListener(ActivateListener {
1673                        address: listener.address,
1674                        proxy: ListenerType::Tcp.into(),
1675                        from_scm: false,
1676                    })
1677                    .into(),
1678                );
1679            }
1680        }
1681
1682        for listener in self.udp_listeners.values() {
1683            v.push(RequestType::AddUdpListener(*listener).into());
1684            if listener.active {
1685                v.push(
1686                    RequestType::ActivateListener(ActivateListener {
1687                        address: listener.address,
1688                        proxy: ListenerType::Udp.into(),
1689                        from_scm: false,
1690                    })
1691                    .into(),
1692                );
1693            }
1694        }
1695
1696        for cluster in self.clusters.values() {
1697            v.push(RequestType::AddCluster(cluster.clone()).into());
1698        }
1699
1700        for front in self.http_fronts.values() {
1701            v.push(RequestType::AddHttpFrontend(front.clone().into()).into());
1702        }
1703
1704        for (front, certs) in self.certificates.iter() {
1705            for certificate_and_key in certs.values() {
1706                v.push(
1707                    RequestType::AddCertificate(AddCertificate {
1708                        address: SocketAddress::from(*front),
1709                        certificate: certificate_and_key.clone(),
1710                        expired_at: None,
1711                    })
1712                    .into(),
1713                );
1714            }
1715        }
1716
1717        for front in self.https_fronts.values() {
1718            v.push(RequestType::AddHttpsFrontend(front.clone().into()).into());
1719        }
1720
1721        for front_list in self.tcp_fronts.values() {
1722            for front in front_list {
1723                v.push(RequestType::AddTcpFrontend(front.clone().into()).into());
1724            }
1725        }
1726
1727        for front_list in self.udp_fronts.values() {
1728            for front in front_list {
1729                v.push(RequestType::AddUdpFrontend(front.clone().into()).into());
1730            }
1731        }
1732
1733        for backend_list in self.backends.values() {
1734            for backend in backend_list {
1735                v.push(RequestType::AddBackend(backend.clone().to_add_backend()).into());
1736            }
1737        }
1738
1739        // Bootstrap round-trip: replaying `generate_requests` into a fresh,
1740        // empty `ConfigState` must reconstruct `self`'s maps exactly. This is
1741        // the property SaveState/LoadState and worker bootstrap depend on. We
1742        // compare the business maps only (not `request_counts`, which is
1743        // bookkeeping mutated by `dispatch`).
1744        #[cfg(debug_assertions)]
1745        {
1746            let mut replayed = ConfigState::new();
1747            for request in &v {
1748                debug_assert!(
1749                    replayed.dispatch(request).is_ok(),
1750                    "every request from generate_requests must replay cleanly"
1751                );
1752            }
1753            debug_assert!(
1754                replayed.clusters == self.clusters
1755                    && replayed.backends == self.backends
1756                    && replayed.http_listeners == self.http_listeners
1757                    && replayed.https_listeners == self.https_listeners
1758                    && replayed.tcp_listeners == self.tcp_listeners
1759                    && replayed.http_fronts == self.http_fronts
1760                    && replayed.https_fronts == self.https_fronts
1761                    && replayed.tcp_fronts == self.tcp_fronts
1762                    && replayed.certificates == self.certificates,
1763                "replaying generate_requests into a fresh state must reproduce self"
1764            );
1765        }
1766
1767        v
1768    }
1769
1770    pub fn generate_activate_requests(&self) -> Vec<Request> {
1771        let mut v: Vec<Request> = Vec::new();
1772        for front in self
1773            .http_listeners
1774            .iter()
1775            .filter(|(_, listener)| listener.active)
1776            .map(|(k, _)| k)
1777        {
1778            v.push(
1779                RequestType::ActivateListener(ActivateListener {
1780                    address: SocketAddress::from(*front),
1781                    proxy: ListenerType::Http.into(),
1782                    from_scm: false,
1783                })
1784                .into(),
1785            );
1786        }
1787
1788        for front in self
1789            .https_listeners
1790            .iter()
1791            .filter(|(_, listener)| listener.active)
1792            .map(|(k, _)| k)
1793        {
1794            v.push(
1795                RequestType::ActivateListener(ActivateListener {
1796                    address: SocketAddress::from(*front),
1797                    proxy: ListenerType::Https.into(),
1798                    from_scm: false,
1799                })
1800                .into(),
1801            );
1802        }
1803        for front in self
1804            .tcp_listeners
1805            .iter()
1806            .filter(|(_, listener)| listener.active)
1807            .map(|(k, _)| k)
1808        {
1809            v.push(
1810                RequestType::ActivateListener(ActivateListener {
1811                    address: SocketAddress::from(*front),
1812                    proxy: ListenerType::Tcp.into(),
1813                    from_scm: false,
1814                })
1815                .into(),
1816            );
1817        }
1818        for front in self
1819            .udp_listeners
1820            .iter()
1821            .filter(|(_, listener)| listener.active)
1822            .map(|(k, _)| k)
1823        {
1824            v.push(
1825                RequestType::ActivateListener(ActivateListener {
1826                    address: SocketAddress::from(*front),
1827                    proxy: ListenerType::Udp.into(),
1828                    from_scm: false,
1829                })
1830                .into(),
1831            );
1832        }
1833
1834        // Symmetry with the active-listener census: exactly one ActivateListener
1835        // is emitted per active listener across the four maps, no more, no less.
1836        #[cfg(debug_assertions)]
1837        {
1838            let active_listeners = self.http_listeners.values().filter(|l| l.active).count()
1839                + self.https_listeners.values().filter(|l| l.active).count()
1840                + self.tcp_listeners.values().filter(|l| l.active).count()
1841                + self.udp_listeners.values().filter(|l| l.active).count();
1842            debug_assert_eq!(
1843                v.len(),
1844                active_listeners,
1845                "generate_activate_requests emits one request per active listener"
1846            );
1847            debug_assert!(
1848                v.iter()
1849                    .all(|r| matches!(r.request_type, Some(RequestType::ActivateListener(_)))),
1850                "generate_activate_requests must emit only ActivateListener requests"
1851            );
1852        }
1853
1854        v
1855    }
1856
1857    pub fn diff(&self, other: &ConfigState) -> Vec<Request> {
1858        //pub tcp_listeners:   HashMap<SocketAddr, (TcpListener, bool)>,
1859        let my_tcp_listeners: HashSet<&SocketAddr> = self.tcp_listeners.keys().collect();
1860        let their_tcp_listeners: HashSet<&SocketAddr> = other.tcp_listeners.keys().collect();
1861        let removed_tcp_listeners = my_tcp_listeners.difference(&their_tcp_listeners);
1862        let added_tcp_listeners = their_tcp_listeners.difference(&my_tcp_listeners);
1863
1864        let my_udp_listeners: HashSet<&SocketAddr> = self.udp_listeners.keys().collect();
1865        let their_udp_listeners: HashSet<&SocketAddr> = other.udp_listeners.keys().collect();
1866        let removed_udp_listeners = my_udp_listeners.difference(&their_udp_listeners);
1867        let added_udp_listeners = their_udp_listeners.difference(&my_udp_listeners);
1868
1869        let my_http_listeners: HashSet<&SocketAddr> = self.http_listeners.keys().collect();
1870        let their_http_listeners: HashSet<&SocketAddr> = other.http_listeners.keys().collect();
1871        let removed_http_listeners = my_http_listeners.difference(&their_http_listeners);
1872        let added_http_listeners = their_http_listeners.difference(&my_http_listeners);
1873
1874        let my_https_listeners: HashSet<&SocketAddr> = self.https_listeners.keys().collect();
1875        let their_https_listeners: HashSet<&SocketAddr> = other.https_listeners.keys().collect();
1876        let removed_https_listeners = my_https_listeners.difference(&their_https_listeners);
1877        let added_https_listeners = their_https_listeners.difference(&my_https_listeners);
1878
1879        let mut v: Vec<Request> = vec![];
1880
1881        for address in removed_tcp_listeners {
1882            if self.tcp_listeners[*address].active {
1883                v.push(
1884                    RequestType::DeactivateListener(DeactivateListener {
1885                        address: SocketAddress::from(**address),
1886                        proxy: ListenerType::Tcp.into(),
1887                        to_scm: false,
1888                    })
1889                    .into(),
1890                );
1891            }
1892
1893            v.push(
1894                RequestType::RemoveListener(RemoveListener {
1895                    address: SocketAddress::from(**address),
1896                    proxy: ListenerType::Tcp.into(),
1897                })
1898                .into(),
1899            );
1900        }
1901
1902        for address in added_tcp_listeners.clone() {
1903            v.push(RequestType::AddTcpListener(other.tcp_listeners[*address]).into());
1904
1905            if other.tcp_listeners[*address].active {
1906                v.push(
1907                    RequestType::ActivateListener(ActivateListener {
1908                        address: SocketAddress::from(**address),
1909                        proxy: ListenerType::Tcp.into(),
1910                        from_scm: false,
1911                    })
1912                    .into(),
1913                );
1914            }
1915        }
1916
1917        for address in removed_udp_listeners {
1918            if self.udp_listeners[*address].active {
1919                v.push(
1920                    RequestType::DeactivateListener(DeactivateListener {
1921                        address: SocketAddress::from(**address),
1922                        proxy: ListenerType::Udp.into(),
1923                        to_scm: false,
1924                    })
1925                    .into(),
1926                );
1927            }
1928
1929            v.push(
1930                RequestType::RemoveListener(RemoveListener {
1931                    address: SocketAddress::from(**address),
1932                    proxy: ListenerType::Udp.into(),
1933                })
1934                .into(),
1935            );
1936        }
1937
1938        for address in added_udp_listeners.clone() {
1939            v.push(RequestType::AddUdpListener(other.udp_listeners[*address]).into());
1940
1941            if other.udp_listeners[*address].active {
1942                v.push(
1943                    RequestType::ActivateListener(ActivateListener {
1944                        address: SocketAddress::from(**address),
1945                        proxy: ListenerType::Udp.into(),
1946                        from_scm: false,
1947                    })
1948                    .into(),
1949                );
1950            }
1951        }
1952
1953        for address in removed_http_listeners {
1954            if self.http_listeners[*address].active {
1955                v.push(
1956                    RequestType::DeactivateListener(DeactivateListener {
1957                        address: SocketAddress::from(**address),
1958                        proxy: ListenerType::Http.into(),
1959                        to_scm: false,
1960                    })
1961                    .into(),
1962                );
1963            }
1964
1965            v.push(
1966                RequestType::RemoveListener(RemoveListener {
1967                    address: SocketAddress::from(**address),
1968                    proxy: ListenerType::Http.into(),
1969                })
1970                .into(),
1971            );
1972        }
1973
1974        for address in added_http_listeners.clone() {
1975            v.push(RequestType::AddHttpListener(other.http_listeners[*address].clone()).into());
1976
1977            if other.http_listeners[*address].active {
1978                v.push(
1979                    RequestType::ActivateListener(ActivateListener {
1980                        address: SocketAddress::from(**address),
1981                        proxy: ListenerType::Http.into(),
1982                        from_scm: false,
1983                    })
1984                    .into(),
1985                );
1986            }
1987        }
1988
1989        for address in removed_https_listeners {
1990            if self.https_listeners[*address].active {
1991                v.push(
1992                    RequestType::DeactivateListener(DeactivateListener {
1993                        address: SocketAddress::from(**address),
1994                        proxy: ListenerType::Https.into(),
1995                        to_scm: false,
1996                    })
1997                    .into(),
1998                );
1999            }
2000
2001            v.push(
2002                RequestType::RemoveListener(RemoveListener {
2003                    address: SocketAddress::from(**address),
2004                    proxy: ListenerType::Https.into(),
2005                })
2006                .into(),
2007            );
2008        }
2009
2010        for address in added_https_listeners.clone() {
2011            v.push(RequestType::AddHttpsListener(other.https_listeners[*address].clone()).into());
2012
2013            if other.https_listeners[*address].active {
2014                v.push(
2015                    RequestType::ActivateListener(ActivateListener {
2016                        address: SocketAddress::from(**address),
2017                        proxy: ListenerType::Https.into(),
2018                        from_scm: false,
2019                    })
2020                    .into(),
2021                );
2022            }
2023        }
2024
2025        for addr in my_tcp_listeners.intersection(&their_tcp_listeners) {
2026            let my_listener = &self.tcp_listeners[*addr];
2027            let their_listener = &other.tcp_listeners[*addr];
2028
2029            if my_listener != their_listener {
2030                v.push(
2031                    RequestType::RemoveListener(RemoveListener {
2032                        address: SocketAddress::from(**addr),
2033                        proxy: ListenerType::Tcp.into(),
2034                    })
2035                    .into(),
2036                );
2037                // any added listener should be unactive
2038                let mut listener_to_add = *their_listener;
2039                listener_to_add.active = false;
2040                v.push(RequestType::AddTcpListener(listener_to_add).into());
2041
2042                // The Remove + Add(active=false) above wipes the listener's
2043                // active state. Re-emit an ActivateListener whenever the target
2044                // state keeps it active, otherwise a config change on a still
2045                // active listener would silently deactivate it on replay. This
2046                // subsumes the newly-active (`!my.active && their.active`) case:
2047                // a differing `active` flag always makes the listeners unequal.
2048                if their_listener.active {
2049                    v.push(
2050                        RequestType::ActivateListener(ActivateListener {
2051                            address: SocketAddress::from(**addr),
2052                            proxy: ListenerType::Tcp.into(),
2053                            from_scm: false,
2054                        })
2055                        .into(),
2056                    );
2057                }
2058            }
2059
2060            if my_listener.active && !their_listener.active {
2061                v.push(
2062                    RequestType::DeactivateListener(DeactivateListener {
2063                        address: SocketAddress::from(**addr),
2064                        proxy: ListenerType::Tcp.into(),
2065                        to_scm: false,
2066                    })
2067                    .into(),
2068                );
2069            }
2070        }
2071
2072        for addr in my_udp_listeners.intersection(&their_udp_listeners) {
2073            let my_listener = &self.udp_listeners[*addr];
2074            let their_listener = &other.udp_listeners[*addr];
2075
2076            if my_listener != their_listener {
2077                v.push(
2078                    RequestType::RemoveListener(RemoveListener {
2079                        address: SocketAddress::from(**addr),
2080                        proxy: ListenerType::Udp.into(),
2081                    })
2082                    .into(),
2083                );
2084                // any added listener should be unactive
2085                let mut listener_to_add = *their_listener;
2086                listener_to_add.active = false;
2087                v.push(RequestType::AddUdpListener(listener_to_add).into());
2088
2089                // The Remove + Add(active=false) above wipes the listener's
2090                // active state. Re-emit an ActivateListener whenever the target
2091                // state keeps it active, otherwise a config change on a still
2092                // active listener would silently deactivate it on replay. This
2093                // subsumes the newly-active (`!my.active && their.active`) case:
2094                // a differing `active` flag always makes the listeners unequal.
2095                if their_listener.active {
2096                    v.push(
2097                        RequestType::ActivateListener(ActivateListener {
2098                            address: SocketAddress::from(**addr),
2099                            proxy: ListenerType::Udp.into(),
2100                            from_scm: false,
2101                        })
2102                        .into(),
2103                    );
2104                }
2105            }
2106
2107            if my_listener.active && !their_listener.active {
2108                v.push(
2109                    RequestType::DeactivateListener(DeactivateListener {
2110                        address: SocketAddress::from(**addr),
2111                        proxy: ListenerType::Udp.into(),
2112                        to_scm: false,
2113                    })
2114                    .into(),
2115                );
2116            }
2117        }
2118
2119        for addr in my_http_listeners.intersection(&their_http_listeners) {
2120            let my_listener = &self.http_listeners[*addr];
2121            let their_listener = &other.http_listeners[*addr];
2122
2123            if my_listener != their_listener {
2124                v.push(
2125                    RequestType::RemoveListener(RemoveListener {
2126                        address: SocketAddress::from(**addr),
2127                        proxy: ListenerType::Http.into(),
2128                    })
2129                    .into(),
2130                );
2131                // any added listener should be unactive
2132                let mut listener_to_add = their_listener.clone();
2133                listener_to_add.active = false;
2134                v.push(RequestType::AddHttpListener(listener_to_add).into());
2135
2136                // The Remove + Add(active=false) above wipes the listener's
2137                // active state. Re-emit an ActivateListener whenever the target
2138                // state keeps it active, otherwise a config change on a still
2139                // active listener would silently deactivate it on replay. This
2140                // subsumes the newly-active (`!my.active && their.active`) case:
2141                // a differing `active` flag always makes the listeners unequal.
2142                if their_listener.active {
2143                    v.push(
2144                        RequestType::ActivateListener(ActivateListener {
2145                            address: SocketAddress::from(**addr),
2146                            proxy: ListenerType::Http.into(),
2147                            from_scm: false,
2148                        })
2149                        .into(),
2150                    );
2151                }
2152            }
2153
2154            if my_listener.active && !their_listener.active {
2155                v.push(
2156                    RequestType::DeactivateListener(DeactivateListener {
2157                        address: SocketAddress::from(**addr),
2158                        proxy: ListenerType::Http.into(),
2159                        to_scm: false,
2160                    })
2161                    .into(),
2162                );
2163            }
2164        }
2165
2166        for addr in my_https_listeners.intersection(&their_https_listeners) {
2167            let my_listener = &self.https_listeners[*addr];
2168            let their_listener = &other.https_listeners[*addr];
2169
2170            if my_listener != their_listener {
2171                v.push(
2172                    RequestType::RemoveListener(RemoveListener {
2173                        address: SocketAddress::from(**addr),
2174                        proxy: ListenerType::Https.into(),
2175                    })
2176                    .into(),
2177                );
2178                // any added listener should be unactive
2179                let mut listener_to_add = their_listener.clone();
2180                listener_to_add.active = false;
2181                v.push(RequestType::AddHttpsListener(listener_to_add).into());
2182
2183                // The Remove + Add(active=false) above wipes the listener's
2184                // active state. Re-emit an ActivateListener whenever the target
2185                // state keeps it active, otherwise a config change on a still
2186                // active listener would silently deactivate it on replay. This
2187                // subsumes the newly-active (`!my.active && their.active`) case:
2188                // a differing `active` flag always makes the listeners unequal.
2189                if their_listener.active {
2190                    v.push(
2191                        RequestType::ActivateListener(ActivateListener {
2192                            address: SocketAddress::from(**addr),
2193                            proxy: ListenerType::Https.into(),
2194                            from_scm: false,
2195                        })
2196                        .into(),
2197                    );
2198                }
2199            }
2200
2201            if my_listener.active && !their_listener.active {
2202                v.push(
2203                    RequestType::DeactivateListener(DeactivateListener {
2204                        address: SocketAddress::from(**addr),
2205                        proxy: ListenerType::Https.into(),
2206                        to_scm: false,
2207                    })
2208                    .into(),
2209                );
2210            }
2211        }
2212
2213        for (cluster_id, res) in diff_map(self.clusters.iter(), other.clusters.iter()) {
2214            match res {
2215                DiffResult::Added | DiffResult::Changed => v.push(
2216                    RequestType::AddCluster(other.clusters.get(cluster_id).unwrap().clone()).into(),
2217                ),
2218                DiffResult::Removed => {
2219                    v.push(RequestType::RemoveCluster(cluster_id.to_string()).into())
2220                }
2221            }
2222        }
2223
2224        for ((cluster_id, backend_id), res) in diff_map(
2225            self.backends.iter().flat_map(|(cluster_id, v)| {
2226                v.iter()
2227                    .map(move |backend| ((cluster_id, &backend.backend_id), backend))
2228            }),
2229            other.backends.iter().flat_map(|(cluster_id, v)| {
2230                v.iter()
2231                    .map(move |backend| ((cluster_id, &backend.backend_id), backend))
2232            }),
2233        ) {
2234            match res {
2235                DiffResult::Added => {
2236                    let backend = other
2237                        .backends
2238                        .get(cluster_id)
2239                        .and_then(|v| v.iter().find(|b| &b.backend_id == backend_id))
2240                        .unwrap();
2241                    v.push(RequestType::AddBackend(backend.clone().to_add_backend()).into());
2242                }
2243                DiffResult::Removed => {
2244                    let backend = self
2245                        .backends
2246                        .get(cluster_id)
2247                        .and_then(|v| v.iter().find(|b| &b.backend_id == backend_id))
2248                        .unwrap();
2249
2250                    v.push(
2251                        RequestType::RemoveBackend(RemoveBackend {
2252                            cluster_id: backend.cluster_id.clone(),
2253                            backend_id: backend.backend_id.clone(),
2254                            address: SocketAddress::from(backend.address),
2255                        })
2256                        .into(),
2257                    );
2258                }
2259                DiffResult::Changed => {
2260                    let backend = self
2261                        .backends
2262                        .get(cluster_id)
2263                        .and_then(|v| v.iter().find(|b| &b.backend_id == backend_id))
2264                        .unwrap();
2265
2266                    v.push(
2267                        RequestType::RemoveBackend(RemoveBackend {
2268                            cluster_id: backend.cluster_id.clone(),
2269                            backend_id: backend.backend_id.clone(),
2270                            address: SocketAddress::from(backend.address),
2271                        })
2272                        .into(),
2273                    );
2274
2275                    let backend = other
2276                        .backends
2277                        .get(cluster_id)
2278                        .and_then(|v| v.iter().find(|b| &b.backend_id == backend_id))
2279                        .unwrap();
2280                    v.push(RequestType::AddBackend(backend.clone().to_add_backend()).into());
2281                }
2282            }
2283        }
2284
2285        let mut my_http_fronts: HashSet<(&str, &HttpFrontend)> = HashSet::new();
2286        for (route, front) in self.http_fronts.iter() {
2287            my_http_fronts.insert((route, front));
2288        }
2289        let mut their_http_fronts: HashSet<(&str, &HttpFrontend)> = HashSet::new();
2290        for (route, front) in other.http_fronts.iter() {
2291            their_http_fronts.insert((route, front));
2292        }
2293
2294        let removed_http_fronts = my_http_fronts.difference(&their_http_fronts);
2295        let added_http_fronts = their_http_fronts.difference(&my_http_fronts);
2296
2297        for &(_, front) in removed_http_fronts {
2298            v.push(RequestType::RemoveHttpFrontend(front.clone().into()).into());
2299        }
2300
2301        for &(_, front) in added_http_fronts {
2302            v.push(RequestType::AddHttpFrontend(front.clone().into()).into());
2303        }
2304
2305        let mut my_https_fronts: HashSet<(&String, &HttpFrontend)> = HashSet::new();
2306        for (route, front) in self.https_fronts.iter() {
2307            my_https_fronts.insert((route, front));
2308        }
2309        let mut their_https_fronts: HashSet<(&String, &HttpFrontend)> = HashSet::new();
2310        for (route, front) in other.https_fronts.iter() {
2311            their_https_fronts.insert((route, front));
2312        }
2313        let removed_https_fronts = my_https_fronts.difference(&their_https_fronts);
2314        let added_https_fronts = their_https_fronts.difference(&my_https_fronts);
2315
2316        for &(_, front) in removed_https_fronts {
2317            v.push(RequestType::RemoveHttpsFrontend(front.clone().into()).into());
2318        }
2319
2320        for &(_, front) in added_https_fronts {
2321            v.push(RequestType::AddHttpsFrontend(front.clone().into()).into());
2322        }
2323
2324        let mut my_tcp_fronts: HashSet<(&ClusterId, &TcpFrontend)> = HashSet::new();
2325        for (cluster_id, front_list) in self.tcp_fronts.iter() {
2326            for front in front_list.iter() {
2327                my_tcp_fronts.insert((cluster_id, front));
2328            }
2329        }
2330        let mut their_tcp_fronts: HashSet<(&ClusterId, &TcpFrontend)> = HashSet::new();
2331        for (cluster_id, front_list) in other.tcp_fronts.iter() {
2332            for front in front_list.iter() {
2333                their_tcp_fronts.insert((cluster_id, front));
2334            }
2335        }
2336
2337        let removed_tcp_fronts = my_tcp_fronts.difference(&their_tcp_fronts);
2338        let added_tcp_fronts = their_tcp_fronts.difference(&my_tcp_fronts);
2339
2340        for &(_, front) in removed_tcp_fronts {
2341            v.push(RequestType::RemoveTcpFrontend(front.clone().into()).into());
2342        }
2343
2344        for &(_, front) in added_tcp_fronts {
2345            v.push(RequestType::AddTcpFrontend(front.clone().into()).into());
2346        }
2347
2348        let mut my_udp_fronts: HashSet<(&ClusterId, &UdpFrontend)> = HashSet::new();
2349        for (cluster_id, front_list) in self.udp_fronts.iter() {
2350            for front in front_list.iter() {
2351                my_udp_fronts.insert((cluster_id, front));
2352            }
2353        }
2354        let mut their_udp_fronts: HashSet<(&ClusterId, &UdpFrontend)> = HashSet::new();
2355        for (cluster_id, front_list) in other.udp_fronts.iter() {
2356            for front in front_list.iter() {
2357                their_udp_fronts.insert((cluster_id, front));
2358            }
2359        }
2360
2361        let removed_udp_fronts = my_udp_fronts.difference(&their_udp_fronts);
2362        let added_udp_fronts = their_udp_fronts.difference(&my_udp_fronts);
2363
2364        for &(_, front) in removed_udp_fronts {
2365            v.push(RequestType::RemoveUdpFrontend(front.clone().into()).into());
2366        }
2367
2368        for &(_, front) in added_udp_fronts {
2369            v.push(RequestType::AddUdpFrontend(front.clone().into()).into());
2370        }
2371
2372        //pub certificates:    HashMap<SocketAddr, HashMap<CertificateFingerprint, (CertificateAndKey, Vec<String>)>>,
2373        let my_certificates: HashSet<(SocketAddr, &Fingerprint)> = HashSet::from_iter(
2374            self.certificates
2375                .iter()
2376                .flat_map(|(addr, certs)| repeat(*addr).zip(certs.keys())),
2377        );
2378        let their_certificates: HashSet<(SocketAddr, &Fingerprint)> = HashSet::from_iter(
2379            other
2380                .certificates
2381                .iter()
2382                .flat_map(|(addr, certs)| repeat(*addr).zip(certs.keys())),
2383        );
2384
2385        let removed_certificates = my_certificates.difference(&their_certificates);
2386        let added_certificates = their_certificates.difference(&my_certificates);
2387
2388        for &(address, fingerprint) in removed_certificates {
2389            v.push(
2390                RequestType::RemoveCertificate(RemoveCertificate {
2391                    address: SocketAddress::from(address),
2392                    fingerprint: fingerprint.to_string(),
2393                })
2394                .into(),
2395            );
2396        }
2397
2398        for &(address, fingerprint) in added_certificates {
2399            if let Some(certificate_and_key) = other
2400                .certificates
2401                .get(&address)
2402                .and_then(|certs| certs.get(fingerprint))
2403            {
2404                v.push(
2405                    RequestType::AddCertificate(AddCertificate {
2406                        address: SocketAddress::from(address),
2407                        certificate: certificate_and_key.clone(),
2408                        expired_at: None,
2409                    })
2410                    .into(),
2411                );
2412            }
2413        }
2414
2415        for address in added_tcp_listeners {
2416            let listener = &other.tcp_listeners[*address];
2417            if listener.active {
2418                v.push(
2419                    RequestType::ActivateListener(ActivateListener {
2420                        address: listener.address,
2421                        proxy: ListenerType::Tcp.into(),
2422                        from_scm: false,
2423                    })
2424                    .into(),
2425                );
2426            }
2427        }
2428
2429        for address in added_udp_listeners {
2430            let listener = &other.udp_listeners[*address];
2431            if listener.active {
2432                v.push(
2433                    RequestType::ActivateListener(ActivateListener {
2434                        address: listener.address,
2435                        proxy: ListenerType::Udp.into(),
2436                        from_scm: false,
2437                    })
2438                    .into(),
2439                );
2440            }
2441        }
2442
2443        // Replay symmetry: the request set `diff` emits, when replayed onto a
2444        // clone of `self`, must reproduce `other`'s routing-relevant maps —
2445        // listeners included, with their `active` flag. This is the property the
2446        // hot-reconfig fan-out relies on — a worker applies these requests and
2447        // must converge on `other`. We verify it in debug by actually replaying;
2448        // `dispatch`'s own invariant sweep runs on every step.
2449        //
2450        // One deliberate normalization: `backends`/`tcp_fronts` are compared
2451        // after dropping empty buckets. `remove_backend`/`remove_tcp_frontend`
2452        // leave an empty `Vec` under a cluster key, whereas `other` may have no
2453        // key at all. An empty bucket emits no requests and is semantically
2454        // equivalent to an absent one, so we normalize it away before comparing.
2455        #[cfg(debug_assertions)]
2456        {
2457            let mut replayed = self.clone();
2458            for request in &v {
2459                // A diff request must always be dispatchable onto `self`.
2460                debug_assert!(
2461                    replayed.dispatch(request).is_ok(),
2462                    "every request emitted by diff must replay cleanly onto self"
2463                );
2464            }
2465            let nonempty = |m: &BTreeMap<ClusterId, Vec<Backend>>| {
2466                m.iter()
2467                    .filter(|(_, v)| !v.is_empty())
2468                    .map(|(k, v)| (k.clone(), v.clone()))
2469                    .collect::<BTreeMap<_, _>>()
2470            };
2471            let nonempty_tcp = |m: &HashMap<ClusterId, Vec<TcpFrontend>>| {
2472                m.iter()
2473                    .filter(|(_, v)| !v.is_empty())
2474                    .map(|(k, v)| (k.clone(), v.clone()))
2475                    .collect::<HashMap<_, _>>()
2476            };
2477            debug_assert!(
2478                replayed.clusters == other.clusters
2479                    && nonempty(&replayed.backends) == nonempty(&other.backends)
2480                    && replayed.http_fronts == other.http_fronts
2481                    && replayed.https_fronts == other.https_fronts
2482                    && nonempty_tcp(&replayed.tcp_fronts) == nonempty_tcp(&other.tcp_fronts)
2483                    && replayed.certificates == other.certificates
2484                    && replayed.http_listeners == other.http_listeners
2485                    && replayed.https_listeners == other.https_listeners
2486                    && replayed.tcp_listeners == other.tcp_listeners
2487                    && replayed.udp_listeners == other.udp_listeners,
2488                "replaying diff(self, other) onto self must reproduce other's clusters/backends/frontends/certificates/listeners"
2489            );
2490        }
2491
2492        v
2493    }
2494
2495    // FIXME: what about deny rules?
2496    pub fn hash_state(&self) -> BTreeMap<ClusterId, u64> {
2497        let mut hm: HashMap<ClusterId, DefaultHasher> = self
2498            .clusters
2499            .keys()
2500            .map(|cluster_id| {
2501                let mut hasher = DefaultHasher::new();
2502                self.clusters.get(cluster_id).hash(&mut hasher);
2503                if let Some(backends) = self.backends.get(cluster_id) {
2504                    backends.iter().collect::<BTreeSet<_>>().hash(&mut hasher)
2505                }
2506                if let Some(tcp_fronts) = self.tcp_fronts.get(cluster_id) {
2507                    tcp_fronts.iter().collect::<BTreeSet<_>>().hash(&mut hasher)
2508                }
2509                (cluster_id.to_owned(), hasher)
2510            })
2511            .collect();
2512
2513        for front in self.http_fronts.values() {
2514            if let Some(cluster_id) = &front.cluster_id
2515                && let Some(hasher) = hm.get_mut(cluster_id)
2516            {
2517                front.hash(hasher);
2518            }
2519        }
2520
2521        for front in self.https_fronts.values() {
2522            if let Some(cluster_id) = &front.cluster_id
2523                && let Some(hasher) = hm.get_mut(cluster_id)
2524            {
2525                front.hash(hasher);
2526            }
2527        }
2528
2529        hm.drain()
2530            .map(|(cluster_id, hasher)| (cluster_id, hasher.finish()))
2531            .collect()
2532    }
2533
2534    /// Gives details about a given cluster.
2535    /// Types like `HttpFrontend` are converted into protobuf ones, like `RequestHttpFrontend`
2536    pub fn cluster_state(&self, cluster_id: &str) -> Option<ClusterInformation> {
2537        let configuration = self.clusters.get(cluster_id).cloned()?;
2538        info!("{:?}", configuration);
2539
2540        let http_frontends: Vec<RequestHttpFrontend> = self
2541            .http_fronts
2542            .values()
2543            .filter(|front| front.cluster_id.as_deref() == Some(cluster_id))
2544            .map(|front| front.clone().into())
2545            .collect();
2546
2547        let https_frontends: Vec<RequestHttpFrontend> = self
2548            .https_fronts
2549            .values()
2550            .filter(|front| front.cluster_id.as_deref() == Some(cluster_id))
2551            .map(|front| front.clone().into())
2552            .collect();
2553
2554        let tcp_frontends: Vec<RequestTcpFrontend> = self
2555            .tcp_fronts
2556            .get(cluster_id)
2557            .cloned()
2558            .unwrap_or_default()
2559            .iter()
2560            .map(|front| front.clone().into())
2561            .collect();
2562
2563        let udp_frontends: Vec<RequestUdpFrontend> = self
2564            .udp_fronts
2565            .get(cluster_id)
2566            .cloned()
2567            .unwrap_or_default()
2568            .iter()
2569            .map(|front| front.clone().into())
2570            .collect();
2571
2572        let backends: Vec<AddBackend> = self
2573            .backends
2574            .get(cluster_id)
2575            .cloned()
2576            .unwrap_or_default()
2577            .iter()
2578            .map(|backend| backend.clone().into())
2579            .collect();
2580
2581        Some(ClusterInformation {
2582            configuration: Some(configuration),
2583            http_frontends,
2584            https_frontends,
2585            tcp_frontends,
2586            backends,
2587            udp_frontends,
2588        })
2589    }
2590
2591    pub fn count_backends(&self) -> usize {
2592        self.backends.values().fold(0, |acc, v| acc + v.len())
2593    }
2594
2595    pub fn count_frontends(&self) -> usize {
2596        self.http_fronts.values().count()
2597            + self.https_fronts.values().count()
2598            + self.tcp_fronts.values().fold(0, |acc, v| acc + v.len())
2599            + self.udp_fronts.values().fold(0, |acc, v| acc + v.len())
2600    }
2601
2602    pub fn get_cluster_ids_by_domain(
2603        &self,
2604        hostname: String,
2605        path: Option<String>,
2606    ) -> HashSet<ClusterId> {
2607        let mut cluster_ids: HashSet<ClusterId> = HashSet::new();
2608
2609        self.http_fronts.values().for_each(|front| {
2610            if domain_check(&front.hostname, &front.path, &hostname, &path)
2611                && let Some(id) = &front.cluster_id
2612            {
2613                cluster_ids.insert(id.to_string());
2614            }
2615        });
2616
2617        self.https_fronts.values().for_each(|front| {
2618            if domain_check(&front.hostname, &front.path, &hostname, &path)
2619                && let Some(id) = &front.cluster_id
2620            {
2621                cluster_ids.insert(id.to_string());
2622            }
2623        });
2624
2625        cluster_ids
2626    }
2627
2628    pub fn get_certificates(
2629        &self,
2630        filters: QueryCertificatesFilters,
2631    ) -> BTreeMap<String, CertificateAndKey> {
2632        self.certificates
2633            .values()
2634            .flat_map(|hash_map| hash_map.iter())
2635            .filter(|(fingerprint, cert)| {
2636                if let Some(domain) = &filters.domain {
2637                    cert.names.contains(domain)
2638                } else if let Some(f) = &filters.fingerprint {
2639                    fingerprint.to_string() == *f
2640                } else {
2641                    true
2642                }
2643            })
2644            .map(|(fingerprint, cert)| (fingerprint.to_string(), cert.to_owned()))
2645            .collect()
2646    }
2647
2648    pub fn list_frontends(&self, filters: FrontendFilters) -> ListedFrontends {
2649        // if no http / https / tcp filter is provided, list all of them
2650        let list_all = !filters.http && !filters.https && !filters.tcp;
2651
2652        let mut listed_frontends = ListedFrontends::default();
2653
2654        if filters.http || list_all {
2655            for http_frontend in self.http_fronts.iter().filter(|f| {
2656                if let Some(domain) = &filters.domain {
2657                    f.1.hostname.contains(domain)
2658                } else {
2659                    true
2660                }
2661            }) {
2662                listed_frontends
2663                    .http_frontends
2664                    .push(http_frontend.1.to_owned().into());
2665            }
2666        }
2667
2668        if filters.https || list_all {
2669            for https_frontend in self.https_fronts.iter().filter(|f| {
2670                if let Some(domain) = &filters.domain {
2671                    f.1.hostname.contains(domain)
2672                } else {
2673                    true
2674                }
2675            }) {
2676                listed_frontends
2677                    .https_frontends
2678                    .push(https_frontend.1.to_owned().into());
2679            }
2680        }
2681
2682        if (filters.tcp || list_all) && filters.domain.is_none() {
2683            for tcp_frontend in self.tcp_fronts.values().flat_map(|v| v.iter()) {
2684                listed_frontends
2685                    .tcp_frontends
2686                    .push(tcp_frontend.to_owned().into())
2687            }
2688        }
2689
2690        // `FrontendFilters` has no dedicated `udp` flag, so UDP frontends ride
2691        // the same default/all-pass path as TCP: surfaced when no protocol
2692        // filter is set (`list_all`) or when the `tcp` filter is requested.
2693        // Datagram frontends carry no hostname, so a `domain` filter excludes
2694        // them (matching the TCP branch).
2695        if (filters.tcp || list_all) && filters.domain.is_none() {
2696            for udp_frontend in self.udp_fronts.values().flat_map(|v| v.iter()) {
2697                listed_frontends
2698                    .udp_frontends
2699                    .push(udp_frontend.to_owned().into())
2700            }
2701        }
2702
2703        listed_frontends
2704    }
2705
2706    pub fn list_listeners(&self) -> ListenersList {
2707        ListenersList {
2708            http_listeners: self
2709                .http_listeners
2710                .iter()
2711                .map(|(addr, listener)| (addr.to_string(), listener.clone()))
2712                .collect(),
2713            https_listeners: self
2714                .https_listeners
2715                .iter()
2716                .map(|(addr, listener)| (addr.to_string(), listener.clone()))
2717                .collect(),
2718            tcp_listeners: self
2719                .tcp_listeners
2720                .iter()
2721                .map(|(addr, listener)| (addr.to_string(), *listener))
2722                .collect(),
2723            udp_listeners: self
2724                .udp_listeners
2725                .iter()
2726                .map(|(addr, listener)| (addr.to_string(), *listener))
2727                .collect(),
2728        }
2729    }
2730
2731    // create requests needed for a worker to recreate the state
2732    pub fn produce_initial_state(&self) -> InitialState {
2733        let mut worker_requests = Vec::new();
2734        for (counter, request) in self.generate_requests().into_iter().enumerate() {
2735            worker_requests.push(WorkerRequest::new(format!("SAVE-{counter}"), request));
2736        }
2737        InitialState {
2738            requests: worker_requests,
2739        }
2740    }
2741
2742    /// generate requests necessary to recreate the state,
2743    /// in protobuf, to a temp file
2744    pub fn write_initial_state_to_file(&self, file: &mut File) -> Result<usize, StateError> {
2745        let initial_state = self.produce_initial_state();
2746        let count = initial_state.requests.len();
2747
2748        let bytes_to_write = initial_state.encode_to_vec();
2749        println!("writing {} in the temp file", bytes_to_write.len());
2750        file.write_all(&bytes_to_write)
2751            .map_err(StateError::FileError)?;
2752
2753        file.sync_all().map_err(StateError::FileError)?;
2754
2755        Ok(count)
2756    }
2757
2758    /// generate requests necessary to recreate the state,
2759    /// write them in a JSON form in a file, separated by \n\0,
2760    /// returns the number of written requests
2761    pub fn write_requests_to_file(&self, file: &mut File) -> Result<usize, StateError> {
2762        let mut counter = 0usize;
2763        let requests = self.generate_requests();
2764
2765        for request in requests {
2766            let message = WorkerRequest::new(format!("SAVE-{counter}"), request);
2767
2768            file.write_all(
2769                &serde_json::to_string(&message)
2770                    .map(|s| s.into_bytes())
2771                    .unwrap_or_default(),
2772            )
2773            .map_err(StateError::FileError)?;
2774
2775            file.write_all(&b"\n\0"[..])
2776                .map_err(StateError::FileError)?;
2777
2778            if counter.is_multiple_of(1000) {
2779                info!("writing {} commands to file", counter);
2780                file.sync_all().map_err(StateError::FileError)?;
2781            }
2782            counter += 1;
2783        }
2784        file.sync_all().map_err(StateError::FileError)?;
2785
2786        Ok(counter)
2787    }
2788}
2789
2790/// Validate all H2 flood knobs in an HTTP listener patch.
2791///
2792/// Every flood-detector knob (including stream-0 WINDOW_UPDATE) requires a
2793/// value `>= 1`. Passing `0` would disable the detector entirely and leave the
2794/// proxy open to CVE-2023-44487 and related attacks. The runtime constructor
2795/// `H2FloodConfig::new()` applies the same `.max(1)` clamping, but a raw
2796/// protobuf client can bypass the CLI layer, so we enforce the bound here too.
2797///
2798/// `h2_max_concurrent_streams` and `h2_stream_shrink_ratio` are connection-
2799/// config knobs that also require `>= 1`.
2800///
2801/// `h2_graceful_shutdown_deadline_seconds = 0` is intentionally **allowed** —
2802/// it means "wait forever (no forced close after GOAWAY)".
2803pub fn validate_h2_flood_knobs_http(patch: &UpdateHttpListenerConfig) -> Result<(), StateError> {
2804    macro_rules! require_ge1 {
2805        ($field:expr, $name:literal) => {
2806            if let Some(0) = $field {
2807                return Err(StateError::InvalidValue {
2808                    field: $name,
2809                    reason: "must be >= 1",
2810                });
2811            }
2812        };
2813    }
2814    require_ge1!(
2815        patch.h2_max_rst_stream_per_window,
2816        "h2_max_rst_stream_per_window"
2817    );
2818    require_ge1!(patch.h2_max_ping_per_window, "h2_max_ping_per_window");
2819    require_ge1!(
2820        patch.h2_max_settings_per_window,
2821        "h2_max_settings_per_window"
2822    );
2823    require_ge1!(
2824        patch.h2_max_empty_data_per_window,
2825        "h2_max_empty_data_per_window"
2826    );
2827    require_ge1!(
2828        patch.h2_max_continuation_frames,
2829        "h2_max_continuation_frames"
2830    );
2831    require_ge1!(patch.h2_max_glitch_count, "h2_max_glitch_count");
2832    require_ge1!(
2833        patch.h2_max_window_update_stream0_per_window,
2834        "h2_max_window_update_stream0_per_window"
2835    );
2836    require_ge1!(patch.h2_max_concurrent_streams, "h2_max_concurrent_streams");
2837    // Shrink ratio runtime floor is 2 (lib/src/protocol/mux/h2.rs ~448 .max(2));
2838    // anything lower is silently promoted so reject at control plane.
2839    if let Some(v) = patch.h2_stream_shrink_ratio
2840        && v < 2
2841    {
2842        return Err(StateError::InvalidValue {
2843            field: "h2_stream_shrink_ratio",
2844            reason: "must be >= 2",
2845        });
2846    }
2847    // Lifetime caps and HPACK limits — must be >= 1 or the runtime trips on the
2848    // first qualifying frame. doc/configure.md advertises "u64 (>= 1)" etc.
2849    require_ge1!(
2850        patch.h2_max_rst_stream_lifetime,
2851        "h2_max_rst_stream_lifetime"
2852    );
2853    require_ge1!(
2854        patch.h2_max_rst_stream_abusive_lifetime,
2855        "h2_max_rst_stream_abusive_lifetime"
2856    );
2857    require_ge1!(
2858        patch.h2_max_rst_stream_emitted_lifetime,
2859        "h2_max_rst_stream_emitted_lifetime"
2860    );
2861    require_ge1!(patch.h2_max_header_list_size, "h2_max_header_list_size");
2862    require_ge1!(patch.h2_max_header_table_size, "h2_max_header_table_size");
2863    require_ge1!(patch.h2_max_header_fields, "h2_max_header_fields");
2864    Ok(())
2865}
2866
2867/// Validate all H2 flood knobs in an HTTPS listener patch (same rules as HTTP).
2868pub fn validate_h2_flood_knobs_https(patch: &UpdateHttpsListenerConfig) -> Result<(), StateError> {
2869    macro_rules! require_ge1 {
2870        ($field:expr, $name:literal) => {
2871            if let Some(0) = $field {
2872                return Err(StateError::InvalidValue {
2873                    field: $name,
2874                    reason: "must be >= 1",
2875                });
2876            }
2877        };
2878    }
2879    require_ge1!(
2880        patch.h2_max_rst_stream_per_window,
2881        "h2_max_rst_stream_per_window"
2882    );
2883    require_ge1!(patch.h2_max_ping_per_window, "h2_max_ping_per_window");
2884    require_ge1!(
2885        patch.h2_max_settings_per_window,
2886        "h2_max_settings_per_window"
2887    );
2888    require_ge1!(
2889        patch.h2_max_empty_data_per_window,
2890        "h2_max_empty_data_per_window"
2891    );
2892    require_ge1!(
2893        patch.h2_max_continuation_frames,
2894        "h2_max_continuation_frames"
2895    );
2896    require_ge1!(patch.h2_max_glitch_count, "h2_max_glitch_count");
2897    require_ge1!(
2898        patch.h2_max_window_update_stream0_per_window,
2899        "h2_max_window_update_stream0_per_window"
2900    );
2901    require_ge1!(patch.h2_max_concurrent_streams, "h2_max_concurrent_streams");
2902    if let Some(v) = patch.h2_stream_shrink_ratio
2903        && v < 2
2904    {
2905        return Err(StateError::InvalidValue {
2906            field: "h2_stream_shrink_ratio",
2907            reason: "must be >= 2",
2908        });
2909    }
2910    require_ge1!(
2911        patch.h2_max_rst_stream_lifetime,
2912        "h2_max_rst_stream_lifetime"
2913    );
2914    require_ge1!(
2915        patch.h2_max_rst_stream_abusive_lifetime,
2916        "h2_max_rst_stream_abusive_lifetime"
2917    );
2918    require_ge1!(
2919        patch.h2_max_rst_stream_emitted_lifetime,
2920        "h2_max_rst_stream_emitted_lifetime"
2921    );
2922    require_ge1!(patch.h2_max_header_list_size, "h2_max_header_list_size");
2923    require_ge1!(patch.h2_max_header_table_size, "h2_max_header_table_size");
2924    require_ge1!(patch.h2_max_header_fields, "h2_max_header_fields");
2925    Ok(())
2926}
2927
2928/// Validate a `sozu_id_header` value against RFC 9110 §5.1 header-name grammar.
2929///
2930/// Rejects empty strings and strings containing CR, LF, colon, space, or tab —
2931/// a conservative approximation of the token grammar that covers all practical
2932/// injection vectors without a full RFC 9110 tokenizer.
2933/// Merge a `CustomHttpAnswers` patch into the listener's stored answers,
2934/// preserving any field not present in the patch.
2935///
2936/// Field-mask semantic: `None` in `patch` means "preserve", `Some` means
2937/// "replace". A no-op patch (all-None) leaves `target` untouched. If `target`
2938/// is currently `None`, initialize it from the patch (any `None` field stays
2939/// `None` so hot-upgrade replay sees the same partial state).
2940pub fn merge_custom_http_answers(
2941    target: &mut Option<CustomHttpAnswers>,
2942    patch: &CustomHttpAnswers,
2943) {
2944    let current = target.get_or_insert_with(CustomHttpAnswers::default);
2945    macro_rules! merge_field {
2946        ($field:ident) => {
2947            if let Some(ref v) = patch.$field {
2948                current.$field = Some(v.clone());
2949            }
2950        };
2951    }
2952    merge_field!(answer_301);
2953    merge_field!(answer_400);
2954    merge_field!(answer_401);
2955    merge_field!(answer_404);
2956    merge_field!(answer_408);
2957    merge_field!(answer_413);
2958    merge_field!(answer_421);
2959    merge_field!(answer_502);
2960    merge_field!(answer_503);
2961    merge_field!(answer_504);
2962    merge_field!(answer_507);
2963}
2964
2965/// Validate an `AlpnProtocols` patch: each value must be "h2" or "http/1.1".
2966/// Empty values vec is allowed (reset-to-default).
2967pub fn validate_alpn_protocols(values: &[String]) -> Result<(), StateError> {
2968    for value in values {
2969        if value != "h2" && value != "http/1.1" {
2970            return Err(StateError::InvalidValue {
2971                field: "alpn_protocols",
2972                reason: "each value must be \"h2\" or \"http/1.1\"",
2973            });
2974        }
2975    }
2976    Ok(())
2977}
2978
2979/// Validate a `sozu_id_header` value against the RFC 9110 §5.1 `token` grammar:
2980///
2981/// ```text
2982/// token  = 1*tchar
2983/// tchar  = "!" / "#" / "$" / "%" / "&" / "'" / "*" / "+" / "-" / "." /
2984///          "^" / "_" / "`" / "|" / "~" / DIGIT / ALPHA
2985/// ```
2986///
2987/// Rejects empty strings, non-ASCII bytes, controls (including CR/LF/tab),
2988/// separators (including colon and space), and any other non-`tchar` byte.
2989pub fn validate_sozu_id_header(value: &str) -> Result<(), StateError> {
2990    if value.is_empty() {
2991        return Err(StateError::InvalidValue {
2992            field: "sozu_id_header",
2993            reason: "must not be empty",
2994        });
2995    }
2996    for b in value.bytes() {
2997        let is_tchar = b.is_ascii_alphanumeric()
2998            || matches!(
2999                b,
3000                b'!' | b'#'
3001                    | b'$'
3002                    | b'%'
3003                    | b'&'
3004                    | b'\''
3005                    | b'*'
3006                    | b'+'
3007                    | b'-'
3008                    | b'.'
3009                    | b'^'
3010                    | b'_'
3011                    | b'`'
3012                    | b'|'
3013                    | b'~'
3014            );
3015        if !is_tchar {
3016            return Err(StateError::InvalidValue {
3017                field: "sozu_id_header",
3018                reason: "must be a valid HTTP header name (RFC 9110 §5.1 token: alphanumeric or one of !#$%&'*+-.^_`|~)",
3019            });
3020        }
3021    }
3022    Ok(())
3023}
3024
3025fn domain_check(
3026    front_hostname: &str,
3027    front_path_rule: &PathRule,
3028    hostname: &str,
3029    path_prefix: &Option<String>,
3030) -> bool {
3031    if hostname != front_hostname {
3032        return false;
3033    }
3034
3035    if let Some(path) = &path_prefix {
3036        return path == &front_path_rule.value;
3037    }
3038
3039    true
3040}
3041
3042/// Canonicalize a TCP frontend's ALPN list into its sorted, deduplicated set
3043/// representation. ALPN matching (the worker's `AlpnMatcher::OneOf` in
3044/// `lib/src/tcp.rs`) is semantically a *set* of protocol names, not an
3045/// ordered list: comparing `Vec`s directly treats `["h2", "http/1.1"]` and
3046/// `["http/1.1", "h2"]` as distinct identities, silently admitting a
3047/// duplicate/overlapping frontend the worker rejects on the next fan-out
3048/// (sozu-proxy/sozu#1290). Used both to canonicalize a frontend for storage
3049/// and to canonicalize a candidate for comparison, so `TcpFrontend.alpn` is
3050/// always already in this form.
3051fn canonical_tcp_alpn(alpn: &[String]) -> Vec<String> {
3052    let mut alpn = alpn.to_vec();
3053    alpn.sort();
3054    alpn.dedup();
3055    // POST: strictly increasing (sorted, no adjacent duplicates) — the
3056    // single property every caller relies on to treat two canonical lists
3057    // as set-equal via plain `==`.
3058    debug_assert!(
3059        alpn.windows(2).all(|pair| pair[0] < pair[1]),
3060        "canonical_tcp_alpn must return a strictly sorted, duplicate-free list"
3061    );
3062    alpn
3063}
3064
3065/// Identity predicate for a TCP frontend: `(address, sni, alpn)`, not the
3066/// whole struct -- two frontends differing only in `tags` still collide
3067/// here even though they compare unequal as full structs, since they'd
3068/// still match the exact same wire traffic. `sni` and `alpn` are expected
3069/// to already be in canonical form (normalized-lowercase sni, sorted+deduped
3070/// alpn via [`canonical_tcp_alpn`]) on both sides -- this is a plain
3071/// equality check, not itself a canonicalizer. Shared by
3072/// `add_tcp_frontend`'s duplicate check and `remove_tcp_frontend`'s
3073/// retain/postcondition so the three call sites can never drift apart
3074/// (sozu-proxy/sozu#1290).
3075fn tcp_frontend_matches(
3076    front: &TcpFrontend,
3077    address: SocketAddr,
3078    sni: &Option<String>,
3079    alpn: &[String],
3080) -> bool {
3081    front.address == address && front.sni == *sni && front.alpn.as_slice() == alpn
3082}
3083
3084struct DiffMap<'a, K: Ord, V, I1, I2> {
3085    my_it: I1,
3086    other_it: I2,
3087    my: Option<(K, &'a V)>,
3088    other: Option<(K, &'a V)>,
3089}
3090
3091//fn diff_map<'a, K:Ord, V: PartialEq>(my: &'a BTreeMap<K,V>, other: &'a BTreeMap<K,V>) -> DiffMap<'a,K,V> {
3092fn diff_map<
3093    'a,
3094    K: Ord,
3095    V: PartialEq,
3096    I1: Iterator<Item = (K, &'a V)>,
3097    I2: Iterator<Item = (K, &'a V)>,
3098>(
3099    my: I1,
3100    other: I2,
3101) -> DiffMap<'a, K, V, I1, I2> {
3102    DiffMap {
3103        my_it: my,
3104        other_it: other,
3105        my: None,
3106        other: None,
3107    }
3108}
3109
3110enum DiffResult {
3111    Added,
3112    Removed,
3113    Changed,
3114}
3115
3116// this will iterate over the keys of both iterators
3117// since keys are sorted, it should be easy to see which ones are in common or not
3118impl<'a, K: Ord, V: PartialEq, I1: Iterator<Item = (K, &'a V)>, I2: Iterator<Item = (K, &'a V)>>
3119    std::iter::Iterator for DiffMap<'a, K, V, I1, I2>
3120{
3121    type Item = (K, DiffResult);
3122
3123    fn next(&mut self) -> Option<Self::Item> {
3124        loop {
3125            if self.my.is_none() {
3126                self.my = self.my_it.next();
3127            }
3128            if self.other.is_none() {
3129                self.other = self.other_it.next();
3130            }
3131
3132            match (self.my.take(), self.other.take()) {
3133                // there are no more elements in my_it, all the next elements in other
3134                // should be added
3135                // if other was none, we will stop the iterator there
3136                (None, other) => return other.map(|(k, _)| (k, DiffResult::Added)),
3137                // there are no more elements in other_it, all the next elements in my
3138                // should be removed
3139                (Some((k, _)), None) => return Some((k, DiffResult::Removed)),
3140                // element is present in my but not other
3141                (Some((k1, _v1)), Some((k2, v2))) if k1 < k2 => {
3142                    self.other = Some((k2, v2));
3143                    return Some((k1, DiffResult::Removed));
3144                }
3145                // element is present in other byt not in my
3146                (Some((k1, v1)), Some((k2, _v2))) if k1 > k2 => {
3147                    self.my = Some((k1, v1));
3148                    return Some((k2, DiffResult::Added));
3149                }
3150                (Some((k1, v1)), Some((_k2, v2))) if v1 != v2 => {
3151                    // key is present in both, if elements have changed
3152                    // return a value, otherwise go to the next key for both maps
3153                    return Some((k1, DiffResult::Changed));
3154                }
3155                _ => {}
3156            }
3157        }
3158    }
3159}
3160
3161#[cfg(test)]
3162mod tests {
3163    use rand::{RngExt, rng, seq::SliceRandom};
3164
3165    use super::*;
3166    use crate::proto::command::{
3167        CustomHttpAnswers, Header, HeaderPosition, HstsConfig, LoadBalancingParams, PathRuleKind,
3168        RedirectPolicy, RedirectScheme, RequestHttpFrontend, RequestTcpFrontend,
3169        RequestUdpFrontend, RulePosition, UdpListenerConfig, UpdateUdpListenerConfig,
3170    };
3171
3172    fn capture_test_logs(run: impl FnOnce() + Send + 'static) -> String {
3173        let receiver = std::net::UdpSocket::bind("127.0.0.1:0")
3174            .expect("test log receiver must bind to a loopback port");
3175        let target = format!(
3176            "udp://{}",
3177            receiver
3178                .local_addr()
3179                .expect("test log receiver must have a local address")
3180        );
3181
3182        std::thread::spawn(move || {
3183            crate::logging::Logger::init(
3184                "state-log-redaction-test".to_owned(),
3185                "info",
3186                &target,
3187                false,
3188                None,
3189                None,
3190                None,
3191            )
3192            .expect("test logger must initialize");
3193            run();
3194        })
3195        .join()
3196        .expect("log-producing test thread must not panic");
3197
3198        receiver
3199            .set_nonblocking(true)
3200            .expect("test log receiver must become nonblocking");
3201        let mut output = String::new();
3202        let mut datagram = vec![0; 65_507];
3203        loop {
3204            match receiver.recv(&mut datagram) {
3205                Ok(length) => output.push_str(&String::from_utf8_lossy(&datagram[..length])),
3206                Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => break,
3207                Err(error) => panic!("test log receiver failed: {error}"),
3208            }
3209        }
3210        assert!(!output.is_empty(), "test log capture received no datagrams");
3211        output
3212    }
3213
3214    #[test]
3215    fn duplicate_certificate_log_redacts_names() {
3216        const NAME_SECRET: &str = "DUPLICATE_CERTIFICATE_NAME_SECRET_SENTINEL";
3217
3218        let name = format!("{NAME_SECRET}{}", "x".repeat(4096));
3219        let name_len = name.len();
3220        let certificate = CertificateAndKey {
3221            certificate: include_str!("../assets/certificate.pem").to_owned(),
3222            key: include_str!("../assets/key.pem").to_owned(),
3223            certificate_chain: Vec::new(),
3224            versions: Vec::new(),
3225            names: vec![name],
3226        };
3227        let fingerprint = certificate
3228            .fingerprint()
3229            .expect("test certificate fingerprint must be computable")
3230            .to_string();
3231        let output = capture_test_logs(move || {
3232            let mut state = ConfigState::default();
3233            let add = AddCertificate {
3234                address: SocketAddress::new_v4(127, 0, 0, 1, 8443),
3235                certificate,
3236                expired_at: None,
3237            };
3238
3239            state
3240                .add_certificate(&add)
3241                .expect("first certificate insertion must succeed");
3242            state
3243                .add_certificate(&add)
3244                .expect("duplicate certificate insertion must be skipped");
3245        });
3246
3247        assert!(
3248            !output.contains(NAME_SECRET),
3249            "duplicate certificate log leaked certificate name {NAME_SECRET}"
3250        );
3251        assert!(
3252            !output.contains(&fingerprint),
3253            "duplicate certificate log leaked certificate fingerprint {fingerprint}"
3254        );
3255        for metadata in [
3256            "fingerprint_bytes=32".to_owned(),
3257            "names_count=1".to_owned(),
3258            format!("names_bytes={name_len}"),
3259        ] {
3260            assert!(
3261                output.contains(&metadata),
3262                "duplicate certificate log omitted bounded metadata {metadata}: {output}"
3263            );
3264        }
3265        assert!(
3266            output.len() <= 512,
3267            "duplicate certificate log is not bounded: {} bytes",
3268            output.len()
3269        );
3270    }
3271
3272    #[test]
3273    fn frontend_state_error_redacts_display_key() {
3274        const HOSTNAME_SECRET: &str = "STATE_ERROR_HOSTNAME_SECRET_SENTINEL";
3275        const PATH_SECRET: &str = "STATE_ERROR_PATH_SECRET_SENTINEL";
3276        const METHOD_SECRET: &str = "STATE_ERROR_METHOD_SECRET_SENTINEL";
3277
3278        let long_value = |marker: &str| format!("{marker}{}", "x".repeat(4096));
3279        let front = RequestHttpFrontend {
3280            cluster_id: Some("cluster".to_owned()),
3281            address: SocketAddress::new_v4(127, 0, 0, 1, 8080),
3282            hostname: long_value(HOSTNAME_SECRET),
3283            path: PathRule {
3284                kind: PathRuleKind::Prefix as i32,
3285                value: long_value(PATH_SECRET),
3286            },
3287            method: Some(long_value(METHOD_SECRET)),
3288            position: RulePosition::Tree as i32,
3289            tags: BTreeMap::new(),
3290            redirect: None,
3291            redirect_scheme: None,
3292            redirect_template: None,
3293            rewrite_host: None,
3294            rewrite_path: None,
3295            rewrite_port: None,
3296            required_auth: None,
3297            headers: Vec::new(),
3298            hsts: None,
3299        };
3300        let mut state = ConfigState::default();
3301        state
3302            .add_http_frontend(&front)
3303            .expect("first frontend insertion must succeed");
3304        assert!(
3305            state
3306                .http_fronts
3307                .keys()
3308                .next()
3309                .expect("raw state key must be retained")
3310                .contains(HOSTNAME_SECRET),
3311            "state storage must retain the raw operator-facing frontend key"
3312        );
3313
3314        let error = state
3315            .add_http_frontend(&front)
3316            .expect_err("duplicate frontend insertion must return StateError::Exists");
3317        let raw_frontend_key = front.to_string();
3318        match &error {
3319            StateError::Exists { id, .. } => assert_eq!(
3320                id, &raw_frontend_key,
3321                "StateError must preserve the public raw frontend identifier"
3322            ),
3323            other => panic!("expected StateError::Exists, got {other:?}"),
3324        }
3325        for (label, output) in [
3326            ("Display", error.to_string()),
3327            ("Debug", format!("{error:?}")),
3328        ] {
3329            for secret in [HOSTNAME_SECRET, PATH_SECRET, METHOD_SECRET] {
3330                assert!(
3331                    !output.contains(secret),
3332                    "StateError {label} leaked frontend marker {secret}"
3333                );
3334            }
3335            let metadata = format!("id_bytes={}", raw_frontend_key.len());
3336            assert!(
3337                output.contains(&metadata),
3338                "StateError {label} omitted bounded metadata {metadata}: {output}"
3339            );
3340            assert!(
3341                output.len() <= 512,
3342                "StateError {label} output is not bounded: {} bytes",
3343                output.len()
3344            );
3345        }
3346    }
3347
3348    #[test]
3349    fn state_error_formatting_bounds_every_string_bearing_variant() {
3350        const SECRET: &str = "STATE_ERROR_VARIANT_SECRET_SENTINEL";
3351
3352        let long_value = || format!("{SECRET}{}", "x".repeat(4096));
3353        let errors = [
3354            StateError::NotFound {
3355                kind: ObjectKind::Backend,
3356                id: long_value(),
3357            },
3358            StateError::Exists {
3359                kind: ObjectKind::Cluster,
3360                id: long_value(),
3361            },
3362            StateError::AddCertificate(CertificateError::ParsePEMCertificate(long_value())),
3363            StateError::RemoveCertificate(long_value()),
3364            StateError::ReplaceCertificate(long_value()),
3365            StateError::FrontendConversion {
3366                frontend: long_value(),
3367                error: long_value(),
3368            },
3369            StateError::FileError(std::io::Error::other(long_value())),
3370            StateError::InvalidTcpFrontend {
3371                address: "127.0.0.1:443"
3372                    .parse()
3373                    .expect("test TCP frontend address must parse"),
3374                reason: long_value(),
3375            },
3376        ];
3377
3378        for error in errors {
3379            for (label, output) in [
3380                ("Display", error.to_string()),
3381                ("Debug", format!("{error:?}")),
3382            ] {
3383                assert!(
3384                    !output.contains(SECRET),
3385                    "StateError {label} leaked a string-bearing variant: {output}"
3386                );
3387                assert!(
3388                    output.len() <= 512,
3389                    "StateError {label} is not bounded: {} bytes",
3390                    output.len()
3391                );
3392            }
3393        }
3394    }
3395
3396    #[test]
3397    fn tcp_frontend_not_found_error_retains_the_raw_identity() {
3398        const TCP_SECRET: &str = "STATE_ERROR_TCP_IDENTITY_SECRET_SENTINEL";
3399
3400        let long_value = |suffix: &str| format!("{TCP_SECRET}_{suffix}{}", "x".repeat(1024));
3401        let frontend = RequestTcpFrontend {
3402            cluster_id: long_value("cluster"),
3403            address: SocketAddress::new_v4(127, 0, 0, 1, 443),
3404            tags: BTreeMap::from([(long_value("tag-key"), long_value("tag-value"))]),
3405            sni: Some(long_value("sni")),
3406            alpn: vec![long_value("alpn")],
3407        };
3408        let error = ConfigState::default()
3409            .remove_tcp_frontend(&frontend)
3410            .expect_err("removing from a missing cluster must return StateError::NotFound");
3411
3412        match &error {
3413            StateError::NotFound { id, .. } => assert!(
3414                id.contains(TCP_SECRET),
3415                "StateError must retain the raw TCP frontend identity: {id}"
3416            ),
3417            other => panic!("expected StateError::NotFound, got {other:?}"),
3418        }
3419        for output in [error.to_string(), format!("{error:?}")] {
3420            assert!(
3421                !output.contains(TCP_SECRET),
3422                "StateError formatting leaked the retained TCP identity: {output}"
3423            );
3424            assert!(
3425                output.len() <= 512,
3426                "StateError formatting is not bounded: {} bytes",
3427                output.len()
3428            );
3429        }
3430    }
3431
3432    #[test]
3433    fn retained_state_debug_redacts_http_frontend_and_collection_data() {
3434        const CLUSTER_SECRET: &str = "RETAINED_CLUSTER_SECRET_SENTINEL";
3435        const HOSTNAME_SECRET: &str = "RETAINED_HOSTNAME_SECRET_SENTINEL";
3436        const PATH_SECRET: &str = "RETAINED_PATH_SECRET_SENTINEL";
3437        const METHOD_SECRET: &str = "RETAINED_METHOD_SECRET_SENTINEL";
3438        const TAG_KEY_SECRET: &str = "RETAINED_TAG_KEY_SECRET_SENTINEL";
3439        const TAG_VALUE_SECRET: &str = "RETAINED_TAG_VALUE_SECRET_SENTINEL";
3440        const REDIRECT_TEMPLATE_SECRET: &str = "RETAINED_REDIRECT_TEMPLATE_SECRET_SENTINEL";
3441        const REWRITE_HOST_SECRET: &str = "RETAINED_REWRITE_HOST_SECRET_SENTINEL";
3442        const REWRITE_PATH_SECRET: &str = "RETAINED_REWRITE_PATH_SECRET_SENTINEL";
3443        const HEADER_KEY_SECRET: &str = "RETAINED_HEADER_KEY_SECRET_SENTINEL";
3444        const HEADER_VALUE_SECRET: &str = "RETAINED_HEADER_VALUE_SECRET_SENTINEL";
3445        const REQUEST_COUNT_SECRET: &str = "RETAINED_REQUEST_COUNT_SECRET_SENTINEL";
3446
3447        let long_value = |marker: &str| format!("{marker}{}", "x".repeat(4096));
3448        let request = RequestHttpFrontend {
3449            cluster_id: Some(long_value(CLUSTER_SECRET)),
3450            address: SocketAddress::new_v4(127, 0, 0, 1, 8443),
3451            hostname: long_value(HOSTNAME_SECRET),
3452            path: PathRule {
3453                kind: PathRuleKind::Regex as i32,
3454                value: long_value(PATH_SECRET),
3455            },
3456            method: Some(long_value(METHOD_SECRET)),
3457            position: RulePosition::Tree as i32,
3458            tags: BTreeMap::from([(long_value(TAG_KEY_SECRET), long_value(TAG_VALUE_SECRET))]),
3459            redirect: Some(RedirectPolicy::PermanentRedirect as i32),
3460            redirect_scheme: Some(RedirectScheme::UseHttps as i32),
3461            redirect_template: Some(long_value(REDIRECT_TEMPLATE_SECRET)),
3462            rewrite_host: Some(long_value(REWRITE_HOST_SECRET)),
3463            rewrite_path: Some(long_value(REWRITE_PATH_SECRET)),
3464            rewrite_port: Some(9443),
3465            required_auth: Some(true),
3466            headers: vec![Header {
3467                position: HeaderPosition::Both as i32,
3468                key: long_value(HEADER_KEY_SECRET),
3469                val: long_value(HEADER_VALUE_SECRET),
3470            }],
3471            hsts: Some(HstsConfig {
3472                enabled: Some(true),
3473                max_age: Some(31_536_000),
3474                include_subdomains: Some(true),
3475                preload: Some(false),
3476                force_replace_backend: Some(true),
3477            }),
3478        };
3479        let retained = request
3480            .clone()
3481            .to_frontend()
3482            .expect("adversarial frontend must convert");
3483        let retained_debug = format!("{retained:?}");
3484
3485        let mut state = ConfigState::default();
3486        state
3487            .dispatch(&RequestType::AddHttpFrontend(request).into())
3488            .expect("adversarial frontend must be retained");
3489        state
3490            .request_counts
3491            .insert(long_value(REQUEST_COUNT_SECRET), 7);
3492        let retained_key = state
3493            .http_fronts
3494            .keys()
3495            .next()
3496            .expect("dispatch must retain the frontend's Display key");
3497        assert!(retained_key.contains(HOSTNAME_SECRET));
3498        assert!(retained_key.contains(PATH_SECRET));
3499        assert!(retained_key.contains(METHOD_SECRET));
3500        let state_debug = format!("{state:?}");
3501
3502        let secrets = [
3503            CLUSTER_SECRET,
3504            HOSTNAME_SECRET,
3505            PATH_SECRET,
3506            METHOD_SECRET,
3507            TAG_KEY_SECRET,
3508            TAG_VALUE_SECRET,
3509            REDIRECT_TEMPLATE_SECRET,
3510            REWRITE_HOST_SECRET,
3511            REWRITE_PATH_SECRET,
3512            HEADER_KEY_SECRET,
3513            HEADER_VALUE_SECRET,
3514            REQUEST_COUNT_SECRET,
3515        ];
3516        for (label, output) in [
3517            ("HttpFrontend", &retained_debug),
3518            ("ConfigState", &state_debug),
3519        ] {
3520            for secret in secrets {
3521                assert!(
3522                    !output.contains(secret),
3523                    "{label} Debug leaked retained marker {secret}"
3524                );
3525            }
3526            assert!(
3527                output.len() <= 2048,
3528                "{label} Debug output is not bounded: {} bytes",
3529                output.len()
3530            );
3531        }
3532
3533        for metadata in [
3534            format!("cluster_id_len: Some({})", long_value(CLUSTER_SECRET).len()),
3535            "address: 127.0.0.1:8443".to_owned(),
3536            format!("hostname_len: {}", long_value(HOSTNAME_SECRET).len()),
3537            "path_kind: 1".to_owned(),
3538            format!("path_len: {}", long_value(PATH_SECRET).len()),
3539            format!("method_len: Some({})", long_value(METHOD_SECRET).len()),
3540            "position: Tree".to_owned(),
3541            "tags_count: Some(1)".to_owned(),
3542            format!(
3543                "tags_len: Some({})",
3544                long_value(TAG_KEY_SECRET).len() + long_value(TAG_VALUE_SECRET).len()
3545            ),
3546            "redirect: Some(4)".to_owned(),
3547            "redirect_scheme: Some(2)".to_owned(),
3548            format!(
3549                "redirect_template_len: Some({})",
3550                long_value(REDIRECT_TEMPLATE_SECRET).len()
3551            ),
3552            format!(
3553                "rewrite_host_len: Some({})",
3554                long_value(REWRITE_HOST_SECRET).len()
3555            ),
3556            format!(
3557                "rewrite_path_len: Some({})",
3558                long_value(REWRITE_PATH_SECRET).len()
3559            ),
3560            "rewrite_port: Some(9443)".to_owned(),
3561            "required_auth: Some(true)".to_owned(),
3562            "headers_count: 1".to_owned(),
3563            format!(
3564                "headers_len: {}",
3565                long_value(HEADER_KEY_SECRET).len() + long_value(HEADER_VALUE_SECRET).len()
3566            ),
3567            "hsts: Some(HstsConfig".to_owned(),
3568        ] {
3569            assert!(
3570                retained_debug.contains(&metadata),
3571                "HttpFrontend Debug omitted safe metadata {metadata}: {retained_debug}"
3572            );
3573        }
3574        for metadata in [
3575            "http_frontends_count: 1",
3576            "backends_count: 0",
3577            "tcp_frontends_count: 0",
3578            "udp_frontends_count: 0",
3579            "certificates_count: 0",
3580            "request_counts_count: 2",
3581        ] {
3582            assert!(
3583                state_debug.contains(metadata),
3584                "ConfigState Debug omitted count metadata {metadata}: {state_debug}"
3585            );
3586        }
3587    }
3588
3589    #[test]
3590    fn serialize() {
3591        let mut state: ConfigState = Default::default();
3592        state
3593            .dispatch(
3594                &RequestType::AddHttpFrontend(RequestHttpFrontend {
3595                    cluster_id: Some(String::from("cluster_1")),
3596                    hostname: String::from("lolcatho.st:8080"),
3597                    path: PathRule::prefix(String::from("/")),
3598                    address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
3599                    position: RulePosition::Tree.into(),
3600                    ..Default::default()
3601                })
3602                .into(),
3603            )
3604            .expect("Could not execute request");
3605        state
3606            .dispatch(
3607                &RequestType::AddHttpFrontend(RequestHttpFrontend {
3608                    cluster_id: Some(String::from("cluster_2")),
3609                    hostname: String::from("test.local"),
3610                    path: PathRule::prefix(String::from("/abc")),
3611                    address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
3612                    position: RulePosition::Pre.into(),
3613                    ..Default::default()
3614                })
3615                .into(),
3616            )
3617            .expect("Could not execute request");
3618        state
3619            .dispatch(
3620                &RequestType::AddBackend(AddBackend {
3621                    cluster_id: String::from("cluster_1"),
3622                    backend_id: String::from("cluster_1-0"),
3623                    address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
3624                    ..Default::default()
3625                })
3626                .into(),
3627            )
3628            .expect("Could not execute request");
3629        state
3630            .dispatch(
3631                &RequestType::AddBackend(AddBackend {
3632                    cluster_id: String::from("cluster_1"),
3633                    backend_id: String::from("cluster_1-1"),
3634                    address: SocketAddress::new_v4(127, 0, 0, 1, 1027),
3635                    ..Default::default()
3636                })
3637                .into(),
3638            )
3639            .expect("Could not execute request");
3640        state
3641            .dispatch(
3642                &RequestType::AddBackend(AddBackend {
3643                    cluster_id: String::from("cluster_2"),
3644                    backend_id: String::from("cluster_2-0"),
3645                    address: SocketAddress::new_v4(192, 167, 1, 2, 1026),
3646                    ..Default::default()
3647                })
3648                .into(),
3649            )
3650            .expect("Could not execute request");
3651        state
3652            .dispatch(
3653                &RequestType::AddBackend(AddBackend {
3654                    cluster_id: String::from("cluster_1"),
3655                    backend_id: String::from("cluster_1-3"),
3656                    address: SocketAddress::new_v4(192, 168, 1, 3, 1027),
3657                    ..Default::default()
3658                })
3659                .into(),
3660            )
3661            .expect("Could not execute request");
3662        state
3663            .dispatch(
3664                &RequestType::RemoveBackend(RemoveBackend {
3665                    cluster_id: String::from("cluster_1"),
3666                    backend_id: String::from("cluster_1-3"),
3667                    address: SocketAddress::new_v4(192, 168, 1, 3, 1027),
3668                })
3669                .into(),
3670            )
3671            .expect("Could not execute request");
3672
3673        /*
3674        let encoded = state.encode();
3675        println!("serialized:\n{}", encoded);
3676
3677        let new_state: Option<HttpProxy> = decode_str(&encoded);
3678        println!("deserialized:\n{:?}", new_state);
3679        assert_eq!(new_state, Some(state));
3680        */
3681        //assert!(false);
3682    }
3683
3684    #[test]
3685    fn diff() {
3686        let mut state: ConfigState = Default::default();
3687        state
3688            .dispatch(
3689                &RequestType::AddHttpFrontend(RequestHttpFrontend {
3690                    cluster_id: Some(String::from("cluster_1")),
3691                    hostname: String::from("lolcatho.st:8080"),
3692                    path: PathRule::prefix(String::from("/")),
3693                    address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
3694                    position: RulePosition::Post.into(),
3695                    ..Default::default()
3696                })
3697                .into(),
3698            )
3699            .expect("Could not execute request");
3700        state
3701            .dispatch(
3702                &RequestType::AddHttpFrontend(RequestHttpFrontend {
3703                    cluster_id: Some(String::from("cluster_2")),
3704                    hostname: String::from("test.local"),
3705                    path: PathRule::prefix(String::from("/abc")),
3706                    address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
3707                    ..Default::default()
3708                })
3709                .into(),
3710            )
3711            .expect("Could not execute request");
3712        state
3713            .dispatch(
3714                &RequestType::AddBackend(AddBackend {
3715                    cluster_id: String::from("cluster_1"),
3716                    backend_id: String::from("cluster_1-0"),
3717                    address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
3718                    load_balancing_parameters: Some(LoadBalancingParams::default()),
3719                    ..Default::default()
3720                })
3721                .into(),
3722            )
3723            .expect("Could not execute request");
3724        state
3725            .dispatch(
3726                &RequestType::AddBackend(AddBackend {
3727                    cluster_id: String::from("cluster_1"),
3728                    backend_id: String::from("cluster_1-1"),
3729                    address: SocketAddress::new_v4(127, 0, 0, 2, 1027),
3730                    load_balancing_parameters: Some(LoadBalancingParams::default()),
3731                    ..Default::default()
3732                })
3733                .into(),
3734            )
3735            .expect("Could not execute request");
3736        state
3737            .dispatch(
3738                &RequestType::AddBackend(AddBackend {
3739                    cluster_id: String::from("cluster_2"),
3740                    backend_id: String::from("cluster_2-0"),
3741                    address: SocketAddress::new_v4(192, 167, 1, 2, 1026),
3742                    load_balancing_parameters: Some(LoadBalancingParams::default()),
3743                    ..Default::default()
3744                })
3745                .into(),
3746            )
3747            .expect("Could not execute request");
3748        state
3749            .dispatch(
3750                &RequestType::AddCluster(Cluster {
3751                    cluster_id: String::from("cluster_2"),
3752                    sticky_session: true,
3753                    https_redirect: true,
3754                    ..Default::default()
3755                })
3756                .into(),
3757            )
3758            .expect("Could not execute request");
3759
3760        let mut state2: ConfigState = Default::default();
3761        state2
3762            .dispatch(
3763                &RequestType::AddHttpFrontend(RequestHttpFrontend {
3764                    cluster_id: Some(String::from("cluster_1")),
3765                    hostname: String::from("lolcatho.st:8080"),
3766                    path: PathRule::prefix(String::from("/")),
3767                    address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
3768                    position: RulePosition::Post.into(),
3769                    ..Default::default()
3770                })
3771                .into(),
3772            )
3773            .expect("Could not execute request");
3774        state2
3775            .dispatch(
3776                &RequestType::AddBackend(AddBackend {
3777                    cluster_id: String::from("cluster_1"),
3778                    backend_id: String::from("cluster_1-0"),
3779                    address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
3780                    load_balancing_parameters: Some(LoadBalancingParams::default()),
3781                    ..Default::default()
3782                })
3783                .into(),
3784            )
3785            .expect("Could not execute request");
3786        state2
3787            .dispatch(
3788                &RequestType::AddBackend(AddBackend {
3789                    cluster_id: String::from("cluster_1"),
3790                    backend_id: String::from("cluster_1-1"),
3791                    address: SocketAddress::new_v4(127, 0, 0, 2, 1027),
3792                    load_balancing_parameters: Some(LoadBalancingParams::default()),
3793                    ..Default::default()
3794                })
3795                .into(),
3796            )
3797            .expect("Could not execute request");
3798        state2
3799            .dispatch(
3800                &RequestType::AddBackend(AddBackend {
3801                    cluster_id: String::from("cluster_1"),
3802                    backend_id: String::from("cluster_1-2"),
3803                    address: SocketAddress::new_v4(127, 0, 0, 2, 1028),
3804                    load_balancing_parameters: Some(LoadBalancingParams::default()),
3805                    ..Default::default()
3806                })
3807                .into(),
3808            )
3809            .expect("Could not execute request");
3810        state2
3811            .dispatch(
3812                &RequestType::AddCluster(Cluster {
3813                    cluster_id: String::from("cluster_3"),
3814                    sticky_session: false,
3815                    https_redirect: false,
3816                    ..Default::default()
3817                })
3818                .into(),
3819            )
3820            .expect("Could not execute request");
3821
3822        let e: Vec<Request> = vec![
3823            RequestType::RemoveHttpFrontend(RequestHttpFrontend {
3824                cluster_id: Some(String::from("cluster_2")),
3825                hostname: String::from("test.local"),
3826                path: PathRule::prefix(String::from("/abc")),
3827                address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
3828                ..Default::default()
3829            })
3830            .into(),
3831            RequestType::RemoveBackend(RemoveBackend {
3832                cluster_id: String::from("cluster_2"),
3833                backend_id: String::from("cluster_2-0"),
3834                address: SocketAddress::new_v4(192, 167, 1, 2, 1026),
3835            })
3836            .into(),
3837            RequestType::AddBackend(AddBackend {
3838                cluster_id: String::from("cluster_1"),
3839                backend_id: String::from("cluster_1-2"),
3840                address: SocketAddress::new_v4(127, 0, 0, 2, 1028),
3841                load_balancing_parameters: Some(LoadBalancingParams::default()),
3842                ..Default::default()
3843            })
3844            .into(),
3845            RequestType::RemoveCluster(String::from("cluster_2")).into(),
3846            RequestType::AddCluster(Cluster {
3847                cluster_id: String::from("cluster_3"),
3848                sticky_session: false,
3849                https_redirect: false,
3850                ..Default::default()
3851            })
3852            .into(),
3853        ];
3854        let expected_diff: HashSet<&Request> = HashSet::from_iter(e.iter());
3855
3856        let d = state.diff(&state2);
3857        let diff = HashSet::from_iter(d.iter());
3858        println!("diff requests:\n{diff:#?}\n");
3859        println!("expected diff requests:\n{expected_diff:#?}\n");
3860
3861        let hash1 = state.hash_state();
3862        let hash2 = state2.hash_state();
3863        let mut state3 = state.clone();
3864        state3
3865            .dispatch(
3866                &RequestType::AddBackend(AddBackend {
3867                    cluster_id: String::from("cluster_1"),
3868                    backend_id: String::from("cluster_1-2"),
3869                    address: SocketAddress::new_v4(127, 0, 0, 2, 1028),
3870                    load_balancing_parameters: Some(LoadBalancingParams::default()),
3871                    ..Default::default()
3872                })
3873                .into(),
3874            )
3875            .expect("Could not execute request");
3876        let hash3 = state3.hash_state();
3877        println!("state 1 hashes: {hash1:#?}");
3878        println!("state 2 hashes: {hash2:#?}");
3879        println!("state 3 hashes: {hash3:#?}");
3880
3881        assert_eq!(diff, expected_diff);
3882    }
3883
3884    #[test]
3885    fn cluster_ids_by_domain() {
3886        let mut config = ConfigState::new();
3887        let http_front_cluster1 = RequestHttpFrontend {
3888            cluster_id: Some(String::from("MyCluster_1")),
3889            hostname: String::from("lolcatho.st"),
3890            path: PathRule::prefix(String::from("")),
3891            address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
3892            ..Default::default()
3893        };
3894
3895        let https_front_cluster1 = RequestHttpFrontend {
3896            cluster_id: Some(String::from("MyCluster_1")),
3897            hostname: String::from("lolcatho.st"),
3898            path: PathRule::prefix(String::from("")),
3899            address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
3900            ..Default::default()
3901        };
3902
3903        let http_front_cluster2 = RequestHttpFrontend {
3904            cluster_id: Some(String::from("MyCluster_2")),
3905            hostname: String::from("lolcatho.st"),
3906            path: PathRule::prefix(String::from("/api")),
3907            address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
3908            ..Default::default()
3909        };
3910
3911        let https_front_cluster2 = RequestHttpFrontend {
3912            cluster_id: Some(String::from("MyCluster_2")),
3913            hostname: String::from("lolcatho.st"),
3914            path: PathRule::prefix(String::from("/api")),
3915            address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
3916            ..Default::default()
3917        };
3918
3919        config
3920            .dispatch(&RequestType::AddHttpFrontend(http_front_cluster1).into())
3921            .expect("Could not execute request");
3922        config
3923            .dispatch(&RequestType::AddHttpFrontend(http_front_cluster2).into())
3924            .expect("Could not execute request");
3925        config
3926            .dispatch(&RequestType::AddHttpsFrontend(https_front_cluster1).into())
3927            .expect("Could not execute request");
3928        config
3929            .dispatch(&RequestType::AddHttpsFrontend(https_front_cluster2).into())
3930            .expect("Could not execute request");
3931
3932        let mut cluster1_cluster2: HashSet<ClusterId> = HashSet::new();
3933        cluster1_cluster2.insert(String::from("MyCluster_1"));
3934        cluster1_cluster2.insert(String::from("MyCluster_2"));
3935
3936        let mut cluster2: HashSet<ClusterId> = HashSet::new();
3937        cluster2.insert(String::from("MyCluster_2"));
3938
3939        let empty: HashSet<ClusterId> = HashSet::new();
3940        assert_eq!(
3941            config.get_cluster_ids_by_domain(String::from("lolcatho.st"), None),
3942            cluster1_cluster2
3943        );
3944        assert_eq!(
3945            config
3946                .get_cluster_ids_by_domain(String::from("lolcatho.st"), Some(String::from("/api"))),
3947            cluster2
3948        );
3949        assert_eq!(
3950            config.get_cluster_ids_by_domain(String::from("lolcathost"), None),
3951            empty
3952        );
3953        assert_eq!(
3954            config
3955                .get_cluster_ids_by_domain(String::from("lolcathost"), Some(String::from("/sozu"))),
3956            empty
3957        );
3958    }
3959
3960    #[test]
3961    fn duplicate_backends() {
3962        let mut state: ConfigState = Default::default();
3963        state
3964            .dispatch(
3965                &RequestType::AddBackend(AddBackend {
3966                    cluster_id: String::from("cluster_1"),
3967                    backend_id: String::from("cluster_1-0"),
3968                    address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
3969                    load_balancing_parameters: Some(LoadBalancingParams::default()),
3970                    ..Default::default()
3971                })
3972                .into(),
3973            )
3974            .expect("Could not execute request");
3975
3976        let b = Backend {
3977            cluster_id: String::from("cluster_1"),
3978            backend_id: String::from("cluster_1-0"),
3979            address: "127.0.0.1:1026".parse().unwrap(),
3980            load_balancing_parameters: Some(LoadBalancingParams::default()),
3981            sticky_id: Some("sticky".to_string()),
3982            backup: None,
3983        };
3984
3985        state
3986            .dispatch(&RequestType::AddBackend(b.clone().to_add_backend()).into())
3987            .expect("Could not execute order");
3988
3989        assert_eq!(state.backends.get("cluster_1").unwrap(), &vec![b]);
3990    }
3991
3992    #[test]
3993    fn remove_backend() {
3994        let mut state: ConfigState = Default::default();
3995        state
3996            .dispatch(
3997                &RequestType::AddCluster(Cluster {
3998                    cluster_id: String::from("cluster_1"),
3999                    ..Default::default()
4000                })
4001                .into(),
4002            )
4003            .expect("Could not execute request");
4004
4005        for i in 0..10 {
4006            state
4007                .dispatch(
4008                    &RequestType::AddBackend(AddBackend {
4009                        cluster_id: String::from("cluster_1"),
4010                        backend_id: format!("cluster_1-{i}"),
4011                        address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
4012                        ..Default::default()
4013                    })
4014                    .into(),
4015                )
4016                .expect("Could not execute request");
4017        }
4018
4019        assert_eq!(state.backends.get("cluster_1").unwrap().len(), 10);
4020
4021        let remove_backend_2 = RequestType::RemoveBackend(RemoveBackend {
4022            cluster_id: String::from("cluster_1"),
4023            backend_id: String::from("cluster_1-0"),
4024            address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
4025        })
4026        .into();
4027
4028        let remove_backend_result = state.dispatch(&remove_backend_2);
4029
4030        assert!(remove_backend_result.is_ok());
4031        assert_eq!(state.backends.get("cluster_1").unwrap().len(), 9);
4032
4033        let redundant_remove = state.dispatch(&remove_backend_2);
4034        assert!(matches!(redundant_remove, Err(StateError::NoChange)));
4035        assert_eq!(state.backends.get("cluster_1").unwrap().len(), 9);
4036    }
4037
4038    #[test]
4039    fn remove_backends_randomly() {
4040        let mut state: ConfigState = Default::default();
4041        state
4042            .dispatch(
4043                &RequestType::AddCluster(Cluster {
4044                    cluster_id: String::from("cluster_1"),
4045                    ..Default::default()
4046                })
4047                .into(),
4048            )
4049            .expect("Could not execute request");
4050
4051        for _ in 0..1000 {
4052            for i in 0..10 {
4053                state
4054                    .dispatch(
4055                        &RequestType::AddBackend(AddBackend {
4056                            cluster_id: String::from("cluster_1"),
4057                            backend_id: format!("cluster_1-{i}"),
4058                            address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
4059                            ..Default::default()
4060                        })
4061                        .into(),
4062                    )
4063                    .expect("Could not execute request");
4064            }
4065
4066            let mut rng = rng();
4067            let mut indexes = vec![0, 1, 2, 3, 4, 5, 6, 7, 8, 9];
4068            indexes.shuffle(&mut rng);
4069            let random_count = rng.random_range(1..indexes.len());
4070            let random_indexes: Vec<i32> = indexes.into_iter().take(random_count).collect();
4071
4072            for j in random_indexes {
4073                let remove_backend_result = state.dispatch(
4074                    &RequestType::RemoveBackend(RemoveBackend {
4075                        cluster_id: String::from("cluster_1"),
4076                        backend_id: format!("cluster_1-{j}"),
4077                        address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
4078                    })
4079                    .into(),
4080                );
4081                assert!(remove_backend_result.is_ok());
4082            }
4083        }
4084    }
4085
4086    #[test]
4087    fn listener_diff() {
4088        let mut state: ConfigState = Default::default();
4089        let custom_http_answers = Some(CustomHttpAnswers {
4090            answer_404: Some("test".to_string()),
4091            ..Default::default()
4092        });
4093        state
4094            .dispatch(
4095                &RequestType::AddTcpListener(TcpListenerConfig {
4096                    address: SocketAddress::new_v4(0, 0, 0, 0, 1234),
4097                    ..Default::default()
4098                })
4099                .into(),
4100            )
4101            .expect("Could not execute request");
4102        state
4103            .dispatch(
4104                &RequestType::ActivateListener(ActivateListener {
4105                    address: SocketAddress::new_v4(0, 0, 0, 0, 1234),
4106                    proxy: ListenerType::Tcp.into(),
4107                    from_scm: false,
4108                })
4109                .into(),
4110            )
4111            .expect("Could not execute request");
4112        state
4113            .dispatch(
4114                &RequestType::AddHttpListener(HttpListenerConfig {
4115                    address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
4116                    ..Default::default()
4117                })
4118                .into(),
4119            )
4120            .expect("Could not execute request");
4121        state
4122            .dispatch(
4123                &RequestType::AddHttpsListener(HttpsListenerConfig {
4124                    address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
4125                    ..Default::default()
4126                })
4127                .into(),
4128            )
4129            .expect("Could not execute request");
4130        state
4131            .dispatch(
4132                &RequestType::ActivateListener(ActivateListener {
4133                    address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
4134                    proxy: ListenerType::Https.into(),
4135                    from_scm: false,
4136                })
4137                .into(),
4138            )
4139            .expect("Could not execute request");
4140
4141        let mut state2: ConfigState = Default::default();
4142        state2
4143            .dispatch(
4144                &RequestType::AddTcpListener(TcpListenerConfig {
4145                    address: SocketAddress::new_v4(0, 0, 0, 0, 1234),
4146                    expect_proxy: true,
4147                    ..Default::default()
4148                })
4149                .into(),
4150            )
4151            .expect("Could not execute request");
4152        state2
4153            .dispatch(
4154                &RequestType::AddHttpListener(HttpListenerConfig {
4155                    address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
4156                    http_answers: custom_http_answers.clone(),
4157                    ..Default::default()
4158                })
4159                .into(),
4160            )
4161            .expect("Could not execute request");
4162        state2
4163            .dispatch(
4164                &RequestType::ActivateListener(ActivateListener {
4165                    address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
4166                    proxy: ListenerType::Http.into(),
4167                    from_scm: false,
4168                })
4169                .into(),
4170            )
4171            .expect("Could not execute request");
4172        state2
4173            .dispatch(
4174                &RequestType::AddHttpsListener(HttpsListenerConfig {
4175                    address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
4176                    http_answers: custom_http_answers.clone(),
4177                    ..Default::default()
4178                })
4179                .into(),
4180            )
4181            .expect("Could not execute request");
4182        state2
4183            .dispatch(
4184                &RequestType::ActivateListener(ActivateListener {
4185                    address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
4186                    proxy: ListenerType::Https.into(),
4187                    from_scm: false,
4188                })
4189                .into(),
4190            )
4191            .expect("Could not execute request");
4192
4193        let e: Vec<Request> = vec![
4194            RequestType::RemoveListener(RemoveListener {
4195                address: SocketAddress::new_v4(0, 0, 0, 0, 1234),
4196                proxy: ListenerType::Tcp.into(),
4197            })
4198            .into(),
4199            RequestType::AddTcpListener(TcpListenerConfig {
4200                address: SocketAddress::new_v4(0, 0, 0, 0, 1234),
4201                expect_proxy: true,
4202                ..Default::default()
4203            })
4204            .into(),
4205            RequestType::DeactivateListener(DeactivateListener {
4206                address: SocketAddress::new_v4(0, 0, 0, 0, 1234),
4207                proxy: ListenerType::Tcp.into(),
4208                to_scm: false,
4209            })
4210            .into(),
4211            RequestType::RemoveListener(RemoveListener {
4212                address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
4213                proxy: ListenerType::Http.into(),
4214            })
4215            .into(),
4216            RequestType::AddHttpListener(HttpListenerConfig {
4217                address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
4218                http_answers: custom_http_answers.clone(),
4219                ..Default::default()
4220            })
4221            .into(),
4222            RequestType::ActivateListener(ActivateListener {
4223                address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
4224                proxy: ListenerType::Http.into(),
4225                from_scm: false,
4226            })
4227            .into(),
4228            RequestType::RemoveListener(RemoveListener {
4229                address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
4230                proxy: ListenerType::Https.into(),
4231            })
4232            .into(),
4233            RequestType::AddHttpsListener(HttpsListenerConfig {
4234                address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
4235                http_answers: custom_http_answers.clone(),
4236                ..Default::default()
4237            })
4238            .into(),
4239            // The 8443 HTTPS listener is active in both states but its content
4240            // changed (custom answers). The Remove + Add(active=false) above
4241            // wipes the active flag, so diff must re-emit an ActivateListener to
4242            // keep the listener live across the hot reconfig. Without it the
4243            // worker would silently deactivate the listener on replay.
4244            RequestType::ActivateListener(ActivateListener {
4245                address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
4246                proxy: ListenerType::Https.into(),
4247                from_scm: false,
4248            })
4249            .into(),
4250        ];
4251
4252        let diff = state.diff(&state2);
4253        //let diff: HashSet<&RequestContent> = HashSet::from_iter(d.iter());
4254        println!("expected diff requests:\n{e:#?}\n");
4255        println!("diff requests:\n{diff:#?}\n");
4256
4257        let _hash1 = state.hash_state();
4258        let _hash2 = state2.hash_state();
4259
4260        assert_eq!(diff, e);
4261
4262        // Round-trip: replaying diff(state -> state2) onto a clone of `state`
4263        // must reproduce `state2`'s listener maps EXACTLY, active flag included.
4264        // This is the hot-reconfig correctness property — a worker applies these
4265        // requests and must converge on the target state. In particular the 8443
4266        // HTTPS listener (active in both states, content changed) must come back
4267        // ACTIVE, which is the bug the ActivateListener re-emission above fixes.
4268        let mut replayed = state.clone();
4269        for request in &diff {
4270            replayed
4271                .dispatch(request)
4272                .expect("every diff request must replay cleanly onto the source state");
4273        }
4274        assert_eq!(
4275            replayed.tcp_listeners, state2.tcp_listeners,
4276            "replayed tcp_listeners must match the target state"
4277        );
4278        assert_eq!(
4279            replayed.http_listeners, state2.http_listeners,
4280            "replayed http_listeners must match the target state"
4281        );
4282        assert_eq!(
4283            replayed.https_listeners, state2.https_listeners,
4284            "replayed https_listeners must match the target state"
4285        );
4286        // Explicitly assert the still-active listener stays active across the
4287        // config change (the core of the fixed bug).
4288        let replayed_8443 = replayed
4289            .https_listeners
4290            .get(&SocketAddr::from(SocketAddress::new_v4(0, 0, 0, 0, 8443)))
4291            .expect("8443 HTTPS listener must exist after replay");
4292        assert!(
4293            replayed_8443.active,
4294            "the 8443 HTTPS listener must stay ACTIVE across a config change"
4295        );
4296    }
4297
4298    #[test]
4299    fn certificate_retrieval() {
4300        let mut state: ConfigState = Default::default();
4301        let certificate_and_key = CertificateAndKey {
4302            certificate: String::from(include_str!("../assets/certificate.pem")),
4303            key: String::from(include_str!("../assets/key.pem")),
4304            certificate_chain: vec![],
4305            versions: vec![],
4306            names: vec!["lolcatho.st".to_string()],
4307        };
4308        let add_certificate = AddCertificate {
4309            address: SocketAddress::new_v4(127, 0, 0, 1, 8080),
4310            certificate: certificate_and_key,
4311            expired_at: None,
4312        };
4313        state
4314            .dispatch(&RequestType::AddCertificate(add_certificate).into())
4315            .expect("Could not add certificate");
4316
4317        println!("state: {state:#?}");
4318
4319        // let fingerprint: Fingerprint = serde_json::from_str(
4320        //     "\"ab2618b674e15243fd02a5618c66509e4840ba60e7d64cebec84cdbfeceee0c5\"",
4321        // )
4322        // .expect("Could not deserialize the fingerprint");
4323
4324        let certificates_found_by_fingerprint = state.get_certificates(QueryCertificatesFilters {
4325            domain: None,
4326            fingerprint: Some(
4327                "ab2618b674e15243fd02a5618c66509e4840ba60e7d64cebec84cdbfeceee0c5".to_string(),
4328            ),
4329        });
4330
4331        println!("found certificate: {certificates_found_by_fingerprint:#?}");
4332
4333        assert!(!certificates_found_by_fingerprint.is_empty());
4334
4335        let certificate_found_by_domain_name = state.get_certificates(QueryCertificatesFilters {
4336            domain: Some("lolcatho.st".to_string()),
4337            fingerprint: None,
4338        });
4339
4340        assert!(!certificate_found_by_domain_name.is_empty());
4341    }
4342
4343    #[test]
4344    fn count_backends_across_clusters() {
4345        let mut state: ConfigState = Default::default();
4346
4347        assert_eq!(state.count_backends(), 0);
4348
4349        state
4350            .dispatch(
4351                &RequestType::AddBackend(AddBackend {
4352                    cluster_id: String::from("cluster_1"),
4353                    backend_id: String::from("cluster_1-0"),
4354                    address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
4355                    ..Default::default()
4356                })
4357                .into(),
4358            )
4359            .expect("Could not execute request");
4360        assert_eq!(state.count_backends(), 1);
4361
4362        state
4363            .dispatch(
4364                &RequestType::AddBackend(AddBackend {
4365                    cluster_id: String::from("cluster_1"),
4366                    backend_id: String::from("cluster_1-1"),
4367                    address: SocketAddress::new_v4(127, 0, 0, 1, 1027),
4368                    ..Default::default()
4369                })
4370                .into(),
4371            )
4372            .expect("Could not execute request");
4373        assert_eq!(state.count_backends(), 2);
4374
4375        // add backend to a second cluster
4376        state
4377            .dispatch(
4378                &RequestType::AddBackend(AddBackend {
4379                    cluster_id: String::from("cluster_2"),
4380                    backend_id: String::from("cluster_2-0"),
4381                    address: SocketAddress::new_v4(192, 168, 1, 1, 8080),
4382                    ..Default::default()
4383                })
4384                .into(),
4385            )
4386            .expect("Could not execute request");
4387        assert_eq!(state.count_backends(), 3);
4388
4389        // remove a backend and verify count decreases
4390        state
4391            .dispatch(
4392                &RequestType::RemoveBackend(RemoveBackend {
4393                    cluster_id: String::from("cluster_1"),
4394                    backend_id: String::from("cluster_1-0"),
4395                    address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
4396                })
4397                .into(),
4398            )
4399            .expect("Could not execute request");
4400        assert_eq!(state.count_backends(), 2);
4401    }
4402
4403    #[test]
4404    fn count_frontends_across_types() {
4405        let mut state: ConfigState = Default::default();
4406
4407        assert_eq!(state.count_frontends(), 0);
4408
4409        // add an HTTP frontend
4410        state
4411            .dispatch(
4412                &RequestType::AddHttpFrontend(RequestHttpFrontend {
4413                    cluster_id: Some(String::from("cluster_1")),
4414                    hostname: String::from("example.com"),
4415                    path: PathRule::prefix(String::from("/")),
4416                    address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
4417                    position: RulePosition::Tree.into(),
4418                    ..Default::default()
4419                })
4420                .into(),
4421            )
4422            .expect("Could not execute request");
4423        assert_eq!(state.count_frontends(), 1);
4424
4425        // add an HTTPS frontend
4426        state
4427            .dispatch(
4428                &RequestType::AddHttpsFrontend(RequestHttpFrontend {
4429                    cluster_id: Some(String::from("cluster_1")),
4430                    hostname: String::from("secure.example.com"),
4431                    path: PathRule::prefix(String::from("/")),
4432                    address: SocketAddress::new_v4(0, 0, 0, 0, 8443),
4433                    position: RulePosition::Tree.into(),
4434                    ..Default::default()
4435                })
4436                .into(),
4437            )
4438            .expect("Could not execute request");
4439        assert_eq!(state.count_frontends(), 2);
4440
4441        // add a TCP frontend
4442        state
4443            .dispatch(
4444                &RequestType::AddTcpFrontend(RequestTcpFrontend {
4445                    cluster_id: String::from("cluster_2"),
4446                    address: SocketAddress::new_v4(0, 0, 0, 0, 5432),
4447                    ..Default::default()
4448                })
4449                .into(),
4450            )
4451            .expect("Could not execute request");
4452        assert_eq!(state.count_frontends(), 3);
4453
4454        // add a second TCP frontend on the same cluster
4455        state
4456            .dispatch(
4457                &RequestType::AddTcpFrontend(RequestTcpFrontend {
4458                    cluster_id: String::from("cluster_2"),
4459                    address: SocketAddress::new_v4(0, 0, 0, 0, 5433),
4460                    ..Default::default()
4461                })
4462                .into(),
4463            )
4464            .expect("Could not execute request");
4465        assert_eq!(state.count_frontends(), 4);
4466
4467        // remove the HTTP frontend
4468        state
4469            .dispatch(
4470                &RequestType::RemoveHttpFrontend(RequestHttpFrontend {
4471                    cluster_id: Some(String::from("cluster_1")),
4472                    hostname: String::from("example.com"),
4473                    path: PathRule::prefix(String::from("/")),
4474                    address: SocketAddress::new_v4(0, 0, 0, 0, 8080),
4475                    position: RulePosition::Tree.into(),
4476                    ..Default::default()
4477                })
4478                .into(),
4479            )
4480            .expect("Could not execute request");
4481        assert_eq!(state.count_frontends(), 3);
4482    }
4483
4484    // ── helpers ────────────────────────────────────────────────────────────────
4485
4486    fn make_https_listener(address: SocketAddress) -> HttpsListenerConfig {
4487        HttpsListenerConfig {
4488            address,
4489            sticky_name: "SOZUBALANCEID".to_owned(),
4490            front_timeout: 60,
4491            back_timeout: 30,
4492            connect_timeout: 3,
4493            request_timeout: 10,
4494            ..Default::default()
4495        }
4496    }
4497
4498    fn make_http_listener(address: SocketAddress) -> HttpListenerConfig {
4499        HttpListenerConfig {
4500            address,
4501            sticky_name: "SOZUBALANCEID".to_owned(),
4502            front_timeout: 60,
4503            back_timeout: 30,
4504            connect_timeout: 3,
4505            request_timeout: 10,
4506            ..Default::default()
4507        }
4508    }
4509
4510    fn make_tcp_listener(address: SocketAddress) -> TcpListenerConfig {
4511        TcpListenerConfig {
4512            address,
4513            front_timeout: 60,
4514            back_timeout: 30,
4515            connect_timeout: 3,
4516            ..Default::default()
4517        }
4518    }
4519
4520    fn make_udp_listener(address: SocketAddress, active: bool) -> UdpListenerConfig {
4521        UdpListenerConfig {
4522            address,
4523            public_address: None,
4524            front_timeout: 30,
4525            back_timeout: 30,
4526            max_rx_datagram_size: 1500,
4527            max_flows: 0,
4528            active,
4529        }
4530    }
4531
4532    /// Mandatory roundtrip guard for the UDP control-plane data model.
4533    ///
4534    /// 1. Build a state holding an active UDP listener + UDP frontend, run
4535    ///    `generate_requests()`, replay every emitted request into a fresh
4536    ///    `ConfigState`, and assert the two states are byte-for-byte equal.
4537    /// 2. Prove a `diff()` between an empty state and the UDP-bearing state
4538    ///    produces requests that, replayed, reconstruct the UDP listener and
4539    ///    frontend (a hot-add path), and that the reverse diff tears them
4540    ///    down again.
4541    #[test]
4542    fn test_udp_state_roundtrip() {
4543        let address = SocketAddress::new_v4(127, 0, 0, 1, 5353);
4544
4545        let mut state = ConfigState::default();
4546        state
4547            .dispatch(&RequestType::AddUdpListener(make_udp_listener(address, true)).into())
4548            .expect("could not add udp listener");
4549        state
4550            .dispatch(
4551                &RequestType::ActivateListener(ActivateListener {
4552                    address,
4553                    proxy: ListenerType::Udp.into(),
4554                    from_scm: false,
4555                })
4556                .into(),
4557            )
4558            .expect("could not activate udp listener");
4559        state
4560            .dispatch(
4561                &RequestType::AddUdpFrontend(RequestUdpFrontend {
4562                    cluster_id: "udp_cluster".to_string(),
4563                    address,
4564                    tags: BTreeMap::from([("owner".to_string(), "team".to_string())]),
4565                })
4566                .into(),
4567            )
4568            .expect("could not add udp frontend");
4569
4570        assert_eq!(state.udp_listeners.len(), 1);
4571        assert!(state.udp_listeners[&address.into()].active);
4572        assert_eq!(
4573            state.udp_fronts.get("udp_cluster").map(Vec::len),
4574            Some(1usize)
4575        );
4576
4577        // `request_counts` is a runtime census side-effect of `dispatch`; it
4578        // diverges by construction whenever the number/shape of replayed
4579        // requests differs from the originals, so compare the logical config
4580        // with the census cleared on both sides.
4581        let logical = |s: &ConfigState| {
4582            let mut c = s.clone();
4583            c.request_counts.clear();
4584            c
4585        };
4586
4587        // 1. generate_requests → replay → equal
4588        let mut replayed = ConfigState::default();
4589        for request in state.generate_requests() {
4590            replayed
4591                .dispatch(&request)
4592                .expect("could not replay generated request");
4593        }
4594        assert_eq!(
4595            logical(&state),
4596            logical(&replayed),
4597            "UDP listener + frontend must survive generate_requests → replay"
4598        );
4599
4600        // 2. diff from empty reconstructs the UDP objects
4601        let empty = ConfigState::default();
4602        let mut from_diff = ConfigState::default();
4603        for request in empty.diff(&state) {
4604            from_diff
4605                .dispatch(&request)
4606                .expect("could not replay diff request");
4607        }
4608        assert_eq!(
4609            logical(&state),
4610            logical(&from_diff),
4611            "diff(empty -> state) must reconstruct the UDP listener + frontend"
4612        );
4613
4614        // reverse diff tears them back down to empty
4615        let mut torn_down = state.clone();
4616        for request in state.diff(&empty) {
4617            torn_down
4618                .dispatch(&request)
4619                .expect("could not replay teardown diff request");
4620        }
4621        assert!(
4622            torn_down.udp_listeners.is_empty(),
4623            "diff(state -> empty) must remove the UDP listener"
4624        );
4625        assert!(
4626            torn_down
4627                .udp_fronts
4628                .get("udp_cluster")
4629                .map(Vec::is_empty)
4630                .unwrap_or(true),
4631            "diff(state -> empty) must remove the UDP frontend"
4632        );
4633
4634        // update path: a partial patch mutates the stored listener in place
4635        state
4636            .dispatch(
4637                &RequestType::UpdateUdpListener(UpdateUdpListenerConfig {
4638                    address,
4639                    max_flows: Some(4096),
4640                    front_timeout: Some(15),
4641                    ..Default::default()
4642                })
4643                .into(),
4644            )
4645            .expect("could not update udp listener");
4646        let updated = &state.udp_listeners[&address.into()];
4647        assert_eq!(updated.max_flows, 4096);
4648        assert_eq!(updated.front_timeout, 15);
4649        assert_eq!(
4650            updated.back_timeout, 30,
4651            "unpatched field must be preserved"
4652        );
4653    }
4654
4655    /// Mandatory roundtrip + identity guard for SNI/ALPN-scoped TCP
4656    /// frontends (sozu-proxy/sozu#1279).
4657    ///
4658    /// 1. Build a state holding a TCP listener with non-default SNI-preread
4659    ///    knobs plus two SNI-scoped frontends on the same address, run
4660    ///    `generate_requests()`, replay every emitted request into a fresh
4661    ///    `ConfigState`, and assert the two states are byte-for-byte equal.
4662    /// 2. Prove `diff()` between an empty state and this state reconstructs
4663    ///    the listener + both frontends (hot-add), and the reverse diff
4664    ///    tears them back down.
4665    /// 3. Prove add/remove identity is `(address, sni, alpn)`: removing one
4666    ///    SNI-scoped sibling leaves the other untouched, and a
4667    ///    address-only removal (mismatched sni/alpn) is rejected rather
4668    ///    than silently nuking every frontend at that address.
4669    #[test]
4670    fn test_tcp_sni_alpn_state_roundtrip() {
4671        let address = SocketAddress::new_v4(127, 0, 0, 1, 6443);
4672
4673        let mut state = ConfigState::default();
4674        state
4675            .dispatch(
4676                &RequestType::AddTcpListener(TcpListenerConfig {
4677                    address,
4678                    front_timeout: 60,
4679                    back_timeout: 30,
4680                    connect_timeout: 3,
4681                    active: true,
4682                    sni_preread_timeout: Some(2),
4683                    sni_preread_max_bytes: Some(8192),
4684                    ..Default::default()
4685                })
4686                .into(),
4687            )
4688            .expect("could not add tcp listener");
4689        state
4690            .dispatch(
4691                &RequestType::ActivateListener(ActivateListener {
4692                    address,
4693                    proxy: ListenerType::Tcp.into(),
4694                    from_scm: false,
4695                })
4696                .into(),
4697            )
4698            .expect("could not activate tcp listener");
4699
4700        let frontend_a = RequestTcpFrontend {
4701            cluster_id: "cluster_sni_a".to_string(),
4702            address,
4703            tags: BTreeMap::new(),
4704            sni: Some("example.com".to_string()),
4705            alpn: vec!["h2".to_string()],
4706        };
4707        let frontend_b = RequestTcpFrontend {
4708            cluster_id: "cluster_sni_b".to_string(),
4709            address,
4710            tags: BTreeMap::new(),
4711            sni: Some("other.example.com".to_string()),
4712            alpn: vec![],
4713        };
4714        state
4715            .dispatch(&RequestType::AddTcpFrontend(frontend_a.clone()).into())
4716            .expect("could not add sni-a tcp frontend");
4717        state
4718            .dispatch(&RequestType::AddTcpFrontend(frontend_b.clone()).into())
4719            .expect("could not add sni-b tcp frontend");
4720
4721        assert_eq!(state.tcp_listeners.len(), 1);
4722        assert!(state.tcp_listeners[&address.into()].active);
4723        assert_eq!(
4724            state.tcp_listeners[&address.into()].sni_preread_timeout,
4725            Some(2)
4726        );
4727        assert_eq!(
4728            state.tcp_listeners[&address.into()].sni_preread_max_bytes,
4729            Some(8192)
4730        );
4731        assert_eq!(state.count_tcp_frontends_raw(), 2);
4732
4733        // `request_counts` is a runtime census side-effect of `dispatch`; it
4734        // diverges by construction whenever the number/shape of replayed
4735        // requests differs from the originals, so compare the logical config
4736        // with the census cleared on both sides.
4737        let logical = |s: &ConfigState| {
4738            let mut c = s.clone();
4739            c.request_counts.clear();
4740            c
4741        };
4742
4743        // 1. generate_requests → replay → equal
4744        let mut replayed = ConfigState::default();
4745        for request in state.generate_requests() {
4746            replayed
4747                .dispatch(&request)
4748                .expect("could not replay generated request");
4749        }
4750        assert_eq!(
4751            logical(&state),
4752            logical(&replayed),
4753            "SNI/ALPN TCP listener + frontends must survive generate_requests → replay"
4754        );
4755
4756        // 2. diff from empty reconstructs the TCP objects
4757        let empty = ConfigState::default();
4758        let mut from_diff = ConfigState::default();
4759        for request in empty.diff(&state) {
4760            from_diff
4761                .dispatch(&request)
4762                .expect("could not replay diff request");
4763        }
4764        assert_eq!(
4765            logical(&state),
4766            logical(&from_diff),
4767            "diff(empty -> state) must reconstruct the tcp listener + both sni frontends"
4768        );
4769
4770        // reverse diff tears them back down to empty
4771        let mut torn_down = state.clone();
4772        for request in state.diff(&empty) {
4773            torn_down
4774                .dispatch(&request)
4775                .expect("could not replay teardown diff request");
4776        }
4777        assert!(
4778            torn_down.tcp_listeners.is_empty(),
4779            "diff(state -> empty) must remove the tcp listener"
4780        );
4781        assert_eq!(
4782            torn_down.count_tcp_frontends_raw(),
4783            0,
4784            "diff(state -> empty) must remove both sni frontends"
4785        );
4786
4787        // 3a. an address-only removal (mismatched sni/alpn) must not match
4788        // either sibling — identity is (address, sni, alpn), not address
4789        // alone.
4790        let mismatched_removal = state.dispatch(
4791            &RequestType::RemoveTcpFrontend(RequestTcpFrontend {
4792                cluster_id: "cluster_sni_a".to_string(),
4793                address,
4794                tags: BTreeMap::new(),
4795                sni: None,
4796                alpn: vec![],
4797            })
4798            .into(),
4799        );
4800        assert!(
4801            mismatched_removal.is_err(),
4802            "removing by address alone (no matching sni/alpn) must be rejected"
4803        );
4804        assert_eq!(
4805            state.count_tcp_frontends_raw(),
4806            2,
4807            "a rejected mismatched removal must not change either sibling"
4808        );
4809
4810        // 3b. removing the exact (address, sni, alpn) identity of frontend_a
4811        // must drop only that entry and leave frontend_b untouched.
4812        state
4813            .dispatch(&RequestType::RemoveTcpFrontend(frontend_a).into())
4814            .expect("could not remove sni-a tcp frontend");
4815        assert_eq!(state.count_tcp_frontends_raw(), 1);
4816        let remaining = state
4817            .tcp_fronts
4818            .get("cluster_sni_b")
4819            .expect("cluster_sni_b bucket must survive");
4820        assert_eq!(remaining.len(), 1);
4821        assert_eq!(remaining[0].sni, Some("other.example.com".to_string()));
4822    }
4823
4824    // ── master-side TCP frontend admission parity (sozu-proxy/sozu#1290) ──────
4825    //
4826    // The command server mutates master `ConfigState` before fanning a
4827    // request out to workers, and a worker NACK never rolls the master back.
4828    // These tests prove the master rejects everything the worker's
4829    // `validate_new_tcp_front` (`lib/src/tcp.rs`) would reject, BEFORE any
4830    // mutation -- every rejection case asserts `hash_state()` and
4831    // `count_tcp_frontends_raw()` (plus the relevant bucket lengths) are
4832    // byte-for-byte unchanged, not just that the call returned `Err`.
4833
4834    fn add_tcp_test_cluster(state: &mut ConfigState, cluster_id: &str) {
4835        state
4836            .dispatch(
4837                &RequestType::AddCluster(Cluster {
4838                    cluster_id: cluster_id.to_string(),
4839                    sticky_session: false,
4840                    https_redirect: false,
4841                    ..Default::default()
4842                })
4843                .into(),
4844            )
4845            .expect("could not add cluster");
4846    }
4847
4848    /// The (address, sni, alpn) identity check must span ALL clusters at a
4849    /// listener address, not just the candidate's own `cluster_id` bucket --
4850    /// otherwise the same identity under a different cluster_id is silently
4851    /// accepted by the master and then rejected by every worker.
4852    #[test]
4853    fn add_tcp_frontend_rejects_cross_cluster_duplicate_identity() {
4854        let address = SocketAddress::new_v4(127, 0, 0, 1, 9001);
4855        let mut state = ConfigState::new();
4856        add_tcp_test_cluster(&mut state, "cluster_x");
4857        add_tcp_test_cluster(&mut state, "cluster_y");
4858
4859        state
4860            .dispatch(
4861                &RequestType::AddTcpFrontend(RequestTcpFrontend {
4862                    cluster_id: "cluster_x".to_string(),
4863                    address,
4864                    tags: BTreeMap::new(),
4865                    sni: Some("example.com".to_string()),
4866                    alpn: vec!["h2".to_string()],
4867                })
4868                .into(),
4869            )
4870            .expect("could not add cluster_x tcp frontend");
4871
4872        let hash_before = state.hash_state();
4873        let count_before = state.count_tcp_frontends_raw();
4874
4875        let result = state.dispatch(
4876            &RequestType::AddTcpFrontend(RequestTcpFrontend {
4877                cluster_id: "cluster_y".to_string(),
4878                address,
4879                tags: BTreeMap::new(),
4880                sni: Some("example.com".to_string()),
4881                alpn: vec!["h2".to_string()],
4882            })
4883            .into(),
4884        );
4885        assert!(
4886            result.is_err(),
4887            "the same (address, sni, alpn) under a different cluster_id must be rejected, \
4888             got: {result:?}"
4889        );
4890        assert_eq!(
4891            state.hash_state(),
4892            hash_before,
4893            "a rejected cross-cluster duplicate must not change the state hash"
4894        );
4895        assert_eq!(
4896            state.count_tcp_frontends_raw(),
4897            count_before,
4898            "a rejected cross-cluster duplicate must not change the frontend count"
4899        );
4900        assert_eq!(state.tcp_fronts.get("cluster_x").map(Vec::len), Some(1));
4901        assert_eq!(
4902            state.tcp_fronts.get("cluster_y").map(Vec::len).unwrap_or(0),
4903            0
4904        );
4905    }
4906
4907    /// ALPN identity is a *set*: the same protocols in a different order
4908    /// must still collide as a duplicate, even within one cluster's bucket.
4909    #[test]
4910    fn add_tcp_frontend_rejects_reordered_alpn_duplicate() {
4911        let address = SocketAddress::new_v4(127, 0, 0, 1, 9002);
4912        let mut state = ConfigState::new();
4913        add_tcp_test_cluster(&mut state, "cluster_reorder");
4914
4915        state
4916            .dispatch(
4917                &RequestType::AddTcpFrontend(RequestTcpFrontend {
4918                    cluster_id: "cluster_reorder".to_string(),
4919                    address,
4920                    tags: BTreeMap::new(),
4921                    sni: Some("example.com".to_string()),
4922                    alpn: vec!["h2".to_string(), "http/1.1".to_string()],
4923                })
4924                .into(),
4925            )
4926            .expect("could not add first tcp frontend");
4927
4928        let hash_before = state.hash_state();
4929        let count_before = state.count_tcp_frontends_raw();
4930
4931        let result = state.dispatch(
4932            &RequestType::AddTcpFrontend(RequestTcpFrontend {
4933                cluster_id: "cluster_reorder".to_string(),
4934                address,
4935                tags: BTreeMap::new(),
4936                sni: Some("example.com".to_string()),
4937                alpn: vec!["http/1.1".to_string(), "h2".to_string()],
4938            })
4939            .into(),
4940        );
4941        assert!(
4942            result.is_err(),
4943            "the same ALPN set in a different order must be rejected as a duplicate \
4944             identity, got: {result:?}"
4945        );
4946        assert_eq!(state.hash_state(), hash_before);
4947        assert_eq!(state.count_tcp_frontends_raw(), count_before);
4948        assert_eq!(
4949            state.tcp_fronts.get("cluster_reorder").map(Vec::len),
4950            Some(1)
4951        );
4952    }
4953
4954    /// Symmetric counterpart: removal must also treat ALPN as a set, so a
4955    /// request differing only in protocol order still matches the stored
4956    /// (canonical) frontend instead of falling through to `NoChange`.
4957    #[test]
4958    fn remove_tcp_frontend_matches_reordered_alpn() {
4959        let address = SocketAddress::new_v4(127, 0, 0, 1, 9003);
4960        let mut state = ConfigState::new();
4961        add_tcp_test_cluster(&mut state, "cluster_remove_reorder");
4962
4963        state
4964            .dispatch(
4965                &RequestType::AddTcpFrontend(RequestTcpFrontend {
4966                    cluster_id: "cluster_remove_reorder".to_string(),
4967                    address,
4968                    tags: BTreeMap::new(),
4969                    sni: Some("example.com".to_string()),
4970                    alpn: vec!["h2".to_string(), "http/1.1".to_string()],
4971                })
4972                .into(),
4973            )
4974            .expect("could not add tcp frontend");
4975
4976        let result = state.dispatch(
4977            &RequestType::RemoveTcpFrontend(RequestTcpFrontend {
4978                cluster_id: "cluster_remove_reorder".to_string(),
4979                address,
4980                tags: BTreeMap::new(),
4981                sni: Some("example.com".to_string()),
4982                alpn: vec!["http/1.1".to_string(), "h2".to_string()],
4983            })
4984            .into(),
4985        );
4986        assert!(
4987            result.is_ok(),
4988            "removing with the same ALPN set in a different order must succeed, got: {result:?}"
4989        );
4990        assert_eq!(state.count_tcp_frontends_raw(), 0);
4991    }
4992
4993    /// ALPN-set overlap on the same (address, normalized sni) must be
4994    /// rejected even across different clusters -- the previous check never
4995    /// looked outside the candidate's own cluster bucket at all.
4996    #[test]
4997    fn add_tcp_frontend_rejects_alpn_overlap_across_clusters_same_sni() {
4998        let address = SocketAddress::new_v4(127, 0, 0, 1, 9004);
4999        let mut state = ConfigState::new();
5000        add_tcp_test_cluster(&mut state, "cluster_overlap_a");
5001        add_tcp_test_cluster(&mut state, "cluster_overlap_b");
5002
5003        state
5004            .dispatch(
5005                &RequestType::AddTcpFrontend(RequestTcpFrontend {
5006                    cluster_id: "cluster_overlap_a".to_string(),
5007                    address,
5008                    tags: BTreeMap::new(),
5009                    sni: Some("example.com".to_string()),
5010                    alpn: vec!["h2".to_string()],
5011                })
5012                .into(),
5013            )
5014            .expect("could not add cluster_overlap_a tcp frontend");
5015
5016        let hash_before = state.hash_state();
5017        let count_before = state.count_tcp_frontends_raw();
5018
5019        let result = state.dispatch(
5020            &RequestType::AddTcpFrontend(RequestTcpFrontend {
5021                cluster_id: "cluster_overlap_b".to_string(),
5022                address,
5023                tags: BTreeMap::new(),
5024                sni: Some("example.com".to_string()),
5025                alpn: vec!["h2".to_string(), "http/1.1".to_string()],
5026            })
5027            .into(),
5028        );
5029        assert!(
5030            result.is_err(),
5031            "an overlapping ALPN set on the same (address, sni) under a different cluster_id \
5032             must be rejected, got: {result:?}"
5033        );
5034        assert_eq!(state.hash_state(), hash_before);
5035        assert_eq!(state.count_tcp_frontends_raw(), count_before);
5036        assert_eq!(
5037            state
5038                .tcp_fronts
5039                .get("cluster_overlap_b")
5040                .map(Vec::len)
5041                .unwrap_or(0),
5042            0
5043        );
5044    }
5045
5046    /// Listener-wide no-SNI/SNI mixing must be enforced in both directions,
5047    /// across cluster buckets.
5048    #[test]
5049    fn add_tcp_frontend_rejects_no_sni_sni_mixing_both_directions() {
5050        let address_a = SocketAddress::new_v4(127, 0, 0, 1, 9005);
5051        let address_b = SocketAddress::new_v4(127, 0, 0, 1, 9006);
5052        let mut state = ConfigState::new();
5053        add_tcp_test_cluster(&mut state, "cluster_mix_1");
5054        add_tcp_test_cluster(&mut state, "cluster_mix_2");
5055        add_tcp_test_cluster(&mut state, "cluster_mix_3");
5056        add_tcp_test_cluster(&mut state, "cluster_mix_4");
5057
5058        // Direction 1: a no-SNI frontend exists; adding an SNI-scoped
5059        // frontend at the SAME address must be rejected.
5060        state
5061            .dispatch(
5062                &RequestType::AddTcpFrontend(RequestTcpFrontend {
5063                    cluster_id: "cluster_mix_1".to_string(),
5064                    address: address_a,
5065                    tags: BTreeMap::new(),
5066                    sni: None,
5067                    alpn: vec![],
5068                })
5069                .into(),
5070            )
5071            .expect("could not add no-sni tcp frontend");
5072        let hash_before_a = state.hash_state();
5073        let count_before_a = state.count_tcp_frontends_raw();
5074        let result_a = state.dispatch(
5075            &RequestType::AddTcpFrontend(RequestTcpFrontend {
5076                cluster_id: "cluster_mix_2".to_string(),
5077                address: address_a,
5078                tags: BTreeMap::new(),
5079                sni: Some("example.com".to_string()),
5080                alpn: vec![],
5081            })
5082            .into(),
5083        );
5084        assert!(
5085            result_a.is_err(),
5086            "an SNI-scoped frontend must not be added to a listener with an existing no-SNI \
5087             front, got: {result_a:?}"
5088        );
5089        assert_eq!(state.hash_state(), hash_before_a);
5090        assert_eq!(state.count_tcp_frontends_raw(), count_before_a);
5091
5092        // Direction 2: an SNI-scoped frontend exists; adding a no-SNI
5093        // frontend at the SAME address must be rejected.
5094        state
5095            .dispatch(
5096                &RequestType::AddTcpFrontend(RequestTcpFrontend {
5097                    cluster_id: "cluster_mix_3".to_string(),
5098                    address: address_b,
5099                    tags: BTreeMap::new(),
5100                    sni: Some("example.com".to_string()),
5101                    alpn: vec![],
5102                })
5103                .into(),
5104            )
5105            .expect("could not add sni-scoped tcp frontend");
5106        let hash_before_b = state.hash_state();
5107        let count_before_b = state.count_tcp_frontends_raw();
5108        let result_b = state.dispatch(
5109            &RequestType::AddTcpFrontend(RequestTcpFrontend {
5110                cluster_id: "cluster_mix_4".to_string(),
5111                address: address_b,
5112                tags: BTreeMap::new(),
5113                sni: None,
5114                alpn: vec![],
5115            })
5116            .into(),
5117        );
5118        assert!(
5119            result_b.is_err(),
5120            "a no-SNI frontend must not be added to a listener with existing SNI-scoped \
5121             fronts, got: {result_b:?}"
5122        );
5123        assert_eq!(state.hash_state(), hash_before_b);
5124        assert_eq!(state.count_tcp_frontends_raw(), count_before_b);
5125    }
5126
5127    /// A malformed SNI pattern must be rejected by the master itself, not
5128    /// merely by config-load's `to_tcp_front` -- a raw `AddTcpFrontend` over
5129    /// the command socket, or a `LoadState` replay, bypasses `config.rs`
5130    /// entirely.
5131    #[test]
5132    fn add_tcp_frontend_rejects_malformed_sni_shape() {
5133        let address = SocketAddress::new_v4(127, 0, 0, 1, 9007);
5134        let mut state = ConfigState::new();
5135        add_tcp_test_cluster(&mut state, "cluster_malformed");
5136
5137        let hash_before = state.hash_state();
5138        let count_before = state.count_tcp_frontends_raw();
5139
5140        let result = state.dispatch(
5141            &RequestType::AddTcpFrontend(RequestTcpFrontend {
5142                cluster_id: "cluster_malformed".to_string(),
5143                address,
5144                tags: BTreeMap::new(),
5145                // Leading empty label -- rejected by `validate_sni_pattern`.
5146                sni: Some(".example.com".to_string()),
5147                alpn: vec![],
5148            })
5149            .into(),
5150        );
5151        assert!(
5152            result.is_err(),
5153            "a malformed SNI pattern must be rejected by the master, not merely by \
5154             config-load, got: {result:?}"
5155        );
5156        assert_eq!(state.hash_state(), hash_before);
5157        assert_eq!(state.count_tcp_frontends_raw(), count_before);
5158        assert_eq!(
5159            state
5160                .tcp_fronts
5161                .get("cluster_malformed")
5162                .map(Vec::len)
5163                .unwrap_or(0),
5164            0
5165        );
5166    }
5167
5168    /// A non-empty `alpn` without `sni` must be rejected by the master:
5169    /// alpn only matches within an SNI-scoped preread, so a no-SNI frontend
5170    /// would silently ignore its alpn list. Config-load (`to_tcp_front`) and
5171    /// the worker both reject this shape; a raw `AddTcpFrontend` over the
5172    /// command socket must not slip past the master.
5173    #[test]
5174    fn add_tcp_frontend_rejects_alpn_without_sni() {
5175        let address = SocketAddress::new_v4(127, 0, 0, 1, 9009);
5176        let mut state = ConfigState::new();
5177        add_tcp_test_cluster(&mut state, "cluster_alpn_no_sni");
5178
5179        let hash_before = state.hash_state();
5180        let count_before = state.count_tcp_frontends_raw();
5181
5182        let result = state.dispatch(
5183            &RequestType::AddTcpFrontend(RequestTcpFrontend {
5184                cluster_id: "cluster_alpn_no_sni".to_string(),
5185                address,
5186                tags: BTreeMap::new(),
5187                sni: None,
5188                alpn: vec!["h2".to_string()],
5189            })
5190            .into(),
5191        );
5192        assert!(
5193            result.is_err(),
5194            "a non-empty alpn without sni must be rejected by the master, got: {result:?}"
5195        );
5196        assert_eq!(
5197            state.hash_state(),
5198            hash_before,
5199            "a rejected alpn-without-sni add must not change the state hash"
5200        );
5201        assert_eq!(
5202            state.count_tcp_frontends_raw(),
5203            count_before,
5204            "a rejected alpn-without-sni add must not change the frontend count"
5205        );
5206        assert_eq!(
5207            state
5208                .tcp_fronts
5209                .get("cluster_alpn_no_sni")
5210                .map(Vec::len)
5211                .unwrap_or(0),
5212            0
5213        );
5214    }
5215
5216    /// Worker-parity guard: an exact catch-all and a wildcard catch-all on
5217    /// the SAME address are DISTINCT (address, sni) keys and must both be
5218    /// accepted -- the listener-wide admission checks above must not
5219    /// over-forbid this legal combination by conflating exact and wildcard
5220    /// SNI as the same key.
5221    #[test]
5222    fn add_tcp_frontend_accepts_exact_and_wildcard_catch_all_siblings() {
5223        let address = SocketAddress::new_v4(127, 0, 0, 1, 9008);
5224        let mut state = ConfigState::new();
5225        add_tcp_test_cluster(&mut state, "cluster_exact_catchall");
5226        add_tcp_test_cluster(&mut state, "cluster_wildcard_catchall");
5227
5228        state
5229            .dispatch(
5230                &RequestType::AddTcpFrontend(RequestTcpFrontend {
5231                    cluster_id: "cluster_exact_catchall".to_string(),
5232                    address,
5233                    tags: BTreeMap::new(),
5234                    sni: Some("example.com".to_string()),
5235                    alpn: vec![],
5236                })
5237                .into(),
5238            )
5239            .expect("could not add exact catch-all tcp frontend");
5240
5241        let result = state.dispatch(
5242            &RequestType::AddTcpFrontend(RequestTcpFrontend {
5243                cluster_id: "cluster_wildcard_catchall".to_string(),
5244                address,
5245                tags: BTreeMap::new(),
5246                sni: Some("*.example.com".to_string()),
5247                alpn: vec![],
5248            })
5249            .into(),
5250        );
5251        assert!(
5252            result.is_ok(),
5253            "an exact catch-all and a wildcard catch-all on the same address must both be \
5254             accepted, got: {result:?}"
5255        );
5256        assert_eq!(state.count_tcp_frontends_raw(), 2);
5257    }
5258
5259    /// `list_frontends` must surface UDP frontends alongside TCP ones. The
5260    /// proto `FrontendFilters` has no `udp` flag, so UDP rides the default
5261    /// (all-pass) and `tcp` filters; a `domain` filter excludes both.
5262    #[test]
5263    fn list_frontends_includes_udp() {
5264        let tcp_addr = SocketAddress::new_v4(0, 0, 0, 0, 6379);
5265        let udp_addr = SocketAddress::new_v4(0, 0, 0, 0, 5353);
5266
5267        let mut state = ConfigState::default();
5268        state
5269            .dispatch(
5270                &RequestType::AddTcpFrontend(RequestTcpFrontend {
5271                    cluster_id: "tcp_cluster".to_string(),
5272                    address: tcp_addr,
5273                    ..Default::default()
5274                })
5275                .into(),
5276            )
5277            .expect("could not add tcp frontend");
5278        state
5279            .dispatch(
5280                &RequestType::AddUdpFrontend(RequestUdpFrontend {
5281                    cluster_id: "udp_cluster".to_string(),
5282                    address: udp_addr,
5283                    ..Default::default()
5284                })
5285                .into(),
5286            )
5287            .expect("could not add udp frontend");
5288
5289        // default filters (all false, no domain) → list everything, incl. UDP
5290        let all = state.list_frontends(FrontendFilters::default());
5291        assert_eq!(all.tcp_frontends.len(), 1, "tcp frontend must be listed");
5292        assert_eq!(
5293            all.udp_frontends.len(),
5294            1,
5295            "udp frontend must be listed under the default all-pass path"
5296        );
5297        assert_eq!(all.udp_frontends[0].cluster_id, "udp_cluster");
5298        assert_eq!(all.udp_frontends[0].address, udp_addr);
5299
5300        // explicit `tcp` filter surfaces both TCP and UDP (no `udp` flag exists)
5301        let tcp_only = state.list_frontends(FrontendFilters {
5302            tcp: true,
5303            ..Default::default()
5304        });
5305        assert_eq!(tcp_only.tcp_frontends.len(), 1);
5306        assert_eq!(
5307            tcp_only.udp_frontends.len(),
5308            1,
5309            "udp frontends ride the tcp filter"
5310        );
5311
5312        // an `http` filter excludes both TCP and UDP
5313        let http_only = state.list_frontends(FrontendFilters {
5314            http: true,
5315            ..Default::default()
5316        });
5317        assert!(http_only.tcp_frontends.is_empty());
5318        assert!(http_only.udp_frontends.is_empty());
5319
5320        // a `domain` filter excludes UDP (no hostname), matching TCP behaviour
5321        let domain_filtered = state.list_frontends(FrontendFilters {
5322            domain: Some("example.com".to_string()),
5323            ..Default::default()
5324        });
5325        assert!(domain_filtered.tcp_frontends.is_empty());
5326        assert!(domain_filtered.udp_frontends.is_empty());
5327    }
5328
5329    // ── update_https_listener ──────────────────────────────────────────────────
5330
5331    /// Happy path: patching two H2 flood knobs updates the map entry; all other
5332    /// fields are left untouched.
5333    #[test]
5334    fn update_https_listener_happy_path_h2_knobs() {
5335        let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
5336        let mut state = ConfigState::new();
5337        state
5338            .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
5339            .unwrap();
5340
5341        let patch = UpdateHttpsListenerConfig {
5342            address: addr,
5343            h2_max_rst_stream_per_window: Some(50),
5344            h2_max_ping_per_window: Some(20),
5345            ..Default::default()
5346        };
5347        state
5348            .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5349            .expect("update must succeed");
5350
5351        let listener = state
5352            .https_listeners
5353            .get(&SocketAddr::from(addr))
5354            .expect("listener must be present");
5355        assert_eq!(listener.h2_max_rst_stream_per_window, Some(50));
5356        assert_eq!(listener.h2_max_ping_per_window, Some(20));
5357        // Untouched fields must be unchanged
5358        assert_eq!(listener.front_timeout, 60);
5359        assert_eq!(listener.h2_max_settings_per_window, None);
5360    }
5361
5362    /// NotFound: patching a listener address that was never registered returns
5363    /// `StateError::NotFound`.
5364    #[test]
5365    fn update_https_listener_not_found() {
5366        let mut state = ConfigState::new();
5367        let patch = UpdateHttpsListenerConfig {
5368            address: SocketAddress::new_v4(1, 2, 3, 4, 9999),
5369            h2_max_rst_stream_per_window: Some(50),
5370            ..Default::default()
5371        };
5372        let err = state
5373            .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5374            .unwrap_err();
5375        assert!(
5376            matches!(
5377                err,
5378                StateError::NotFound {
5379                    kind: ObjectKind::HttpsListener,
5380                    ..
5381                }
5382            ),
5383            "expected NotFound, got: {err}"
5384        );
5385    }
5386
5387    /// No-op: a patch with only `address` set (all options None) is Ok and does
5388    /// not change any field.
5389    #[test]
5390    fn update_https_listener_noop_patch() {
5391        let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
5392        let mut state = ConfigState::new();
5393        let original = make_https_listener(addr);
5394        state
5395            .dispatch(&RequestType::AddHttpsListener(original.clone()).into())
5396            .unwrap();
5397
5398        let patch = UpdateHttpsListenerConfig {
5399            address: addr,
5400            ..Default::default()
5401        };
5402        state
5403            .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5404            .expect("no-op patch must succeed");
5405
5406        let listener = state.https_listeners.get(&SocketAddr::from(addr)).unwrap();
5407        assert_eq!(listener.front_timeout, original.front_timeout);
5408        assert_eq!(
5409            listener.h2_max_rst_stream_per_window,
5410            original.h2_max_rst_stream_per_window
5411        );
5412    }
5413
5414    /// InvalidValue: setting a flood knob to 0 must be rejected.
5415    #[test]
5416    fn update_https_listener_invalid_value_flood_knob_zero() {
5417        let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
5418        let mut state = ConfigState::new();
5419        state
5420            .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
5421            .unwrap();
5422
5423        let patch = UpdateHttpsListenerConfig {
5424            address: addr,
5425            h2_max_rst_stream_per_window: Some(0),
5426            ..Default::default()
5427        };
5428        let err = state
5429            .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5430            .unwrap_err();
5431        assert!(
5432            matches!(
5433                err,
5434                StateError::InvalidValue {
5435                    field: "h2_max_rst_stream_per_window",
5436                    ..
5437                }
5438            ),
5439            "expected InvalidValue for flood knob 0, got: {err}"
5440        );
5441    }
5442
5443    /// AddCluster: an inline `cluster.health_check` with a CRLF-bearing URI
5444    /// must be rejected before the cluster lands in `ConfigState`. Without
5445    /// this guard, TOML reload / SaveState / direct API AddCluster requests
5446    /// bypass the SetHealthCheck-side check and let an attacker-controlled
5447    /// health-check URI smuggle CR/LF into outbound HTTP/1.1 probes.
5448    #[test]
5449    fn add_cluster_invalid_health_check_uri_rejected() {
5450        use crate::proto::command::HealthCheckConfig;
5451
5452        let mut state = ConfigState::new();
5453        let err = state
5454            .dispatch(
5455                &RequestType::AddCluster(Cluster {
5456                    cluster_id: String::from("evil_cluster"),
5457                    health_check: Some(HealthCheckConfig {
5458                        uri: String::from("/foo\r\nGET /admin"),
5459                        interval: 5_000,
5460                        timeout: 1_000,
5461                        healthy_threshold: 2,
5462                        unhealthy_threshold: 2,
5463                        ..Default::default()
5464                    }),
5465                    ..Default::default()
5466                })
5467                .into(),
5468            )
5469            .unwrap_err();
5470
5471        assert!(
5472            matches!(
5473                err,
5474                StateError::InvalidValue {
5475                    field: "health_check",
5476                    ..
5477                }
5478            ),
5479            "expected InvalidValue for CRLF-bearing health-check URI, got: {err}"
5480        );
5481        assert!(
5482            !state.clusters.contains_key("evil_cluster"),
5483            "cluster must not be inserted when health_check fails validation",
5484        );
5485    }
5486
5487    /// ALPN validation: reject unknown ALPN values.
5488    #[test]
5489    fn update_https_listener_alpn_unknown_value_rejected() {
5490        use crate::proto::command::AlpnProtocols;
5491
5492        let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
5493        let mut state = ConfigState::new();
5494        state
5495            .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
5496            .unwrap();
5497
5498        let patch = UpdateHttpsListenerConfig {
5499            address: addr,
5500            alpn_protocols: Some(AlpnProtocols {
5501                values: vec!["h3".to_owned()],
5502            }),
5503            ..Default::default()
5504        };
5505        let err = state
5506            .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5507            .unwrap_err();
5508        assert!(
5509            matches!(
5510                err,
5511                StateError::InvalidValue {
5512                    field: "alpn_protocols",
5513                    ..
5514                }
5515            ),
5516            "expected InvalidValue for unknown ALPN, got: {err}"
5517        );
5518    }
5519
5520    /// ALPN validation: empty values vec = reset to default, must be accepted.
5521    #[test]
5522    fn update_https_listener_alpn_empty_reset_accepted() {
5523        use crate::proto::command::AlpnProtocols;
5524
5525        let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
5526        let mut state = ConfigState::new();
5527        let mut listener = make_https_listener(addr);
5528        listener.alpn_protocols = vec!["h2".to_owned()];
5529        state
5530            .dispatch(&RequestType::AddHttpsListener(listener).into())
5531            .unwrap();
5532
5533        let patch = UpdateHttpsListenerConfig {
5534            address: addr,
5535            alpn_protocols: Some(AlpnProtocols { values: vec![] }),
5536            ..Default::default()
5537        };
5538        state
5539            .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5540            .expect("empty ALPN reset must succeed");
5541
5542        let listener = state.https_listeners.get(&SocketAddr::from(addr)).unwrap();
5543        assert!(
5544            listener.alpn_protocols.is_empty(),
5545            "ALPN must have been reset to empty"
5546        );
5547    }
5548
5549    /// ALPN validation: valid values ["h2", "http/1.1"] must be accepted.
5550    #[test]
5551    fn update_https_listener_alpn_valid_values_accepted() {
5552        use crate::proto::command::AlpnProtocols;
5553
5554        let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
5555        let mut state = ConfigState::new();
5556        state
5557            .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
5558            .unwrap();
5559
5560        let patch = UpdateHttpsListenerConfig {
5561            address: addr,
5562            alpn_protocols: Some(AlpnProtocols {
5563                values: vec!["h2".to_owned(), "http/1.1".to_owned()],
5564            }),
5565            ..Default::default()
5566        };
5567        state
5568            .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5569            .expect("valid ALPN must be accepted");
5570
5571        let listener = state.https_listeners.get(&SocketAddr::from(addr)).unwrap();
5572        assert_eq!(listener.alpn_protocols, vec!["h2", "http/1.1"]);
5573    }
5574
5575    /// ALPN absent wrapper: when `alpn_protocols` is None in the patch, the
5576    /// listener's ALPN field must not be touched.
5577    #[test]
5578    fn update_https_listener_alpn_absent_preserves_existing() {
5579        let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
5580        let mut state = ConfigState::new();
5581        let mut listener = make_https_listener(addr);
5582        listener.alpn_protocols = vec!["h2".to_owned()];
5583        state
5584            .dispatch(&RequestType::AddHttpsListener(listener).into())
5585            .unwrap();
5586
5587        // patch with no alpn_protocols field
5588        let patch = UpdateHttpsListenerConfig {
5589            address: addr,
5590            front_timeout: Some(10),
5591            ..Default::default()
5592        };
5593        state
5594            .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5595            .unwrap();
5596
5597        let listener = state.https_listeners.get(&SocketAddr::from(addr)).unwrap();
5598        assert_eq!(
5599            listener.alpn_protocols,
5600            vec!["h2"],
5601            "ALPN must be preserved when not patched"
5602        );
5603    }
5604
5605    /// sozu_id_header validation: empty string must be rejected.
5606    #[test]
5607    fn update_https_listener_sozu_id_header_empty_rejected() {
5608        let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
5609        let mut state = ConfigState::new();
5610        state
5611            .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
5612            .unwrap();
5613
5614        let patch = UpdateHttpsListenerConfig {
5615            address: addr,
5616            sozu_id_header: Some(String::new()),
5617            ..Default::default()
5618        };
5619        let err = state
5620            .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5621            .unwrap_err();
5622        assert!(
5623            matches!(
5624                err,
5625                StateError::InvalidValue {
5626                    field: "sozu_id_header",
5627                    ..
5628                }
5629            ),
5630            "expected InvalidValue for empty header name, got: {err}"
5631        );
5632    }
5633
5634    /// sozu_id_header validation: value containing colon must be rejected.
5635    #[test]
5636    fn update_https_listener_sozu_id_header_colon_rejected() {
5637        let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
5638        let mut state = ConfigState::new();
5639        state
5640            .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
5641            .unwrap();
5642
5643        let patch = UpdateHttpsListenerConfig {
5644            address: addr,
5645            sozu_id_header: Some("bad: value".to_owned()),
5646            ..Default::default()
5647        };
5648        let err = state
5649            .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5650            .unwrap_err();
5651        assert!(
5652            matches!(
5653                err,
5654                StateError::InvalidValue {
5655                    field: "sozu_id_header",
5656                    ..
5657                }
5658            ),
5659            "expected InvalidValue for header name with colon, got: {err}"
5660        );
5661    }
5662
5663    /// sozu_id_header validation: well-formed token must be accepted.
5664    #[test]
5665    fn update_https_listener_sozu_id_header_valid_accepted() {
5666        let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
5667        let mut state = ConfigState::new();
5668        state
5669            .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
5670            .unwrap();
5671
5672        let patch = UpdateHttpsListenerConfig {
5673            address: addr,
5674            sozu_id_header: Some("X-Edge-Id".to_owned()),
5675            ..Default::default()
5676        };
5677        state
5678            .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5679            .expect("valid header name must be accepted");
5680
5681        let listener = state.https_listeners.get(&SocketAddr::from(addr)).unwrap();
5682        assert_eq!(listener.sozu_id_header.as_deref(), Some("X-Edge-Id"));
5683    }
5684
5685    /// h2_graceful_shutdown_deadline_seconds = 0 must be allowed (means "wait forever").
5686    #[test]
5687    fn update_https_listener_graceful_shutdown_zero_allowed() {
5688        let addr = SocketAddress::new_v4(0, 0, 0, 0, 8443);
5689        let mut state = ConfigState::new();
5690        state
5691            .dispatch(&RequestType::AddHttpsListener(make_https_listener(addr)).into())
5692            .unwrap();
5693
5694        let patch = UpdateHttpsListenerConfig {
5695            address: addr,
5696            h2_graceful_shutdown_deadline_seconds: Some(0),
5697            ..Default::default()
5698        };
5699        state
5700            .dispatch(&RequestType::UpdateHttpsListener(patch).into())
5701            .expect("graceful_shutdown_deadline=0 must be allowed");
5702
5703        let listener = state.https_listeners.get(&SocketAddr::from(addr)).unwrap();
5704        assert_eq!(listener.h2_graceful_shutdown_deadline_seconds, Some(0));
5705    }
5706
5707    // ── update_http_listener ───────────────────────────────────────────────────
5708
5709    /// Happy path for HTTP listener patch.
5710    #[test]
5711    fn update_http_listener_happy_path() {
5712        let addr = SocketAddress::new_v4(0, 0, 0, 0, 8080);
5713        let mut state = ConfigState::new();
5714        state
5715            .dispatch(&RequestType::AddHttpListener(make_http_listener(addr)).into())
5716            .unwrap();
5717
5718        let patch = UpdateHttpListenerConfig {
5719            address: addr,
5720            front_timeout: Some(15),
5721            h2_max_rst_stream_per_window: Some(25),
5722            ..Default::default()
5723        };
5724        state
5725            .dispatch(&RequestType::UpdateHttpListener(patch).into())
5726            .expect("HTTP update must succeed");
5727
5728        let listener = state.http_listeners.get(&SocketAddr::from(addr)).unwrap();
5729        assert_eq!(listener.front_timeout, 15);
5730        assert_eq!(listener.h2_max_rst_stream_per_window, Some(25));
5731        // untouched
5732        assert_eq!(listener.back_timeout, 30);
5733    }
5734
5735    /// HTTP listener: flood knob 0 is rejected.
5736    #[test]
5737    fn update_http_listener_flood_knob_zero_rejected() {
5738        let addr = SocketAddress::new_v4(0, 0, 0, 0, 8080);
5739        let mut state = ConfigState::new();
5740        state
5741            .dispatch(&RequestType::AddHttpListener(make_http_listener(addr)).into())
5742            .unwrap();
5743
5744        let patch = UpdateHttpListenerConfig {
5745            address: addr,
5746            h2_max_window_update_stream0_per_window: Some(0),
5747            ..Default::default()
5748        };
5749        let err = state
5750            .dispatch(&RequestType::UpdateHttpListener(patch).into())
5751            .unwrap_err();
5752        assert!(
5753            matches!(
5754                err,
5755                StateError::InvalidValue {
5756                    field: "h2_max_window_update_stream0_per_window",
5757                    ..
5758                }
5759            ),
5760            "expected InvalidValue, got: {err}"
5761        );
5762    }
5763
5764    // ── update_tcp_listener ────────────────────────────────────────────────────
5765
5766    /// Happy path for TCP listener patch.
5767    #[test]
5768    fn update_tcp_listener_happy_path() {
5769        let addr = SocketAddress::new_v4(0, 0, 0, 0, 9000);
5770        let mut state = ConfigState::new();
5771        state
5772            .dispatch(&RequestType::AddTcpListener(make_tcp_listener(addr)).into())
5773            .unwrap();
5774
5775        let patch = UpdateTcpListenerConfig {
5776            address: addr,
5777            front_timeout: Some(5),
5778            ..Default::default()
5779        };
5780        state
5781            .dispatch(&RequestType::UpdateTcpListener(patch).into())
5782            .expect("TCP update must succeed");
5783
5784        let listener = state.tcp_listeners.get(&SocketAddr::from(addr)).unwrap();
5785        assert_eq!(listener.front_timeout, 5);
5786        assert_eq!(listener.back_timeout, 30); // untouched
5787    }
5788
5789    /// TCP listener: NotFound when address is unknown.
5790    #[test]
5791    fn update_tcp_listener_not_found() {
5792        let mut state = ConfigState::new();
5793        let patch = UpdateTcpListenerConfig {
5794            address: SocketAddress::new_v4(9, 9, 9, 9, 9999),
5795            front_timeout: Some(5),
5796            ..Default::default()
5797        };
5798        let err = state
5799            .dispatch(&RequestType::UpdateTcpListener(patch).into())
5800            .unwrap_err();
5801        assert!(
5802            matches!(
5803                err,
5804                StateError::NotFound {
5805                    kind: ObjectKind::TcpListener,
5806                    ..
5807                }
5808            ),
5809            "expected NotFound, got: {err}"
5810        );
5811    }
5812
5813    /// `ConfigState::dispatch` MUST treat `SetMetricDetail` as a
5814    /// runtime-only verb (no persisted state mutation). A future
5815    /// refactor that drops the variant from the no-op match arm and
5816    /// falls through to the catch-all would silently re-break the
5817    /// SetMetricDetail dispatch path with `UndispatchableRequest`.
5818    #[test]
5819    fn dispatch_passes_through_set_metric_detail() {
5820        use crate::proto::command::{MetricDetail, SetMetricDetail};
5821        let mut state = ConfigState::new();
5822        let req: Request = RequestType::SetMetricDetail(SetMetricDetail {
5823            client_id: "test:1".to_owned(),
5824            detail: Some(MetricDetail::DetailBackend as i32),
5825            ttl_seconds: Some(60),
5826            clear: Some(false),
5827            reason: Some("regression-guard".to_owned()),
5828            peer_pid: None,
5829            peer_session_ulid: None,
5830        })
5831        .into();
5832        state
5833            .dispatch(&req)
5834            .expect("SetMetricDetail must traverse dispatch without UndispatchableRequest");
5835    }
5836}