1use std::{
3 cell::RefCell,
4 collections::{HashMap, HashSet, VecDeque, hash_map::Entry},
5 hash::{DefaultHasher, Hash, Hasher},
6 io::Error as IoError,
7 net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr},
8 os::unix::io::{AsRawFd, FromRawFd, RawFd},
9 rc::{Rc, Weak},
10 str::FromStr,
11 sync::LazyLock,
12 time::{Duration, Instant},
13};
14
15use mio::{
16 Events, Interest, Poll, Token,
17 net::{TcpListener as MioTcpListener, TcpStream, UdpSocket as MioUdpSocket},
18};
19use slab::Slab;
20use sozu_command::{
21 channel::{Channel, ChannelError},
22 config::MetricDetailLevel,
23 listener_key::ListenerKey,
24 logging,
25 proto::command::{
26 ActivateListener, AddBackend, CertificatesWithFingerprints, ClosedSessions, Cluster,
27 ClusterHashes, ClusterInformations, DeactivateListener, Event, EventKind,
28 HttpListenerConfig, HttpsListenerConfig, InitialState, ListenerType,
29 LoadBalancingAlgorithms, LoadMetric, MetricDetail, MetricsConfiguration,
30 ProtectedHealthCheckPhase, RemoveBackend, Request, ResponseContent, ResponseStatus,
31 ServerConfig, TcpListenerConfig as CommandTcpListener,
32 UdpListenerConfig as CommandUdpListener, UpdateHttpListenerConfig,
33 UpdateHttpsListenerConfig, UpdateTcpListenerConfig, UpdateUdpListenerConfig, WorkerRequest,
34 WorkerResponse, request::RequestType, response_content::ContentType,
35 },
36 ready::Ready,
37 scm_socket::{Listeners, ScmSocket, ScmSocketError},
38 state::{ClusterId, ConfigState, StateError},
39};
40
41use crate::metrics::names;
42use crate::{
43 AcceptError, Protocol, ProxyConfiguration, ProxySession, SessionIsToBeClosed, SessionTarget,
44 backends::{Backend, BackendMap},
45 features::FEATURES,
46 health_check::HealthChecker,
47 http, https,
48 load_balancing::ShuffleSharding,
49 metrics::METRICS,
50 pool::Pool,
51 protected_health_check::ProtectedWorker,
52 tcp,
53 timer::Timer,
54 udp,
55};
56
57pub const PER_SOURCE_BUCKETS: usize = 256;
71
72static PER_SOURCE_BUCKET_KEYS: LazyLock<[&'static str; PER_SOURCE_BUCKETS]> = LazyLock::new(|| {
77 let mut keys: [&'static str; PER_SOURCE_BUCKETS] = [""; PER_SOURCE_BUCKETS];
78 for (i, slot) in keys.iter_mut().enumerate() {
79 let owned = format!("client.connect.per_source.bucket_{i:03}");
81 *slot = Box::leak(owned.into_boxed_str());
82 }
83 keys
84});
85
86fn per_source_bucket(peer: &SocketAddr) -> &'static str {
91 let mut hasher = DefaultHasher::new();
92 match peer.ip() {
93 IpAddr::V4(v4) => {
94 let octets = v4.octets();
95 let masked = Ipv4Addr::new(octets[0], octets[1], octets[2], 0);
97 masked.hash(&mut hasher);
98 }
99 IpAddr::V6(v6) => {
100 let octets = v6.octets();
101 let mut masked_octets = [0u8; 16];
103 masked_octets[..6].copy_from_slice(&octets[..6]);
104 Ipv6Addr::from(masked_octets).hash(&mut hasher);
105 }
106 }
107 let idx = (hasher.finish() as usize) % PER_SOURCE_BUCKETS;
108 PER_SOURCE_BUCKET_KEYS[idx]
109}
110
111const ACCEPT_SATURATION_TICK: Duration = Duration::from_secs(1);
116
117pub type ProxyChannel = Channel<WorkerResponse, WorkerRequest>;
118
119thread_local! {
120 pub static QUEUE: RefCell<VecDeque<WorkerResponse>> = const { RefCell::new(VecDeque::new()) };
121}
122
123thread_local! {
124 pub static TIMER: RefCell<Timer<Token>> = RefCell::new(Timer::default());
125}
126
127pub fn push_queue(message: WorkerResponse) {
128 QUEUE.with(|queue| {
129 (*queue.borrow_mut()).push_back(message);
130 });
131}
132
133pub fn push_event(event: Event) {
134 QUEUE.with(|queue| {
135 (*queue.borrow_mut()).push_back(WorkerResponse {
136 id: "EVENT".to_string(),
137 message: String::new(),
138 status: ResponseStatus::Processing.into(),
139 content: Some(ContentType::Event(event).into()),
140 });
141 });
142}
143
144fn is_frontend_removal(request: &Request) -> bool {
146 matches!(
147 request.request_type,
148 Some(
149 RequestType::RemoveHttpFrontend(_)
150 | RequestType::RemoveHttpFrontendWithHeaderPredicates(_)
151 | RequestType::RemoveHttpsFrontend(_)
152 | RequestType::RemoveHttpsFrontendWithHeaderPredicates(_)
153 | RequestType::RemoveTcpFrontend(_)
154 | RequestType::RemoveUdpFrontend(_)
155 )
156 )
157}
158
159fn worker_metric_detail_status_content(
165 configured: MetricDetailLevel,
166 effective: MetricDetailLevel,
167 previous_effective: MetricDetailLevel,
168 active_lease_count: u32,
169) -> ResponseContent {
170 use sozu_command::proto::command::WorkerMetricDetailStatus;
171 ContentType::WorkerMetricDetailStatus(WorkerMetricDetailStatus {
172 configured: MetricDetail::from(configured) as i32,
173 effective: MetricDetail::from(effective) as i32,
174 previous_effective: MetricDetail::from(previous_effective) as i32,
175 active_lease_count,
176 })
177 .into()
178}
179
180fn push_metric_detail_transition(
187 previous: MetricDetailLevel,
188 effective: MetricDetailLevel,
189 transition_kind: &'static str,
190 client_id: Option<String>,
191) {
192 use sozu_command::proto::command::MetricDetailTransition;
193 if previous == effective {
198 return;
199 }
200 push_event(Event {
201 kind: EventKind::MetricDetailChanged as i32,
202 cluster_id: None,
203 backend_id: None,
204 address: None,
205 metric_detail: Some(MetricDetailTransition {
206 previous_effective: MetricDetail::from(previous) as i32,
207 effective: MetricDetail::from(effective) as i32,
208 transition_kind: transition_kind.to_owned(),
209 client_id,
210 }),
211 });
212}
213
214fn tick_metric_detail_leases(now: Instant) {
222 let lease_tick_transition = METRICS.with(|metrics| {
229 let mut m = metrics.borrow_mut();
230 if !m.lease_tick_due(now) {
231 return None;
232 }
233 let previous = m.lease_tick(now)?;
234 let effective = m.detail_effective();
235 Some((previous, effective))
236 });
237 if let Some((previous, effective)) = lease_tick_transition {
238 push_metric_detail_transition(previous, effective, "lease_tick_expired", None);
243 }
244}
245
246#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
247pub struct ListenToken(pub usize);
248#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
249pub struct SessionToken(pub usize);
250
251impl From<usize> for ListenToken {
252 fn from(val: usize) -> ListenToken {
253 ListenToken(val)
254 }
255}
256
257impl From<ListenToken> for usize {
258 fn from(val: ListenToken) -> usize {
259 val.0
260 }
261}
262
263impl From<usize> for SessionToken {
264 fn from(val: usize) -> SessionToken {
265 SessionToken(val)
266 }
267}
268
269impl From<SessionToken> for usize {
270 fn from(val: SessionToken) -> usize {
271 val.0
272 }
273}
274
275fn mask_ipv4(address: Ipv4Addr, prefix: u32) -> Ipv4Addr {
289 if prefix >= u32::BITS {
290 return address;
291 }
292 if prefix == 0 {
293 return Ipv4Addr::UNSPECIFIED;
294 }
295 debug_assert!(
298 prefix > 0 && prefix < u32::BITS,
299 "both shift-overflowing ends must have returned before the shift"
300 );
301 let mask = u32::MAX << (u32::BITS - prefix);
302 let masked = Ipv4Addr::from(u32::from(address) & mask);
303 debug_assert_eq!(
307 u32::from(masked),
308 u32::from(address) & mask,
309 "masking must keep exactly the network bits of the source address"
310 );
311 debug_assert_eq!(
312 u32::from(masked) & !mask,
313 0,
314 "masking must clear every host bit below the prefix"
315 );
316 masked
317}
318
319fn mask_ipv6(address: Ipv6Addr, prefix: u32) -> Ipv6Addr {
323 if prefix >= u128::BITS {
324 return address;
325 }
326 if prefix == 0 {
327 return Ipv6Addr::UNSPECIFIED;
328 }
329 debug_assert!(
330 prefix > 0 && prefix < u128::BITS,
331 "both shift-overflowing ends must have returned before the shift"
332 );
333 let mask = u128::MAX << (u128::BITS - prefix);
334 let masked = Ipv6Addr::from(u128::from(address) & mask);
335 debug_assert_eq!(
336 u128::from(masked),
337 u128::from(address) & mask,
338 "masking must keep exactly the network bits of the source address"
339 );
340 debug_assert_eq!(
341 u128::from(masked) & !mask,
342 0,
343 "masking must clear every host bit below the prefix"
344 );
345 masked
346}
347
348pub fn subnet_key(address: &IpAddr, ipv4_prefix: u32, ipv6_prefix: u32) -> IpAddr {
359 let canonical_v4 = match address {
360 IpAddr::V4(v4) => Some(*v4),
361 IpAddr::V6(v6) => v6.to_ipv4_mapped(),
362 };
363 let key = match (canonical_v4, address) {
364 (Some(v4), _) => IpAddr::V4(mask_ipv4(v4, ipv4_prefix)),
365 (None, IpAddr::V6(v6)) => IpAddr::V6(mask_ipv6(*v6, ipv6_prefix)),
366 (None, IpAddr::V4(v4)) => IpAddr::V4(mask_ipv4(*v4, ipv4_prefix)),
368 };
369 debug_assert_eq!(
373 key.is_ipv4(),
374 canonical_v4.is_some(),
375 "subnet key family must follow the canonicalised source family"
376 );
377 debug_assert_eq!(
378 key.is_ipv6(),
379 canonical_v4.is_none(),
380 "a genuine IPv6 source must never produce an IPv4 subnet key"
381 );
382 key
383}
384
385pub struct SessionManager {
386 pub max_connections: usize,
387 pub nb_connections: usize,
388 pub can_accept: bool,
389 pub slab: Slab<Rc<RefCell<dyn ProxySession>>>,
390 pub max_connections_per_ip: u64,
394 pub retry_after: u32,
398 connections_per_cluster_ip: HashMap<ClusterId, HashMap<IpAddr, usize>>,
422 cluster_ip_tracks: HashMap<Token, HashMap<ClusterId, HashSet<IpAddr>>>,
429 pub max_connections_per_subnet: u64,
443 pub subnet_ipv4_prefix: u32,
447 pub subnet_ipv6_prefix: u32,
450 connections_per_cluster_subnet: HashMap<ClusterId, HashMap<IpAddr, usize>>,
463 cluster_subnet_tracks: HashMap<Token, HashMap<ClusterId, HashSet<IpAddr>>>,
466 pub max_connection_attempts: u32,
473}
474
475impl SessionManager {
476 pub fn new(
477 slab: Slab<Rc<RefCell<dyn ProxySession>>>,
478 max_connections: usize,
479 max_connections_per_ip: u64,
480 retry_after: u32,
481 max_connections_per_subnet: u64,
482 subnet_ipv4_prefix: u32,
483 subnet_ipv6_prefix: u32,
484 ) -> Rc<RefCell<Self>> {
485 Rc::new(RefCell::new(SessionManager {
486 max_connections,
487 nb_connections: 0,
488 can_accept: true,
489 slab,
490 max_connections_per_ip,
491 retry_after,
492 connections_per_cluster_ip: HashMap::new(),
493 cluster_ip_tracks: HashMap::new(),
494 max_connections_per_subnet,
495 subnet_ipv4_prefix,
496 subnet_ipv6_prefix,
497 connections_per_cluster_subnet: HashMap::new(),
498 cluster_subnet_tracks: HashMap::new(),
499 max_connection_attempts: sozu_command::config::DEFAULT_MAX_CONNECTION_ATTEMPTS,
500 }))
501 }
502
503 pub fn effective_max_connection_attempts(&self, override_value: Option<u32>) -> u32 {
507 override_value.unwrap_or(self.max_connection_attempts)
508 }
509
510 pub fn effective_max_connections_per_ip(&self, override_value: Option<u64>) -> u64 {
515 override_value.unwrap_or(self.max_connections_per_ip)
516 }
517
518 pub fn effective_retry_after(&self, override_value: Option<u32>) -> u32 {
522 override_value.unwrap_or(self.retry_after)
523 }
524
525 pub fn effective_max_connections_per_subnet(&self, override_value: Option<u64>) -> u64 {
530 override_value.unwrap_or(self.max_connections_per_subnet)
531 }
532
533 pub fn subnet_key(&self, ip: &IpAddr) -> IpAddr {
536 subnet_key(ip, self.subnet_ipv4_prefix, self.subnet_ipv6_prefix)
537 }
538
539 pub fn cluster_subnet_at_limit(
549 &self,
550 token: Token,
551 cluster_id: &str,
552 ip: &IpAddr,
553 override_value: Option<u64>,
554 ) -> bool {
555 let limit = self.effective_max_connections_per_subnet(override_value);
556 if limit == 0 {
557 return false;
558 }
559 debug_assert!(
560 limit > 0,
561 "limit==0 (unlimited) must have returned before reaching the bounded check"
562 );
563 let subnet = self.subnet_key(ip);
564 let already_tracked = self
565 .cluster_subnet_tracks
566 .get(&token)
567 .and_then(|by_cluster| by_cluster.get(cluster_id))
568 .is_some_and(|subnets| subnets.contains(&subnet));
569 if already_tracked {
570 debug_assert!(
571 self.connections_per_cluster_subnet
572 .get(cluster_id)
573 .and_then(|by_subnet| by_subnet.get(&subnet))
574 .is_some_and(|c| *c > 0),
575 "a tracked (token, cluster, subnet) slot must have a positive forward count"
576 );
577 return false;
578 }
579 self.connections_per_cluster_subnet
580 .get(cluster_id)
581 .and_then(|by_subnet| by_subnet.get(&subnet))
582 .is_some_and(|c| (*c as u64) >= limit)
583 }
584
585 pub fn cluster_connection_at_limit(
596 &self,
597 token: Token,
598 cluster_id: &str,
599 ip: &IpAddr,
600 ip_override: Option<u64>,
601 subnet_override: Option<u64>,
602 ) -> bool {
603 self.cluster_ip_at_limit(token, cluster_id, ip, ip_override)
604 || self.cluster_subnet_at_limit(token, cluster_id, ip, subnet_override)
605 }
606
607 pub fn cluster_ip_at_limit(
619 &self,
620 token: Token,
621 cluster_id: &str,
622 ip: &IpAddr,
623 override_value: Option<u64>,
624 ) -> bool {
625 let limit = self.effective_max_connections_per_ip(override_value);
626 if limit == 0 {
627 return false;
628 }
629 debug_assert!(
632 limit > 0,
633 "limit==0 (unlimited) must have returned before reaching the bounded check"
634 );
635 let already_tracked = self
636 .cluster_ip_tracks
637 .get(&token)
638 .and_then(|by_cluster| by_cluster.get(cluster_id))
639 .is_some_and(|ips| ips.contains(ip));
640 if already_tracked {
641 debug_assert!(
645 self.connections_per_cluster_ip
646 .get(cluster_id)
647 .and_then(|by_ip| by_ip.get(ip))
648 .is_some_and(|c| *c > 0),
649 "a tracked (token, cluster, ip) slot must have a positive forward count"
650 );
651 return false;
652 }
653 self.connections_per_cluster_ip
654 .get(cluster_id)
655 .and_then(|by_ip| by_ip.get(ip))
656 .is_some_and(|c| (*c as u64) >= limit)
657 }
658
659 pub fn track_cluster_ip(&mut self, token: Token, cluster_id: ClusterId, ip: IpAddr) {
739 let count_before = self
743 .connections_per_cluster_ip
744 .get(&cluster_id)
745 .and_then(|by_ip| by_ip.get(&ip))
746 .copied()
747 .unwrap_or(0);
748 let inserted = self
749 .cluster_ip_tracks
750 .entry(token)
751 .or_default()
752 .entry(cluster_id.clone())
753 .or_default()
754 .insert(ip);
755 if inserted {
756 *self
757 .connections_per_cluster_ip
758 .entry(cluster_id.clone())
759 .or_default()
760 .entry(ip)
761 .or_insert(0) += 1;
762 }
763 debug_assert!(
767 self.cluster_ip_tracks
768 .get(&token)
769 .and_then(|by_cluster| by_cluster.get(&cluster_id))
770 .is_some_and(|ips| ips.contains(&ip)),
771 "track must leave the (token, cluster, ip) recorded in the reverse index"
772 );
773 debug_assert_eq!(
774 self.connections_per_cluster_ip
775 .get(&cluster_id)
776 .and_then(|by_ip| by_ip.get(&ip))
777 .copied()
778 .unwrap_or(0),
779 count_before + inserted as usize,
780 "forward count must advance by exactly 1 on first track, 0 on a repeat"
781 );
782 #[cfg(debug_assertions)]
783 self.check_invariants();
784 }
785
786 pub fn track_cluster_subnet(
800 &mut self,
801 token: Token,
802 cluster_id: ClusterId,
803 ip: IpAddr,
804 override_value: Option<u64>,
805 ) {
806 if self.effective_max_connections_per_subnet(override_value) == 0 {
807 return;
815 }
816 let subnet = self.subnet_key(&ip);
817 let count_before = self
821 .connections_per_cluster_subnet
822 .get(&cluster_id)
823 .and_then(|by_subnet| by_subnet.get(&subnet))
824 .copied()
825 .unwrap_or(0);
826 let inserted = self
827 .cluster_subnet_tracks
828 .entry(token)
829 .or_default()
830 .entry(cluster_id.clone())
831 .or_default()
832 .insert(subnet);
833 if inserted {
834 *self
835 .connections_per_cluster_subnet
836 .entry(cluster_id.clone())
837 .or_default()
838 .entry(subnet)
839 .or_insert(0) += 1;
840 }
841 debug_assert!(
842 self.cluster_subnet_tracks
843 .get(&token)
844 .and_then(|by_cluster| by_cluster.get(&cluster_id))
845 .is_some_and(|subnets| subnets.contains(&subnet)),
846 "track must leave the (token, cluster, subnet) recorded in the reverse index"
847 );
848 debug_assert_eq!(
849 self.connections_per_cluster_subnet
850 .get(&cluster_id)
851 .and_then(|by_subnet| by_subnet.get(&subnet))
852 .copied()
853 .unwrap_or(0),
854 count_before + inserted as usize,
855 "forward count must advance by exactly 1 on first track, 0 on a repeat"
856 );
857 #[cfg(debug_assertions)]
858 self.check_invariants();
859 }
860
861 pub fn track_cluster_connection(
867 &mut self,
868 token: Token,
869 cluster_id: ClusterId,
870 ip: IpAddr,
871 subnet_override: Option<u64>,
872 ) {
873 self.track_cluster_ip(token, cluster_id.clone(), ip);
874 self.track_cluster_subnet(token, cluster_id, ip, subnet_override);
875 }
876
877 pub fn untrack_all_cluster_ip(&mut self, token: Token) {
891 self.untrack_all_cluster_subnet(token);
892 let Some(by_cluster) = self.cluster_ip_tracks.remove(&token) else {
893 return;
894 };
895 debug_assert!(
898 !self.cluster_ip_tracks.contains_key(&token),
899 "untrack_all must evict the token from the reverse index"
900 );
901 for (cluster_id, ips) in by_cluster {
902 let Entry::Occupied(mut outer) = self.connections_per_cluster_ip.entry(cluster_id)
903 else {
904 continue;
905 };
906 for ip in ips {
907 if let Entry::Occupied(mut inner) = outer.get_mut().entry(ip) {
908 let count = inner.get_mut();
909 *count = count.saturating_sub(1);
910 if *count == 0 {
911 inner.remove();
912 }
913 }
914 }
915 if outer.get().is_empty() {
916 outer.remove();
917 }
918 }
919 debug_assert!(
922 self.connections_per_cluster_ip
923 .values()
924 .all(|by_ip| !by_ip.is_empty() && by_ip.values().all(|&c| c > 0)),
925 "untrack_all must not leave empty inner maps or zero-count ips behind"
926 );
927 #[cfg(debug_assertions)]
928 self.check_invariants();
929 }
930
931 fn untrack_all_cluster_subnet(&mut self, token: Token) {
936 let Some(by_cluster) = self.cluster_subnet_tracks.remove(&token) else {
937 return;
938 };
939 debug_assert!(
940 !self.cluster_subnet_tracks.contains_key(&token),
941 "untrack_all must evict the token from the subnet reverse index"
942 );
943 for (cluster_id, subnets) in by_cluster {
944 let Entry::Occupied(mut outer) = self.connections_per_cluster_subnet.entry(cluster_id)
945 else {
946 continue;
947 };
948 for subnet in subnets {
949 if let Entry::Occupied(mut inner) = outer.get_mut().entry(subnet) {
950 let count = inner.get_mut();
951 *count = count.saturating_sub(1);
952 if *count == 0 {
953 inner.remove();
954 }
955 }
956 }
957 if outer.get().is_empty() {
958 outer.remove();
959 }
960 }
961 debug_assert!(
962 self.connections_per_cluster_subnet
963 .values()
964 .all(|by_subnet| !by_subnet.is_empty() && by_subnet.values().all(|&c| c > 0)),
965 "untrack_all must not leave empty inner maps or zero-count subnets behind"
966 );
967 }
968
969 pub fn subnet_tracking_is_empty(&self) -> bool {
975 self.cluster_subnet_tracks.is_empty() && self.connections_per_cluster_subnet.is_empty()
976 }
977
978 pub fn clear_cluster_subnet_tracking(&mut self) {
984 self.cluster_subnet_tracks.clear();
985 self.connections_per_cluster_subnet.clear();
986 debug_assert!(
987 self.cluster_subnet_tracks.is_empty() && self.connections_per_cluster_subnet.is_empty(),
988 "clear must wipe both the subnet reverse index and the subnet forward count map"
989 );
990 #[cfg(debug_assertions)]
991 self.check_invariants();
992 }
993
994 pub fn clear_cluster_ip_tracking(&mut self) {
999 self.cluster_ip_tracks.clear();
1000 self.connections_per_cluster_ip.clear();
1001 debug_assert!(
1004 self.cluster_ip_tracks.is_empty() && self.connections_per_cluster_ip.is_empty(),
1005 "clear must wipe both the reverse index and the forward count map"
1006 );
1007 #[cfg(debug_assertions)]
1008 self.check_invariants();
1009 }
1010
1011 pub fn at_capacity(&self) -> bool {
1013 self.slab.len() >= self.accept_slab_threshold()
1014 }
1015
1016 pub fn accept_slab_threshold(&self) -> usize {
1028 let threshold = 10 + 2 * self.max_connections;
1029 debug_assert!(
1033 threshold > self.max_connections,
1034 "accept gate must sit strictly above max_connections to reserve system slots"
1035 );
1036 threshold
1037 }
1038
1039 pub fn check_limits(&mut self) -> bool {
1042 debug_assert!(
1045 self.nb_connections <= self.max_connections,
1046 "nb_connections must never exceed max_connections"
1047 );
1048 if self.nb_connections >= self.max_connections {
1049 error!("max number of session connection reached, flushing the accept queue");
1050 gauge!(names::accept_queue::BACKPRESSURE, 1);
1051 self.can_accept = false;
1052 debug_assert!(
1054 !self.can_accept,
1055 "refusing at the cap must clear can_accept"
1056 );
1057 return false;
1058 }
1059
1060 if self.at_capacity() {
1061 error!("not enough memory to accept another session, flushing the accept queue");
1062 error!(
1063 "nb_connections: {}, max_connections: {}",
1064 self.nb_connections, self.max_connections
1065 );
1066 gauge!(names::accept_queue::BACKPRESSURE, 1);
1067 self.can_accept = false;
1068
1069 debug_assert!(
1070 !self.can_accept,
1071 "refusing at slab capacity must clear can_accept"
1072 );
1073 return false;
1074 }
1075
1076 debug_assert!(
1078 self.nb_connections < self.max_connections && !self.at_capacity(),
1079 "check_limits returned room while a gate was actually saturated"
1080 );
1081 true
1082 }
1083
1084 pub fn to_session(token: Token) -> SessionToken {
1085 SessionToken(token.0)
1086 }
1087
1088 pub fn incr(&mut self) {
1089 let before = self.nb_connections;
1090 self.nb_connections += 1;
1091 assert!(self.nb_connections <= self.max_connections);
1092 debug_assert_eq!(
1094 self.nb_connections,
1095 before + 1,
1096 "incr must raise nb_connections by exactly one"
1097 );
1098 gauge!(names::client::CONNECTIONS, self.nb_connections);
1104 }
1105
1106 pub fn decr(&mut self) {
1109 assert!(self.nb_connections != 0);
1110 let before = self.nb_connections;
1111 self.nb_connections -= 1;
1112 debug_assert_eq!(
1114 self.nb_connections,
1115 before - 1,
1116 "decr must lower nb_connections by exactly one"
1117 );
1118 gauge!(names::client::CONNECTIONS, self.nb_connections);
1119
1120 if !self.can_accept && self.nb_connections < self.max_connections * 90 / 100 {
1122 debug!(
1123 "nb_connections = {}, max_connections = {}, starting to accept again",
1124 self.nb_connections, self.max_connections
1125 );
1126 gauge!(names::accept_queue::BACKPRESSURE, 0);
1127 self.can_accept = true;
1128 }
1129 }
1130
1131 #[cfg(debug_assertions)]
1136 fn check_invariants(&self) {
1137 debug_assert!(
1139 self.nb_connections <= self.max_connections,
1140 "nb_connections {} exceeds max_connections {}",
1141 self.nb_connections,
1142 self.max_connections
1143 );
1144 debug_assert!(
1147 self.connections_per_cluster_ip
1148 .values()
1149 .all(|by_ip| !by_ip.is_empty() && by_ip.values().all(|&c| c > 0)),
1150 "connections_per_cluster_ip holds an empty inner map or a zero count"
1151 );
1152 debug_assert!(
1157 self.cluster_ip_tracks.values().all(|by_cluster| {
1158 by_cluster.iter().all(|(cluster_id, ips)| {
1159 ips.iter().all(|ip| {
1160 self.connections_per_cluster_ip
1161 .get(cluster_id)
1162 .and_then(|by_ip| by_ip.get(ip))
1163 .is_some_and(|&c| c > 0)
1164 })
1165 })
1166 }),
1167 "a tracked (token, cluster, ip) slot has no positive forward count"
1168 );
1169 debug_assert!(
1171 self.cluster_ip_tracks.values().all(|by_cluster| {
1172 !by_cluster.is_empty() && by_cluster.values().all(|ips| !ips.is_empty())
1173 }),
1174 "cluster_ip_tracks retains an empty per-token or per-cluster entry"
1175 );
1176 debug_assert!(
1178 self.connections_per_cluster_subnet
1179 .values()
1180 .all(|by_subnet| !by_subnet.is_empty() && by_subnet.values().all(|&c| c > 0)),
1181 "connections_per_cluster_subnet holds an empty inner map or a zero count"
1182 );
1183 debug_assert!(
1184 self.cluster_subnet_tracks.values().all(|by_cluster| {
1185 by_cluster.iter().all(|(cluster_id, subnets)| {
1186 subnets.iter().all(|subnet| {
1187 self.connections_per_cluster_subnet
1188 .get(cluster_id)
1189 .and_then(|by_subnet| by_subnet.get(subnet))
1190 .is_some_and(|&c| c > 0)
1191 })
1192 })
1193 }),
1194 "a tracked (token, cluster, subnet) slot has no positive forward count"
1195 );
1196 debug_assert!(
1197 self.cluster_subnet_tracks.values().all(|by_cluster| {
1198 !by_cluster.is_empty() && by_cluster.values().all(|subnets| !subnets.is_empty())
1199 }),
1200 "cluster_subnet_tracks retains an empty per-token or per-cluster entry"
1201 );
1202 }
1203}
1204
1205#[derive(thiserror::Error, Debug)]
1206pub enum ServerError {
1207 #[error("could not create event loop with MIO poll: {0}")]
1208 CreatePoll(IoError),
1209 #[error("could not clone the MIO registry: {0}")]
1210 CloneRegistry(IoError),
1211 #[error("could not register the channel: {0}")]
1212 RegisterChannel(IoError),
1213 #[error("{msg}:{scm_err}")]
1214 ScmSocket {
1215 msg: String,
1216 scm_err: ScmSocketError,
1217 },
1218 #[error("could not read the initial message of the main process: {0}")]
1219 ReadInitialMessage(ChannelError),
1220 #[error("could not answer the initial message of the main process: {0}")]
1221 WriteInitialAnswer(ChannelError),
1222}
1223
1224#[derive(Debug, Clone, PartialEq, Eq)]
1226enum BootstrapStage {
1227 Start,
1229 Ready,
1231 Prepared { cluster: String, version: u64 },
1233 Committed { cluster: String, version: u64 },
1235}
1236
1237struct BootstrapStep {
1240 phase: ProtectedHealthCheckPhase,
1241 cluster: String,
1242 version: u64,
1243 next: BootstrapStage,
1244}
1245
1246impl BootstrapStage {
1247 fn step(&self, request: &RequestType) -> Option<BootstrapStep> {
1250 use ProtectedHealthCheckPhase as Phase;
1251 match request {
1252 RequestType::QueryProtectedHealthCheckSupport(_) if *self == Self::Start => {
1253 Some(BootstrapStep {
1254 phase: Phase::ProtectedHealthCheckCapability,
1255 cluster: String::new(),
1256 version: 0,
1257 next: Self::Ready,
1258 })
1259 }
1260 RequestType::ProtectedHealthCheckOrder(order) => {
1261 let phase = Phase::try_from(order.phase).ok()?;
1262 let (cluster, version) = (order.cluster_id.to_owned(), order.version);
1263 let outstanding = match self {
1264 Self::Prepared {
1265 cluster: held,
1266 version: staged,
1267 }
1268 | Self::Committed {
1269 cluster: held,
1270 version: staged,
1271 } => *held == cluster && *staged == version,
1272 _ => false,
1273 };
1274 let next = match (self, phase) {
1275 (Self::Ready, Phase::ProtectedHealthCheckPrepare) => Self::Prepared {
1276 cluster: cluster.to_owned(),
1277 version,
1278 },
1279 (Self::Prepared { .. }, Phase::ProtectedHealthCheckCommit) if outstanding => {
1280 Self::Committed {
1281 cluster: cluster.to_owned(),
1282 version,
1283 }
1284 }
1285 (Self::Committed { .. }, Phase::ProtectedHealthCheckRelease) if outstanding => {
1286 Self::Ready
1287 }
1288 (
1289 Self::Prepared { .. } | Self::Committed { .. },
1290 Phase::ProtectedHealthCheckAbort,
1291 ) if outstanding => Self::Ready,
1292 _ => return None,
1293 };
1294 Some(BootstrapStep {
1295 phase,
1296 cluster,
1297 version,
1298 next,
1299 })
1300 }
1301 _ => None,
1302 }
1303 }
1304
1305 fn accepts_status(&self) -> bool {
1308 matches!(self, Self::Start | Self::Ready)
1309 }
1310}
1311
1312impl BootstrapStep {
1313 fn is_answered_by(&self, response: &WorkerResponse) -> bool {
1315 response.status == ResponseStatus::Ok as i32
1316 && matches!(
1317 response
1318 .content
1319 .as_ref()
1320 .and_then(|content| content.content_type.as_ref()),
1321 Some(ContentType::ProtectedHealthCheckReply(reply))
1322 if reply.phase == self.phase as i32
1323 && reply.cluster_id == self.cluster
1324 && reply.version == self.version
1325 )
1326 }
1327}
1328
1329pub struct Server {
1346 accept_queue_timeout: Duration,
1347 accept_queue: VecDeque<(TcpStream, ListenToken, Protocol, Instant, SocketAddr)>,
1354 evict_on_queue_full: bool,
1358 accept_ready: HashSet<ListenToken>,
1359 backends: Rc<RefCell<BackendMap>>,
1360 base_sessions_count: usize,
1361 channel: ProxyChannel,
1362 config_state: ConfigState,
1363 current_poll_errors: i32,
1364 health_checker: HealthChecker,
1365 http: Rc<RefCell<http::HttpProxy>>,
1366 https: Rc<RefCell<https::HttpsProxy>>,
1367 last_sessions_len: usize,
1368 last_shutting_down_message: Option<Instant>,
1369 last_zombie_check: Instant,
1370 loop_start: Instant,
1371 started_at: Instant,
1375 last_saturation_tick: Instant,
1380 max_poll_errors: i32, pool: Rc<RefCell<Pool>>,
1387 pub poll: Poll,
1388 protected_health_check: ProtectedWorker,
1390 poll_timeout: Option<Duration>, scm_listeners: Option<Listeners>,
1392 scm: ScmSocket,
1393 sessions: Rc<RefCell<SessionManager>>,
1394 should_poll_at: Option<Instant>,
1395 shutting_down: Option<String>,
1396 closing_sessions: Vec<(Token, Weak<RefCell<dyn ProxySession>>)>,
1404 yielded_sessions: Vec<(Token, Weak<RefCell<dyn ProxySession>>)>,
1410 tcp: Rc<RefCell<tcp::TcpProxy>>,
1411 udp: Rc<RefCell<udp::UdpProxy>>,
1412 zombie_check_interval: Duration,
1413}
1414
1415impl Server {
1416 pub fn try_new_from_config(
1417 worker_to_main_channel: ProxyChannel,
1418 worker_to_main_scm: ScmSocket,
1419 config: ServerConfig,
1420 initial_state: InitialState,
1421 expects_initial_status: bool,
1422 ) -> Result<Self, ServerError> {
1423 let event_loop = Poll::new().map_err(ServerError::CreatePoll)?;
1424 if let Some(cap) = config.basic_auth_max_credential_bytes {
1431 crate::protocol::mux::auth::set_max_decoded_credential_bytes(cap as usize);
1432 }
1433 #[cfg(all(target_os = "linux", feature = "splice"))]
1440 if let Some(cap) = config.splice_pipe_capacity_bytes {
1441 crate::splice::set_pipe_capacity(cap as usize);
1442 }
1443 let pool = Rc::new(RefCell::new(Pool::with_capacity(
1444 config.min_buffers as usize,
1445 config.max_buffers as usize,
1446 config.buffer_size as usize,
1447 )));
1448 let backends = Rc::new(RefCell::new(BackendMap::new()));
1449
1450 let sessions: Rc<RefCell<SessionManager>> = SessionManager::new(
1455 Slab::with_capacity(config.slab_capacity() as usize),
1456 config.max_connections as usize,
1457 config
1458 .max_connections_per_ip
1459 .unwrap_or(sozu_command::config::DEFAULT_MAX_CONNECTIONS_PER_IP),
1460 config
1461 .retry_after
1462 .unwrap_or(sozu_command::config::DEFAULT_RETRY_AFTER),
1463 config
1464 .max_connections_per_subnet
1465 .unwrap_or(sozu_command::config::DEFAULT_MAX_CONNECTIONS_PER_SUBNET),
1466 config
1467 .subnet_ipv4_prefix
1468 .unwrap_or(sozu_command::config::DEFAULT_SUBNET_IPV4_PREFIX),
1469 config
1470 .subnet_ipv6_prefix
1471 .unwrap_or(sozu_command::config::DEFAULT_SUBNET_IPV6_PREFIX),
1472 );
1473 sessions.borrow_mut().max_connection_attempts = match config.max_connection_attempts {
1477 Some(attempts)
1478 if sozu_command::config::validate_max_connection_attempts(attempts).is_ok() =>
1479 {
1480 attempts
1481 }
1482 Some(attempts) => {
1483 error!(
1484 "max_connection_attempts = {} is out of 1..=255, keeping {}",
1485 attempts,
1486 sozu_command::config::DEFAULT_MAX_CONNECTION_ATTEMPTS
1487 );
1488 sozu_command::config::DEFAULT_MAX_CONNECTION_ATTEMPTS
1489 }
1490 None => sozu_command::config::DEFAULT_MAX_CONNECTION_ATTEMPTS,
1491 };
1492 {
1493 let mut s = sessions.borrow_mut();
1494 let entry = s.slab.vacant_entry();
1495 trace!("taking token {:?} for channel", SessionToken(entry.key()));
1496 entry.insert(Rc::new(RefCell::new(ListenSession {
1497 protocol: Protocol::Channel,
1498 })));
1499 }
1500 {
1501 let mut s = sessions.borrow_mut();
1502 let entry = s.slab.vacant_entry();
1503 trace!("taking token {:?} for metrics", SessionToken(entry.key()));
1504 entry.insert(Rc::new(RefCell::new(ListenSession {
1505 protocol: Protocol::Timer,
1506 })));
1507 }
1508 {
1509 let mut s = sessions.borrow_mut();
1510 let entry = s.slab.vacant_entry();
1511 trace!("taking token {:?} for metrics", SessionToken(entry.key()));
1512 entry.insert(Rc::new(RefCell::new(ListenSession {
1513 protocol: Protocol::Metrics,
1514 })));
1515 }
1516
1517 Server::new(
1518 event_loop,
1519 worker_to_main_channel,
1520 worker_to_main_scm,
1521 sessions,
1522 pool,
1523 backends,
1524 None,
1525 None,
1526 None,
1527 config,
1528 Some(initial_state),
1529 expects_initial_status,
1530 )
1531 }
1532
1533 #[allow(clippy::too_many_arguments)]
1534 pub fn new(
1535 poll: Poll,
1536 mut channel: ProxyChannel,
1537 scm: ScmSocket,
1538 sessions: Rc<RefCell<SessionManager>>,
1539 pool: Rc<RefCell<Pool>>,
1540 backends: Rc<RefCell<BackendMap>>,
1541 http: Option<http::HttpProxy>,
1542 https: Option<https::HttpsProxy>,
1543 tcp: Option<tcp::TcpProxy>,
1544 server_config: ServerConfig,
1545 initial_state: Option<InitialState>,
1546 expects_initial_status: bool,
1547 ) -> Result<Self, ServerError> {
1548 FEATURES.with(|_features| {
1549 });
1551
1552 poll.registry()
1553 .register(
1554 &mut channel,
1555 Token(0),
1556 Interest::READABLE | Interest::WRITABLE,
1557 )
1558 .map_err(ServerError::RegisterChannel)?;
1559
1560 METRICS.with(|metrics| {
1561 if let Some(sock) = (*metrics.borrow_mut()).socket_mut() {
1562 poll.registry()
1563 .register(sock, Token(2), Interest::WRITABLE)
1564 .expect("should register the metrics socket");
1565 }
1566 });
1567
1568 let base_sessions_count = sessions.borrow().slab.len();
1569
1570 let http = Rc::new(RefCell::new(match http {
1571 Some(http) => http,
1572 None => {
1573 let registry = poll
1574 .registry()
1575 .try_clone()
1576 .map_err(ServerError::CloneRegistry)?;
1577
1578 http::HttpProxy::new(registry, sessions.clone(), pool.clone(), backends.clone())
1579 }
1580 }));
1581
1582 let https = Rc::new(RefCell::new(match https {
1583 Some(https) => https,
1584 None => {
1585 let registry = poll
1586 .registry()
1587 .try_clone()
1588 .map_err(ServerError::CloneRegistry)?;
1589
1590 https::HttpsProxy::new(registry, sessions.clone(), pool.clone(), backends.clone())
1591 }
1592 }));
1593
1594 let tcp = Rc::new(RefCell::new(match tcp {
1595 Some(tcp) => tcp,
1596 None => {
1597 let registry = poll
1598 .registry()
1599 .try_clone()
1600 .map_err(ServerError::CloneRegistry)?;
1601
1602 tcp::TcpProxy::new(registry, sessions.clone(), pool.clone(), backends.clone())
1603 }
1604 }));
1605
1606 let udp = Rc::new(RefCell::new({
1610 let registry = poll
1611 .registry()
1612 .try_clone()
1613 .map_err(ServerError::CloneRegistry)?;
1614
1615 udp::UdpProxy::new(
1616 registry,
1617 sessions.clone(),
1618 pool.clone(),
1619 backends.clone(),
1620 server_config.max_connections as usize,
1621 server_config.buffer_size as usize,
1622 )
1623 }));
1624
1625 let mut server = Server {
1626 accept_queue_timeout: Duration::from_secs(u64::from(
1627 server_config.accept_queue_timeout,
1628 )),
1629 accept_queue: VecDeque::new(),
1630 evict_on_queue_full: server_config.evict_on_queue_full.unwrap_or(false),
1631 accept_ready: HashSet::new(),
1632 backends,
1633 base_sessions_count,
1634 channel,
1635 config_state: ConfigState::new(),
1636 current_poll_errors: 0,
1637 health_checker: HealthChecker::new(),
1638 http,
1639 https,
1640 last_sessions_len: 0, last_shutting_down_message: None,
1642 last_zombie_check: Instant::now(), loop_start: Instant::now(), started_at: Instant::now(), last_saturation_tick: Instant::now(), max_poll_errors: 10000, pool,
1648 poll_timeout: Some(Duration::from_millis(1000)), poll,
1650 scm_listeners: None,
1651 scm,
1652 sessions,
1653 should_poll_at: None,
1654 shutting_down: None,
1655 closing_sessions: Vec::new(),
1656 yielded_sessions: Vec::new(),
1657 protected_health_check: ProtectedWorker::default(),
1658 tcp,
1659 udp,
1660 zombie_check_interval: Duration::from_secs(u64::from(
1661 server_config.zombie_check_interval,
1662 )),
1663 };
1664
1665 info!("will try to receive listeners");
1686 server
1687 .scm
1688 .set_blocking(true)
1689 .map_err(|scm_err| ServerError::ScmSocket {
1690 msg: "Could not set the scm socket to blocking".to_string(),
1691 scm_err,
1692 })?;
1693 let listeners =
1694 server
1695 .scm
1696 .receive_listeners()
1697 .map_err(|scm_err| ServerError::ScmSocket {
1698 msg: "could not receive listeners from the scm socket".to_string(),
1699 scm_err,
1700 })?;
1701 server
1702 .scm
1703 .set_blocking(false)
1704 .map_err(|scm_err| ServerError::ScmSocket {
1705 msg: "Could not set the scm socket to unblocking".to_string(),
1706 scm_err,
1707 })?;
1708 info!("received listeners: {:?}", listeners);
1709 server.scm_listeners = Some(listeners);
1710
1711 if let Some(state) = initial_state {
1713 for request in state.requests {
1714 trace!("generating initial config request: {:#?}", request);
1715 server.notify_proxys(request);
1716 }
1717
1718 QUEUE.with(|queue| {
1720 (*queue.borrow_mut()).clear();
1721 });
1722 }
1723
1724 server.report_unclaimed_inherited_sockets();
1725
1726 if expects_initial_status {
1727 server.block_channel();
1731 server.await_initial_status()?;
1732 server.unblock_channel();
1733 }
1734
1735 Ok(server)
1736 }
1737
1738 fn await_initial_status(&mut self) -> Result<(), ServerError> {
1750 let mut stage = BootstrapStage::Start;
1751 loop {
1752 let WorkerRequest { id, mut content } = self
1753 .channel
1754 .read_message()
1755 .map_err(ServerError::ReadInitialMessage)?;
1756 debug!("got initial message {}", id);
1758
1759 let Some(request_type) = content.request_type.as_mut() else {
1760 self.answer_initial(worker_response_error(id, REFUSED_INITIAL_REQUEST))?;
1761 continue;
1762 };
1763
1764 if matches!(request_type, RequestType::Status(_)) {
1765 if !stage.accepts_status() {
1766 self.answer_initial(worker_response_error(id, REFUSED_INITIAL_REQUEST))?;
1767 continue;
1768 }
1769 self.answer_initial(WorkerResponse::ok(id))?;
1770 return Ok(());
1771 }
1772
1773 let Some(step) = stage.step(request_type) else {
1774 self.answer_initial(worker_response_error(id, REFUSED_INITIAL_REQUEST))?;
1775 continue;
1776 };
1777 let response = self.protected_health_check.handle(
1778 &id,
1779 request_type,
1780 &mut self.config_state,
1781 &mut self.backends.borrow_mut(),
1782 );
1783 match response {
1784 Some(response) if step.is_answered_by(&response) => {
1785 stage = step.next;
1786 self.answer_initial(response)?;
1787 }
1788 Some(response) if response.status != ResponseStatus::Ok as i32 => {
1790 self.answer_initial(response)?;
1791 }
1792 _ => {
1793 self.answer_initial(worker_response_error(id, REFUSED_INITIAL_REQUEST))?;
1794 }
1795 }
1796 }
1797 }
1798
1799 fn answer_initial(&mut self, response: WorkerResponse) -> Result<(), ServerError> {
1800 self.channel
1801 .write_message(&response)
1802 .map_err(ServerError::WriteInitialAnswer)
1803 }
1804
1805 pub fn run(&mut self) {
1807 let mut events = Events::with_capacity(1024); let mut event_owners: Vec<Option<Rc<RefCell<dyn ProxySession>>>> =
1812 Vec::with_capacity(events.capacity());
1813 self.last_sessions_len = self.sessions.borrow().slab.len();
1814
1815 self.last_zombie_check = Instant::now();
1816 self.loop_start = Instant::now();
1817
1818 loop {
1819 self.check_for_poll_errors();
1820
1821 let timeout = self.reset_loop_time_and_get_timeout();
1822
1823 match self.poll.poll(&mut events, timeout) {
1824 Ok(_) => self.current_poll_errors = 0,
1825 Err(error) => {
1826 error!("Error while polling events: {:?}", error);
1827 self.current_poll_errors += 1;
1828 continue;
1829 }
1830 }
1831
1832 let yielded = std::mem::take(&mut self.yielded_sessions);
1836
1837 let after_epoll = Instant::now();
1838 time!(
1839 names::event_loop::EPOLL_TIME,
1840 (after_epoll - self.loop_start).as_millis()
1841 );
1842 self.loop_start = after_epoll;
1843
1844 tick_metric_detail_leases(after_epoll);
1846
1847 self.send_queue();
1848
1849 self.snapshot_event_owners(events.iter().map(|event| event.token()), &mut event_owners);
1850 for (event, owner) in events.iter().zip(event_owners.drain(..)) {
1851 match event.token() {
1852 Token(0) => {
1854 if event.is_error() {
1860 error!(
1861 "error on command channel, the main process can no longer reach this worker"
1862 );
1863 return;
1864 }
1865 if event.is_read_closed() || event.is_write_closed() {
1866 error!("command channel was closed");
1867 return;
1868 }
1869 let ready = Ready::from(event);
1870 self.channel.handle_events(ready);
1871
1872 loop {
1874 QUEUE.with(|queue| {
1875 if !(*queue.borrow()).is_empty() {
1876 self.channel.interest.insert(Ready::WRITABLE);
1877 }
1878 });
1879
1880 if self.channel.readiness() == Ready::EMPTY {
1883 break;
1884 }
1885
1886 if self.read_channel_messages_and_notify() {
1888 return;
1889 }
1890
1891 QUEUE.with(|queue| {
1892 if !(*queue.borrow()).is_empty() {
1893 self.channel.interest.insert(Ready::WRITABLE);
1894 }
1895 });
1896
1897 self.send_queue();
1898 }
1899 }
1900 Token(1) => {
1902 while let Some(t) = TIMER.with(|timer| timer.borrow_mut().poll()) {
1903 self.timeout(t);
1904 }
1905 }
1906 Token(2) => METRICS.with(|metrics| {
1908 (*metrics.borrow_mut()).writable();
1909 }),
1910 token if self.health_checker.owns_token(token) => {
1913 self.health_checker.ready(token);
1914 }
1915 token if self.udp.borrow().health_owns_token(token) => {
1916 self.udp.borrow_mut().health_ready(token);
1917 }
1918 token => self.ready_if_still_owned(token, Ready::from(event), owner),
1919 }
1920 }
1921 self.resume_yielded_sessions(yielded);
1922
1923 if let Some(t) = self.should_poll_at.as_ref()
1924 && *t <= Instant::now()
1925 {
1926 while let Some(t) = TIMER.with(|timer| timer.borrow_mut().poll()) {
1927 self.timeout(t);
1929 }
1930 }
1931 self.handle_remaining_readiness();
1932 self.create_sessions();
1933
1934 self.should_poll_at = TIMER.with(|timer| timer.borrow().next_poll_date());
1935
1936 self.zombie_check();
1937 self.health_checker
1938 .poll(&self.backends, self.poll.registry());
1939 self.udp.borrow_mut().health_poll();
1943
1944 {
1957 let sessions = self.sessions.borrow();
1958 let nb_connections = sessions.nb_connections;
1959 let max_connections = sessions.max_connections;
1960 let slab_len = sessions.slab.len();
1961 let slab_capacity = sessions.slab.capacity();
1962 let accept_threshold = sessions.accept_slab_threshold();
1963
1964 gauge!(names::client::CONNECTIONS, nb_connections);
1965 gauge!(names::client::CONNECTIONS_MAX, max_connections);
1966 if let Some(percent) = (nb_connections * 100).checked_div(max_connections) {
1967 gauge!(names::client::CONNECTIONS_PERCENT, percent);
1968 }
1969
1970 gauge!(names::slab::ENTRIES, slab_len);
1971 gauge!(names::slab::CAPACITY, slab_capacity);
1972 if let Some(percent) = (slab_len * 100).checked_div(slab_capacity) {
1973 gauge!(names::slab::USAGE_PERCENT, percent);
1974 }
1975 if let Some(percent) = (slab_len * 100).checked_div(accept_threshold) {
1976 gauge!(names::slab::ACCEPT_THRESHOLD_PERCENT, percent);
1977 }
1978 }
1979 {
1985 let pool = self.pool.borrow();
1986 let used = pool.inner.used();
1987 let capacity = pool.inner.capacity();
1988 gauge!(names::buffer::IN_USE, used);
1989 gauge!(names::buffer::CAPACITY, capacity);
1990 if let Some(percent) = (used * 100).checked_div(capacity) {
1991 gauge!(names::buffer::USAGE_PERCENT, percent);
1992 }
1993 }
1994 let now = Instant::now();
2002 if now.duration_since(self.last_saturation_tick) >= ACCEPT_SATURATION_TICK {
2003 if !self.sessions.borrow().can_accept {
2004 incr!(names::accept_queue::SATURATED_SECONDS);
2005 }
2006 self.last_saturation_tick = now;
2007 }
2008 gauge!(
2012 names::process::UPTIME_SECONDS,
2013 self.started_at.elapsed().as_secs() as usize
2014 );
2015 gauge!(
2020 names::server::LIVE,
2021 if self.shutting_down.is_some() { 0 } else { 1 }
2022 );
2023 METRICS.with(|metrics| {
2024 (*metrics.borrow_mut()).send_data();
2025 });
2026
2027 if self.shutting_down.is_some() && self.shut_down_sessions() {
2028 return;
2029 }
2030 if !self.closing_sessions.is_empty() {
2031 self.drive_closing_sessions();
2032 }
2033 }
2034 }
2035
2036 fn check_for_poll_errors(&mut self) {
2037 if self.current_poll_errors >= self.max_poll_errors {
2038 error!(
2039 "Something is going very wrong. Last {} poll() calls failed, crashing..",
2040 self.current_poll_errors
2041 );
2042 panic!(
2043 "poll() calls failed {} times in a row",
2044 self.current_poll_errors
2045 );
2046 }
2047 }
2048
2049 fn reset_loop_time_and_get_timeout(&mut self) -> Option<Duration> {
2050 let now = Instant::now();
2051 time!(
2052 names::event_loop::EVENT_LOOP_TIME,
2053 (now - self.loop_start).as_millis()
2054 );
2055
2056 let mut timeout = match self.should_poll_at.as_ref() {
2057 None => self.poll_timeout,
2058 Some(i) => {
2059 if *i <= now {
2060 self.poll_timeout
2061 } else {
2062 let dur = *i - now;
2063 match self.poll_timeout {
2064 None => Some(dur),
2065 Some(t) => {
2066 if t < dur {
2067 Some(t)
2068 } else {
2069 Some(dur)
2070 }
2071 }
2072 }
2073 }
2074 }
2075 };
2076
2077 if self.shutting_down.is_some() || !self.closing_sessions.is_empty() {
2081 let shutdown_tick = Duration::from_millis(100);
2082 timeout = match timeout {
2083 None => Some(shutdown_tick),
2084 Some(current) => Some(current.min(shutdown_tick)),
2085 };
2086 }
2087
2088 if !self.yielded_sessions.is_empty() {
2092 timeout = Some(Duration::ZERO);
2093 }
2094
2095 self.loop_start = now;
2096 timeout
2097 }
2098
2099 fn read_channel_messages_and_notify(&mut self) -> bool {
2101 if !self.channel.readiness().is_readable() {
2102 return false;
2103 }
2104
2105 if let Err(e) = self.channel.readable() {
2106 error!("error reading from channel: {:?}", e);
2107 }
2108
2109 loop {
2110 let request = self.channel.read_message();
2111 debug!("Received request {:?}", request);
2112 match request {
2113 Ok(request) => match request.content.request_type {
2114 Some(RequestType::HardStop(_)) => {
2115 let req_id = request.id.clone();
2116 self.notify(request);
2117 if let Err(e) = self.channel.write_message(&WorkerResponse::ok(req_id)) {
2118 error!("Could not send ok response to the main process: {}", e);
2119 }
2120 if let Err(e) = self.channel.run() {
2121 error!("Error while running the server channel: {}", e);
2122 }
2123 return true;
2124 }
2125 Some(RequestType::SoftStop(_)) => {
2126 self.shutting_down = Some(request.id.clone());
2127 self.last_sessions_len = self.sessions.borrow().slab.len();
2128 self.notify(request);
2129 }
2130 Some(RequestType::ReturnListenSockets(_)) => {
2131 info!("received ReturnListenSockets order");
2132 match self.return_listen_sockets() {
2133 Ok(_) => push_queue(WorkerResponse::ok(request.id)),
2134 Err(error) => push_queue(worker_response_error(
2135 request.id,
2136 format!("Could not send listeners on scm socket: {error:?}"),
2137 )),
2138 }
2139 }
2140 _ => self.notify(request),
2141 },
2142 Err(_) => {
2144 if (self.channel.interest & self.channel.readiness).is_readable() {
2146 if let Err(e) = self.channel.readable() {
2147 error!("error reading from channel: {:?}", e);
2148 }
2149 continue;
2150 }
2151 break;
2152 }
2153 }
2154 }
2155 false
2156 }
2157
2158 fn zombie_check(&mut self) {
2160 let now = Instant::now();
2161 if now - self.last_zombie_check < self.zombie_check_interval {
2162 return;
2163 }
2164 info!("zombie check");
2165 debug_assert!(
2168 now >= self.last_zombie_check,
2169 "zombie-check timestamp must never move backwards"
2170 );
2171 self.last_zombie_check = now;
2172
2173 let mut zombie_tokens = HashSet::new();
2174
2175 for (_index, session) in self
2177 .sessions
2178 .borrow_mut()
2179 .slab
2180 .iter_mut()
2181 .filter(|(_, c)| now - c.borrow().last_event() > self.zombie_check_interval)
2182 {
2183 let session_token = session.borrow().frontend_token();
2184 if !zombie_tokens.contains(&session_token) {
2185 session.borrow().print_session();
2186 zombie_tokens.insert(session_token);
2187 }
2188 }
2189
2190 debug_assert!(
2195 !self.sessions.borrow().slab.iter().any(|(_, session)| {
2196 let s = session.borrow();
2197 zombie_tokens.contains(&s.frontend_token())
2198 && matches!(
2199 s.protocol(),
2200 Protocol::HTTPListen | Protocol::HTTPSListen | Protocol::TCPListen
2201 )
2202 }),
2203 "zombie reaping must never target a listener session"
2204 );
2205
2206 let zombie_count = zombie_tokens.len() as i64;
2207 count!(names::misc::ZOMBIES, zombie_count);
2208
2209 let remaining_count = self.shut_down_sessions_by_frontend_tokens(zombie_tokens);
2210 info!(
2211 "removing {} zombies ({} remaining entries after close)",
2212 zombie_count, remaining_count
2213 );
2214 }
2215
2216 fn shut_down_sessions_by_frontend_tokens(&self, tokens: HashSet<Token>) -> usize {
2219 if tokens.is_empty() {
2220 return 0;
2221 }
2222
2223 for token in &tokens {
2225 if self.sessions.borrow().slab.contains(token.0) {
2226 let slab_before = self.sessions.borrow().slab.len();
2227 let session = { self.sessions.borrow_mut().slab.remove(token.0) };
2228 session.borrow_mut().close();
2229 self.sessions.borrow_mut().decr();
2230 debug_assert!(
2235 !self.sessions.borrow().slab.contains(token.0),
2236 "removed token must be absent from the slab"
2237 );
2238 debug_assert!(
2239 self.sessions.borrow().slab.len() < slab_before,
2240 "removing a present session must free at least its own slab slot"
2241 );
2242 }
2243 }
2244
2245 let mut dangling_entries = HashSet::new();
2247 for (entry_key, session) in &self.sessions.borrow().slab {
2248 if tokens.contains(&session.borrow().frontend_token()) {
2249 dangling_entries.insert(entry_key);
2250 }
2251 }
2252
2253 let mut dangling_entries_count = 0;
2255 for entry_key in dangling_entries {
2256 let mut sessions = self.sessions.borrow_mut();
2257 if sessions.slab.contains(entry_key) {
2258 sessions.slab.remove(entry_key);
2259 dangling_entries_count += 1;
2260 }
2261 }
2262 debug_assert!(
2266 !self
2267 .sessions
2268 .borrow()
2269 .slab
2270 .iter()
2271 .any(|(_, session)| tokens.contains(&session.borrow().frontend_token())),
2272 "no slab entry may reference a closed frontend token after teardown"
2273 );
2274 dangling_entries_count
2275 }
2276
2277 fn shut_down_sessions(&mut self) -> bool {
2279 let sessions_count = self.sessions.borrow().slab.len();
2280 let mut sessions_to_shut_down = HashSet::new();
2281
2282 for (_key, session) in &self.sessions.borrow().slab {
2283 let mut session = session.borrow_mut();
2284 if session.shutting_down() {
2285 debug!(
2286 "Server killing session from shutting_down: token={:?}, protocol={:?}",
2287 session.frontend_token(),
2288 session.protocol()
2289 );
2290 sessions_to_shut_down.insert(Token(session.frontend_token().0));
2291 }
2292 }
2293 let _ = self.shut_down_sessions_by_frontend_tokens(sessions_to_shut_down);
2294
2295 let new_sessions_count = self.sessions.borrow().slab.len();
2296
2297 if new_sessions_count < sessions_count {
2298 let now = Instant::now();
2299 if let Some(last) = self.last_shutting_down_message
2300 && (now - last) > Duration::from_secs(5)
2301 {
2302 info!(
2303 "closed {} sessions, {} sessions left, base_sessions_count = {}",
2304 sessions_count - new_sessions_count,
2305 new_sessions_count,
2306 self.base_sessions_count
2307 );
2308 }
2309 self.last_shutting_down_message = Some(now);
2310 }
2311
2312 if new_sessions_count <= self.base_sessions_count {
2313 info!("last session stopped, shutting down!");
2314 if let Err(e) = self.channel.run() {
2315 error!("Error while running the server channel: {}", e);
2316 }
2317 let id = self
2319 .shutting_down
2320 .take()
2321 .expect("should have shut down correctly"); debug!("Responding OK to main process for request {}", id);
2324
2325 let proxy_response = WorkerResponse::ok(id);
2326 if let Err(e) = self.channel.write_message(&proxy_response) {
2327 error!("Could not write response to the main process: {}", e);
2328 }
2329 if let Err(e) = self.channel.run() {
2330 error!("Error while running the server channel: {}", e);
2331 }
2332 return true;
2333 }
2334
2335 if new_sessions_count < self.last_sessions_len {
2336 info!(
2337 "shutting down, {} slab elements remaining (base: {})",
2338 new_sessions_count - self.base_sessions_count,
2339 self.base_sessions_count
2340 );
2341 self.last_sessions_len = new_sessions_count;
2342 }
2343
2344 false
2345 }
2346
2347 fn close_sessions(&mut self, target: &SessionTarget) -> ClosedSessions {
2353 let mut targeted: HashMap<Token, Rc<RefCell<dyn ProxySession>>> = HashMap::new();
2356 for (_key, session) in &self.sessions.borrow().slab {
2357 let borrowed = session.borrow();
2358 if borrowed.targets(target) {
2359 targeted
2360 .entry(borrowed.frontend_token())
2361 .or_insert_with(|| session.clone());
2362 }
2363 }
2364
2365 let mut to_close = HashSet::new();
2366 let mut draining = 0u64;
2367 for (token, session) in targeted {
2368 if session.borrow_mut().shutting_down() {
2369 to_close.insert(token);
2370 } else {
2371 draining += 1;
2372 let already_closing = self.closing_sessions.iter().any(|(_, pending)| {
2373 pending.upgrade().is_some_and(|p| Rc::ptr_eq(&p, &session))
2374 });
2375 if !already_closing {
2376 self.closing_sessions.push((token, Rc::downgrade(&session)));
2377 }
2378 }
2379 }
2380 let closed = to_close.len() as u64;
2381 self.shut_down_sessions_by_frontend_tokens(to_close);
2382
2383 debug_assert!(
2385 self.closing_sessions.len() as u64 >= draining,
2386 "every draining session must be watched by drive_closing_sessions"
2387 );
2388 ClosedSessions { closed, draining }
2389 }
2390
2391 fn drive_closing_sessions(&mut self) {
2395 let pending = std::mem::take(&mut self.closing_sessions);
2396 let mut to_close = HashSet::new();
2397 for (token, weak) in pending {
2398 let Some(session) = weak.upgrade() else {
2399 continue;
2400 };
2401 let still_there = self
2403 .sessions
2404 .borrow()
2405 .slab
2406 .get(token.0)
2407 .is_some_and(|entry| Rc::ptr_eq(entry, &session));
2408 if !still_there {
2409 continue;
2410 }
2411 if session.borrow_mut().shutting_down() {
2412 to_close.insert(token);
2413 } else {
2414 self.closing_sessions.push((token, weak));
2415 }
2416 }
2417 debug_assert!(
2419 self.closing_sessions
2420 .iter()
2421 .all(|(token, _)| !to_close.contains(token)),
2422 "a session closed this tick must not stay watched"
2423 );
2424 self.shut_down_sessions_by_frontend_tokens(to_close);
2425 }
2426
2427 fn queue_if_yielded(&mut self, session: &Rc<RefCell<dyn ProxySession>>) {
2430 if !session.borrow_mut().yielded() {
2431 return;
2432 }
2433 let queued = self
2434 .yielded_sessions
2435 .iter()
2436 .any(|(_, weak)| weak.upgrade().is_some_and(|q| Rc::ptr_eq(&q, session)));
2437 if !queued {
2438 let token = session.borrow().frontend_token();
2439 self.yielded_sessions.push((token, Rc::downgrade(session)));
2440 }
2441 }
2442
2443 fn resume_yielded_sessions(&mut self, pending: Vec<(Token, Weak<RefCell<dyn ProxySession>>)>) {
2450 for (token, weak) in pending {
2451 let Some(session) = weak.upgrade() else {
2452 continue;
2453 };
2454 let still_there = self
2455 .sessions
2456 .borrow()
2457 .slab
2458 .get(token.0)
2459 .is_some_and(|entry| Rc::ptr_eq(entry, &session));
2460 if !still_there {
2461 trace!(
2462 "PROXY\t{:?} dropping the resumption of a yielded session that closed since",
2463 token
2464 );
2465 continue;
2466 }
2467 if session.borrow_mut().ready(session.clone()) {
2468 debug!(
2469 "Server killing session from a resumed ready: token={:?}, protocol={:?}",
2470 token,
2471 session.borrow().protocol()
2472 );
2473 self.kill_session(session);
2474 } else {
2475 self.queue_if_yielded(&session);
2476 }
2477 }
2478 }
2479
2480 fn kill_session(&self, session: Rc<RefCell<dyn ProxySession>>) {
2481 let token = session.borrow().frontend_token();
2482 let _ = self.shut_down_sessions_by_frontend_tokens(HashSet::from([token]));
2483 }
2484
2485 fn send_queue(&mut self) {
2486 if self.channel.readiness.is_writable() {
2487 QUEUE.with(|q| {
2488 let mut queue = q.borrow_mut();
2489 loop {
2490 if let Some(resp) = queue.pop_front() {
2491 debug!("Sending response {:?}", resp);
2492 if let Err(e) = self.channel.write_message(&resp) {
2493 error!("Could not write message {} on the channel: {}", resp, e);
2494 queue.push_front(resp);
2495 }
2496 }
2497
2498 if self.channel.back_buf.available_data() > 0
2499 && let Err(e) = self.channel.writable()
2500 {
2501 error!("error writing to channel: {:?}", e);
2502 }
2503
2504 if !self.channel.readiness.is_writable() {
2505 break;
2506 }
2507
2508 if self.channel.back_buf.available_data() == 0 && queue.is_empty() {
2509 break;
2510 }
2511 }
2512 });
2513 }
2514 }
2515
2516 fn notify(&mut self, message: WorkerRequest) {
2517 let now = std::time::Instant::now();
2523 tick_metric_detail_leases(now);
2524 match &message.content.request_type {
2525 Some(RequestType::ConfigureMetrics(configuration)) => {
2526 match MetricsConfiguration::try_from(*configuration) {
2527 Ok(metrics_config) => {
2528 METRICS.with(|metrics| {
2529 (*metrics.borrow_mut()).configure(&metrics_config);
2530 push_queue(WorkerResponse::ok(message.id));
2531 });
2532 }
2533 Err(e) => {
2534 error!("Error configuring metrics: {}", e);
2535 push_queue(WorkerResponse::error(message.id, e));
2536 }
2537 }
2538 return;
2539 }
2540 Some(RequestType::QueryMetrics(query_metrics_options)) => {
2541 METRICS.with(|metrics| {
2542 match (*metrics.borrow_mut()).query(query_metrics_options) {
2543 Ok(c) => push_queue(WorkerResponse::ok_with_content(message.id, c)),
2544 Err(e) => {
2545 error!("Error querying metrics: {}", e);
2546 push_queue(WorkerResponse::error(message.id, e))
2547 }
2548 }
2549 });
2550 return;
2551 }
2552 Some(RequestType::SetMetricDetail(req)) => {
2560 let presented_binding = crate::metrics::PeerBinding {
2567 pid: req.peer_pid,
2568 session_ulid: req.peer_session_ulid.as_deref().and_then(|s| {
2574 rusty_ulid::Ulid::from_str(s)
2575 .map(u128::from)
2576 .ok()
2577 .or_else(|| u128::from_str_radix(s.trim_start_matches("0x"), 16).ok())
2578 }),
2579 };
2580 if req.clear.unwrap_or(false) {
2581 if req.client_id.len() > crate::metrics::LEASE_CLIENT_ID_MAX_BYTES {
2592 let msg = format!(
2593 "SetMetricDetail: clear client_id length {} exceeds {} bytes",
2594 req.client_id.len(),
2595 crate::metrics::LEASE_CLIENT_ID_MAX_BYTES,
2596 );
2597 error!("{}", msg);
2598 push_queue(WorkerResponse::error(message.id.clone(), msg));
2599 return;
2600 }
2601 let (outcome, effective_after, configured_after, lease_count_after) = METRICS
2609 .with(|metrics| {
2610 let mut m = metrics.borrow_mut();
2611 let outcome = m.lease_clear(&req.client_id, presented_binding);
2612 (
2613 outcome,
2614 m.detail_effective(),
2615 m.detail_configured(),
2616 m.lease_count(),
2617 )
2618 });
2619 match outcome {
2620 crate::metrics::LeaseClearOutcome::Cleared { previous_effective } => {
2621 push_metric_detail_transition(
2622 previous_effective,
2623 effective_after,
2624 "lease_clear",
2625 Some(req.client_id.clone()),
2626 );
2627 push_queue(WorkerResponse::ok_with_content(
2628 message.id.clone(),
2629 worker_metric_detail_status_content(
2630 configured_after,
2631 effective_after,
2632 previous_effective,
2633 lease_count_after,
2634 ),
2635 ));
2636 }
2637 crate::metrics::LeaseClearOutcome::NotFound => {
2638 push_queue(WorkerResponse::ok_with_content(
2642 message.id.clone(),
2643 worker_metric_detail_status_content(
2644 configured_after,
2645 effective_after,
2646 effective_after,
2647 lease_count_after,
2648 ),
2649 ));
2650 }
2651 crate::metrics::LeaseClearOutcome::Unauthorized => {
2652 let msg = "SetMetricDetail: clear refused (peer \
2663 binding does not match the apply-time owner)"
2664 .to_owned();
2665 error!("{}", msg);
2666 push_queue(WorkerResponse::error(message.id.clone(), msg));
2667 }
2668 }
2669 return;
2670 }
2671 let detail_proto = match req.detail {
2672 Some(d) => d,
2673 None => {
2674 let msg = "SetMetricDetail without `detail` and without `clear`".to_owned();
2681 error!("{}", msg);
2682 push_queue(WorkerResponse::error(message.id.clone(), msg));
2683 return;
2684 }
2685 };
2686 let detail_enum = match MetricDetail::try_from(detail_proto) {
2687 Ok(d) => d,
2688 Err(e) => {
2689 let msg =
2690 format!("SetMetricDetail: invalid MetricDetail variant {detail_proto}");
2691 error!("{}: {}", msg, e);
2692 push_queue(WorkerResponse::error(message.id.clone(), msg));
2693 return;
2694 }
2695 };
2696 let level = MetricDetailLevel::from(detail_enum);
2697 if let Some(t) = req.ttl_seconds
2706 && u64::from(t) > crate::metrics::LEASE_TTL_MAX.as_secs()
2707 {
2708 let msg = format!(
2709 "SetMetricDetail: ttl_seconds={t} exceeds LEASE_TTL_MAX={}",
2710 crate::metrics::LEASE_TTL_MAX.as_secs()
2711 );
2712 error!("{}", msg);
2713 push_queue(WorkerResponse::error(message.id.clone(), msg));
2714 return;
2715 }
2716 let ttl_seconds = req.ttl_seconds.filter(|&t| t > 0).unwrap_or_else(|| {
2717 u32::try_from(crate::metrics::LEASE_TTL_DEFAULT.as_secs()).unwrap_or(60)
2724 });
2725 let ttl = std::time::Duration::from_secs(ttl_seconds.into());
2726 let (outcome, configured_after, lease_count_after) = METRICS.with(|metrics| {
2727 let mut m = metrics.borrow_mut();
2728 let outcome =
2729 m.lease_apply(req.client_id.clone(), level, ttl, presented_binding, now);
2730 (outcome, m.detail_configured(), m.lease_count())
2731 });
2732 match outcome {
2733 crate::metrics::LeaseApplyOutcome::Applied {
2734 previous_effective,
2735 new_effective,
2736 } => {
2737 push_metric_detail_transition(
2738 previous_effective,
2739 new_effective,
2740 "lease_apply",
2741 Some(req.client_id.clone()),
2742 );
2743 push_queue(WorkerResponse::ok_with_content(
2744 message.id.clone(),
2745 worker_metric_detail_status_content(
2746 configured_after,
2747 new_effective,
2748 previous_effective,
2749 lease_count_after,
2750 ),
2751 ));
2752 }
2753 crate::metrics::LeaseApplyOutcome::ClientIdTooLong => {
2754 let msg = format!(
2755 "SetMetricDetail: client_id length {} exceeds {} bytes",
2756 req.client_id.len(),
2757 crate::metrics::LEASE_CLIENT_ID_MAX_BYTES,
2758 );
2759 error!("{}", msg);
2760 push_queue(WorkerResponse::error(message.id.clone(), msg));
2761 }
2762 crate::metrics::LeaseApplyOutcome::TableFull => {
2763 let msg = format!(
2770 "SetMetricDetail: lease table at capacity ({} entries); reject new \
2771 apply — operators must retry after an active lease expires or is \
2772 cleared",
2773 crate::metrics::LEASE_TABLE_CAP,
2774 );
2775 error!("{}", msg);
2776 push_queue(WorkerResponse::error(message.id.clone(), msg));
2777 }
2778 crate::metrics::LeaseApplyOutcome::TtlOutOfRange => {
2779 let msg = format!(
2785 "SetMetricDetail: ttl exceeds LEASE_TTL_MAX={} (internal contract \
2786 violation: dispatch gate should have rejected)",
2787 crate::metrics::LEASE_TTL_MAX.as_secs(),
2788 );
2789 error!("{}", msg);
2790 push_queue(WorkerResponse::error(message.id.clone(), msg));
2791 }
2792 crate::metrics::LeaseApplyOutcome::Unauthorized => {
2793 let msg = "SetMetricDetail: renewal refused (peer binding does not \
2802 match the apply-time owner)"
2803 .to_owned();
2804 error!("{}", msg);
2805 push_queue(WorkerResponse::error(message.id.clone(), msg));
2806 }
2807 }
2808 return;
2809 }
2810 Some(RequestType::Logging(logging_filter)) => {
2811 info!(
2812 "{} changing logging filter to {}",
2813 message.id, logging_filter
2814 );
2815 let (directives, _errors) = logging::parse_logging_spec(logging_filter);
2817 logging::LOGGER.with(|logger| {
2818 logger.borrow_mut().set_directives(directives);
2819 });
2820 push_queue(WorkerResponse::ok(message.id));
2821 return;
2822 }
2823 Some(RequestType::QueryClustersHashes(_)) => {
2824 push_queue(WorkerResponse::ok_with_content(
2825 message.id.clone(),
2826 ContentType::ClusterHashes(ClusterHashes {
2827 map: self
2828 .config_state
2829 .hash_state()
2830 .into_iter()
2831 .map(|(cluster_id, hash)| (cluster_id.to_string(), hash))
2832 .collect(),
2833 })
2834 .into(),
2835 ));
2836 return;
2837 }
2838 Some(RequestType::QueryClusterById(cluster_id)) => {
2839 push_queue(WorkerResponse::ok_with_content(
2840 message.id.clone(),
2841 ContentType::Clusters(ClusterInformations {
2842 vec: self
2843 .config_state
2844 .cluster_state(cluster_id)
2845 .map_or(vec![], |ci| vec![ci]),
2846 })
2847 .into(),
2848 ));
2849 }
2850 Some(RequestType::SetMaxConnectionsPerIp(limit)) => {
2851 let mut sessions = self.sessions.borrow_mut();
2852 let previous = sessions.max_connections_per_ip;
2853 sessions.max_connections_per_ip = *limit;
2854 if *limit == 0 {
2859 sessions.clear_cluster_ip_tracking();
2860 }
2861 info!(
2862 "{} updated global max_connections_per_ip from {} to {}",
2863 message.id, previous, limit
2864 );
2865 push_queue(WorkerResponse::ok(message.id));
2866 return;
2867 }
2868 Some(RequestType::QueryMaxConnectionsPerIp(_)) => {
2869 let limit = self.sessions.borrow().max_connections_per_ip;
2870 push_queue(WorkerResponse::ok_with_content(
2871 message.id,
2872 ContentType::MaxConnectionsPerIpLimit(
2873 sozu_command::proto::command::MaxConnectionsPerIpLimit { limit },
2874 )
2875 .into(),
2876 ));
2877 return;
2878 }
2879 Some(RequestType::SetMaxConnectionsPerSubnet(limit)) => {
2880 let mut sessions = self.sessions.borrow_mut();
2881 let previous = sessions.max_connections_per_subnet;
2882 sessions.max_connections_per_subnet = *limit;
2883 if *limit == 0 {
2889 sessions.clear_cluster_subnet_tracking();
2890 }
2891 info!(
2892 "{} updated global max_connections_per_subnet from {} to {}",
2893 message.id, previous, limit
2894 );
2895 push_queue(WorkerResponse::ok(message.id));
2896 return;
2897 }
2898 Some(RequestType::SetMaxConnectionAttempts(attempts)) => {
2899 if let Err(reason) =
2900 sozu_command::config::validate_max_connection_attempts(*attempts)
2901 {
2902 push_queue(WorkerResponse::error(message.id, reason));
2903 return;
2904 }
2905 let mut sessions = self.sessions.borrow_mut();
2906 let previous = sessions.max_connection_attempts;
2907 sessions.max_connection_attempts = *attempts;
2908 info!(
2909 "{} updated global max_connection_attempts from {} to {}",
2910 message.id, previous, attempts
2911 );
2912 push_queue(WorkerResponse::ok(message.id));
2913 return;
2914 }
2915 Some(RequestType::CloseSessions(request)) => {
2916 let target = SessionTarget::from(request);
2917 let closed = self.close_sessions(&target);
2918 info!(
2919 "{} CloseSessions on {:?}: {} sessions closed, {} draining",
2920 message.id, target, closed.closed, closed.draining
2921 );
2922 push_queue(WorkerResponse::ok_with_content(
2923 message.id,
2924 ContentType::ClosedSessions(closed).into(),
2925 ));
2926 return;
2927 }
2928 Some(RequestType::QueryMaxConnectionsPerSubnet(_)) => {
2929 let sessions = self.sessions.borrow();
2930 let limit = sessions.max_connections_per_subnet;
2931 let ipv4_prefix = sessions.subnet_ipv4_prefix;
2935 let ipv6_prefix = sessions.subnet_ipv6_prefix;
2936 drop(sessions);
2937 push_queue(WorkerResponse::ok_with_content(
2938 message.id,
2939 ContentType::MaxConnectionsPerSubnetLimit(
2940 sozu_command::proto::command::MaxConnectionsPerSubnetLimit {
2941 limit,
2942 ipv4_prefix,
2943 ipv6_prefix,
2944 },
2945 )
2946 .into(),
2947 ));
2948 return;
2949 }
2950 Some(RequestType::QueryClustersByDomain(domain)) => {
2951 let cluster_ids = self
2952 .config_state
2953 .get_cluster_ids_by_domain(domain.hostname.clone(), domain.path.clone());
2954 let vec = cluster_ids
2955 .iter()
2956 .filter_map(|cluster_id| self.config_state.cluster_state(cluster_id))
2957 .collect();
2958
2959 push_queue(WorkerResponse::ok_with_content(
2960 message.id.clone(),
2961 ContentType::Clusters(ClusterInformations { vec }).into(),
2962 ));
2963 return;
2964 }
2965 Some(RequestType::QueryCertificatesFromWorkers(filters))
2976 if filters.fingerprint.is_some() =>
2977 {
2978 let certs = self.config_state.get_certificates(filters.clone());
2979 let response = if !certs.is_empty() {
2980 WorkerResponse::ok_with_content(
2981 message.id.clone(),
2982 ContentType::CertificatesWithFingerprints(CertificatesWithFingerprints {
2983 certs,
2984 })
2985 .into(),
2986 )
2987 } else {
2988 worker_response_error(
2989 message.id.clone(),
2990 "Could not find certificate for this fingerprint",
2991 )
2992 };
2993 push_queue(response);
2994 return;
2995 }
2996 _other_request => {}
2999 }
3000 self.notify_proxys(message);
3001 }
3002
3003 pub fn notify_proxys(&mut self, mut request: WorkerRequest) {
3004 if matches!(
3008 request.content.request_type,
3009 Some(RequestType::SetProtectedHealthCheck(_))
3010 ) {
3011 push_queue(worker_response_error(
3012 request.id,
3013 "SetProtectedHealthCheck is not a worker order",
3014 ));
3015 return;
3016 }
3017 if let Some(request_type) = request.content.request_type.as_mut()
3018 && let Some(response) = self.protected_health_check.handle(
3019 &request.id,
3020 request_type,
3021 &mut self.config_state,
3022 &mut self.backends.borrow_mut(),
3023 )
3024 {
3025 push_queue(response);
3026 return;
3027 }
3028
3029 if let Some(reason) = request
3035 .content
3036 .request_type
3037 .as_ref()
3038 .and_then(|request_type| {
3039 self.config_state
3040 .health_check_port_unix_conflict(request_type)
3041 })
3042 {
3043 push_queue(worker_response_error(request.id, reason));
3044 return;
3045 }
3046
3047 let cluster_objects = match &request.content.request_type {
3051 Some(RequestType::RemoveCluster(cluster_id)) => {
3052 self.config_state.cluster_removal_requests(cluster_id)
3053 }
3054 _ => Vec::new(),
3055 };
3056
3057 let rollback_candidate = request
3061 .content
3062 .request_type
3063 .as_ref()
3064 .and_then(|request_type| self.config_state.rollback_for(request_type));
3065 let mut rollback = None;
3066 if let Err(e) = self.config_state.dispatch(&request.content) {
3067 if self
3068 .config_state
3069 .removed_with_its_cluster(&request.content, &e)
3070 {
3071 debug!("{} already removed with its cluster: {}", request.id, e);
3077 push_queue(WorkerResponse::ok(request.id));
3078 return;
3079 }
3080 if is_frontend_removal(&request.content)
3081 && matches!(e, StateError::NotFound { .. } | StateError::NoChange)
3082 {
3083 push_queue(WorkerResponse::error(
3093 request.id,
3094 format!("Could not execute order on config state: {e}"),
3095 ));
3096 return;
3097 }
3098 if matches!(e, StateError::ClusterNotFound { .. }) {
3099 push_queue(WorkerResponse::error(
3105 request.id,
3106 format!("Could not execute order on config state: {e}"),
3107 ));
3108 return;
3109 }
3110 error!("Could not execute order on config state: {}", e);
3111 } else {
3112 rollback = rollback_candidate;
3113 }
3114
3115 let req_id = request.id.clone();
3116
3117 match request.content.request_type {
3118 Some(RequestType::AddCluster(ref cluster)) => {
3119 if let Some(hc) = cluster.health_check.as_ref()
3127 && let Err(reason) = sozu_command::config::validate_health_check_config(hc)
3128 {
3129 push_queue(worker_response_error(req_id, reason));
3130 return;
3131 }
3132 if let Err(reason) = sozu_command::config::validate_shuffle_sharding(
3135 cluster.shard_percent,
3136 cluster.shard_min_backends,
3137 cluster.shard_mode,
3138 ) {
3139 push_queue(worker_response_error(req_id, reason));
3140 return;
3141 }
3142 if let Some(Err(reason)) = cluster
3143 .max_connection_attempts
3144 .map(sozu_command::config::validate_max_connection_attempts)
3145 {
3146 push_queue(worker_response_error(req_id, reason));
3147 return;
3148 }
3149 self.protected_health_check
3150 .forget_cluster(&cluster.cluster_id);
3151 self.add_cluster(cluster);
3152 if cluster.health_check.is_none() {
3156 self.health_checker.remove_cluster(&cluster.cluster_id);
3157 }
3158 METRICS.with(|metrics| {
3163 (*metrics.borrow_mut()).add_cluster(&cluster.cluster_id);
3164 });
3165 }
3167 Some(RequestType::RemoveCluster(ref cluster_id)) => {
3168 self.protected_health_check.forget_cluster(cluster_id);
3169 self.remove_cluster_objects(&req_id, cluster_id, cluster_objects);
3170 self.remove_health_check_state(cluster_id);
3171 METRICS.with(|metrics| {
3172 (*metrics.borrow_mut()).remove_cluster(cluster_id);
3173 });
3174 }
3176 Some(RequestType::SetHealthCheck(ref set)) => {
3177 if let Err(reason) = sozu_command::config::validate_health_check_config(&set.config)
3178 {
3179 push_queue(worker_response_error(req_id, reason));
3180 return;
3181 }
3182 self.protected_health_check.forget_cluster(&set.cluster_id);
3183 self.backends
3184 .borrow_mut()
3185 .set_health_check_config(&set.cluster_id, Some(set.config.to_owned()));
3186 push_queue(WorkerResponse::ok(req_id));
3187 return;
3188 }
3189 Some(RequestType::RemoveHealthCheck(ref cluster_id)) => {
3190 self.remove_health_check_state(cluster_id);
3191 push_queue(WorkerResponse::ok(req_id));
3192 return;
3193 }
3194 Some(RequestType::AddBackend(ref backend))
3195 | Some(RequestType::AddUnixBackend(ref backend)) => {
3196 push_queue(self.add_backend(&req_id, backend));
3197 return;
3198 }
3199 Some(RequestType::RemoveBackend(ref remove_backend)) => {
3200 push_queue(self.remove_backend(&req_id, remove_backend));
3201 return;
3202 }
3203 _ => {}
3204 };
3205
3206 let proxy_destinations = request.content.get_destinations();
3207 let mut notify_response = None;
3208 if proxy_destinations.to_http_proxy {
3209 notify_response = Some(self.http.borrow_mut().notify(request.clone()));
3210 }
3211 if proxy_destinations.to_https_proxy {
3212 let http_proxy_response = self.https.borrow_mut().notify(request.clone());
3213 if http_proxy_response.is_failure() || notify_response.is_none() {
3214 notify_response = Some(http_proxy_response);
3215 }
3216 }
3217 if proxy_destinations.to_tcp_proxy {
3218 let tcp_proxy_response = self.tcp.borrow_mut().notify(request.clone());
3219 if tcp_proxy_response.is_failure() || notify_response.is_none() {
3220 notify_response = Some(tcp_proxy_response);
3221 }
3222 }
3223 if proxy_destinations.to_udp_proxy {
3224 let udp_proxy_response = self.udp.borrow_mut().notify(request.clone());
3225 if udp_proxy_response.is_failure() || notify_response.is_none() {
3226 notify_response = Some(udp_proxy_response);
3227 }
3228 }
3229 if let Some(response) = notify_response {
3230 self.finish_proxy_order(&mut rollback, response);
3231 }
3232
3233 match request.content.request_type {
3234 Some(RequestType::AddHttpListener(listener)) => {
3236 let response = self.notify_add_http_listener(&req_id, listener);
3237 self.finish_proxy_order(&mut rollback, response);
3238 }
3239 Some(
3240 RequestType::AddHttpsListener(listener)
3241 | RequestType::AddHttpsListenerWithClientAuth(listener),
3242 ) => {
3243 let response = self.notify_add_https_listener(&req_id, listener);
3244 self.finish_proxy_order(&mut rollback, response);
3245 }
3246 Some(RequestType::AddTcpListener(listener)) => {
3247 let response = self.notify_add_tcp_listener(&req_id, listener);
3248 self.finish_proxy_order(&mut rollback, response);
3249 }
3250 Some(RequestType::AddUdpListener(listener)) => {
3251 let response = self.notify_add_udp_listener(&req_id, listener);
3252 self.finish_proxy_order(&mut rollback, response);
3253 }
3254 Some(RequestType::UpdateHttpListener(patch)) => {
3255 let response = self.notify_update_http_listener(&req_id, patch);
3256 self.finish_proxy_order(&mut rollback, response);
3257 }
3258 Some(
3259 RequestType::UpdateHttpsListener(patch)
3260 | RequestType::UpdateHttpsListenerWithClientAuth(patch),
3261 ) => {
3262 let response = self.notify_update_https_listener(&req_id, patch);
3263 self.finish_proxy_order(&mut rollback, response);
3264 }
3265 Some(RequestType::UpdateTcpListener(patch)) => {
3266 let response = self.notify_update_tcp_listener(&req_id, patch);
3267 self.finish_proxy_order(&mut rollback, response);
3268 }
3269 Some(RequestType::UpdateUdpListener(patch)) => {
3270 let response = self.notify_update_udp_listener(&req_id, patch);
3271 self.finish_proxy_order(&mut rollback, response);
3272 }
3273 Some(RequestType::RemoveListener(ref remove)) => {
3274 debug!("{} remove {:?} listener {:?}", req_id, remove.proxy, remove);
3275 debug_assert!(
3278 self.base_sessions_count > 0,
3279 "removing a listener with base_sessions_count == 0 would underflow"
3280 );
3281 self.base_sessions_count -= 1;
3282 let listener_type = ListenerType::try_from(remove.proxy);
3283 let address = remove.listener_key();
3291 let listen_token = match listener_type {
3292 Ok(ListenerType::Http) => self.http.borrow().listener_token(&address),
3293 Ok(ListenerType::Https) => self.https.borrow().listener_token(&address),
3294 Ok(ListenerType::Tcp) => self.tcp.borrow().listener_token(&address),
3295 Ok(ListenerType::Udp) => self.udp.borrow().listener_token(&address),
3296 Err(_) => None,
3297 };
3298 let response = match listener_type {
3299 Ok(ListenerType::Http) => self.http.borrow_mut().notify(request),
3300 Ok(ListenerType::Https) => self.https.borrow_mut().notify(request),
3301 Ok(ListenerType::Tcp) => self.tcp.borrow_mut().notify(request),
3302 Ok(ListenerType::Udp) => self.udp.borrow_mut().notify(request),
3303 Err(_) => WorkerResponse::error(req_id, "Wrong variant ListenerType"),
3304 };
3305 if let Some(token) = listen_token {
3306 self.sessions.borrow_mut().slab.try_remove(token.0);
3307 self.accept_ready.remove(&ListenToken(token.0));
3310 info!("released listen token {:?}", token);
3311 }
3312 self.finish_proxy_order(&mut rollback, response);
3313 }
3314 Some(RequestType::ActivateListener(ref activate)) => {
3315 let response = self.notify_activate_listener(&req_id, activate);
3316 self.finish_proxy_order(&mut rollback, response);
3317 }
3318 Some(RequestType::DeactivateListener(ref deactivate)) => {
3319 let response = self.notify_deactivate_listener(&req_id, deactivate);
3320 self.finish_proxy_order(&mut rollback, response);
3321 }
3322 _other_request => {}
3323 };
3324 }
3325
3326 fn finish_proxy_order(&mut self, rollback: &mut Option<Request>, response: WorkerResponse) {
3329 if response.is_failure()
3330 && let Some(rollback) = rollback.take()
3331 {
3332 debug_assert!(
3333 rollback.request_type.is_some(),
3334 "a ConfigState rollback must carry one request"
3335 );
3336 if let Err(error) = self.config_state.dispatch(&rollback) {
3337 error!(
3338 "Could not roll back worker config state after a proxy refused an order: {}",
3339 error
3340 );
3341 }
3342 debug_assert!(
3343 rollback.request_type.is_some(),
3344 "dispatching a rollback must not consume its request type"
3345 );
3346 }
3347
3348 push_queue(response);
3349 }
3350
3351 fn add_cluster(&mut self, cluster: &Cluster) {
3352 let mut backends = self.backends.borrow_mut();
3353 backends.set_load_balancing_policy_for_cluster(
3354 &cluster.cluster_id,
3355 LoadBalancingAlgorithms::try_from(cluster.load_balancing).unwrap_or_default(),
3356 cluster
3357 .load_metric
3358 .and_then(|n| LoadMetric::try_from(n).ok()),
3359 );
3360 backends.set_shuffle_sharding_for_cluster(
3361 &cluster.cluster_id,
3362 ShuffleSharding::from_cluster(
3363 cluster.shard_percent,
3364 cluster.shard_min_backends,
3365 cluster.shard_mode,
3366 ),
3367 );
3368 backends.set_health_check_config(&cluster.cluster_id, cluster.health_check.to_owned());
3369 backends.set_cluster_http2(&cluster.cluster_id, cluster.http2.unwrap_or(false));
3370 }
3371
3372 fn add_backend(&mut self, req_id: &str, add_backend: &AddBackend) -> WorkerResponse {
3373 if let Some(Err(reason)) = add_backend
3374 .unix_path
3375 .as_deref()
3376 .map(sozu_command::config::parse_backend_unix_path)
3377 {
3378 return WorkerResponse::error(req_id, format!("invalid unix_path: {reason}"));
3379 }
3380 let new_backend = Backend::new(
3381 &add_backend.backend_id,
3382 add_backend.address.into(),
3383 add_backend.sticky_id.clone(),
3384 add_backend.load_balancing_parameters,
3385 add_backend.backup,
3386 )
3387 .with_unix_path(add_backend.unix_path.clone());
3388 self.backends
3389 .borrow_mut()
3390 .add_backend(&add_backend.cluster_id, new_backend);
3391
3392 WorkerResponse::ok(req_id)
3393 }
3394
3395 fn remove_cluster_objects(&mut self, req_id: &str, cluster_id: &str, requests: Vec<Request>) {
3407 for request in requests {
3408 let result = match request.request_type {
3409 Some(
3410 RequestType::RemoveHttpFrontend(front)
3411 | RequestType::RemoveHttpFrontendWithHeaderPredicates(front),
3412 ) => self.http.borrow_mut().remove_http_frontend(front),
3413 Some(
3414 RequestType::RemoveHttpsFrontend(front)
3415 | RequestType::RemoveHttpsFrontendWithHeaderPredicates(front),
3416 ) => self
3417 .https
3418 .borrow_mut()
3419 .remove_https_frontend(front)
3420 .map(|_| ()),
3421 Some(RequestType::RemoveTcpFrontend(front)) => {
3422 self.tcp.borrow_mut().remove_tcp_front(front)
3423 }
3424 Some(RequestType::RemoveUdpFrontend(front)) => {
3425 self.udp.borrow_mut().remove_udp_front(front)
3426 }
3427 Some(RequestType::RemoveBackend(backend)) => {
3428 let response = self.remove_backend(req_id, &backend);
3429 debug_assert_eq!(
3430 response.status,
3431 ResponseStatus::Ok as i32,
3432 "Server::remove_backend always answers ok"
3433 );
3434 Ok(())
3435 }
3436 other => {
3437 debug_assert!(
3438 false,
3439 "cluster_removal_requests only returns frontend and backend removals, got {other:?}"
3440 );
3441 Ok(())
3442 }
3443 };
3444 if let Err(error) = result {
3445 warn!(
3446 "{} removing cluster {}: could not remove one of its objects: {}",
3447 req_id, cluster_id, error
3448 );
3449 }
3450 }
3451 self.backends.borrow_mut().remove_cluster(cluster_id);
3452 debug_assert!(
3453 !self.backends.borrow().backends.contains_key(cluster_id),
3454 "a removed cluster must leave no backend list behind"
3455 );
3456 }
3457
3458 fn remove_health_check_state(&mut self, cluster_id: &str) {
3463 self.protected_health_check.forget_cluster(cluster_id);
3464 self.health_checker.remove_cluster(cluster_id);
3465 self.backends
3466 .borrow_mut()
3467 .set_health_check_config(cluster_id, None);
3468 }
3469
3470 fn remove_backend(&mut self, req_id: &str, backend: &RemoveBackend) -> WorkerResponse {
3471 let address = backend.address.into();
3472 let id_still_live = {
3479 let mut backends = self.backends.borrow_mut();
3480 backends.remove_backend(&backend.cluster_id, &backend.backend_id, &address);
3481 backends.has_backend_id(&backend.cluster_id, &backend.backend_id)
3482 };
3483 if !id_still_live {
3484 METRICS.with(|metrics| {
3485 (*metrics.borrow_mut()).remove_backend(&backend.cluster_id, &backend.backend_id);
3486 });
3487 }
3488
3489 WorkerResponse::ok(req_id)
3490 }
3491
3492 fn notify_add_http_listener(
3493 &mut self,
3494 req_id: &str,
3495 listener: HttpListenerConfig,
3496 ) -> WorkerResponse {
3497 debug!("{} add http listener {:?}", req_id, listener);
3498
3499 let address: SocketAddr = listener.address.into();
3500 let retained_frontends = self
3501 .config_state
3502 .http_fronts
3503 .values()
3504 .filter(|frontend| frontend.address == address)
3505 .cloned()
3506 .map(Into::into)
3507 .collect();
3508
3509 if self.sessions.borrow().at_capacity() {
3510 return worker_response_error(req_id, "session list is full, cannot add a listener");
3511 }
3512
3513 let mut session_manager = self.sessions.borrow_mut();
3514 let slab_before = session_manager.slab.len();
3516 debug_assert!(
3517 !session_manager
3518 .slab
3519 .contains(session_manager.slab.vacant_key()),
3520 "the next vacant slab key must be free before insertion"
3521 );
3522 let entry = session_manager.slab.vacant_entry();
3523 let token = Token(entry.key());
3524
3525 match self.http.borrow_mut().add_listener_with_retained_frontends(
3526 listener,
3527 token,
3528 retained_frontends,
3529 ) {
3530 Ok(_token) => {
3531 entry.insert(Rc::new(RefCell::new(ListenSession {
3532 protocol: Protocol::HTTPListen,
3533 })));
3534 debug_assert!(
3537 session_manager.slab.contains(token.0),
3538 "listener insert must occupy the token's slab key"
3539 );
3540 debug_assert_eq!(
3541 session_manager.slab.len(),
3542 slab_before + 1,
3543 "adding a listener must occupy exactly one slab slot"
3544 );
3545 self.base_sessions_count += 1;
3546 WorkerResponse::ok(req_id)
3547 }
3548 Err(e) => worker_response_error(req_id, format!("Could not add HTTP listener: {e}")),
3549 }
3550 }
3551
3552 fn notify_add_https_listener(
3553 &mut self,
3554 req_id: &str,
3555 listener: HttpsListenerConfig,
3556 ) -> WorkerResponse {
3557 debug!("{} add https listener {:?}", req_id, listener);
3558
3559 let wire_address = listener.address;
3560 let address: SocketAddr = wire_address.into();
3561 let retained_frontends = self
3562 .config_state
3563 .https_fronts
3564 .values()
3565 .filter(|frontend| frontend.address == address)
3566 .cloned()
3567 .map(Into::into)
3568 .collect();
3569 let retained_certificates = self
3570 .config_state
3571 .certificate_add_requests_for_address(address);
3572
3573 if self.sessions.borrow().at_capacity() {
3574 return worker_response_error(req_id, "session list is full, cannot add a listener");
3575 }
3576
3577 let mut session_manager = self.sessions.borrow_mut();
3578 let slab_before = session_manager.slab.len();
3579 debug_assert!(
3580 !session_manager
3581 .slab
3582 .contains(session_manager.slab.vacant_key()),
3583 "the next vacant slab key must be free before insertion"
3584 );
3585 let entry = session_manager.slab.vacant_entry();
3586 let token = Token(entry.key());
3587
3588 match self.https.borrow_mut().add_listener_with_retained_state(
3589 listener.clone(),
3590 token,
3591 retained_frontends,
3592 retained_certificates,
3593 ) {
3594 Ok(_token) => {
3595 entry.insert(Rc::new(RefCell::new(ListenSession {
3596 protocol: Protocol::HTTPSListen,
3597 })));
3598 debug_assert!(
3599 session_manager.slab.contains(token.0),
3600 "listener insert must occupy the token's slab key"
3601 );
3602 debug_assert_eq!(
3603 session_manager.slab.len(),
3604 slab_before + 1,
3605 "adding a listener must occupy exactly one slab slot"
3606 );
3607 self.base_sessions_count += 1;
3608 WorkerResponse::ok(req_id)
3609 }
3610 Err(e) => worker_response_error(req_id, format!("Could not add HTTPS listener: {e}")),
3611 }
3612 }
3613
3614 fn notify_add_tcp_listener(
3615 &mut self,
3616 req_id: &str,
3617 listener: CommandTcpListener,
3618 ) -> WorkerResponse {
3619 debug!("{} add tcp listener {:?}", req_id, listener);
3620
3621 let address: SocketAddr = listener.address.into();
3622 let retained_frontends = self
3623 .config_state
3624 .tcp_fronts
3625 .values()
3626 .flatten()
3627 .filter(|frontend| frontend.address == address)
3628 .cloned()
3629 .map(Into::into)
3630 .collect();
3631
3632 if self.sessions.borrow().at_capacity() {
3633 return worker_response_error(req_id, "session list is full, cannot add a listener");
3634 }
3635
3636 let mut session_manager = self.sessions.borrow_mut();
3637 let slab_before = session_manager.slab.len();
3638 debug_assert!(
3639 !session_manager
3640 .slab
3641 .contains(session_manager.slab.vacant_key()),
3642 "the next vacant slab key must be free before insertion"
3643 );
3644 let entry = session_manager.slab.vacant_entry();
3645 let token = Token(entry.key());
3646
3647 match self.tcp.borrow_mut().add_listener_with_retained_frontends(
3648 listener,
3649 token,
3650 retained_frontends,
3651 ) {
3652 Ok(_token) => {
3653 entry.insert(Rc::new(RefCell::new(ListenSession {
3654 protocol: Protocol::TCPListen,
3655 })));
3656 debug_assert!(
3657 session_manager.slab.contains(token.0),
3658 "listener insert must occupy the token's slab key"
3659 );
3660 debug_assert_eq!(
3661 session_manager.slab.len(),
3662 slab_before + 1,
3663 "adding a listener must occupy exactly one slab slot"
3664 );
3665 self.base_sessions_count += 1;
3666 WorkerResponse::ok(req_id)
3667 }
3668 Err(e) => worker_response_error(req_id, format!("Could not add TCP listener: {e}")),
3669 }
3670 }
3671
3672 fn notify_add_udp_listener(
3673 &mut self,
3674 req_id: &str,
3675 listener: CommandUdpListener,
3676 ) -> WorkerResponse {
3677 debug!("{} add udp listener {:?}", req_id, listener);
3678
3679 let address: SocketAddr = listener.address.into();
3680 let retained_frontends = self
3681 .config_state
3682 .udp_fronts
3683 .values()
3684 .flatten()
3685 .filter(|frontend| frontend.address == address)
3686 .cloned()
3687 .map(Into::into)
3688 .collect();
3689
3690 if self.sessions.borrow().at_capacity() {
3691 return worker_response_error(req_id, "session list is full, cannot add a listener");
3692 }
3693
3694 let mut session_manager = self.sessions.borrow_mut();
3695 let entry = session_manager.slab.vacant_entry();
3696 let token = Token(entry.key());
3697
3698 match self.udp.borrow_mut().add_listener_with_retained_frontends(
3699 listener,
3700 token,
3701 retained_frontends,
3702 ) {
3703 Ok(_token) => {
3704 entry.insert(Rc::new(RefCell::new(ListenSession {
3705 protocol: Protocol::UDPListen,
3706 })));
3707 self.base_sessions_count += 1;
3708 WorkerResponse::ok(req_id)
3709 }
3710 Err(e) => worker_response_error(req_id, format!("Could not add UDP listener: {e}")),
3711 }
3712 }
3713
3714 fn notify_update_udp_listener(
3715 &mut self,
3716 req_id: &str,
3717 patch: UpdateUdpListenerConfig,
3718 ) -> WorkerResponse {
3719 debug!("{} update udp listener {:?}", req_id, patch.address);
3720 match self.udp.borrow_mut().update_listener(patch) {
3721 Ok(()) => WorkerResponse::ok(req_id),
3722 Err(e) => worker_response_error(req_id, format!("Could not update UDP listener: {e}")),
3723 }
3724 }
3725
3726 fn notify_update_http_listener(
3727 &mut self,
3728 req_id: &str,
3729 patch: UpdateHttpListenerConfig,
3730 ) -> WorkerResponse {
3731 debug!("{} update http listener {:?}", req_id, patch.address);
3732 match self.http.borrow_mut().update_listener(patch) {
3733 Ok(()) => WorkerResponse::ok(req_id),
3734 Err(e) => worker_response_error(req_id, format!("Could not update HTTP listener: {e}")),
3735 }
3736 }
3737
3738 fn notify_update_https_listener(
3739 &mut self,
3740 req_id: &str,
3741 patch: UpdateHttpsListenerConfig,
3742 ) -> WorkerResponse {
3743 debug!("{} update https listener {:?}", req_id, patch.address);
3744 match self.https.borrow_mut().update_listener(patch) {
3745 Ok(()) => WorkerResponse::ok(req_id),
3746 Err(e) => {
3747 worker_response_error(req_id, format!("Could not update HTTPS listener: {e}"))
3748 }
3749 }
3750 }
3751
3752 fn notify_update_tcp_listener(
3753 &mut self,
3754 req_id: &str,
3755 patch: UpdateTcpListenerConfig,
3756 ) -> WorkerResponse {
3757 debug!("{} update tcp listener {:?}", req_id, patch.address);
3758 match self.tcp.borrow_mut().update_listener(patch) {
3759 Ok(()) => WorkerResponse::ok(req_id),
3760 Err(e) => worker_response_error(req_id, format!("Could not update TCP listener: {e}")),
3761 }
3762 }
3763
3764 fn reserve_listen_token(&mut self, token: Token, protocol: Protocol) {
3799 let mut sessions = self.sessions.borrow_mut();
3800 match sessions.slab.get_mut(token.0) {
3801 Some(slot) => {
3802 *slot = Rc::new(RefCell::new(ListenSession { protocol }));
3803 info!("reserved listen token {:?} for {:?}", token, protocol);
3804 }
3805 None => error!(
3809 "listen token {:?} ({:?}) has no slab slot to reserve; the listener would not be reachable if reactivated",
3810 token, protocol
3811 ),
3812 }
3813 }
3814
3815 fn report_unclaimed_inherited_sockets(&self) {
3835 let Some(scm_listeners) = self.scm_listeners.as_ref() else {
3836 return;
3837 };
3838 let leftovers: [(&str, &Vec<(ListenerKey, RawFd)>); 4] = [
3839 ("HTTP", &scm_listeners.http),
3840 ("HTTPS", &scm_listeners.tls),
3841 ("TCP", &scm_listeners.tcp),
3842 ("UDP", &scm_listeners.udp),
3843 ];
3844 for (label, table) in leftovers {
3845 for (address, fd) in table {
3846 let has_listener = match label {
3847 "HTTP" => self.http.borrow().listener_token(address).is_some(),
3848 "HTTPS" => self.https.borrow().listener_token(address).is_some(),
3849 "TCP" => self.tcp.borrow().listener_token(address).is_some(),
3850 _ => self.udp.borrow().listener_token(address).is_some(),
3851 };
3852 if has_listener {
3853 info!(
3854 "inherited {} listening socket for {} (fd {}) is held for an \
3855 ActivateListener this worker has not processed yet",
3856 label, address, fd
3857 );
3858 } else {
3859 warn!(
3860 "inherited {} listening socket for {} (fd {}) is held with no listener \
3861 at that address: until an AddListener plus ActivateListener for {} \
3862 adopts it, it is expected to keep taking a share of new connections \
3863 that nothing accepts (inferred from SO_REUSEPORT semantics, not \
3864 measured here)",
3865 label, address, fd, address
3866 );
3867 }
3868 }
3869 }
3870 }
3871
3872 fn notify_activate_listener(
3873 &mut self,
3874 req_id: &str,
3875 activate: &ActivateListener,
3876 ) -> WorkerResponse {
3877 use crate::InheritedSocketFate;
3878
3879 debug!(
3880 "{} activate {:?} listener {:?}",
3881 req_id, activate.proxy, activate
3882 );
3883
3884 let address = activate.listener_key();
3885
3886 match ListenerType::try_from(activate.proxy) {
3887 Ok(ListenerType::Http) => {
3888 let listener = match self.http.borrow().inherited_socket_fate(&address) {
3889 InheritedSocketFate::Adopted => self
3890 .scm_listeners
3891 .as_mut()
3892 .and_then(|listeners| listeners.get_http(&address))
3893 .map(|fd| unsafe { MioTcpListener::from_raw_fd(fd) }),
3898 InheritedSocketFate::Refused => {
3899 discard_inherited_socket::<MioTcpListener>(
3900 self.scm_listeners
3901 .as_mut()
3902 .and_then(|listeners| listeners.get_http(&address)),
3903 &address,
3904 "HTTP",
3905 );
3906 None
3907 }
3908 InheritedSocketFate::Unclaimed => None,
3911 };
3912
3913 let activated_token = self.http.borrow_mut().activate_listener(&address, listener);
3914 match activated_token {
3915 Ok(token) => {
3916 self.accept(ListenToken(token.0), Protocol::HTTPListen);
3917 WorkerResponse::ok(req_id)
3918 }
3919 Err(activate_error) => worker_response_error(
3920 req_id,
3921 format!("Could not activate HTTP listener: {activate_error}"),
3922 ),
3923 }
3924 }
3925 Ok(ListenerType::Https) => {
3926 let listener = match self.https.borrow().inherited_socket_fate(&address) {
3927 InheritedSocketFate::Adopted => self
3928 .scm_listeners
3929 .as_mut()
3930 .and_then(|listeners| listeners.get_https(&address))
3931 .map(|fd| unsafe { MioTcpListener::from_raw_fd(fd) }),
3936 InheritedSocketFate::Refused => {
3937 discard_inherited_socket::<MioTcpListener>(
3938 self.scm_listeners
3939 .as_mut()
3940 .and_then(|listeners| listeners.get_https(&address)),
3941 &address,
3942 "HTTPS",
3943 );
3944 None
3945 }
3946 InheritedSocketFate::Unclaimed => None,
3949 };
3950
3951 let activated_token = self
3952 .https
3953 .borrow_mut()
3954 .activate_listener(&address, listener);
3955 match activated_token {
3956 Ok(token) => {
3957 self.accept(ListenToken(token.0), Protocol::HTTPSListen);
3958 WorkerResponse::ok(req_id)
3959 }
3960 Err(activate_error) => worker_response_error(
3961 req_id,
3962 format!("Could not activate HTTPS listener: {activate_error}"),
3963 ),
3964 }
3965 }
3966 Ok(ListenerType::Tcp) => {
3967 let listener = match self.tcp.borrow().inherited_socket_fate(&address) {
3968 InheritedSocketFate::Adopted => self
3969 .scm_listeners
3970 .as_mut()
3971 .and_then(|listeners| listeners.get_tcp(&address))
3972 .map(|fd| unsafe { MioTcpListener::from_raw_fd(fd) }),
3977 InheritedSocketFate::Refused => {
3978 discard_inherited_socket::<MioTcpListener>(
3979 self.scm_listeners
3980 .as_mut()
3981 .and_then(|listeners| listeners.get_tcp(&address)),
3982 &address,
3983 "TCP",
3984 );
3985 None
3986 }
3987 InheritedSocketFate::Unclaimed => None,
3990 };
3991
3992 let listener_token = self.tcp.borrow_mut().activate_listener(&address, listener);
3993 match listener_token {
3994 Ok(token) => {
3995 self.accept(ListenToken(token.0), Protocol::TCPListen);
3996 WorkerResponse::ok(req_id)
3997 }
3998 Err(activate_error) => worker_response_error(
3999 req_id,
4000 format!("Could not activate TCP listener: {activate_error}"),
4001 ),
4002 }
4003 }
4004 Ok(ListenerType::Udp) => {
4005 let socket = match self.udp.borrow().inherited_socket_fate(&address) {
4006 InheritedSocketFate::Adopted => self
4007 .scm_listeners
4008 .as_mut()
4009 .and_then(|listeners| listeners.get_udp(&address))
4010 .map(|fd| unsafe { MioUdpSocket::from_raw_fd(fd) }),
4018 InheritedSocketFate::Refused => {
4019 discard_inherited_socket::<MioUdpSocket>(
4020 self.scm_listeners
4021 .as_mut()
4022 .and_then(|listeners| listeners.get_udp(&address)),
4023 &address,
4024 "UDP",
4025 );
4026 None
4027 }
4028 InheritedSocketFate::Unclaimed => None,
4031 };
4032
4033 let activated_token = self.udp.borrow_mut().activate_listener(&address, socket);
4034 match activated_token {
4035 Ok(token) => {
4036 let session_installed = self.udp.borrow().has_listener_session(token)
4086 && self.sessions.borrow().slab.contains(token.0);
4087 if !session_installed {
4088 let session = match self.udp.borrow_mut().build_session(token) {
4095 Some(session) => session,
4096 None => {
4097 return worker_response_error(
4098 req_id,
4099 format!(
4100 "Could not build the UDP listener session for {address}"
4101 ),
4102 );
4103 }
4104 };
4105 let mut sessions = self.sessions.borrow_mut();
4106 match sessions.slab.get_mut(token.0) {
4107 Some(slot) => *slot = session,
4108 None => {
4109 return worker_response_error(
4110 req_id,
4111 format!(
4112 "UDP listener {address} has no session slot at {token:?}"
4113 ),
4114 );
4115 }
4116 }
4117 }
4118 WorkerResponse::ok(req_id)
4119 }
4120 Err(activate_error) => worker_response_error(
4121 req_id,
4122 format!("Could not activate UDP listener: {activate_error}"),
4123 ),
4124 }
4125 }
4126 Err(_) => worker_response_error(req_id, "Wrong variant for ListenerType on request"),
4127 }
4128 }
4129
4130 fn notify_deactivate_listener(
4131 &mut self,
4132 req_id: &str,
4133 deactivate: &DeactivateListener,
4134 ) -> WorkerResponse {
4135 debug!(
4136 "{} deactivate {:?} listener {:?}",
4137 req_id, deactivate.proxy, deactivate
4138 );
4139
4140 let address = deactivate.listener_key();
4141
4142 match ListenerType::try_from(deactivate.proxy) {
4143 Ok(ListenerType::Http) => {
4144 let (token, mut listener) =
4145 match self.http.borrow_mut().give_back_listener(&address) {
4146 Ok((token, listener)) => (token, listener),
4147 Err(e) => {
4148 return worker_response_error(
4149 req_id,
4150 format!("Couldn't deactivate HTTP listener at {address}: {e}"),
4151 );
4152 }
4153 };
4154
4155 if let Err(e) = self.poll.registry().deregister(&mut listener) {
4156 error!(
4157 "error deregistering HTTP listen socket({:?}): {:?}",
4158 deactivate, e
4159 );
4160 }
4161
4162 self.reserve_listen_token(token, Protocol::HTTPListen);
4167 self.accept_ready.remove(&ListenToken(token.0));
4173
4174 if deactivate.to_scm {
4175 self.unblock_scm_socket();
4176 let listeners = Listeners {
4177 http: vec![(address.clone(), listener.as_raw_fd())],
4178 tls: vec![],
4179 tcp: vec![],
4180 udp: vec![],
4181 };
4182 info!("sending HTTP listener: {:?}", listeners);
4183 let res = self.scm.send_listeners(&listeners);
4184
4185 self.block_scm_socket();
4186
4187 info!("sent HTTP listener: {:?}", res);
4188 }
4189 WorkerResponse::ok(req_id)
4190 }
4191 Ok(ListenerType::Https) => {
4192 let (token, mut listener) =
4193 match self.https.borrow_mut().give_back_listener(&address) {
4194 Ok((token, listener)) => (token, listener),
4195 Err(e) => {
4196 return worker_response_error(
4197 req_id,
4198 format!("Couldn't deactivate HTTPS listener at {address}: {e}",),
4199 );
4200 }
4201 };
4202 if let Err(e) = self.poll.registry().deregister(&mut listener) {
4203 error!(
4204 "error deregistering HTTPS listen socket({:?}): {:?}",
4205 deactivate, e
4206 );
4207 }
4208 self.reserve_listen_token(token, Protocol::HTTPSListen);
4211 self.accept_ready.remove(&ListenToken(token.0));
4212
4213 if deactivate.to_scm {
4214 self.unblock_scm_socket();
4215 let listeners = Listeners {
4216 http: vec![],
4217 tls: vec![(address.clone(), listener.as_raw_fd())],
4218 tcp: vec![],
4219 udp: vec![],
4220 };
4221 info!("sending HTTPS listener: {:?}", listeners);
4222 let res = self.scm.send_listeners(&listeners);
4223
4224 self.block_scm_socket();
4225
4226 info!("sent HTTPS listener: {:?}", res);
4227 }
4228 WorkerResponse::ok(req_id)
4229 }
4230 Ok(ListenerType::Tcp) => {
4231 let (token, mut listener) = match self.tcp.borrow_mut().give_back_listener(&address)
4232 {
4233 Ok((token, listener)) => (token, listener),
4234 Err(e) => {
4235 return worker_response_error(
4236 req_id,
4237 format!("Could not deactivate TCP listener at {address}: {e}"),
4238 );
4239 }
4240 };
4241
4242 if let Err(e) = self.poll.registry().deregister(&mut listener) {
4243 error!(
4244 "error deregistering TCP listen socket({:?}): {:?}",
4245 deactivate, e
4246 );
4247 }
4248 self.reserve_listen_token(token, Protocol::TCPListen);
4251 self.accept_ready.remove(&ListenToken(token.0));
4252
4253 if deactivate.to_scm {
4254 self.unblock_scm_socket();
4255 let listeners = Listeners {
4256 http: vec![],
4257 tls: vec![],
4258 tcp: vec![(address.clone(), listener.as_raw_fd())],
4259 udp: vec![],
4260 };
4261 info!("sending TCP listener: {:?}", listeners);
4262 let res = self.scm.send_listeners(&listeners);
4263
4264 self.block_scm_socket();
4265
4266 info!("sent TCP listener: {:?}", res);
4267 }
4268 WorkerResponse::ok(req_id)
4269 }
4270 Ok(ListenerType::Udp) => {
4271 let (token, mut listener) = match self.udp.borrow_mut().give_back_listener(&address)
4272 {
4273 Ok((token, listener)) => (token, listener),
4274 Err(e) => {
4275 return worker_response_error(
4276 req_id,
4277 format!("Could not deactivate UDP listener at {address}: {e}"),
4278 );
4279 }
4280 };
4281
4282 if let Err(e) = self.poll.registry().deregister(&mut listener) {
4283 error!(
4284 "error deregistering UDP listen socket({:?}): {:?}",
4285 deactivate, e
4286 );
4287 }
4288 self.reserve_listen_token(token, Protocol::UDPListen);
4293 self.accept_ready.remove(&ListenToken(token.0));
4297
4298 if deactivate.to_scm {
4299 self.unblock_scm_socket();
4300 let listeners = Listeners {
4301 http: vec![],
4302 tls: vec![],
4303 tcp: vec![],
4304 udp: vec![(address.clone(), listener.as_raw_fd())],
4305 };
4306 info!("sending UDP listener: {:?}", listeners);
4307 let res = self.scm.send_listeners(&listeners);
4308
4309 self.block_scm_socket();
4310
4311 info!("sent UDP listener: {:?}", res);
4312 }
4313 WorkerResponse::ok(req_id)
4314 }
4315 Err(_) => worker_response_error(req_id, "Wrong variant for ListenerType on request"),
4316 }
4317 }
4318
4319 pub fn return_listen_sockets(&mut self) -> Result<(), ScmSocketError> {
4321 self.unblock_scm_socket();
4322
4323 let mut http_listeners = self.http.borrow_mut().give_back_listeners();
4324 for &mut (_, ref mut sock) in http_listeners.iter_mut() {
4325 if let Err(e) = self.poll.registry().deregister(sock) {
4326 error!(
4327 "error deregistering HTTP listen socket({:?}): {:?}",
4328 sock, e
4329 );
4330 }
4331 }
4332
4333 let mut https_listeners = self.https.borrow_mut().give_back_listeners();
4334 for &mut (_, ref mut sock) in https_listeners.iter_mut() {
4335 if let Err(e) = self.poll.registry().deregister(sock) {
4336 error!(
4337 "error deregistering HTTPS listen socket({:?}): {:?}",
4338 sock, e
4339 );
4340 }
4341 }
4342
4343 let mut tcp_listeners = self.tcp.borrow_mut().give_back_listeners();
4344 for &mut (_, ref mut sock) in tcp_listeners.iter_mut() {
4345 if let Err(e) = self.poll.registry().deregister(sock) {
4346 error!("error deregistering TCP listen socket({:?}): {:?}", sock, e);
4347 }
4348 }
4349
4350 let mut udp_listeners = self.udp.borrow_mut().give_back_listeners();
4351 for &mut (_, ref mut sock) in udp_listeners.iter_mut() {
4352 if let Err(e) = self.poll.registry().deregister(sock) {
4353 error!("error deregistering UDP listen socket({:?}): {:?}", sock, e);
4354 }
4355 }
4356
4357 let listeners = Listeners {
4359 http: http_listeners
4360 .iter()
4361 .map(|(key, listener)| (key.clone(), listener.as_raw_fd()))
4362 .collect(),
4363 tls: https_listeners
4364 .iter()
4365 .map(|(key, listener)| (key.clone(), listener.as_raw_fd()))
4366 .collect(),
4367 tcp: tcp_listeners
4368 .iter()
4369 .map(|(key, listener)| (key.clone(), listener.as_raw_fd()))
4370 .collect(),
4371 udp: udp_listeners
4372 .iter()
4373 .map(|(key, listener)| (key.clone(), listener.as_raw_fd()))
4374 .collect(),
4375 };
4376 debug_assert_eq!(
4380 listeners.http.len(),
4381 http_listeners.len(),
4382 "every HTTP listener must be collected exactly once"
4383 );
4384 debug_assert_eq!(
4385 listeners.tls.len(),
4386 https_listeners.len(),
4387 "every HTTPS listener must be collected exactly once"
4388 );
4389 debug_assert_eq!(
4390 listeners.tcp.len(),
4391 tcp_listeners.len(),
4392 "every TCP listener must be collected exactly once"
4393 );
4394 info!("sending default listeners: {:?}", listeners);
4395 let res = self.scm.send_listeners(&listeners);
4396
4397 self.block_scm_socket();
4398
4399 info!("sent default listeners: {:?}", res);
4400 res
4401 }
4402
4403 fn block_scm_socket(&mut self) {
4404 if let Err(e) = self.scm.set_blocking(true) {
4405 error!("Could not block scm socket: {}", e);
4406 }
4407 }
4408
4409 fn unblock_scm_socket(&mut self) {
4410 if let Err(e) = self.scm.set_blocking(false) {
4411 error!("Could not unblock scm socket: {}", e);
4412 }
4413 }
4414
4415 pub fn to_session(&self, token: Token) -> SessionToken {
4416 SessionToken(token.0)
4417 }
4418
4419 pub fn from_session(&self, token: SessionToken) -> Token {
4420 Token(token.0)
4421 }
4422
4423 pub fn accept(&mut self, token: ListenToken, protocol: Protocol) {
4424 let (proto_key, accepted_protocol) = match protocol {
4437 Protocol::TCPListen => ("listener.accepted.tcp", Protocol::TCPListen),
4438 Protocol::HTTPListen => ("listener.accepted.http", Protocol::HTTPListen),
4439 Protocol::HTTPSListen => ("listener.accepted.https", Protocol::HTTPSListen),
4440 other => {
4441 warn!(
4442 "accept() called with non-listen protocol {:?} on token {:?}; skipping",
4443 other, token
4444 );
4445 return;
4446 }
4447 };
4448
4449 debug_assert!(
4452 matches!(
4453 accepted_protocol,
4454 Protocol::TCPListen | Protocol::HTTPListen | Protocol::HTTPSListen
4455 ),
4456 "accept dispatch must run with a listen protocol only"
4457 );
4458
4459 loop {
4460 let result = match accepted_protocol {
4461 Protocol::TCPListen => self.tcp.borrow_mut().accept(token),
4462 Protocol::HTTPListen => self.http.borrow_mut().accept(token),
4463 Protocol::HTTPSListen => self.https.borrow_mut().accept(token),
4464 other => unreachable!(
4468 "accept dispatch reached non-listen protocol {:?} after outer guard",
4469 other
4470 ),
4471 };
4472 match result {
4473 Ok((sock, peer)) => {
4474 incr!(names::listener::ACCEPTED_TOTAL);
4478 incr!(proto_key);
4479 incr!(per_source_bucket(&peer));
4480 let queue_before = self.accept_queue.len();
4481 self.accept_queue.push_back((
4482 sock,
4483 token,
4484 accepted_protocol,
4485 Instant::now(),
4486 peer,
4487 ));
4488 debug_assert_eq!(
4490 self.accept_queue.len(),
4491 queue_before + 1,
4492 "each accepted socket must enqueue exactly one entry"
4493 );
4494 }
4495 Err(AcceptError::WouldBlock) => {
4496 self.accept_ready.remove(&token);
4497 break;
4498 }
4499 Err(other) => {
4500 error!(
4501 "error accepting {:?} sockets: {:?}",
4502 accepted_protocol, other
4503 );
4504 self.accept_ready.remove(&token);
4505 break;
4506 }
4507 }
4508 }
4509
4510 gauge!(names::accept_queue::CONNECTIONS, self.accept_queue.len());
4511 }
4512
4513 pub fn create_sessions(&mut self) {
4514 while let Some((sock, token, protocol, timestamp, peer)) = self.accept_queue.pop_back() {
4515 let wait_time = Instant::now() - timestamp;
4516 time!(names::accept_queue::WAIT_TIME, wait_time.as_millis());
4517 if wait_time > self.accept_queue_timeout {
4518 incr!(names::accept_queue::TIMEOUT);
4519 continue;
4520 }
4521
4522 if !self.sessions.borrow_mut().check_limits() {
4523 if !self.evict_on_queue_full {
4529 incr!(names::listener::CONNECTION_CAPPED);
4530 break;
4531 }
4532
4533 if self.shutting_down.is_some() {
4537 incr!(names::listener::CONNECTION_CAPPED);
4538 break;
4539 }
4540
4541 let to_evict = (self.sessions.borrow().max_connections / 100).max(1);
4550 let evicted = self.evict_least_active_sessions(to_evict);
4551 if evicted == 0 {
4552 warn!("evict_on_queue_full enabled but no candidate sessions to evict");
4558 incr!(names::listener::CONNECTION_CAPPED);
4559 break;
4560 }
4561
4562 count!(names::sessions::EVICTED, evicted as i64);
4563 warn!(
4564 "evicted {} least recently active sessions to make room",
4565 evicted
4566 );
4567
4568 if !self.sessions.borrow_mut().check_limits() {
4569 incr!(names::listener::CONNECTION_CAPPED);
4570 break;
4571 }
4572 }
4573
4574 debug_assert!(
4581 matches!(
4582 protocol,
4583 Protocol::TCPListen | Protocol::HTTPListen | Protocol::HTTPSListen
4584 ),
4585 "accept queue must only hold listen protocols, got {protocol:?}"
4586 );
4587 match protocol {
4588 Protocol::TCPListen => {
4589 let proxy = self.tcp.clone();
4590 if self
4591 .tcp
4592 .borrow_mut()
4593 .create_session(sock, peer, token, wait_time, proxy)
4594 .is_err()
4595 {
4596 break;
4597 }
4598 }
4599 Protocol::HTTPListen => {
4600 let proxy = self.http.clone();
4601 if self
4602 .http
4603 .borrow_mut()
4604 .create_session(sock, peer, token, wait_time, proxy)
4605 .is_err()
4606 {
4607 break;
4608 }
4609 }
4610 Protocol::HTTPSListen => {
4611 if self
4612 .https
4613 .borrow_mut()
4614 .create_session(sock, peer, token, wait_time, self.https.clone())
4615 .is_err()
4616 {
4617 break;
4618 }
4619 }
4620 _ => panic!("should not call accept() on a HTTP, HTTPS or TCP session"),
4621 };
4622 let nb_before = self.sessions.borrow().nb_connections;
4623 self.sessions.borrow_mut().incr();
4624 debug_assert_eq!(
4626 self.sessions.borrow().nb_connections,
4627 nb_before + 1,
4628 "create_sessions must account exactly one new connection per created session"
4629 );
4630 }
4631
4632 gauge!(names::accept_queue::CONNECTIONS, self.accept_queue.len());
4633 }
4634
4635 fn snapshot_event_owners(
4638 &self,
4639 tokens: impl Iterator<Item = Token>,
4640 owners: &mut Vec<Option<Rc<RefCell<dyn ProxySession>>>>,
4641 ) {
4642 debug_assert!(
4643 owners.is_empty(),
4644 "the previous turn must have consumed every event owner"
4645 );
4646 let sessions = self.sessions.borrow();
4647 owners.extend(tokens.map(|token| sessions.slab.get(token.0).cloned()));
4648 }
4649
4650 fn ready_if_still_owned(
4672 &mut self,
4673 token: Token,
4674 events: Ready,
4675 owner: Option<Rc<RefCell<dyn ProxySession>>>,
4676 ) {
4677 let still_owned = owner.is_some_and(|owner| {
4678 self.sessions
4679 .borrow()
4680 .slab
4681 .get(token.0)
4682 .is_some_and(|current| Rc::ptr_eq(current, &owner))
4683 });
4684 if !still_owned {
4685 trace!(
4686 "PROXY\t{:?} dropping events {:?} from a socket whose session closed earlier in this epoll result",
4687 token, events
4688 );
4689 return;
4690 }
4691 self.ready(token, events);
4692 }
4693
4694 pub fn ready(&mut self, token: Token, events: Ready) {
4695 trace!("PROXY\t{:?} got events: {:?}", token, events);
4696
4697 let session_token = token.0;
4698 if self.sessions.borrow().slab.contains(session_token) {
4699 let protocol = self.sessions.borrow().slab[session_token]
4701 .borrow()
4702 .protocol();
4703 match protocol {
4712 Protocol::HTTPListen | Protocol::HTTPSListen | Protocol::TCPListen => {
4713 if events.is_readable() {
4715 self.accept_ready.insert(ListenToken(token.0));
4716 if self.sessions.borrow().can_accept {
4717 self.accept(ListenToken(token.0), protocol);
4718 }
4719 return;
4720 }
4721
4722 if events.is_writable() {
4723 error!(
4724 "received writable for listener {:?}, this should not happen",
4725 token
4726 );
4727 return;
4728 }
4729
4730 if events.is_hup() {
4731 error!("should not happen: server {:?} closed", token);
4732 return;
4733 }
4734
4735 unreachable!();
4736 }
4737 _ => {}
4738 }
4739
4740 let session = self.sessions.borrow_mut().slab[session_token].clone();
4741 session.borrow_mut().update_readiness(token, events);
4742 if session.borrow_mut().ready(session.clone()) {
4743 debug!(
4744 "Server killing session from ready: token={:?}, protocol={:?}, events={:?}",
4745 token, protocol, events
4746 );
4747 self.kill_session(session);
4748 } else {
4749 self.queue_if_yielded(&session);
4750 }
4751 }
4752 }
4753
4754 pub fn timeout(&mut self, token: Token) {
4755 trace!("PROXY\t{:?} got timeout", token);
4756
4757 let session_token = token.0;
4758 if self.sessions.borrow().slab.contains(session_token) {
4759 let session = self.sessions.borrow_mut().slab[session_token].clone();
4760 if session.borrow_mut().timeout(token) {
4761 debug!(
4762 "Server killing session from timeout: token={:?}, protocol={:?}",
4763 token,
4764 session.borrow().protocol()
4765 );
4766 self.kill_session(session);
4767 } else if session.borrow().needs_ready_pass() {
4768 if session.borrow_mut().ready(session.clone()) {
4773 debug!(
4774 "Server killing session from ready after timeout: token={:?}, protocol={:?}",
4775 token,
4776 session.borrow().protocol()
4777 );
4778 self.kill_session(session);
4779 } else {
4780 self.queue_if_yielded(&session);
4781 }
4782 }
4783 }
4784 }
4785
4786 pub fn handle_remaining_readiness(&mut self) {
4787 if self.sessions.borrow().can_accept && !self.accept_ready.is_empty() {
4790 while let Some(token) = self
4791 .accept_ready
4792 .iter()
4793 .next()
4794 .map(|token| ListenToken(token.0))
4795 {
4796 let protocol = self
4813 .sessions
4814 .borrow()
4815 .slab
4816 .get(token.0)
4817 .map(|session| session.borrow().protocol());
4818 let Some(protocol) = protocol else {
4819 error!(
4820 "accept_ready holds listen token {:?}, which no longer has a session; dropping it",
4821 token
4822 );
4823 self.accept_ready.remove(&token);
4824 continue;
4825 };
4826 self.accept(token, protocol);
4827 if !self.sessions.borrow().can_accept || self.accept_ready.is_empty() {
4828 break;
4829 }
4830 }
4831 }
4832 }
4833 fn block_channel(&mut self) {
4834 if let Err(e) = self.channel.blocking() {
4835 error!("Could not block channel: {}", e);
4836 }
4837 }
4838 fn unblock_channel(&mut self) {
4839 if let Err(e) = self.channel.nonblocking() {
4840 error!("Could not block channel: {}", e);
4841 }
4842 }
4843
4844 fn evict_least_active_sessions(&self, count: usize) -> usize {
4856 if count == 0 {
4857 return 0;
4858 }
4859
4860 let tokens = {
4861 let sessions = self.sessions.borrow();
4862 let mut candidates: Vec<(Token, Instant)> = sessions
4863 .slab
4864 .iter()
4865 .filter(|(_, session)| {
4866 !matches!(
4867 session.borrow().protocol(),
4868 Protocol::HTTPListen
4869 | Protocol::HTTPSListen
4870 | Protocol::TCPListen
4871 | Protocol::UDPListen
4872 | Protocol::Channel
4873 | Protocol::Metrics
4874 | Protocol::Timer
4875 )
4876 })
4877 .map(|(_, session)| {
4878 let s = session.borrow();
4879 (s.frontend_token(), s.last_event())
4880 })
4881 .collect();
4882
4883 if candidates.is_empty() {
4886 return 0;
4887 }
4888
4889 let pivot = count.min(candidates.len()) - 1;
4890 candidates.select_nth_unstable_by_key(pivot, |&(_, last_event)| last_event);
4891
4892 candidates[..=pivot]
4893 .iter()
4894 .map(|&(token, _)| token)
4895 .collect::<HashSet<Token>>()
4896 };
4897
4898 let evicted = tokens.len();
4899 self.shut_down_sessions_by_frontend_tokens(tokens);
4900 evicted
4901 }
4902}
4903
4904fn discard_inherited_socket<T: FromRawFd>(
4924 fd: Option<RawFd>,
4925 address: &ListenerKey,
4926 protocol: &str,
4927) {
4928 let Some(fd) = fd else {
4929 return;
4930 };
4931 warn!(
4932 "closing the inherited {} listening socket for {}: this worker already has an active \
4933 listener there, so nothing will adopt this descriptor unless that listener is \
4934 deactivated and reactivated, and until then it is expected to keep taking a share of \
4935 new connections that nothing accepts (inferred from SO_REUSEPORT semantics, not \
4936 measured here)",
4937 protocol, address
4938 );
4939 drop(unsafe { T::from_raw_fd(fd) });
4943}
4944
4945const REFUSED_INITIAL_REQUEST: &str = "unexpected request before the initial status";
4947
4948fn worker_response_error<S: ToString, T: ToString>(request_id: S, error: T) -> WorkerResponse {
4951 error!(
4952 "error on request {}, {}",
4953 request_id.to_string(),
4954 error.to_string()
4955 );
4956 WorkerResponse::error(request_id, error)
4957}
4958
4959pub struct ListenSession {
4960 pub protocol: Protocol,
4961}
4962
4963impl ProxySession for ListenSession {
4964 fn last_event(&self) -> Instant {
4965 Instant::now()
4966 }
4967
4968 fn print_session(&self) {}
4969
4970 fn frontend_token(&self) -> Token {
4971 Token(0)
4972 }
4973
4974 fn protocol(&self) -> Protocol {
4975 self.protocol
4976 }
4977
4978 fn ready(&mut self, _session: Rc<RefCell<dyn ProxySession>>) -> SessionIsToBeClosed {
4979 false
4980 }
4981
4982 fn shutting_down(&mut self) -> SessionIsToBeClosed {
4983 false
4984 }
4985
4986 fn update_readiness(&mut self, _token: Token, _events: Ready) {}
4987
4988 fn close(&mut self) {}
4989
4990 fn timeout(&mut self, _token: Token) -> SessionIsToBeClosed {
4991 error!(
4992 "called ProxySession::timeout(token={:?}, time) on ListenSession {{ protocol: {:?} }}",
4993 _token, self.protocol
4994 );
4995 false
4996 }
4997}
4998
4999#[cfg(test)]
5000mod accept_telemetry_tests {
5001 use super::*;
5002
5003 #[test]
5006 fn per_source_bucket_collapses_ipv4_slash24() {
5007 let a: SocketAddr = "203.0.113.5:1234".parse().unwrap();
5008 let b: SocketAddr = "203.0.113.250:9999".parse().unwrap();
5009 assert_eq!(
5010 per_source_bucket(&a),
5011 per_source_bucket(&b),
5012 "addresses in the same /24 must land in the same bucket"
5013 );
5014 }
5015
5016 #[test]
5018 fn per_source_bucket_collapses_ipv6_slash48() {
5019 let a: SocketAddr = "[2001:db8:1234::1]:443".parse().unwrap();
5020 let b: SocketAddr = "[2001:db8:1234:abcd::ffff]:8443".parse().unwrap();
5021 assert_eq!(
5022 per_source_bucket(&a),
5023 per_source_bucket(&b),
5024 "addresses in the same /48 must land in the same bucket"
5025 );
5026 }
5027
5028 #[test]
5031 fn per_source_bucket_keys_are_bounded() {
5032 assert_eq!(PER_SOURCE_BUCKET_KEYS.len(), PER_SOURCE_BUCKETS);
5033 for (i, key) in PER_SOURCE_BUCKET_KEYS.iter().enumerate() {
5034 let expected = format!("client.connect.per_source.bucket_{i:03}");
5035 assert_eq!(*key, expected.as_str());
5036 }
5037 }
5038
5039 #[test]
5042 fn per_source_bucket_distributes_distinct_subnets() {
5043 let mut hits = std::collections::HashSet::new();
5044 for i in 0..200u8 {
5045 let addr: SocketAddr = format!("10.0.{i}.42:80").parse().unwrap();
5046 hits.insert(per_source_bucket(&addr));
5047 }
5048 assert!(
5052 hits.len() >= 100,
5053 "expected at least 100 distinct buckets across 200 /24s, got {}",
5054 hits.len()
5055 );
5056 }
5057}
5058
5059#[cfg(test)]
5060mod state_error_log_tests {
5061 use sozu_command::state::StateError;
5062
5063 #[test]
5064 fn worker_state_dispatch_error_sink_bounds_the_reason() {
5065 const REASON_SECRET: &str = "WORKER_STATE_ERROR_REASON_SECRET_SENTINEL";
5066
5067 let reason = format!("{REASON_SECRET}{}", "x".repeat(4096));
5068 let reason_len = reason.len();
5069 let output = crate::capture_test_logs(move || {
5070 let error = StateError::InvalidTcpFrontend {
5071 address: "127.0.0.1:443"
5072 .parse()
5073 .expect("test TCP frontend address must parse"),
5074 reason,
5075 };
5076 error!("Could not execute order on config state: {}", error);
5077 });
5078
5079 assert!(
5080 !output.contains(REASON_SECRET),
5081 "worker state-error sink leaked the reason: {output}"
5082 );
5083 assert!(
5084 output.contains(&format!("reason_bytes={reason_len}")),
5085 "worker state-error sink omitted bounded reason metadata: {output}"
5086 );
5087 assert!(
5088 output.len() <= 512,
5089 "worker state-error sink is not bounded: {} bytes",
5090 output.len()
5091 );
5092 }
5093}
5094
5095#[cfg(test)]
5096mod eviction_tests {
5097 use std::collections::HashSet;
5098 use std::time::{Duration, Instant};
5099
5100 use mio::Token;
5101
5102 #[test]
5106 fn select_nth_finds_oldest_sessions() {
5107 let now = Instant::now();
5108 let mut candidates = [
5109 (Token(1), now - Duration::from_secs(10)), (Token(2), now - Duration::from_secs(50)), (Token(3), now - Duration::from_secs(5)), (Token(4), now - Duration::from_secs(30)), (Token(5), now - Duration::from_secs(20)), ];
5115
5116 let count = 2;
5117 let pivot = count.min(candidates.len()) - 1;
5118 candidates.select_nth_unstable_by_key(pivot, |&(_, last_event)| last_event);
5119
5120 let selected: HashSet<Token> = candidates[..=pivot]
5121 .iter()
5122 .map(|&(token, _)| token)
5123 .collect();
5124
5125 assert_eq!(selected.len(), 2);
5126 assert!(
5127 selected.contains(&Token(2)),
5128 "should contain 50s-old session"
5129 );
5130 assert!(
5131 selected.contains(&Token(4)),
5132 "should contain 30s-old session"
5133 );
5134 }
5135
5136 #[test]
5140 fn select_nth_with_count_exceeding_candidates() {
5141 let now = Instant::now();
5142 let mut candidates = [(Token(1), now - Duration::from_secs(10))];
5143
5144 let count = 5;
5145 let pivot = count.min(candidates.len()) - 1;
5146 candidates.select_nth_unstable_by_key(pivot, |&(_, last_event)| last_event);
5147
5148 let selected: HashSet<Token> = candidates[..=pivot]
5149 .iter()
5150 .map(|&(token, _)| token)
5151 .collect();
5152
5153 assert_eq!(selected.len(), 1);
5154 assert!(selected.contains(&Token(1)));
5155 }
5156}
5157
5158#[cfg(test)]
5159mod accept_ready_tests {
5160 use sozu_command::{
5161 config::ListenerBuilder,
5162 proto::command::{DeactivateListener, ListenerType, SocketAddress},
5163 };
5164
5165 use super::*;
5166 use crate::testing::{ServerParts, prebuild_server, provide_port};
5167
5168 pub(super) fn server_with_tcp_listener() -> (Server, Token, SocketAddress) {
5173 let ServerParts {
5174 event_loop,
5175 registry,
5176 sessions,
5177 pool,
5178 backends,
5179 server_scm_socket,
5180 server_config,
5181 ..
5182 } = prebuild_server(16, 16384, true).expect("could not prebuild a test server");
5187 let (_command_channel, proxy_channel) =
5188 Channel::generate(1000, 10000).expect("could not generate a test channel");
5189
5190 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
5191 let listener_config = ListenerBuilder::new_tcp(address)
5192 .to_tcp(None)
5193 .expect("could not build a TcpListenerConfig for the test");
5194
5195 let listen_token = {
5196 let mut session_manager = sessions.borrow_mut();
5197 let entry = session_manager.slab.vacant_entry();
5198 let token = Token(entry.key());
5199 entry.insert(Rc::new(RefCell::new(ListenSession {
5200 protocol: Protocol::TCPListen,
5201 })));
5202 token
5203 };
5204
5205 let mut tcp_proxy =
5206 tcp::TcpProxy::new(registry, sessions.clone(), pool.clone(), backends.clone());
5207 tcp_proxy
5208 .add_listener(listener_config, listen_token)
5209 .expect("could not add the test TCP listener");
5210 tcp_proxy
5211 .activate_listener(&address.into(), None)
5212 .expect("could not activate the test TCP listener");
5213
5214 let server = Server::new(
5215 event_loop,
5216 proxy_channel,
5217 server_scm_socket,
5218 sessions,
5219 pool,
5220 backends,
5221 None,
5222 None,
5223 Some(tcp_proxy),
5224 server_config,
5225 None,
5226 false,
5227 )
5228 .expect("could not build the test server");
5229
5230 (server, listen_token, address)
5233 }
5234
5235 #[test]
5247 fn deactivating_a_listener_drops_its_pending_accept_token() {
5248 let (mut server, listen_token, address) = server_with_tcp_listener();
5249
5250 server.accept_ready.insert(ListenToken(listen_token.0));
5252
5253 let response = server.notify_deactivate_listener(
5254 "test-deactivate",
5255 &DeactivateListener {
5256 interface: None,
5257 address,
5258 proxy: ListenerType::Tcp as i32,
5259 to_scm: false,
5260 },
5261 );
5262 assert_eq!(
5263 response.status,
5264 ResponseStatus::Ok as i32,
5265 "deactivating an activated TCP listener must succeed: {response:?}"
5266 );
5267 assert!(
5268 server.sessions.borrow().slab.contains(listen_token.0),
5269 "deactivating a listener must KEEP its slab slot reserved — see \
5270 `reserve_listen_token`; the accept purge below is what stops the \
5271 deferred pass from accepting on a socket-less listener"
5272 );
5273 assert!(
5274 !server.accept_ready.contains(&ListenToken(listen_token.0)),
5275 "a deactivated listener must not stay queued for a deferred accept"
5276 );
5277
5278 server.handle_remaining_readiness();
5280 }
5281
5282 #[test]
5293 fn a_stale_accept_token_is_dropped_instead_of_panicking() {
5294 let (mut server, listen_token, _address) = server_with_tcp_listener();
5295
5296 server.sessions.borrow_mut().slab.remove(listen_token.0);
5297 server.accept_ready.insert(ListenToken(listen_token.0));
5298
5299 server.handle_remaining_readiness();
5300
5301 assert!(
5302 server.accept_ready.is_empty(),
5303 "a listen token with no session must be dropped from accept_ready"
5304 );
5305 }
5306}
5307#[cfg(test)]
5312mod event_owner_tests {
5313 use super::*;
5314
5315 struct ClosingSession {
5318 sessions: Rc<RefCell<SessionManager>>,
5319 frontend: Token,
5320 backend: Token,
5321 }
5322
5323 impl ProxySession for ClosingSession {
5324 fn protocol(&self) -> Protocol {
5325 Protocol::TCP
5326 }
5327 fn ready(&mut self, _session: Rc<RefCell<dyn ProxySession>>) -> SessionIsToBeClosed {
5328 true
5329 }
5330 fn shutting_down(&mut self) -> SessionIsToBeClosed {
5331 true
5332 }
5333 fn update_readiness(&mut self, _token: Token, _events: Ready) {}
5334 fn close(&mut self) {
5335 self.sessions.borrow_mut().slab.try_remove(self.backend.0);
5336 }
5337 fn timeout(&mut self, _token: Token) -> SessionIsToBeClosed {
5338 false
5339 }
5340 fn last_event(&self) -> Instant {
5341 Instant::now()
5342 }
5343 fn print_session(&self) {}
5344 fn frontend_token(&self) -> Token {
5345 self.frontend
5346 }
5347 }
5348
5349 struct DialingSession {
5353 sessions: Rc<RefCell<SessionManager>>,
5354 frontend: Token,
5355 backend: Option<Token>,
5356 delivered: Vec<Token>,
5357 }
5358
5359 impl ProxySession for DialingSession {
5360 fn protocol(&self) -> Protocol {
5361 Protocol::TCP
5362 }
5363 fn ready(&mut self, session: Rc<RefCell<dyn ProxySession>>) -> SessionIsToBeClosed {
5364 if self.backend.is_none() {
5365 self.backend = Some(Token(self.sessions.borrow_mut().slab.insert(session)));
5366 }
5367 false
5368 }
5369 fn shutting_down(&mut self) -> SessionIsToBeClosed {
5370 true
5371 }
5372 fn update_readiness(&mut self, token: Token, _events: Ready) {
5373 self.delivered.push(token);
5374 }
5375 fn close(&mut self) {}
5376 fn timeout(&mut self, _token: Token) -> SessionIsToBeClosed {
5377 false
5378 }
5379 fn last_event(&self) -> Instant {
5380 Instant::now()
5381 }
5382 fn print_session(&self) {}
5383 fn frontend_token(&self) -> Token {
5384 self.frontend
5385 }
5386 }
5387
5388 #[test]
5396 fn an_event_of_a_closed_session_is_not_delivered_to_the_session_reusing_its_token() {
5397 let (mut server, _listen_token, _address) =
5398 super::accept_ready_tests::server_with_tcp_listener();
5399 let sessions = server.sessions.clone();
5400
5401 let closing = Rc::new(RefCell::new(ClosingSession {
5402 sessions: sessions.clone(),
5403 frontend: Token(0),
5404 backend: Token(0),
5405 }));
5406 let dialing = Rc::new(RefCell::new(DialingSession {
5407 sessions: sessions.clone(),
5408 frontend: Token(0),
5409 backend: None,
5410 delivered: Vec::new(),
5411 }));
5412 {
5413 let mut sessions = sessions.borrow_mut();
5414 let closing_session: Rc<RefCell<dyn ProxySession>> = closing.clone();
5415 let frontend = Token(sessions.slab.insert(closing_session.clone()));
5416 let backend = Token(sessions.slab.insert(closing_session));
5417 let mut closing = closing.borrow_mut();
5418 closing.frontend = frontend;
5419 closing.backend = backend;
5420 let dialing_session: Rc<RefCell<dyn ProxySession>> = dialing.clone();
5421 dialing.borrow_mut().frontend = Token(sessions.slab.insert(dialing_session));
5422 sessions.incr();
5423 sessions.incr();
5424 }
5425 let closed_frontend = closing.borrow().frontend;
5426 let closed_backend = closing.borrow().backend;
5427 let dialing_frontend = dialing.borrow().frontend;
5428
5429 let batch = [
5430 (closed_frontend, Ready::READABLE | Ready::HUP),
5431 (dialing_frontend, Ready::READABLE | Ready::WRITABLE),
5432 (
5433 closed_backend,
5434 Ready::READABLE | Ready::WRITABLE | Ready::HUP,
5435 ),
5436 ];
5437 let mut owners = Vec::new();
5438 server.snapshot_event_owners(batch.iter().map(|(token, _)| *token), &mut owners);
5439 for ((token, events), owner) in batch.into_iter().zip(owners.drain(..)) {
5440 server.ready_if_still_owned(token, events, owner);
5441 }
5442
5443 assert_eq!(
5444 dialing.borrow().backend,
5445 Some(closed_backend),
5446 "precondition: the dial must reuse the key the closed session freed in this batch"
5447 );
5448 assert!(
5449 !sessions.borrow().slab.contains(closed_frontend.0),
5450 "the closed session must have left the slab"
5451 );
5452 assert_eq!(
5453 dialing.borrow().delivered,
5454 vec![dialing_frontend],
5455 "the closed backend's HUP must not reach the session that reused its token"
5456 );
5457 }
5458}
5459#[cfg(test)]
5463mod closing_sessions_tests {
5464 use std::cell::Cell;
5465
5466 use super::*;
5467
5468 struct ScriptedSession {
5471 frontend: Token,
5472 can_close_now: bool,
5473 closed: Rc<Cell<bool>>,
5474 }
5475
5476 impl ProxySession for ScriptedSession {
5477 fn protocol(&self) -> Protocol {
5478 Protocol::HTTP
5479 }
5480 fn ready(&mut self, _session: Rc<RefCell<dyn ProxySession>>) -> SessionIsToBeClosed {
5481 false
5482 }
5483 fn shutting_down(&mut self) -> SessionIsToBeClosed {
5484 self.can_close_now
5485 }
5486 fn update_readiness(&mut self, _token: Token, _events: Ready) {}
5487 fn close(&mut self) {
5488 self.closed.set(true);
5489 }
5490 fn timeout(&mut self, _token: Token) -> SessionIsToBeClosed {
5491 false
5492 }
5493 fn last_event(&self) -> Instant {
5494 Instant::now()
5495 }
5496 fn print_session(&self) {}
5497 fn frontend_token(&self) -> Token {
5498 self.frontend
5499 }
5500 }
5501
5502 fn insert_scripted(
5503 server: &Server,
5504 can_close_now: bool,
5505 ) -> (Token, Rc<RefCell<dyn ProxySession>>, Rc<Cell<bool>>) {
5506 let closed = Rc::new(Cell::new(false));
5507 let mut sessions = server.sessions.borrow_mut();
5508 let entry = sessions.slab.vacant_entry();
5509 let token = Token(entry.key());
5510 let session: Rc<RefCell<dyn ProxySession>> = Rc::new(RefCell::new(ScriptedSession {
5511 frontend: token,
5512 can_close_now,
5513 closed: closed.clone(),
5514 }));
5515 entry.insert(session.clone());
5516 sessions.incr();
5517 (token, session, closed)
5518 }
5519
5520 #[test]
5528 fn a_draining_sessions_reused_slab_key_is_not_closed() {
5529 let (mut server, _listen_token, _address) =
5530 super::accept_ready_tests::server_with_tcp_listener();
5531
5532 let (token, draining, draining_closed) = insert_scripted(&server, true);
5533 server
5534 .closing_sessions
5535 .push((token, Rc::downgrade(&draining)));
5536
5537 server.sessions.borrow_mut().slab.remove(token.0);
5539 let (reused, unrelated, unrelated_closed) = insert_scripted(&server, true);
5540 assert_eq!(reused, token, "precondition: the slab must reuse the key");
5541 assert!(
5542 !Rc::ptr_eq(&draining, &unrelated),
5543 "precondition: the two sessions are distinct"
5544 );
5545
5546 server.drive_closing_sessions();
5547
5548 assert!(
5549 server.sessions.borrow().slab.contains(token.0),
5550 "the unrelated session that reused the key must stay in the slab"
5551 );
5552 assert!(
5553 !unrelated_closed.get() && !draining_closed.get(),
5554 "neither session may be closed by the stale watch"
5555 );
5556 assert!(
5557 server.closing_sessions.is_empty(),
5558 "the stale watch must be dropped, not kept"
5559 );
5560 }
5561
5562 #[test]
5565 fn a_draining_session_still_in_its_slot_is_closed_when_ready() {
5566 let (mut server, _listen_token, _address) =
5567 super::accept_ready_tests::server_with_tcp_listener();
5568
5569 let (token, session, closed) = insert_scripted(&server, true);
5570 server
5571 .closing_sessions
5572 .push((token, Rc::downgrade(&session)));
5573
5574 server.drive_closing_sessions();
5575
5576 assert!(
5577 closed.get(),
5578 "a still-owned closable session must be closed"
5579 );
5580 assert!(!server.sessions.borrow().slab.contains(token.0));
5581 assert!(server.closing_sessions.is_empty());
5582 }
5583}
5584#[cfg(test)]
5589mod yielded_sessions_tests {
5590 use std::cell::Cell;
5591
5592 use super::*;
5593
5594 struct YieldingSession {
5597 frontend: Token,
5598 yields: Rc<Cell<u32>>,
5599 yielded: bool,
5600 ready_calls: Rc<Cell<u32>>,
5601 readiness_updates: Rc<Cell<u32>>,
5602 }
5603
5604 impl ProxySession for YieldingSession {
5605 fn protocol(&self) -> Protocol {
5606 Protocol::HTTP
5607 }
5608 fn ready(&mut self, _session: Rc<RefCell<dyn ProxySession>>) -> SessionIsToBeClosed {
5609 self.ready_calls.set(self.ready_calls.get() + 1);
5610 self.yielded = self.yields.get() > 0;
5611 if self.yielded {
5612 self.yields.set(self.yields.get() - 1);
5613 }
5614 false
5615 }
5616 fn yielded(&mut self) -> bool {
5617 std::mem::take(&mut self.yielded)
5618 }
5619 fn shutting_down(&mut self) -> SessionIsToBeClosed {
5620 true
5621 }
5622 fn update_readiness(&mut self, _token: Token, _events: Ready) {
5623 self.readiness_updates.set(self.readiness_updates.get() + 1);
5624 }
5625 fn close(&mut self) {}
5626 fn timeout(&mut self, _token: Token) -> SessionIsToBeClosed {
5627 false
5628 }
5629 fn last_event(&self) -> Instant {
5630 Instant::now()
5631 }
5632 fn print_session(&self) {}
5633 fn frontend_token(&self) -> Token {
5634 self.frontend
5635 }
5636 }
5637
5638 struct Counters {
5639 ready_calls: Rc<Cell<u32>>,
5640 readiness_updates: Rc<Cell<u32>>,
5641 }
5642
5643 fn insert_yielding(
5644 server: &Server,
5645 yields: u32,
5646 ) -> (Token, Rc<RefCell<dyn ProxySession>>, Counters) {
5647 let ready_calls = Rc::new(Cell::new(0));
5648 let readiness_updates = Rc::new(Cell::new(0));
5649 let mut sessions = server.sessions.borrow_mut();
5650 let entry = sessions.slab.vacant_entry();
5651 let token = Token(entry.key());
5652 let session: Rc<RefCell<dyn ProxySession>> = Rc::new(RefCell::new(YieldingSession {
5653 frontend: token,
5654 yields: Rc::new(Cell::new(yields)),
5655 yielded: false,
5656 ready_calls: ready_calls.clone(),
5657 readiness_updates: readiness_updates.clone(),
5658 }));
5659 entry.insert(session.clone());
5660 sessions.incr();
5661 (
5662 token,
5663 session,
5664 Counters {
5665 ready_calls,
5666 readiness_updates,
5667 },
5668 )
5669 }
5670
5671 #[test]
5675 fn a_yielded_session_is_resumed_without_a_new_event_until_it_stops_yielding() {
5676 let (mut server, _listen_token, _address) =
5677 super::accept_ready_tests::server_with_tcp_listener();
5678 let (token, _session, counters) = insert_yielding(&server, 2);
5679
5680 server.ready(token, Ready::READABLE);
5681 assert_eq!(server.yielded_sessions.len(), 1, "the yield must be queued");
5682 assert_eq!(
5683 server.reset_loop_time_and_get_timeout(),
5684 Some(Duration::ZERO),
5685 "a queued yield must make the next poll non-blocking"
5686 );
5687
5688 let pending = std::mem::take(&mut server.yielded_sessions);
5689 server.resume_yielded_sessions(pending);
5690 assert_eq!(counters.ready_calls.get(), 2, "the session must be resumed");
5691 assert_eq!(
5692 counters.readiness_updates.get(),
5693 1,
5694 "a resumption merges no new readiness"
5695 );
5696 assert_eq!(
5697 server.yielded_sessions.len(),
5698 1,
5699 "a session that yields again is queued for the next turn"
5700 );
5701
5702 let pending = std::mem::take(&mut server.yielded_sessions);
5703 server.resume_yielded_sessions(pending);
5704 assert_eq!(counters.ready_calls.get(), 3);
5705 assert!(
5706 server.yielded_sessions.is_empty(),
5707 "a session that did not yield leaves the queue"
5708 );
5709 assert!(server.sessions.borrow().slab.contains(token.0));
5710 }
5711
5712 #[test]
5721 fn a_yielded_session_that_closed_before_its_resumption_is_not_resumed() {
5722 let (mut server, _listen_token, _address) =
5723 super::accept_ready_tests::server_with_tcp_listener();
5724 let (token, yielded, yielded_counters) = insert_yielding(&server, 1);
5725
5726 server.ready(token, Ready::READABLE);
5727 assert_eq!(server.yielded_sessions.len(), 1, "precondition: queued");
5728
5729 server.sessions.borrow_mut().slab.remove(token.0);
5731 let (reused, tenant, tenant_counters) = insert_yielding(&server, 0);
5732 assert_eq!(reused, token, "precondition: the slab must reuse the key");
5733 assert!(!Rc::ptr_eq(&yielded, &tenant));
5734
5735 let pending = std::mem::take(&mut server.yielded_sessions);
5736 server.resume_yielded_sessions(pending);
5737
5738 assert_eq!(
5739 yielded_counters.ready_calls.get(),
5740 1,
5741 "a closed session must not be resumed"
5742 );
5743 assert_eq!(
5744 tenant_counters.ready_calls.get(),
5745 0,
5746 "the session that reused the key must not be run on the stale entry"
5747 );
5748 assert!(server.yielded_sessions.is_empty());
5749 assert!(server.sessions.borrow().slab.contains(token.0));
5750 }
5751}
5752#[cfg(test)]
5757mod listener_lifecycle_tests {
5758 use std::net::UdpSocket as StdUdpSocket;
5759
5760 use sozu_command::{
5761 config::ListenerBuilder,
5762 proto::command::{
5763 ActivateListener, DeactivateListener, ListenerType, LoadBalancingParams,
5764 RemoveListener, RequestUdpFrontend, SocketAddress, filtered_metrics,
5765 },
5766 };
5767
5768 use super::accept_ready_tests::server_with_tcp_listener;
5769 use super::*;
5770 use crate::{
5771 metrics::{METRICS, names},
5772 testing::{ServerParts, prebuild_server, provide_port},
5773 };
5774
5775 pub(super) fn bare_server() -> Server {
5780 let ServerParts {
5781 event_loop,
5782 sessions,
5783 pool,
5784 backends,
5785 server_scm_socket,
5786 server_config,
5787 ..
5788 } = prebuild_server(16, 16384, true).expect("could not prebuild a test server");
5789 let (_command_channel, proxy_channel) =
5790 Channel::generate(1000, 10000).expect("could not generate a test channel");
5791
5792 Server::new(
5793 event_loop,
5794 proxy_channel,
5795 server_scm_socket,
5796 sessions,
5797 pool,
5798 backends,
5799 None,
5800 None,
5801 None,
5802 server_config,
5803 None,
5804 false,
5805 )
5806 .expect("could not build the test server")
5807 }
5808
5809 fn add_udp_listener(server: &mut Server, address: SocketAddress) -> Token {
5810 let listener_config = ListenerBuilder::new_udp(address)
5811 .to_udp(None)
5812 .expect("could not build a UdpListenerConfig for the test");
5813 let response = server.notify_add_udp_listener("test-add-udp", listener_config);
5814 assert_eq!(
5815 response.status,
5816 ResponseStatus::Ok as i32,
5817 "adding a UDP listener must succeed: {response:?}"
5818 );
5819 server
5820 .udp
5821 .borrow()
5822 .listener_token(&address.into())
5823 .expect("the added UDP listener must own a token")
5824 }
5825
5826 fn activate(server: &mut Server, address: SocketAddress, proxy: ListenerType) {
5827 let response = server.notify_activate_listener(
5828 "test-activate",
5829 &ActivateListener {
5830 interface: None,
5831 address,
5832 proxy: proxy as i32,
5833 from_scm: false,
5834 },
5835 );
5836 assert_eq!(
5837 response.status,
5838 ResponseStatus::Ok as i32,
5839 "activating a {proxy:?} listener must succeed: {response:?}"
5840 );
5841 }
5842
5843 fn deactivate(server: &mut Server, address: SocketAddress, proxy: ListenerType) {
5844 let response = server.notify_deactivate_listener(
5845 "test-deactivate",
5846 &DeactivateListener {
5847 interface: None,
5848 address,
5849 proxy: proxy as i32,
5850 to_scm: false,
5851 },
5852 );
5853 assert_eq!(
5854 response.status,
5855 ResponseStatus::Ok as i32,
5856 "deactivating a {proxy:?} listener must succeed: {response:?}"
5857 );
5858 }
5859
5860 fn slot_state(server: &Server, token: Token) -> Option<(Protocol, bool)> {
5866 let sessions = server.sessions.borrow();
5867 let session = sessions.slab.get(token.0)?;
5868 let session = session.borrow();
5869 Some((session.protocol(), session.frontend_token() == Token(0)))
5870 }
5871
5872 #[test]
5893 fn a_reactivated_tcp_listener_is_reachable_from_the_event_loop() {
5894 let (mut server, listen_token, address) = server_with_tcp_listener();
5895
5896 deactivate(&mut server, address, ListenerType::Tcp);
5897 activate(&mut server, address, ListenerType::Tcp);
5898
5899 server.sessions.borrow_mut().can_accept = false;
5900 server.ready(listen_token, Ready::READABLE);
5901
5902 assert!(
5903 server.accept_ready.contains(&ListenToken(listen_token.0)),
5904 "a reactivated listener must still be dispatched by `ready()`"
5905 );
5906 }
5907
5908 #[test]
5920 fn a_reactivated_udp_listener_gets_its_session_back() {
5921 let mut server = bare_server();
5922 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
5923 let listen_token = add_udp_listener(&mut server, address);
5924
5925 assert_eq!(
5926 slot_state(&server, listen_token),
5927 Some((Protocol::UDPListen, true)),
5928 "an added, not-yet-activated UDP listener holds the placeholder"
5929 );
5930
5931 activate(&mut server, address, ListenerType::Udp);
5932 assert_eq!(
5933 slot_state(&server, listen_token),
5934 Some((Protocol::UDPListen, false)),
5935 "activating must install the real UdpListenerSession"
5936 );
5937
5938 deactivate(&mut server, address, ListenerType::Udp);
5939 activate(&mut server, address, ListenerType::Udp);
5940
5941 assert_eq!(
5942 slot_state(&server, listen_token),
5943 Some((Protocol::UDPListen, false)),
5944 "reactivating must reinstall the real UdpListenerSession"
5945 );
5946 }
5947
5948 #[test]
5972 fn a_deactivated_udp_listeners_key_is_not_handed_to_another_session() {
5973 let mut server = bare_server();
5974 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
5975 let listen_token = add_udp_listener(&mut server, address);
5976 activate(&mut server, address, ListenerType::Udp);
5977 deactivate(&mut server, address, ListenerType::Udp);
5978
5979 let interloper = {
5982 let mut sessions = server.sessions.borrow_mut();
5983 let entry = sessions.slab.vacant_entry();
5984 let key = entry.key();
5985 entry.insert(Rc::new(RefCell::new(ListenSession {
5986 protocol: Protocol::Channel,
5987 })));
5988 Token(key)
5989 };
5990 assert_ne!(
5991 interloper, listen_token,
5992 "a deactivated listener's key must not be handed to another session"
5993 );
5994
5995 activate(&mut server, address, ListenerType::Udp);
5996
5997 assert_eq!(
5998 slot_state(&server, interloper),
5999 Some((Protocol::Channel, true)),
6000 "reactivating a listener must not overwrite another session's slot"
6001 );
6002 }
6003
6004 const UDP_FLOW_CLUSTER: &str = "udp-duplicate-activate";
6006
6007 fn slab_session(server: &Server, token: Token) -> Rc<RefCell<dyn ProxySession>> {
6013 server
6014 .sessions
6015 .borrow()
6016 .slab
6017 .get(token.0)
6018 .expect("the listener must own a slab slot")
6019 .clone()
6020 }
6021
6022 fn active_flows_gauge() -> i64 {
6026 METRICS.with(|metrics| {
6027 metrics
6028 .borrow_mut()
6029 .dump_local_proxy_metrics()
6030 .get(names::udp::ACTIVE_FLOWS)
6031 .and_then(|metric| match metric.inner {
6032 Some(filtered_metrics::Inner::Gauge(value)) => Some(value as i64),
6033 _ => None,
6034 })
6035 .unwrap_or(0)
6036 })
6037 }
6038
6039 fn udp_worker_with_backend(
6045 server: &mut Server,
6046 address: SocketAddress,
6047 ) -> (StdUdpSocket, StdUdpSocket) {
6048 server.notify_proxys(WorkerRequest {
6050 id: "test-add-udp-cluster".to_owned(),
6051 content: RequestType::AddCluster(Cluster {
6052 cluster_id: UDP_FLOW_CLUSTER.to_owned(),
6053 ..Default::default()
6054 })
6055 .into(),
6056 });
6057 server.notify_proxys(WorkerRequest {
6058 id: "test-add-udp-front".to_owned(),
6059 content: RequestType::AddUdpFrontend(RequestUdpFrontend {
6060 cluster_id: UDP_FLOW_CLUSTER.to_owned(),
6061 address,
6062 tags: Default::default(),
6063 })
6064 .into(),
6065 });
6066
6067 let backend_socket =
6068 StdUdpSocket::bind("127.0.0.1:0").expect("could not bind the test UDP backend");
6069 backend_socket
6070 .set_read_timeout(Some(Duration::from_millis(500)))
6071 .expect("could not set the backend read timeout");
6072 let backend_address = backend_socket
6073 .local_addr()
6074 .expect("the test UDP backend must have a local address");
6075
6076 server.notify_proxys(WorkerRequest {
6077 id: "test-add-udp-backend".to_owned(),
6078 content: RequestType::AddBackend(AddBackend {
6079 cluster_id: UDP_FLOW_CLUSTER.to_owned(),
6080 backend_id: format!("{UDP_FLOW_CLUSTER}-0"),
6081 address: backend_address.into(),
6082 load_balancing_parameters: Some(LoadBalancingParams::default()),
6083 sticky_id: None,
6084 backup: None,
6085 unix_path: None,
6086 })
6087 .into(),
6088 });
6089
6090 let client_socket =
6091 StdUdpSocket::bind("127.0.0.1:0").expect("could not bind the test UDP client");
6092 (client_socket, backend_socket)
6093 }
6094
6095 fn forward_one(
6101 server: &mut Server,
6102 listen_token: Token,
6103 listen_address: SocketAddress,
6104 client: &StdUdpSocket,
6105 backend: &StdUdpSocket,
6106 payload: &[u8],
6107 ) -> Option<Vec<u8>> {
6108 let target: std::net::SocketAddr = listen_address.into();
6109 client
6110 .send_to(payload, target)
6111 .expect("the test client must reach the UDP listener");
6112 server.ready(listen_token, Ready::READABLE);
6113
6114 let mut received = [0u8; 64];
6115 backend
6116 .recv_from(&mut received)
6117 .ok()
6118 .map(|(len, _)| received[..len].to_vec())
6119 }
6120
6121 #[test]
6136 fn a_duplicate_udp_activate_keeps_the_live_listener_session() {
6137 let mut server = bare_server();
6138 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
6139 let listen_token = add_udp_listener(&mut server, address);
6140 activate(&mut server, address, ListenerType::Udp);
6141
6142 let installed = slab_session(&server, listen_token);
6143
6144 let response = server.notify_activate_listener(
6145 "test-activate-twice",
6146 &ActivateListener {
6147 interface: None,
6148 address,
6149 proxy: ListenerType::Udp as i32,
6150 from_scm: false,
6151 },
6152 );
6153 assert_eq!(
6160 response.status,
6161 ResponseStatus::Ok as i32,
6162 "a repeated ActivateListener on an active UDP listener must answer ok: {response:?}"
6163 );
6164 assert_eq!(
6165 slot_state(&server, listen_token),
6166 Some((Protocol::UDPListen, false)),
6167 "the real UdpListenerSession must still be the one in the slab"
6168 );
6169 assert!(
6170 Rc::ptr_eq(&installed, &slab_session(&server, listen_token)),
6171 "a repeated ActivateListener must not replace the live UDP listener session"
6172 );
6173 }
6174
6175 #[test]
6187 fn in_flight_udp_flows_survive_a_duplicate_udp_activate() {
6188 let mut server = bare_server();
6189 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
6190 let listen_token = add_udp_listener(&mut server, address);
6191 activate(&mut server, address, ListenerType::Udp);
6192 let (client, backend) = udp_worker_with_backend(&mut server, address);
6193
6194 assert_eq!(
6195 forward_one(
6196 &mut server,
6197 listen_token,
6198 address,
6199 &client,
6200 &backend,
6201 b"before"
6202 )
6203 .as_deref(),
6204 Some(&b"before"[..]),
6205 "the flow must be established before the duplicate activate"
6206 );
6207 assert_eq!(
6208 active_flows_gauge(),
6209 1,
6210 "one established flow must be on the gauge"
6211 );
6212
6213 activate(&mut server, address, ListenerType::Udp);
6214
6215 assert_eq!(
6220 active_flows_gauge(),
6221 1,
6222 "a repeated ActivateListener must neither open nor evict a flow"
6223 );
6224 assert_eq!(
6225 forward_one(
6226 &mut server,
6227 listen_token,
6228 address,
6229 &client,
6230 &backend,
6231 b"after"
6232 )
6233 .as_deref(),
6234 Some(&b"after"[..]),
6235 "an in-flight UDP flow must survive a repeated ActivateListener"
6236 );
6237 }
6238
6239 #[test]
6250 fn a_duplicate_udp_activate_strands_no_upstream_slab_slot() {
6251 let mut server = bare_server();
6252 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
6253 let listen_token = add_udp_listener(&mut server, address);
6254 activate(&mut server, address, ListenerType::Udp);
6255 let (client, backend) = udp_worker_with_backend(&mut server, address);
6256
6257 let baseline = server.sessions.borrow().slab.len();
6260
6261 assert!(
6262 forward_one(
6263 &mut server,
6264 listen_token,
6265 address,
6266 &client,
6267 &backend,
6268 b"flow"
6269 )
6270 .is_some(),
6271 "the flow must be established before the duplicate activate"
6272 );
6273 assert_eq!(
6274 server.sessions.borrow().slab.len(),
6275 baseline + 1,
6276 "an established flow owns exactly one upstream slab slot"
6277 );
6278
6279 activate(&mut server, address, ListenerType::Udp);
6280 deactivate(&mut server, address, ListenerType::Udp);
6281
6282 assert_eq!(
6283 server.sessions.borrow().slab.len(),
6284 baseline,
6285 "tearing the listener down must release every per-flow slab slot"
6286 );
6287 assert_eq!(
6288 active_flows_gauge(),
6289 0,
6290 "tearing the listener down must return the active-flows gauge to zero"
6291 );
6292 }
6293
6294 #[test]
6310 fn replaying_activate_listener_is_a_clean_no_op_for_an_active_udp_listener() {
6311 let mut server = bare_server();
6312 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
6313 let listener_config = ListenerBuilder::new_udp(address)
6314 .to_udp(None)
6315 .expect("could not build a UdpListenerConfig for the test");
6316
6317 server.notify_proxys(WorkerRequest {
6320 id: "test-add-udp".to_owned(),
6321 content: RequestType::AddUdpListener(listener_config).into(),
6322 });
6323 server.notify_proxys(WorkerRequest {
6324 id: "test-activate".to_owned(),
6325 content: RequestType::ActivateListener(ActivateListener {
6326 interface: None,
6327 address,
6328 proxy: ListenerType::Udp as i32,
6329 from_scm: false,
6330 })
6331 .into(),
6332 });
6333 let listen_token = server
6334 .udp
6335 .borrow()
6336 .listener_token(&address.into())
6337 .expect("the added UDP listener must own a token");
6338 let installed = slab_session(&server, listen_token);
6339
6340 let replayed: Vec<ActivateListener> = server
6341 .config_state
6342 .generate_activate_requests()
6343 .into_iter()
6344 .filter_map(|request| match request.request_type {
6345 Some(RequestType::ActivateListener(activate)) => Some(activate),
6346 _ => None,
6347 })
6348 .collect();
6349 assert_eq!(
6350 replayed.len(),
6351 1,
6352 "an active UDP listener must be replayed exactly once: {replayed:?}"
6353 );
6354
6355 for activate in &replayed {
6356 let response = server.notify_activate_listener("test-replay", activate);
6357 assert_eq!(
6358 response.status,
6359 ResponseStatus::Ok as i32,
6360 "a replayed ActivateListener must answer ok: {response:?}"
6361 );
6362 }
6363
6364 assert_eq!(
6365 slot_state(&server, listen_token),
6366 Some((Protocol::UDPListen, false)),
6367 "the replay must leave the real UdpListenerSession in the slab"
6368 );
6369 assert!(
6370 Rc::ptr_eq(&installed, &slab_session(&server, listen_token)),
6371 "replaying ActivateListener must leave the live UDP listener session in place"
6372 );
6373 }
6374
6375 #[test]
6394 fn removing_a_listener_releases_its_reserved_slab_slot() {
6395 let mut server = bare_server();
6396 let baseline = server.sessions.borrow().slab.len();
6397 let base_sessions_count = server.base_sessions_count;
6398
6399 for cycle in 0..3 {
6400 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
6401 let listen_token = add_udp_listener(&mut server, address);
6402 activate(&mut server, address, ListenerType::Udp);
6403 if cycle % 2 == 0 {
6404 deactivate(&mut server, address, ListenerType::Udp);
6405 }
6406
6407 server.notify_proxys(WorkerRequest {
6408 id: "test-remove".to_owned(),
6409 content: RequestType::RemoveListener(RemoveListener {
6410 interface: None,
6411 address,
6412 proxy: ListenerType::Udp as i32,
6413 })
6414 .into(),
6415 });
6416
6417 assert!(
6418 !server.sessions.borrow().slab.contains(listen_token.0),
6419 "cycle {cycle}: removing a listener must release its reserved slot"
6420 );
6421 assert_eq!(
6422 server.sessions.borrow().slab.len(),
6423 baseline,
6424 "cycle {cycle}: the slab must return to its baseline occupancy"
6425 );
6426 assert_eq!(
6427 server.base_sessions_count, base_sessions_count,
6428 "cycle {cycle}: base_sessions_count must track the reserved slots"
6429 );
6430 }
6431 }
6432
6433 #[test]
6444 fn udp_flow_limits_come_from_the_cluster_not_the_global_defaults() {
6445 use std::{net::SocketAddr, time::Instant};
6446
6447 use crate::protocol::udp::{
6448 BackendId, BackendSource, ManagerInput, MetricEvent, Output, UdpManager,
6449 };
6450
6451 struct OneBackend;
6452 impl BackendSource for OneBackend {
6453 fn select(
6454 &mut self,
6455 _cluster: &str,
6456 _key: Option<u64>,
6457 _now: Instant,
6458 ) -> Option<(BackendId, SocketAddr)> {
6459 Some(("b".to_owned(), SocketAddr::from(([127, 0, 0, 1], 5353))))
6460 }
6461 }
6462 fn admit(manager: &RefCell<UdpManager>, port: u16) -> bool {
6465 let mut manager = manager.borrow_mut();
6466 manager.handle_input(
6467 ManagerInput::ClientDatagram {
6468 src: SocketAddr::from(([192, 0, 2, 1], port)),
6469 payload: b"q",
6470 backends: &mut OneBackend,
6471 },
6472 Instant::now(),
6473 );
6474 let mut opened = false;
6475 let mut shed = false;
6476 while let Some(out) = manager.poll_output() {
6477 opened |= matches!(out, Output::OpenUpstream { .. });
6478 shed |= matches!(out, Output::Metric(MetricEvent::FlowShedSourceLimit));
6479 }
6480 assert_ne!(opened, shed, "a datagram is either admitted or shed");
6481 opened
6482 }
6483 fn add_cluster(server: &mut Server, max_connections_per_ip: Option<u64>) {
6484 server.notify_proxys(WorkerRequest {
6485 id: "test-add-cluster".to_owned(),
6486 content: RequestType::AddCluster(Cluster {
6487 cluster_id: "udp-limits".to_owned(),
6488 max_connections_per_ip,
6489 ..Default::default()
6490 })
6491 .into(),
6492 });
6493 }
6494
6495 let mut server = bare_server();
6496 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
6497 let token = add_udp_listener(&mut server, address);
6498 server.notify(WorkerRequest {
6499 id: "test-set-per-ip".to_owned(),
6500 content: RequestType::SetMaxConnectionsPerIp(1).into(),
6501 });
6502 server.notify(WorkerRequest {
6503 id: "test-set-per-subnet".to_owned(),
6504 content: RequestType::SetMaxConnectionsPerSubnet(1).into(),
6505 });
6506 add_cluster(&mut server, None);
6507 server.notify_proxys(WorkerRequest {
6508 id: "test-add-udp-front".to_owned(),
6509 content: RequestType::AddUdpFrontend(RequestUdpFrontend {
6510 cluster_id: "udp-limits".to_owned(),
6511 address,
6512 tags: Default::default(),
6513 })
6514 .into(),
6515 });
6516 let manager = server
6517 .udp
6518 .borrow()
6519 .manager(token)
6520 .expect("the UDP listener must own a flow manager");
6521
6522 {
6523 let manager = manager.borrow();
6524 let cfg = manager.cluster_config();
6525 assert_eq!(
6526 cfg.max_flows_per_ip, 0,
6527 "the global per-IP default must not apply to UDP"
6528 );
6529 assert_eq!(
6530 cfg.max_flows_per_subnet, 0,
6531 "the global per-subnet default must not apply to UDP"
6532 );
6533 }
6534 assert!(admit(&manager, 1000));
6535 assert!(
6536 admit(&manager, 1001),
6537 "a global limit of 1 must not shed a second UDP flow"
6538 );
6539
6540 add_cluster(&mut server, Some(2));
6541 assert_eq!(manager.borrow().cluster_config().max_flows_per_ip, 2);
6542 assert!(
6543 !admit(&manager, 1002),
6544 "the cluster's own limit of 2 counts the two open flows and sheds a third"
6545 );
6546 }
6547}
6548
6549#[cfg(test)]
6550mod add_cluster_validation_tests {
6551 use sozu_command::proto::command::{LoadBalancingParams, ShardMode};
6552
6553 use super::listener_lifecycle_tests::bare_server;
6554 use super::*;
6555 use crate::{backends::ShardOutcome, load_balancing::affinity_key_from_ip};
6556
6557 const CLUSTER: &str = "sharded";
6558 const ADD_CLUSTER_ID: &str = "test-add-invalid-shard";
6559
6560 #[test]
6570 fn an_add_cluster_with_an_invalid_shard_percent_is_refused_and_unsharded() {
6571 let mut server = bare_server();
6572 server.notify_proxys(WorkerRequest {
6576 id: "test-add-unsharded-cluster".to_owned(),
6577 content: RequestType::AddCluster(Cluster {
6578 cluster_id: CLUSTER.to_owned(),
6579 load_balancing: LoadBalancingAlgorithms::RoundRobin as i32,
6580 ..Default::default()
6581 })
6582 .into(),
6583 });
6584 server.notify_proxys(WorkerRequest {
6585 id: ADD_CLUSTER_ID.to_owned(),
6586 content: RequestType::AddCluster(Cluster {
6587 cluster_id: CLUSTER.to_owned(),
6588 load_balancing: LoadBalancingAlgorithms::RoundRobin as i32,
6589 shard_percent: Some(0),
6592 shard_min_backends: Some(2),
6593 shard_mode: Some(ShardMode::Strict as i32),
6594 ..Default::default()
6595 })
6596 .into(),
6597 });
6598 let response = QUEUE
6599 .with(|queue| {
6600 queue
6601 .borrow()
6602 .iter()
6603 .find(|response| response.id == ADD_CLUSTER_ID)
6604 .cloned()
6605 })
6606 .expect("the worker must answer the AddCluster");
6607 assert_eq!(
6608 response.status,
6609 ResponseStatus::Failure as i32,
6610 "an invalid shard_percent must be refused: {response:?}"
6611 );
6612 assert_eq!(response.message, "shard_percent must lie in 1..=100");
6613
6614 for index in 0..4u16 {
6615 server.notify_proxys(WorkerRequest {
6616 id: format!("test-add-backend-{index}"),
6617 content: RequestType::AddBackend(AddBackend {
6618 cluster_id: CLUSTER.to_owned(),
6619 backend_id: format!("{CLUSTER}-{index}"),
6620 address: SocketAddr::from(([127, 0, 0, 1], 23_000 + index)).into(),
6621 load_balancing_parameters: Some(LoadBalancingParams::default()),
6622 sticky_id: None,
6623 backup: None,
6624 unix_path: None,
6625 })
6626 .into(),
6627 });
6628 }
6629 let key = affinity_key_from_ip(IpAddr::V4(Ipv4Addr::LOCALHOST));
6630 let mut backends = server.backends.borrow_mut();
6631 let list = backends
6632 .backends
6633 .get_mut(CLUSTER)
6634 .expect("the backends must have created the cluster's list");
6635 assert_eq!(
6636 list.select_with_key(Some(key), Instant::now()).1,
6637 ShardOutcome::Unsharded,
6638 "a refused AddCluster must not arm a shard"
6639 );
6640 }
6641}
6642
6643#[cfg(test)]
6647mod health_check_port_unix_tests {
6648 use sozu_command::{
6649 config::HEALTH_CHECK_PORT_UNIX_REFUSED,
6650 proto::command::{HealthCheckConfig, HealthCheckMode, SetHealthCheck, SocketAddress},
6651 };
6652
6653 use super::listener_lifecycle_tests::bare_server;
6654 use super::*;
6655
6656 const CLUSTER: &str = "port-unix-cluster";
6657
6658 fn probe(port: Option<u32>) -> HealthCheckConfig {
6659 HealthCheckConfig {
6660 interval: 10,
6661 timeout: 5,
6662 healthy_threshold: 3,
6663 unhealthy_threshold: 3,
6664 mode: HealthCheckMode::Tcp as i32,
6665 port,
6666 ..Default::default()
6667 }
6668 }
6669
6670 fn cluster(health_check: Option<HealthCheckConfig>) -> RequestType {
6671 RequestType::AddCluster(Cluster {
6672 cluster_id: CLUSTER.to_owned(),
6673 health_check,
6674 ..Default::default()
6675 })
6676 }
6677
6678 fn unix_backend() -> RequestType {
6679 RequestType::AddUnixBackend(AddBackend {
6680 cluster_id: CLUSTER.to_owned(),
6681 backend_id: format!("{CLUSTER}-0"),
6682 address: SocketAddress::new_v4(127, 0, 0, 1, 1),
6683 unix_path: Some("/run/app.sock".to_owned()),
6684 ..Default::default()
6685 })
6686 }
6687
6688 fn send(server: &mut Server, request: RequestType) -> Vec<WorkerResponse> {
6689 server.notify_proxys(WorkerRequest {
6690 id: "test-request".to_owned(),
6691 content: request.into(),
6692 });
6693 QUEUE.with(|queue| {
6694 queue
6695 .borrow_mut()
6696 .drain(..)
6697 .filter(|response| response.id != "EVENT")
6698 .collect()
6699 })
6700 }
6701
6702 fn snapshot(server: &Server) -> (String, Vec<*const RefCell<Backend>>, Vec<String>, usize) {
6706 let backends = server.backends.borrow();
6707 (
6708 format!(
6709 "{:?}{:?}",
6710 server.config_state.clusters, server.config_state.backends
6711 ),
6712 backends
6713 .backends
6714 .get(CLUSTER)
6715 .map(|list| list.backends.iter().map(Rc::as_ptr).collect())
6716 .unwrap_or_default(),
6717 backends
6718 .health_check_configs
6719 .values()
6720 .map(|config| format!("{config:?}"))
6721 .collect(),
6722 server.health_checker.in_flight_len(),
6723 )
6724 }
6725
6726 fn assert_refused(server: &mut Server, request: RequestType) {
6727 let before = snapshot(server);
6728 let responses = send(server, request);
6729 assert_eq!(responses.len(), 1, "one answer expected: {responses:?}");
6730 assert_eq!(responses[0].status, ResponseStatus::Failure as i32);
6731 assert!(
6732 responses[0]
6733 .message
6734 .contains(HEALTH_CHECK_PORT_UNIX_REFUSED),
6735 "the answer names the refusal: {responses:?}"
6736 );
6737 assert_eq!(before, snapshot(server), "a refusal changes nothing");
6738 }
6739
6740 #[test]
6741 fn a_probe_port_already_set_refuses_a_unix_backend() {
6742 let mut server = bare_server();
6743 send(&mut server, cluster(Some(probe(Some(8081)))));
6744 assert_refused(&mut server, unix_backend());
6745 assert!(snapshot(&server).1.is_empty(), "no backend was added");
6746 }
6747
6748 #[test]
6749 fn a_unix_backend_already_there_refuses_a_probe_port() {
6750 let mut server = bare_server();
6751 send(&mut server, cluster(Some(probe(None))));
6752 send(&mut server, unix_backend());
6753 assert_eq!(snapshot(&server).1.len(), 1, "test setup: the unix backend");
6754
6755 assert_refused(
6756 &mut server,
6757 RequestType::SetHealthCheck(SetHealthCheck {
6758 cluster_id: CLUSTER.to_owned(),
6759 config: probe(Some(8081)),
6760 }),
6761 );
6762 assert_refused(&mut server, cluster(Some(probe(Some(8081)))));
6763 assert_eq!(
6764 server.backends.borrow().health_check_configs[CLUSTER].port,
6765 None,
6766 "the previous probe config stays"
6767 );
6768 }
6769}
6770
6771#[cfg(test)]
6774mod remove_health_check_tests {
6775 use sozu_command::proto::command::{HealthCheckConfig, LoadBalancingParams};
6776
6777 use super::listener_lifecycle_tests::bare_server;
6778 use super::*;
6779
6780 const CLUSTER: &str = "health-checked";
6781
6782 fn health_check() -> HealthCheckConfig {
6783 HealthCheckConfig {
6784 uri: "/livez".to_owned(),
6785 interval: 1,
6786 timeout: 1,
6787 healthy_threshold: 1,
6788 unhealthy_threshold: 1,
6789 ..Default::default()
6790 }
6791 }
6792
6793 fn add_cluster(server: &mut Server, health_check: Option<HealthCheckConfig>) {
6794 server.notify_proxys(WorkerRequest {
6795 id: "test-add-cluster".to_owned(),
6796 content: RequestType::AddCluster(Cluster {
6797 cluster_id: CLUSTER.to_owned(),
6798 health_check,
6799 ..Default::default()
6800 })
6801 .into(),
6802 });
6803 }
6804
6805 fn server_with_down_backend() -> Server {
6808 let mut server = bare_server();
6809 add_cluster(&mut server, Some(health_check()));
6810 server.notify_proxys(WorkerRequest {
6811 id: "test-add-backend".to_owned(),
6812 content: RequestType::AddBackend(AddBackend {
6813 cluster_id: CLUSTER.to_owned(),
6814 backend_id: format!("{CLUSTER}-0"),
6815 address: SocketAddr::from(([127, 0, 0, 1], 23_100)).into(),
6816 load_balancing_parameters: Some(LoadBalancingParams::default()),
6817 sticky_id: None,
6818 backup: None,
6819 unix_path: None,
6820 })
6821 .into(),
6822 });
6823 server.backends.borrow().backends[CLUSTER].backends[0]
6824 .borrow_mut()
6825 .health
6826 .record_failure(1);
6827 assert!(!backend_is_healthy(&server), "test setup: backend DOWN");
6828 server
6829 }
6830
6831 fn backend_is_healthy(server: &Server) -> bool {
6832 server.backends.borrow().backends[CLUSTER].backends[0]
6833 .borrow()
6834 .health
6835 .is_healthy()
6836 }
6837
6838 #[test]
6842 fn remove_health_check_resets_a_backend_marked_down() {
6843 let mut server = server_with_down_backend();
6844 server.notify_proxys(WorkerRequest {
6845 id: "test-remove-health-check".to_owned(),
6846 content: RequestType::RemoveHealthCheck(CLUSTER.to_owned()).into(),
6847 });
6848 assert!(
6849 !server
6850 .backends
6851 .borrow()
6852 .health_check_configs
6853 .contains_key(CLUSTER),
6854 "RemoveHealthCheck must forget the cluster's configuration"
6855 );
6856 assert!(
6857 backend_is_healthy(&server),
6858 "RemoveHealthCheck must reset a backend a probe marked DOWN"
6859 );
6860 }
6861
6862 #[test]
6868 fn add_cluster_without_health_check_resets_a_backend_marked_down() {
6869 let mut server = server_with_down_backend();
6870 add_cluster(&mut server, None);
6871 assert!(
6872 !server
6873 .backends
6874 .borrow()
6875 .health_check_configs
6876 .contains_key(CLUSTER),
6877 "an AddCluster without a health check must forget the configuration"
6878 );
6879 assert!(
6880 backend_is_healthy(&server),
6881 "an AddCluster without a health check must reset a backend marked DOWN"
6882 );
6883 }
6884
6885 #[test]
6892 fn an_idempotent_reload_keeps_a_backend_marked_down_and_its_counters() {
6893 let ported = HealthCheckConfig {
6894 port: Some(8081),
6895 ..health_check()
6896 };
6897 let mut server = bare_server();
6898 add_cluster(&mut server, Some(ported.clone()));
6899 let add_backend = || WorkerRequest {
6900 id: "test-add-backend".to_owned(),
6901 content: RequestType::AddBackend(AddBackend {
6902 cluster_id: CLUSTER.to_owned(),
6903 backend_id: format!("{CLUSTER}-0"),
6904 address: SocketAddr::from(([127, 0, 0, 1], 23_100)).into(),
6905 load_balancing_parameters: Some(LoadBalancingParams::default()),
6906 sticky_id: None,
6907 backup: None,
6908 unix_path: None,
6909 })
6910 .into(),
6911 };
6912 server.notify_proxys(add_backend());
6913 {
6914 let backends = server.backends.borrow();
6915 let mut backend = backends.backends[CLUSTER].backends[0].borrow_mut();
6916 for _ in 0..3 {
6919 backend.health.record_failure(3);
6920 }
6921 }
6922 let before = server.backends.borrow().backends[CLUSTER].backends[0]
6923 .borrow()
6924 .health
6925 .clone();
6926 assert!(!before.is_healthy(), "test setup: backend DOWN");
6927 assert_eq!(before.consecutive_failures, 3, "test setup: counters");
6928
6929 add_cluster(&mut server, Some(ported.clone()));
6931 server.notify_proxys(add_backend());
6932
6933 let after = server.backends.borrow().backends[CLUSTER].backends[0]
6934 .borrow()
6935 .health
6936 .clone();
6937 assert_eq!(after, before, "a reload must not reset the runtime health");
6938 assert_eq!(
6939 server.backends.borrow().health_check_configs[CLUSTER],
6940 ported,
6941 "the health check configuration stays"
6942 );
6943 }
6944}
6945
6946#[cfg(test)]
6953mod remove_cluster_cascade_tests {
6954 use sozu_command::{
6955 config::ListenerBuilder,
6956 proto::command::{
6957 LoadBalancingParams, PathRule, RequestHttpFrontend, RequestTcpFrontend,
6958 RequestUdpFrontend, RulePosition, SocketAddress,
6959 },
6960 };
6961
6962 use super::listener_lifecycle_tests::bare_server;
6963 use super::*;
6964 use crate::{L7ListenerHandler, protocol::http::parser::Method, testing::provide_port};
6965
6966 const REMOVED: &str = "cascade_removed";
6967 const KEPT: &str = "cascade_kept";
6968
6969 fn send(server: &mut Server, id: &str, request: RequestType) -> Vec<WorkerResponse> {
6973 server.notify_proxys(WorkerRequest {
6974 id: id.to_owned(),
6975 content: request.into(),
6976 });
6977 QUEUE.with(|queue| {
6978 queue
6979 .borrow_mut()
6980 .drain(..)
6981 .filter(|response| response.id != "EVENT")
6982 .collect()
6983 })
6984 }
6985
6986 fn http_front(cluster_id: Option<&str>, hostname: &str, path: &str) -> RequestHttpFrontend {
6987 RequestHttpFrontend {
6988 cluster_id: cluster_id.map(str::to_owned),
6989 hostname: hostname.to_owned(),
6990 path: PathRule::prefix(path.to_owned()),
6991 address: SocketAddress::new_v4(127, 0, 0, 1, 0),
6992 ..Default::default()
6993 }
6994 }
6995
6996 #[test]
7001 fn a_proxy_rejection_rolls_back_reversible_worker_state() {
7002 let cluster_id = "worker-rollback";
7003 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7004 let http = http_front(Some(cluster_id), "rollback.test", "/");
7005
7006 let requests = [
7007 RequestType::AddHttpFrontend(RequestHttpFrontend {
7008 address,
7009 ..http.clone()
7010 }),
7011 RequestType::AddHttpsFrontend(RequestHttpFrontend { address, ..http }),
7012 RequestType::AddTcpFrontend(RequestTcpFrontend {
7013 cluster_id: cluster_id.to_owned(),
7014 address,
7015 ..Default::default()
7016 }),
7017 RequestType::AddUdpFrontend(RequestUdpFrontend {
7018 cluster_id: cluster_id.to_owned(),
7019 address,
7020 ..Default::default()
7021 }),
7022 ];
7023
7024 for (index, request) in requests.into_iter().enumerate() {
7025 let mut server = bare_server();
7026 for response in send(
7027 &mut server,
7028 &format!("cluster-{index}"),
7029 RequestType::AddCluster(Cluster {
7030 cluster_id: cluster_id.to_owned(),
7031 ..Default::default()
7032 }),
7033 ) {
7034 assert_eq!(response.status, ResponseStatus::Ok as i32);
7035 }
7036
7037 let responses = send(&mut server, &format!("frontend-{index}"), request);
7038 assert_eq!(
7039 responses
7040 .iter()
7041 .map(|response| response.status)
7042 .collect::<Vec<_>>(),
7043 vec![ResponseStatus::Failure as i32],
7044 "a frontend with no listener must be refused: {responses:?}"
7045 );
7046
7047 let cluster = server
7048 .config_state
7049 .cluster_state(cluster_id)
7050 .expect("the setup cluster must stay in worker state");
7051 assert!(
7052 cluster.http_frontends.is_empty()
7053 && cluster.https_frontends.is_empty()
7054 && cluster.tcp_frontends.is_empty()
7055 && cluster.udp_frontends.is_empty(),
7056 "a refused frontend must not remain in worker ConfigState: {cluster:?}"
7057 );
7058 }
7059 }
7060
7061 #[test]
7066 fn a_refused_frontend_removal_restores_the_exact_stored_frontend() {
7067 let cluster_id = "worker-remove-rollback";
7068 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7069 let tags = std::collections::BTreeMap::from([("owner".to_owned(), "preserved".to_owned())]);
7070 let mut http = http_front(Some(cluster_id), "remove.test", "/api");
7071 http.address = address;
7072 http.method = Some("POST".to_owned());
7073 http.position = RulePosition::Post.into();
7074 http.tags = tags.clone();
7075 http.rewrite_host = Some("internal.test".to_owned());
7076
7077 let mut http_selector = http.clone();
7078 http_selector.position = RulePosition::Tree.into();
7079 http_selector.tags.clear();
7080 http_selector.rewrite_host = None;
7081
7082 let mut https = http.clone();
7083 https.hostname = "secure-remove.test".to_owned();
7084 let mut https_selector = https.clone();
7085 https_selector.position = RulePosition::Tree.into();
7086 https_selector.tags.clear();
7087 https_selector.rewrite_host = None;
7088
7089 let tcp = RequestTcpFrontend {
7090 cluster_id: cluster_id.to_owned(),
7091 address,
7092 tags: tags.clone(),
7093 sni: Some("REMOVE.TEST".to_owned()),
7094 alpn: vec!["http/1.1".to_owned(), "h2".to_owned()],
7095 };
7096 let tcp_selector = RequestTcpFrontend {
7097 tags: Default::default(),
7098 sni: Some("remove.test".to_owned()),
7099 alpn: vec!["h2".to_owned(), "http/1.1".to_owned()],
7100 ..tcp.clone()
7101 };
7102
7103 let udp = RequestUdpFrontend {
7104 cluster_id: cluster_id.to_owned(),
7105 address,
7106 tags,
7107 };
7108
7109 let cases = [
7110 (
7111 "http",
7112 RequestType::AddHttpFrontend(http),
7113 RequestType::RemoveHttpFrontend(http_selector),
7114 ),
7115 (
7116 "https",
7117 RequestType::AddHttpsFrontend(https),
7118 RequestType::RemoveHttpsFrontend(https_selector),
7119 ),
7120 (
7121 "tcp",
7122 RequestType::AddTcpFrontend(tcp),
7123 RequestType::RemoveTcpFrontend(tcp_selector),
7124 ),
7125 (
7126 "udp",
7127 RequestType::AddUdpFrontend(udp.clone()),
7128 RequestType::RemoveUdpFrontend(udp),
7129 ),
7130 ];
7131
7132 for (protocol, add, remove) in cases {
7133 let mut server = bare_server();
7134 server
7135 .config_state
7136 .dispatch(
7137 &RequestType::AddCluster(Cluster {
7138 cluster_id: cluster_id.to_owned(),
7139 ..Default::default()
7140 })
7141 .into(),
7142 )
7143 .expect("the setup cluster must enter worker state");
7144 server
7145 .config_state
7146 .dispatch(&add.into())
7147 .unwrap_or_else(|error| {
7148 panic!("the {protocol} setup frontend must be stored: {error}")
7149 });
7150 let before = server
7151 .config_state
7152 .cluster_state(cluster_id)
7153 .expect("the setup cluster must expose its frontend");
7154
7155 let responses = send(&mut server, &format!("remove-{protocol}"), remove);
7156 assert_eq!(
7157 responses
7158 .iter()
7159 .map(|response| response.status)
7160 .collect::<Vec<_>>(),
7161 vec![ResponseStatus::Failure as i32],
7162 "a frontend removal with no runtime listener must be refused: {responses:?}"
7163 );
7164 assert_eq!(
7165 server.config_state.cluster_state(cluster_id),
7166 Some(before),
7167 "a refused {protocol} removal must restore the complete stored frontend"
7168 );
7169 }
7170 }
7171
7172 #[test]
7177 fn a_refused_duplicate_frontend_add_preserves_the_preexisting_frontend() {
7178 let cluster_id = "worker-add-rollback";
7179 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7180 let tags =
7181 std::collections::BTreeMap::from([("owner".to_owned(), "preexisting".to_owned())]);
7182 let mut http = http_front(Some(cluster_id), "duplicate.test", "/api");
7183 http.address = address;
7184 http.method = Some("PATCH".to_owned());
7185 http.position = RulePosition::Pre.into();
7186 http.tags = tags.clone();
7187
7188 let mut https = http.clone();
7189 https.hostname = "secure-duplicate.test".to_owned();
7190 let tcp = RequestTcpFrontend {
7191 cluster_id: cluster_id.to_owned(),
7192 address,
7193 tags: tags.clone(),
7194 sni: Some("duplicate.test".to_owned()),
7195 alpn: vec!["h2".to_owned()],
7196 };
7197 let udp = RequestUdpFrontend {
7198 cluster_id: cluster_id.to_owned(),
7199 address,
7200 tags,
7201 };
7202
7203 let cases = [
7204 ("http", RequestType::AddHttpFrontend(http)),
7205 ("https", RequestType::AddHttpsFrontend(https)),
7206 ("tcp", RequestType::AddTcpFrontend(tcp)),
7207 ("udp", RequestType::AddUdpFrontend(udp)),
7208 ];
7209
7210 for (protocol, add) in cases {
7211 let mut server = bare_server();
7212 server
7213 .config_state
7214 .dispatch(
7215 &RequestType::AddCluster(Cluster {
7216 cluster_id: cluster_id.to_owned(),
7217 ..Default::default()
7218 })
7219 .into(),
7220 )
7221 .expect("the setup cluster must enter worker state");
7222 server
7223 .config_state
7224 .dispatch(&add.clone().into())
7225 .unwrap_or_else(|error| {
7226 panic!("the {protocol} setup frontend must be stored: {error}")
7227 });
7228 let before = server
7229 .config_state
7230 .cluster_state(cluster_id)
7231 .expect("the setup cluster must expose its frontend");
7232
7233 let responses = send(&mut server, &format!("duplicate-{protocol}"), add);
7234 assert_eq!(
7235 responses
7236 .iter()
7237 .map(|response| response.status)
7238 .collect::<Vec<_>>(),
7239 vec![ResponseStatus::Failure as i32],
7240 "a duplicate frontend with no runtime listener must be refused: {responses:?}"
7241 );
7242 assert_eq!(
7243 server.config_state.cluster_state(cluster_id),
7244 Some(before),
7245 "a refused duplicate {protocol} add must preserve the preexisting frontend"
7246 );
7247 }
7248 }
7249
7250 #[test]
7254 fn a_refused_duplicate_listener_add_preserves_the_preexisting_listener() {
7255 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7256 let listener = ListenerBuilder::new_http(address)
7257 .to_http(None)
7258 .expect("the test HTTP listener configuration must build");
7259 let mut server = bare_server();
7260 server
7261 .config_state
7262 .dispatch(&RequestType::AddHttpListener(listener.clone()).into())
7263 .expect("the setup listener must enter worker state");
7264 let before = server.config_state.http_listeners.clone();
7265
7266 let threshold = server.sessions.borrow().accept_slab_threshold();
7269 while server.sessions.borrow().slab.len() < threshold {
7270 server
7271 .sessions
7272 .borrow_mut()
7273 .slab
7274 .insert(Rc::new(RefCell::new(ListenSession {
7275 protocol: Protocol::HTTPListen,
7276 })));
7277 }
7278 assert!(server.sessions.borrow().at_capacity());
7279
7280 let responses = send(
7281 &mut server,
7282 "duplicate-listener",
7283 RequestType::AddHttpListener(listener),
7284 );
7285 assert_eq!(
7286 responses
7287 .iter()
7288 .map(|response| response.status)
7289 .collect::<Vec<_>>(),
7290 vec![ResponseStatus::Failure as i32],
7291 "the full session slab must refuse the duplicate listener: {responses:?}"
7292 );
7293 assert_eq!(
7294 server.config_state.http_listeners, before,
7295 "a refused duplicate listener add must preserve the preexisting listener"
7296 );
7297 }
7298
7299 fn backend(cluster_id: &str, port: u16) -> AddBackend {
7300 AddBackend {
7301 cluster_id: cluster_id.to_owned(),
7302 backend_id: format!("{cluster_id}-0"),
7303 address: SocketAddress::new_v4(127, 0, 0, 1, port),
7304 load_balancing_parameters: Some(LoadBalancingParams::default()),
7305 ..Default::default()
7306 }
7307 }
7308
7309 #[test]
7320 fn remove_then_readd_same_cluster_id_preserves_and_releases_live_admission_leases() {
7321 let mut server = bare_server();
7322 let cluster_id = ClusterId::from(REMOVED);
7323 let source: IpAddr = "203.0.113.9".parse().expect("test address must parse");
7324 let subnet = server.sessions.borrow().subnet_key(&source);
7325 let old_token = Token(60_001);
7326 let new_token = Token(60_002);
7327
7328 for response in send(
7329 &mut server,
7330 "add-old-incarnation",
7331 RequestType::AddCluster(Cluster {
7332 cluster_id: REMOVED.to_owned(),
7333 max_connections_per_ip: Some(2),
7334 max_connections_per_subnet: Some(2),
7335 ..Default::default()
7336 }),
7337 ) {
7338 assert_eq!(response.status, ResponseStatus::Ok as i32);
7339 }
7340 server.sessions.borrow_mut().track_cluster_connection(
7341 old_token,
7342 cluster_id.clone(),
7343 source,
7344 Some(2),
7345 );
7346
7347 for response in send(
7348 &mut server,
7349 "remove-old-incarnation",
7350 RequestType::RemoveCluster(REMOVED.to_owned()),
7351 ) {
7352 assert_eq!(response.status, ResponseStatus::Ok as i32);
7353 }
7354 for response in send(
7355 &mut server,
7356 "add-new-incarnation",
7357 RequestType::AddCluster(Cluster {
7358 cluster_id: REMOVED.to_owned(),
7359 max_connections_per_ip: Some(2),
7360 max_connections_per_subnet: Some(2),
7361 ..Default::default()
7362 }),
7363 ) {
7364 assert_eq!(response.status, ResponseStatus::Ok as i32);
7365 }
7366
7367 {
7368 let mut sessions = server.sessions.borrow_mut();
7369 assert_eq!(sessions.connections_per_cluster_ip[REMOVED][&source], 1);
7370 assert_eq!(sessions.connections_per_cluster_subnet[REMOVED][&subnet], 1);
7371 sessions.track_cluster_connection(new_token, cluster_id, source, Some(2));
7372 assert_eq!(sessions.connections_per_cluster_ip[REMOVED][&source], 2);
7373 assert_eq!(sessions.connections_per_cluster_subnet[REMOVED][&subnet], 2);
7374
7375 sessions.untrack_all_cluster_ip(old_token);
7376 assert_eq!(sessions.connections_per_cluster_ip[REMOVED][&source], 1);
7377 assert_eq!(sessions.connections_per_cluster_subnet[REMOVED][&subnet], 1);
7378 assert!(sessions.cluster_ip_tracks.contains_key(&new_token));
7379 assert!(sessions.cluster_subnet_tracks.contains_key(&new_token));
7380
7381 sessions.untrack_all_cluster_ip(new_token);
7382 assert!(!sessions.connections_per_cluster_ip.contains_key(REMOVED));
7383 assert!(
7384 !sessions
7385 .connections_per_cluster_subnet
7386 .contains_key(REMOVED)
7387 );
7388 }
7389 }
7390
7391 fn http_route(
7395 server: &Server,
7396 address: SocketAddress,
7397 hostname: &str,
7398 path: &str,
7399 ) -> Option<Option<String>> {
7400 let http = server.http.borrow();
7401 let token = http
7402 .listener_token(&SocketAddr::from(address).into())
7403 .expect("the test HTTP listener must own a token");
7404 let listener = http
7405 .get_listener(&token)
7406 .expect("the test HTTP listener must exist");
7407 listener
7408 .borrow()
7409 .frontend_from_request(hostname, path, &Method::Get, &crate::router::NoHeaders)
7410 .ok()
7411 .map(|route| route.cluster_id.map(|id| id.to_string()))
7412 }
7413
7414 #[test]
7421 fn remove_cluster_removes_its_routes_and_backends_from_the_worker() {
7422 let mut server = bare_server();
7423 let http_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7424 let tcp_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7425
7426 let mut removed_front = http_front(Some(REMOVED), "removed.test", "/");
7427 removed_front.address = http_address;
7428 removed_front.position = RulePosition::Pre.into();
7431 let mut deny_front = http_front(None, "removed.test", "/deny");
7432 deny_front.address = http_address;
7433 let mut kept_front = http_front(Some(KEPT), "kept.test", "/");
7434 kept_front.address = http_address;
7435 let removed_tcp_front = RequestTcpFrontend {
7436 cluster_id: REMOVED.to_owned(),
7437 address: tcp_address,
7438 ..Default::default()
7439 };
7440
7441 let setup = vec![
7442 RequestType::AddHttpListener(
7443 ListenerBuilder::new_http(http_address)
7444 .to_http(None)
7445 .expect("could not build the test HTTP listener config"),
7446 ),
7447 RequestType::AddTcpListener(
7448 ListenerBuilder::new_tcp(tcp_address)
7449 .to_tcp(None)
7450 .expect("could not build the test TCP listener config"),
7451 ),
7452 RequestType::AddCluster(Cluster {
7453 cluster_id: REMOVED.to_owned(),
7454 ..Default::default()
7455 }),
7456 RequestType::AddCluster(Cluster {
7457 cluster_id: KEPT.to_owned(),
7458 ..Default::default()
7459 }),
7460 RequestType::AddHttpFrontend(removed_front.clone()),
7461 RequestType::AddHttpFrontend(deny_front),
7462 RequestType::AddHttpFrontend(kept_front),
7463 RequestType::AddTcpFrontend(removed_tcp_front.clone()),
7464 RequestType::AddBackend(backend(REMOVED, 1026)),
7465 RequestType::AddBackend(backend(KEPT, 1027)),
7466 ];
7467 for (index, request) in setup.into_iter().enumerate() {
7468 for response in send(&mut server, &format!("setup-{index}"), request) {
7469 assert_eq!(
7470 response.status,
7471 ResponseStatus::Ok as i32,
7472 "setup must succeed: {response:?}"
7473 );
7474 }
7475 }
7476 assert_eq!(
7477 http_route(&server, http_address, "removed.test", "/"),
7478 Some(Some(REMOVED.to_owned())),
7479 "premise: the removed cluster's hostname routes to it"
7480 );
7481
7482 let responses = send(
7483 &mut server,
7484 "remove-cluster",
7485 RequestType::RemoveCluster(REMOVED.to_owned()),
7486 );
7487 assert_eq!(
7488 responses.iter().map(|r| r.id.as_str()).collect::<Vec<_>>(),
7489 vec!["remove-cluster"],
7490 "the worker must answer the RemoveCluster exactly once: {responses:?}"
7491 );
7492 assert_eq!(
7493 responses[0].status,
7494 ResponseStatus::Ok as i32,
7495 "RemoveCluster must succeed: {responses:?}"
7496 );
7497
7498 assert_eq!(
7499 http_route(&server, http_address, "removed.test", "/"),
7500 None,
7501 "the removed cluster's hostname must no longer route"
7502 );
7503 assert_eq!(
7504 http_route(&server, http_address, "removed.test", "/deny"),
7505 Some(None),
7506 "a route that names no cluster must survive the removal of a cluster on its hostname"
7507 );
7508 assert_eq!(
7509 http_route(&server, http_address, "kept.test", "/"),
7510 Some(Some(KEPT.to_owned())),
7511 "another cluster's route must stay"
7512 );
7513 let backends = server.backends.borrow();
7514 assert!(
7515 !backends.backends.contains_key(REMOVED),
7516 "the removed cluster's backends must be gone from the BackendMap"
7517 );
7518 assert!(
7519 backends.backends.contains_key(KEPT),
7520 "another cluster's backends must stay"
7521 );
7522 drop(backends);
7523
7524 for (index, request) in [
7525 RequestType::RemoveHttpFrontend(removed_front),
7526 RequestType::RemoveTcpFrontend(removed_tcp_front),
7527 RequestType::RemoveBackend(RemoveBackend {
7528 cluster_id: REMOVED.to_owned(),
7529 backend_id: format!("{REMOVED}-0"),
7530 address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
7531 }),
7532 ]
7533 .into_iter()
7534 .enumerate()
7535 {
7536 let responses = send(&mut server, &format!("follow-up-{index}"), request);
7537 assert!(
7538 !responses.is_empty()
7539 && responses
7540 .iter()
7541 .all(|r| r.status == ResponseStatus::Ok as i32),
7542 "a removal the cascade already performed must answer ok: {responses:?}"
7543 );
7544 }
7545 }
7546
7547 #[test]
7551 fn a_backend_of_an_unknown_cluster_is_refused_by_the_worker() {
7552 const ORPHAN: &str = "orphan";
7553 let mut server = bare_server();
7554 let assert_refused = |server: &mut Server, id: &str, port: u16| {
7555 let responses = send(server, id, RequestType::AddBackend(backend(ORPHAN, port)));
7556 assert!(
7557 !responses.is_empty()
7558 && responses
7559 .iter()
7560 .all(|r| r.status == ResponseStatus::Failure as i32),
7561 "a backend of an unknown cluster must be refused: {responses:?}"
7562 );
7563 assert!(
7564 !server.backends.borrow().backends.contains_key(ORPHAN),
7565 "a refused backend must not reach the BackendMap"
7566 );
7567 };
7568
7569 assert_refused(&mut server, "never-added", 1026);
7570
7571 for (index, request) in [
7572 RequestType::AddCluster(Cluster {
7573 cluster_id: ORPHAN.to_owned(),
7574 ..Default::default()
7575 }),
7576 RequestType::AddBackend(backend(ORPHAN, 1026)),
7577 RequestType::RemoveCluster(ORPHAN.to_owned()),
7578 ]
7579 .into_iter()
7580 .enumerate()
7581 {
7582 for response in send(&mut server, &format!("step-{index}"), request) {
7583 assert_eq!(
7584 response.status,
7585 ResponseStatus::Ok as i32,
7586 "step must succeed: {response:?}"
7587 );
7588 }
7589 }
7590
7591 assert_refused(&mut server, "removed", 1027);
7592 }
7593
7594 #[test]
7599 fn a_stale_frontend_removal_never_removes_another_clusters_route() {
7600 let http_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7601 let route = |cluster_id: &str| {
7602 let mut front = http_front(Some(cluster_id), "reused.test", "/");
7603 front.address = http_address;
7604 front
7605 };
7606 let cluster = |cluster_id: &str| {
7607 RequestType::AddCluster(Cluster {
7608 cluster_id: cluster_id.to_owned(),
7609 ..Default::default()
7610 })
7611 };
7612 let listener = || {
7613 RequestType::AddHttpListener(
7614 ListenerBuilder::new_http(http_address)
7615 .to_http(None)
7616 .expect("could not build the test HTTP listener config"),
7617 )
7618 };
7619
7620 for (a_is_gone, setup, expected_status) in [
7621 (
7622 true,
7623 vec![
7624 listener(),
7625 cluster("stale_a"),
7626 RequestType::AddHttpFrontend(route("stale_a")),
7627 RequestType::RemoveCluster("stale_a".to_owned()),
7628 cluster("stale_b"),
7629 RequestType::AddHttpFrontend(route("stale_b")),
7630 ],
7631 ResponseStatus::Ok,
7632 ),
7633 (
7634 false,
7635 vec![
7636 listener(),
7637 cluster("stale_a"),
7638 cluster("stale_b"),
7639 RequestType::AddHttpFrontend(route("stale_b")),
7640 ],
7641 ResponseStatus::Failure,
7642 ),
7643 ] {
7644 let mut server = bare_server();
7645 for (index, request) in setup.into_iter().enumerate() {
7646 for response in send(&mut server, &format!("setup-{index}"), request) {
7647 assert_eq!(
7648 response.status,
7649 ResponseStatus::Ok as i32,
7650 "setup must succeed: {response:?}"
7651 );
7652 }
7653 }
7654 assert_eq!(
7655 http_route(&server, http_address, "reused.test", "/"),
7656 Some(Some("stale_b".to_owned())),
7657 "premise: cluster B owns the route"
7658 );
7659
7660 let responses = send(
7661 &mut server,
7662 "stale-removal",
7663 RequestType::RemoveHttpFrontend(route("stale_a")),
7664 );
7665 assert_eq!(
7666 responses.iter().map(|r| r.status).collect::<Vec<_>>(),
7667 vec![expected_status as i32],
7668 "A gone: {a_is_gone}, answers: {responses:?}"
7669 );
7670 assert_eq!(
7671 http_route(&server, http_address, "reused.test", "/"),
7672 Some(Some("stale_b".to_owned())),
7673 "cluster B's route must survive a stale removal naming cluster A (A gone: {a_is_gone})"
7674 );
7675 }
7676 }
7677
7678 #[test]
7684 fn a_stale_tcp_or_udp_frontend_removal_never_unmaps_another_clusters_listener() {
7685 let cluster = |cluster_id: &str| {
7686 RequestType::AddCluster(Cluster {
7687 cluster_id: cluster_id.to_owned(),
7688 ..Default::default()
7689 })
7690 };
7691 for udp in [false, true] {
7692 let x = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7693 let y = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7694 let listener = |address: SocketAddress| {
7695 if udp {
7696 RequestType::AddUdpListener(
7697 ListenerBuilder::new_udp(address)
7698 .to_udp(None)
7699 .expect("could not build the test UDP listener config"),
7700 )
7701 } else {
7702 RequestType::AddTcpListener(
7703 ListenerBuilder::new_tcp(address)
7704 .to_tcp(None)
7705 .expect("could not build the test TCP listener config"),
7706 )
7707 }
7708 };
7709 let front = |cluster_id: &str, address: SocketAddress| {
7710 if udp {
7711 (
7712 RequestType::AddUdpFrontend(RequestUdpFrontend {
7713 cluster_id: cluster_id.to_owned(),
7714 address,
7715 ..Default::default()
7716 }),
7717 RequestType::RemoveUdpFrontend(RequestUdpFrontend {
7718 cluster_id: cluster_id.to_owned(),
7719 address,
7720 ..Default::default()
7721 }),
7722 )
7723 } else {
7724 (
7725 RequestType::AddTcpFrontend(RequestTcpFrontend {
7726 cluster_id: cluster_id.to_owned(),
7727 address,
7728 ..Default::default()
7729 }),
7730 RequestType::RemoveTcpFrontend(RequestTcpFrontend {
7731 cluster_id: cluster_id.to_owned(),
7732 address,
7733 ..Default::default()
7734 }),
7735 )
7736 }
7737 };
7738 let owner = |server: &Server| {
7739 let key = SocketAddr::from(x).into();
7740 if udp {
7741 server.udp.borrow().listener_cluster(&key)
7742 } else {
7743 server.tcp.borrow().listener_cluster(&key)
7744 }
7745 .map(|id| id.to_string())
7746 };
7747
7748 let mut server = bare_server();
7749 for (index, request) in [
7750 listener(x),
7751 listener(y),
7752 cluster("stale_a"),
7753 front("stale_a", x).0,
7754 RequestType::RemoveCluster("stale_a".to_owned()),
7755 cluster("stale_b"),
7756 front("stale_b", x).0,
7757 cluster("stale_a"),
7758 front("stale_a", y).0,
7759 ]
7760 .into_iter()
7761 .enumerate()
7762 {
7763 for response in send(&mut server, &format!("setup-{index}"), request) {
7764 assert_eq!(
7765 response.status,
7766 ResponseStatus::Ok as i32,
7767 "setup must succeed (udp: {udp}): {response:?}"
7768 );
7769 }
7770 }
7771 assert_eq!(
7772 owner(&server).as_deref(),
7773 Some("stale_b"),
7774 "premise: cluster B owns the listener on X (udp: {udp})"
7775 );
7776
7777 let responses = send(&mut server, "stale-removal", front("stale_a", x).1);
7778 assert_eq!(
7779 responses.iter().map(|r| r.status).collect::<Vec<_>>(),
7780 vec![ResponseStatus::Failure as i32],
7781 "the stale removal must be refused (udp: {udp}): {responses:?}"
7782 );
7783 assert_eq!(
7784 owner(&server).as_deref(),
7785 Some("stale_b"),
7786 "cluster B must keep the listener on X after a stale removal naming A (udp: {udp})"
7787 );
7788 }
7789 }
7790}
7791
7792#[cfg(test)]
7793mod scm_listener_handoff_tests {
7794 use std::{
7795 net::{TcpStream as StdTcpStream, UdpSocket as StdUdpSocket},
7796 os::fd::{IntoRawFd, RawFd},
7797 };
7798
7799 use sozu_command::{config::ListenerBuilder, proto::command::SocketAddress};
7800
7801 use super::*;
7802 use crate::{
7803 ProxyError,
7804 socket::{server_bind, udp_bind},
7805 testing::{ServerParts, prebuild_server, provide_port},
7806 };
7807
7808 fn state_with_four_active_listeners(
7811 http_address: SocketAddress,
7812 https_address: SocketAddress,
7813 tcp_address: SocketAddress,
7814 udp_address: SocketAddress,
7815 ) -> ConfigState {
7816 let mut state = ConfigState::new();
7817 let adds: Vec<(Request, SocketAddress, ListenerType)> = vec![
7818 (
7819 RequestType::AddHttpListener(
7820 ListenerBuilder::new_http(http_address)
7821 .to_http(None)
7822 .expect("could not build the test HTTP listener config"),
7823 )
7824 .into(),
7825 http_address,
7826 ListenerType::Http,
7827 ),
7828 (
7829 RequestType::AddHttpsListener(
7830 ListenerBuilder::new_https(https_address)
7831 .to_tls(None)
7832 .expect("could not build the test HTTPS listener config"),
7833 )
7834 .into(),
7835 https_address,
7836 ListenerType::Https,
7837 ),
7838 (
7839 RequestType::AddTcpListener(
7840 ListenerBuilder::new_tcp(tcp_address)
7841 .to_tcp(None)
7842 .expect("could not build the test TCP listener config"),
7843 )
7844 .into(),
7845 tcp_address,
7846 ListenerType::Tcp,
7847 ),
7848 (
7849 RequestType::AddUdpListener(
7850 ListenerBuilder::new_udp(udp_address)
7851 .to_udp(None)
7852 .expect("could not build the test UDP listener config"),
7853 )
7854 .into(),
7855 udp_address,
7856 ListenerType::Udp,
7857 ),
7858 ];
7859 for (add, address, proxy) in adds {
7860 state.dispatch(&add).expect("could not add the listener");
7861 state
7862 .dispatch(
7863 &RequestType::ActivateListener(ActivateListener {
7864 interface: None,
7865 address,
7866 proxy: proxy.into(),
7867 from_scm: false,
7868 })
7869 .into(),
7870 )
7871 .expect("could not mark the listener active");
7872 }
7873 state
7874 }
7875
7876 fn inherited_table<'a>(server: &'a Server, label: &str) -> &'a Vec<(ListenerKey, RawFd)> {
7878 let scm_listeners = server
7879 .scm_listeners
7880 .as_ref()
7881 .expect("scm_listeners must be present");
7882 match label {
7883 "HTTP" => &scm_listeners.http,
7884 "HTTPS" => &scm_listeners.tls,
7885 "TCP" => &scm_listeners.tcp,
7886 "UDP" => &scm_listeners.udp,
7887 other => panic!("unknown protocol label {other}"),
7888 }
7889 }
7890
7891 fn descriptor_is_socket_on_port(fd: RawFd, port: u16) -> bool {
7901 let mut storage: libc::sockaddr_storage = unsafe { std::mem::zeroed() };
7902 let mut length = std::mem::size_of::<libc::sockaddr_storage>() as libc::socklen_t;
7903 let result = unsafe {
7907 libc::getsockname(
7908 fd,
7909 (&raw mut storage).cast::<libc::sockaddr>(),
7910 &raw mut length,
7911 )
7912 };
7913 if result != 0 || storage.ss_family != libc::AF_INET as libc::sa_family_t {
7914 return false;
7915 }
7916 let inet = unsafe { &*(&raw const storage).cast::<libc::sockaddr_in>() };
7919 u16::from_be(inet.sin_port) == port
7920 }
7921
7922 const QUEUED_DATAGRAM: &[u8] = b"queued-before-the-handoff";
7924
7925 #[test]
7955 fn inherited_listener_sockets_are_adopted_by_the_initial_activation() {
7956 let ServerParts {
7957 event_loop,
7958 sessions,
7959 pool,
7960 backends,
7961 client_scm_socket,
7962 server_scm_socket,
7963 server_config,
7964 ..
7965 } = prebuild_server(16, 16384, false).expect("could not prebuild a test server");
7966 let (_command_channel, proxy_channel) =
7967 Channel::generate(1000, 10000).expect("could not generate a test channel");
7968
7969 let http_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7970 let https_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7971 let tcp_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7972 let udp_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7973
7974 let http_listener = server_bind(http_address.into(), None)
7977 .expect("could not bind the inherited HTTP socket");
7978 let https_listener = server_bind(https_address.into(), None)
7979 .expect("could not bind the inherited HTTPS socket");
7980 let tcp_listener =
7981 server_bind(tcp_address.into(), None).expect("could not bind the inherited TCP socket");
7982 let udp_socket =
7983 udp_bind(udp_address.into(), None).expect("could not bind the inherited UDP socket");
7984
7985 let http_client = StdTcpStream::connect::<SocketAddr>(http_address.into())
7990 .expect("could not queue a connection on the inherited HTTP socket");
7991 let https_client = StdTcpStream::connect::<SocketAddr>(https_address.into())
7992 .expect("could not queue a connection on the inherited HTTPS socket");
7993 let tcp_client = StdTcpStream::connect::<SocketAddr>(tcp_address.into())
7994 .expect("could not queue a connection on the inherited TCP socket");
7995 let datagram_sender =
7996 StdUdpSocket::bind("127.0.0.1:0").expect("could not bind the test datagram sender");
7997 datagram_sender
7998 .send_to(QUEUED_DATAGRAM, SocketAddr::from(udp_address))
7999 .expect("could not queue a datagram on the inherited UDP socket");
8000
8001 let listeners = Listeners {
8004 http: vec![(http_address.into(), http_listener.into_raw_fd())],
8005 tls: vec![(https_address.into(), https_listener.into_raw_fd())],
8006 tcp: vec![(tcp_address.into(), tcp_listener.into_raw_fd())],
8007 udp: vec![(udp_address.into(), udp_socket.into_raw_fd())],
8008 };
8009 client_scm_socket
8010 .send_listeners(&listeners)
8011 .expect("could not send the inherited listeners");
8012 listeners.close();
8013
8014 let initial_state =
8020 state_with_four_active_listeners(http_address, https_address, tcp_address, udp_address)
8021 .produce_initial_state();
8022 assert_eq!(
8023 initial_state.requests.len(),
8024 8,
8025 "the generator must emit one Add and one Activate per active listener"
8026 );
8027
8028 let server = Server::new(
8029 event_loop,
8030 proxy_channel,
8031 server_scm_socket,
8032 sessions,
8033 pool,
8034 backends,
8035 None,
8036 None,
8037 None,
8038 server_config,
8039 Some(initial_state),
8040 false,
8041 )
8042 .expect("could not build the test server");
8043
8044 let accepted: Vec<Protocol> = server
8045 .accept_queue
8046 .iter()
8047 .map(|(_, _, protocol, _, _)| *protocol)
8048 .collect();
8049 assert!(
8050 accepted.contains(&Protocol::HTTPListen),
8051 "the connection queued on the inherited HTTP socket must survive the hand-off, got {accepted:?}"
8052 );
8053 assert!(
8054 accepted.contains(&Protocol::HTTPSListen),
8055 "the connection queued on the inherited HTTPS socket must survive the hand-off, got {accepted:?}"
8056 );
8057 assert!(
8058 accepted.contains(&Protocol::TCPListen),
8059 "the connection queued on the inherited TCP socket must survive the hand-off, got {accepted:?}"
8060 );
8061
8062 for (protocol, client) in [
8070 (Protocol::HTTPListen, &http_client),
8071 (Protocol::HTTPSListen, &https_client),
8072 (Protocol::TCPListen, &tcp_client),
8073 ] {
8074 let (sock, _, _, _, peer) = server
8075 .accept_queue
8076 .iter()
8077 .find(|(_, _, queued, _, _)| *queued == protocol)
8078 .expect("the queued connection was asserted above");
8079 assert_eq!(
8080 socket2::SockRef::from(sock).tcp_nodelay().ok(),
8081 Some(true),
8082 "a {protocol:?} connection queued before the hand-off must carry TCP_NODELAY"
8083 );
8084 assert_eq!(
8085 *peer,
8086 client.local_addr().expect("the test client has an address"),
8087 "the {protocol:?} peer address must be the client's"
8088 );
8089 }
8090
8091 let (_, adopted_udp_socket) = server
8092 .udp
8093 .borrow_mut()
8094 .give_back_listeners()
8095 .pop()
8096 .expect("the UDP listener must hold a socket after activation");
8097 let mut buffer = [0u8; 64];
8098 let received = adopted_udp_socket.recv_from(&mut buffer);
8099 match received {
8100 Ok((length, _)) => assert_eq!(
8101 &buffer[..length],
8102 QUEUED_DATAGRAM,
8103 "the datagram queued on the inherited UDP socket must survive the hand-off"
8104 ),
8105 Err(error) => panic!(
8106 "the datagram queued on the inherited UDP socket must survive the hand-off, got {error:?}"
8107 ),
8108 }
8109 }
8110
8111 #[test]
8135 fn a_repeated_activation_closes_the_descriptor_it_cannot_adopt_now() {
8136 let ServerParts {
8137 event_loop,
8138 sessions,
8139 pool,
8140 backends,
8141 server_scm_socket,
8142 server_config,
8143 ..
8144 } = prebuild_server(16, 16384, true).expect("could not prebuild a test server");
8145 let (_command_channel, proxy_channel) =
8146 Channel::generate(1000, 10000).expect("could not generate a test channel");
8147
8148 let http_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8149 let https_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8150 let tcp_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8151 let udp_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8152
8153 let initial_state =
8156 state_with_four_active_listeners(http_address, https_address, tcp_address, udp_address)
8157 .produce_initial_state();
8158 let mut server = Server::new(
8159 event_loop,
8160 proxy_channel,
8161 server_scm_socket,
8162 sessions,
8163 pool,
8164 backends,
8165 None,
8166 None,
8167 None,
8168 server_config,
8169 Some(initial_state),
8170 false,
8171 )
8172 .expect("could not build the test server");
8173
8174 let http_fd = server_bind(http_address.into(), None)
8176 .expect("could not bind the spare HTTP socket")
8177 .into_raw_fd();
8178 let https_fd = server_bind(https_address.into(), None)
8179 .expect("could not bind the spare HTTPS socket")
8180 .into_raw_fd();
8181 let tcp_fd = server_bind(tcp_address.into(), None)
8182 .expect("could not bind the spare TCP socket")
8183 .into_raw_fd();
8184 let udp_fd = udp_bind(udp_address.into(), None)
8185 .expect("could not bind the spare UDP socket")
8186 .into_raw_fd();
8187 {
8188 let scm_listeners = server
8189 .scm_listeners
8190 .as_mut()
8191 .expect("Server::new must have populated scm_listeners");
8192 scm_listeners.http.push((http_address.into(), http_fd));
8193 scm_listeners.tls.push((https_address.into(), https_fd));
8194 scm_listeners.tcp.push((tcp_address.into(), tcp_fd));
8195 scm_listeners.udp.push((udp_address.into(), udp_fd));
8196 }
8197
8198 for (address, proxy) in [
8199 (http_address, ListenerType::Http),
8200 (https_address, ListenerType::Https),
8201 (tcp_address, ListenerType::Tcp),
8202 (udp_address, ListenerType::Udp),
8203 ] {
8204 let response = server.notify_activate_listener(
8205 "test-repeat",
8206 &ActivateListener {
8207 interface: None,
8208 address,
8209 proxy: proxy.into(),
8210 from_scm: false,
8211 },
8212 );
8213 assert_eq!(
8214 response.status,
8215 ResponseStatus::Ok as i32,
8216 "a repeated {proxy:?} activation must still answer ok: {response:?}"
8217 );
8218 }
8219
8220 let scm_listeners = server
8221 .scm_listeners
8222 .as_ref()
8223 .expect("scm_listeners must still be present");
8224 for (label, table, fd, port) in [
8225 (
8226 "HTTP",
8227 &scm_listeners.http,
8228 http_fd,
8229 http_address.port as u16,
8230 ),
8231 (
8232 "HTTPS",
8233 &scm_listeners.tls,
8234 https_fd,
8235 https_address.port as u16,
8236 ),
8237 ("TCP", &scm_listeners.tcp, tcp_fd, tcp_address.port as u16),
8238 ("UDP", &scm_listeners.udp, udp_fd, udp_address.port as u16),
8239 ] {
8240 assert!(
8241 !table.iter().any(|(_, entry)| *entry == fd),
8242 "the inherited {label} descriptor this activation cannot adopt must leave the SCM table"
8243 );
8244 assert!(
8245 !descriptor_is_socket_on_port(fd, port),
8246 "the inherited {label} descriptor this activation cannot adopt must be closed, \
8247 not left bound in the address's SO_REUSEPORT group, where it is expected — \
8248 inferred, not measured here — to go on taking a share of new connections"
8249 );
8250 }
8251 }
8252
8253 #[test]
8270 fn an_inherited_descriptor_with_no_listener_yet_is_adopted_by_a_later_activation() {
8271 let ServerParts {
8272 event_loop,
8273 sessions,
8274 pool,
8275 backends,
8276 server_scm_socket,
8277 server_config,
8278 ..
8279 } = prebuild_server(16, 16384, true).expect("could not prebuild a test server");
8280 let (_command_channel, proxy_channel) =
8281 Channel::generate(1000, 10000).expect("could not generate a test channel");
8282
8283 let http_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8284 let https_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8285 let tcp_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8286 let udp_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8287
8288 let mut server = Server::new(
8289 event_loop,
8290 proxy_channel,
8291 server_scm_socket,
8292 sessions,
8293 pool,
8294 backends,
8295 None,
8296 None,
8297 None,
8298 server_config,
8299 None,
8300 false,
8301 )
8302 .expect("could not build the test server");
8303
8304 let http_listener = server_bind(http_address.into(), None)
8308 .expect("could not bind the inherited HTTP socket");
8309 let https_listener = server_bind(https_address.into(), None)
8310 .expect("could not bind the inherited HTTPS socket");
8311 let tcp_listener =
8312 server_bind(tcp_address.into(), None).expect("could not bind the inherited TCP socket");
8313 let udp_socket =
8314 udp_bind(udp_address.into(), None).expect("could not bind the inherited UDP socket");
8315 let _http_client = StdTcpStream::connect::<SocketAddr>(http_address.into())
8316 .expect("could not queue a connection on the inherited HTTP socket");
8317 let _https_client = StdTcpStream::connect::<SocketAddr>(https_address.into())
8318 .expect("could not queue a connection on the inherited HTTPS socket");
8319 let _tcp_client = StdTcpStream::connect::<SocketAddr>(tcp_address.into())
8320 .expect("could not queue a connection on the inherited TCP socket");
8321 let (http_fd, https_fd, tcp_fd, udp_fd) = (
8322 http_listener.into_raw_fd(),
8323 https_listener.into_raw_fd(),
8324 tcp_listener.into_raw_fd(),
8325 udp_socket.into_raw_fd(),
8326 );
8327 {
8328 let scm_listeners = server
8329 .scm_listeners
8330 .as_mut()
8331 .expect("Server::new must have populated scm_listeners");
8332 scm_listeners.http.push((http_address.into(), http_fd));
8333 scm_listeners.tls.push((https_address.into(), https_fd));
8334 scm_listeners.tcp.push((tcp_address.into(), tcp_fd));
8335 scm_listeners.udp.push((udp_address.into(), udp_fd));
8336 }
8337
8338 let add_requests =
8339 state_with_four_active_listeners(http_address, https_address, tcp_address, udp_address);
8340 let cases = [
8341 ("HTTP", http_address, ListenerType::Http, http_fd),
8342 ("HTTPS", https_address, ListenerType::Https, https_fd),
8343 ("TCP", tcp_address, ListenerType::Tcp, tcp_fd),
8344 ("UDP", udp_address, ListenerType::Udp, udp_fd),
8345 ];
8346
8347 for (label, address, proxy, fd) in cases {
8350 let response = server.notify_activate_listener(
8351 "test-early",
8352 &ActivateListener {
8353 interface: None,
8354 address,
8355 proxy: proxy.into(),
8356 from_scm: false,
8357 },
8358 );
8359 assert_ne!(
8360 response.status,
8361 ResponseStatus::Ok as i32,
8362 "activating {label} with no listener must fail: {response:?}"
8363 );
8364 assert!(
8365 inherited_table(&server, label)
8366 .iter()
8367 .any(|(_, e)| *e == fd),
8368 "the {label} descriptor must stay in the SCM table while no listener owns \
8369 its address"
8370 );
8371 assert!(
8372 descriptor_is_socket_on_port(fd, address.port as u16),
8373 "the {label} descriptor must stay open while no listener owns its address"
8374 );
8375 }
8376
8377 for request in add_requests.produce_initial_state().requests {
8380 server.notify_proxys(request);
8381 }
8382 for (label, _address, _proxy, fd) in cases {
8383 assert!(
8384 !inherited_table(&server, label)
8385 .iter()
8386 .any(|(_, e)| *e == fd),
8387 "the retained {label} descriptor must be adopted by the later activation"
8388 );
8389 }
8390 let accepted: Vec<Protocol> = server
8391 .accept_queue
8392 .iter()
8393 .map(|(_, _, protocol, _, _)| *protocol)
8394 .collect();
8395 for expected in [
8396 Protocol::HTTPListen,
8397 Protocol::HTTPSListen,
8398 Protocol::TCPListen,
8399 ] {
8400 assert!(
8401 accepted.contains(&expected),
8402 "the connection queued on the retained {expected:?} descriptor must be served \
8403 by the later activation, got {accepted:?}"
8404 );
8405 }
8406 }
8407
8408 #[test]
8433 fn a_socket_parked_by_a_failed_registration_is_not_mistaken_for_a_live_one() {
8434 let ServerParts {
8435 registry,
8436 sessions,
8437 pool,
8438 backends,
8439 ..
8440 } = prebuild_server(16, 16384, true).expect("could not prebuild a test server");
8441
8442 let token_for = |protocol: Protocol| {
8443 let mut manager = sessions.borrow_mut();
8444 let entry = manager.slab.vacant_entry();
8445 let token = Token(entry.key());
8446 entry.insert(Rc::new(RefCell::new(ListenSession { protocol })));
8447 token
8448 };
8449 let pre_registered = |address: SocketAddress| {
8452 let mut socket = server_bind(address.into(), None)
8453 .expect("could not bind the test listening socket");
8454 registry
8455 .register(&mut socket, Token(usize::MAX - 2), Interest::READABLE)
8456 .expect("could not pre-register the test socket");
8457 socket
8458 };
8459
8460 let http_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8461 let mut http_proxy = http::HttpProxy::new(
8462 registry.try_clone().expect("could not clone the registry"),
8463 sessions.clone(),
8464 pool.clone(),
8465 backends.clone(),
8466 );
8467 http_proxy
8468 .add_listener(
8469 ListenerBuilder::new_http(http_address)
8470 .to_http(None)
8471 .expect("could not build the test HTTP listener config"),
8472 token_for(Protocol::HTTPListen),
8473 )
8474 .expect("could not add the test HTTP listener");
8475 assert!(
8476 http_proxy
8477 .activate_listener(&http_address.into(), Some(pre_registered(http_address)))
8478 .is_err(),
8479 "the pre-registered descriptor must make the HTTP registration fail"
8480 );
8481 assert!(
8482 matches!(
8483 http_proxy.give_back_listener(&http_address.into()),
8484 Err(ProxyError::UnactivatedListener)
8485 ),
8486 "an HTTP listener holding only a parked socket must not answer as activated"
8487 );
8488 assert!(
8489 http_proxy.soft_stop().is_ok(),
8490 "a parked socket must not fail an HTTP soft stop"
8491 );
8492
8493 let https_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8494 let mut https_proxy = https::HttpsProxy::new(
8495 registry.try_clone().expect("could not clone the registry"),
8496 sessions.clone(),
8497 pool.clone(),
8498 backends.clone(),
8499 );
8500 https_proxy
8501 .add_listener(
8502 ListenerBuilder::new_https(https_address)
8503 .to_tls(None)
8504 .expect("could not build the test HTTPS listener config"),
8505 token_for(Protocol::HTTPSListen),
8506 )
8507 .expect("could not add the test HTTPS listener");
8508 assert!(
8509 https_proxy
8510 .activate_listener(&https_address.into(), Some(pre_registered(https_address)))
8511 .is_err(),
8512 "the pre-registered descriptor must make the HTTPS registration fail"
8513 );
8514 assert!(
8515 matches!(
8516 https_proxy.give_back_listener(&https_address.into()),
8517 Err(ProxyError::UnactivatedListener)
8518 ),
8519 "an HTTPS listener holding only a parked socket must not answer as activated"
8520 );
8521 assert!(
8522 https_proxy.hard_stop().is_ok(),
8523 "a parked socket must not fail an HTTPS hard stop"
8524 );
8525
8526 let tcp_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8527 let mut tcp_proxy = tcp::TcpProxy::new(
8528 registry.try_clone().expect("could not clone the registry"),
8529 sessions.clone(),
8530 pool.clone(),
8531 backends.clone(),
8532 );
8533 tcp_proxy
8534 .add_listener(
8535 ListenerBuilder::new_tcp(tcp_address)
8536 .to_tcp(None)
8537 .expect("could not build the test TCP listener config"),
8538 token_for(Protocol::TCPListen),
8539 )
8540 .expect("could not add the test TCP listener");
8541 assert!(
8542 tcp_proxy
8543 .activate_listener(&tcp_address.into(), Some(pre_registered(tcp_address)))
8544 .is_err(),
8545 "the pre-registered descriptor must make the TCP registration fail"
8546 );
8547 assert!(
8548 matches!(
8549 tcp_proxy.give_back_listener(&tcp_address.into()),
8550 Err(ProxyError::UnactivatedListener)
8551 ),
8552 "a TCP listener holding only a parked socket must not answer as activated"
8553 );
8554
8555 let udp_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8556 let mut udp_proxy = udp::UdpProxy::new(
8557 registry.try_clone().expect("could not clone the registry"),
8558 sessions.clone(),
8559 pool.clone(),
8560 backends.clone(),
8561 16,
8562 16384,
8563 );
8564 udp_proxy
8565 .add_listener(
8566 ListenerBuilder::new_udp(udp_address)
8567 .to_udp(None)
8568 .expect("could not build the test UDP listener config"),
8569 token_for(Protocol::UDP),
8570 )
8571 .expect("could not add the test UDP listener");
8572 let mut udp_socket =
8573 udp_bind(udp_address.into(), None).expect("could not bind the test UDP socket");
8574 registry
8575 .register(&mut udp_socket, Token(usize::MAX - 2), Interest::READABLE)
8576 .expect("could not pre-register the test UDP socket");
8577 assert!(
8578 udp_proxy
8579 .activate_listener(&udp_address.into(), Some(udp_socket))
8580 .is_err(),
8581 "the pre-registered descriptor must make the UDP registration fail"
8582 );
8583 assert!(
8584 matches!(
8585 udp_proxy.give_back_listener(&udp_address.into()),
8586 Err(ProxyError::UnactivatedListener)
8587 ),
8588 "a UDP listener holding only a parked socket must not answer as activated"
8589 );
8590 }
8591
8592 #[test]
8606 fn a_failed_registration_does_not_close_the_socket_it_was_handed() {
8607 let ServerParts {
8608 registry,
8609 sessions,
8610 pool,
8611 backends,
8612 ..
8613 } = prebuild_server(16, 16384, true).expect("could not prebuild a test server");
8614
8615 let token_for = |protocol: Protocol| {
8616 let mut manager = sessions.borrow_mut();
8617 let entry = manager.slab.vacant_entry();
8618 let token = Token(entry.key());
8619 entry.insert(Rc::new(RefCell::new(ListenSession { protocol })));
8620 token
8621 };
8622
8623 let handed_out = |address: SocketAddress| {
8626 let mut socket = server_bind(address.into(), None)
8627 .expect("could not bind the test listening socket");
8628 registry
8629 .register(&mut socket, Token(usize::MAX - 1), Interest::READABLE)
8630 .expect("could not pre-register the test socket");
8631 let fd = socket.as_raw_fd();
8632 (socket, fd)
8633 };
8634
8635 let http_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8636 let token = token_for(Protocol::HTTPListen);
8637 let mut http_proxy = http::HttpProxy::new(
8638 registry.try_clone().expect("could not clone the registry"),
8639 sessions.clone(),
8640 pool.clone(),
8641 backends.clone(),
8642 );
8643 http_proxy
8644 .add_listener(
8645 ListenerBuilder::new_http(http_address)
8646 .to_http(None)
8647 .expect("could not build the test HTTP listener config"),
8648 token,
8649 )
8650 .expect("could not add the test HTTP listener");
8651 let (socket, http_fd) = handed_out(http_address);
8652 let result = http_proxy.activate_listener(&http_address.into(), Some(socket));
8653 assert!(
8654 result.is_err(),
8655 "the pre-registered descriptor must make the HTTP registration fail"
8656 );
8657 assert!(
8658 descriptor_is_socket_on_port(http_fd, http_address.port as u16),
8659 "a failed HTTP registration must not close the socket it was handed"
8660 );
8661
8662 let tcp_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8663 let token = token_for(Protocol::TCPListen);
8664 let mut tcp_proxy = tcp::TcpProxy::new(
8665 registry.try_clone().expect("could not clone the registry"),
8666 sessions.clone(),
8667 pool.clone(),
8668 backends.clone(),
8669 );
8670 tcp_proxy
8671 .add_listener(
8672 ListenerBuilder::new_tcp(tcp_address)
8673 .to_tcp(None)
8674 .expect("could not build the test TCP listener config"),
8675 token,
8676 )
8677 .expect("could not add the test TCP listener");
8678 let (socket, tcp_fd) = handed_out(tcp_address);
8679 let result = tcp_proxy.activate_listener(&tcp_address.into(), Some(socket));
8680 assert!(
8681 result.is_err(),
8682 "the pre-registered descriptor must make the TCP registration fail"
8683 );
8684 assert!(
8685 descriptor_is_socket_on_port(tcp_fd, tcp_address.port as u16),
8686 "a failed TCP registration must not close the socket it was handed"
8687 );
8688
8689 let https_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8690 let token = token_for(Protocol::HTTPSListen);
8691 let mut https_proxy = https::HttpsProxy::new(
8692 registry.try_clone().expect("could not clone the registry"),
8693 sessions.clone(),
8694 pool.clone(),
8695 backends.clone(),
8696 );
8697 https_proxy
8698 .add_listener(
8699 ListenerBuilder::new_https(https_address)
8700 .to_tls(None)
8701 .expect("could not build the test HTTPS listener config"),
8702 token,
8703 )
8704 .expect("could not add the test HTTPS listener");
8705 let (socket, https_fd) = handed_out(https_address);
8706 let result = https_proxy.activate_listener(&https_address.into(), Some(socket));
8707 assert!(
8708 result.is_err(),
8709 "the pre-registered descriptor must make the HTTPS registration fail"
8710 );
8711 assert!(
8712 descriptor_is_socket_on_port(https_fd, https_address.port as u16),
8713 "a failed HTTPS registration must not close the socket it was handed"
8714 );
8715
8716 let udp_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8717 let token = token_for(Protocol::UDP);
8718 let mut udp_proxy = udp::UdpProxy::new(
8719 registry.try_clone().expect("could not clone the registry"),
8720 sessions.clone(),
8721 pool.clone(),
8722 backends.clone(),
8723 16,
8724 16384,
8725 );
8726 udp_proxy
8727 .add_listener(
8728 ListenerBuilder::new_udp(udp_address)
8729 .to_udp(None)
8730 .expect("could not build the test UDP listener config"),
8731 token,
8732 )
8733 .expect("could not add the test UDP listener");
8734 let mut socket =
8735 udp_bind(udp_address.into(), None).expect("could not bind the test UDP socket");
8736 registry
8737 .register(&mut socket, Token(usize::MAX - 1), Interest::READABLE)
8738 .expect("could not pre-register the test UDP socket");
8739 let udp_fd = socket.as_raw_fd();
8740 let result = udp_proxy.activate_listener(&udp_address.into(), Some(socket));
8741 assert!(
8742 result.is_err(),
8743 "the pre-registered descriptor must make the UDP registration fail"
8744 );
8745 assert!(
8746 descriptor_is_socket_on_port(udp_fd, udp_address.port as u16),
8747 "a failed UDP registration must not close the socket it was handed"
8748 );
8749 }
8750}
8751
8752#[cfg(test)]
8753mod tests {
8754 use super::{ClusterId, SessionManager, mask_ipv4, mask_ipv6, subnet_key};
8755 use mio::Token;
8756 use slab::Slab;
8757 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
8758
8759 #[test]
8766 fn a_cluster_override_dropping_to_zero_after_a_track_is_a_no_op() {
8767 let manager = SessionManager::new(Slab::with_capacity(8), 8, 0, 0, 0, 24, 56);
8768 let token = Token(1);
8769 let cluster = ClusterId::from("api");
8770 let ip: IpAddr = "203.0.113.9".parse().expect("test address must parse");
8771
8772 manager
8774 .borrow_mut()
8775 .track_cluster_subnet(token, cluster.clone(), ip, Some(5));
8776 assert!(
8777 !manager.borrow().subnet_tracking_is_empty(),
8778 "premise: the positive override must have recorded a slot"
8779 );
8780
8781 manager
8785 .borrow_mut()
8786 .track_cluster_subnet(token, cluster.clone(), ip, Some(0));
8787
8788 manager.borrow_mut().untrack_all_cluster_ip(token);
8790 assert!(
8791 manager.borrow().subnet_tracking_is_empty(),
8792 "teardown must drain the slot even though the cap was disabled in between"
8793 );
8794 }
8795
8796 #[test]
8801 fn masking_handles_both_shift_overflowing_prefix_bounds() {
8802 let v4: Ipv4Addr = "203.0.113.45".parse().expect("test address must parse");
8803 assert_eq!(
8804 mask_ipv4(v4, 32),
8805 v4,
8806 "a /32 must be the address itself, not a wrapped-shift artefact"
8807 );
8808 assert_eq!(
8809 mask_ipv4(v4, 0),
8810 Ipv4Addr::UNSPECIFIED,
8811 "a /0 must mask every bit away"
8812 );
8813
8814 let v6: Ipv6Addr = "2001:db8:dead:beef::1"
8815 .parse()
8816 .expect("test address must parse");
8817 assert_eq!(
8818 mask_ipv6(v6, 128),
8819 v6,
8820 "a /128 must be the address itself, not a wrapped-shift artefact"
8821 );
8822 assert_eq!(
8823 mask_ipv6(v6, 0),
8824 Ipv6Addr::UNSPECIFIED,
8825 "a /0 must mask every bit away"
8826 );
8827
8828 assert_eq!(
8831 mask_ipv4(v4, 33),
8832 v4,
8833 "an over-wide IPv4 prefix must clamp to the address, not panic or wrap"
8834 );
8835 assert_eq!(
8836 mask_ipv6(v6, 129),
8837 v6,
8838 "an over-wide IPv6 prefix must clamp to the address, not panic or wrap"
8839 );
8840 }
8841
8842 #[test]
8845 fn masking_groups_addresses_at_the_recommended_prefixes() {
8846 let parse_v4 = |s: &str| -> Ipv4Addr { s.parse().expect("test address must parse") };
8847 assert_eq!(
8848 mask_ipv4(parse_v4("203.0.113.1"), 24),
8849 mask_ipv4(parse_v4("203.0.113.254"), 24),
8850 "two hosts in one /24 must share a subnet key"
8851 );
8852 assert_ne!(
8853 mask_ipv4(parse_v4("203.0.113.1"), 24),
8854 mask_ipv4(parse_v4("203.0.114.1"), 24),
8855 "hosts in different /24s must NOT share a subnet key"
8856 );
8857
8858 let parse_v6 = |s: &str| -> Ipv6Addr { s.parse().expect("test address must parse") };
8859 assert_eq!(
8860 mask_ipv6(parse_v6("2001:db8:abcd:0000::1"), 56),
8861 mask_ipv6(parse_v6("2001:db8:abcd:00ff::9999"), 56),
8862 "two addresses in one /56 must share a subnet key"
8863 );
8864 assert_ne!(
8865 mask_ipv6(parse_v6("2001:db8:abcd:0000::1"), 56),
8866 mask_ipv6(parse_v6("2001:db8:abcd:0100::1"), 56),
8867 "addresses in different /56s must NOT share a subnet key"
8868 );
8869
8870 assert_eq!(
8873 mask_ipv6(parse_v6("2001:db8:abcd:1234::1"), 56),
8874 mask_ipv6(parse_v6("2001:db8:abcd:1234:ffff:ffff:ffff:ffff"), 56),
8875 "every address a /64 holder can mint must collapse to one /56 key — that \
8876 collapse IS the mitigation for sozu-proxy/sozu#1270"
8877 );
8878 }
8879
8880 #[test]
8885 fn an_ipv4_mapped_source_is_charged_to_the_ipv4_prefix() {
8886 let mapped: IpAddr = "::ffff:203.0.113.7".parse().expect("must parse");
8887 let native: IpAddr = "203.0.113.7".parse().expect("must parse");
8888 assert_eq!(
8889 subnet_key(&mapped, 24, 56),
8890 subnet_key(&native, 24, 56),
8891 "a mapped IPv4 source must key identically to the same address arriving natively"
8892 );
8893 assert!(
8894 subnet_key(&mapped, 24, 56).is_ipv4(),
8895 "the key for a mapped source must be an IPv4 key"
8896 );
8897
8898 let other: IpAddr = "::ffff:198.51.100.7".parse().expect("must parse");
8902 assert_ne!(
8903 subnet_key(&mapped, 24, 56),
8904 subnet_key(&other, 24, 56),
8905 "mapped sources from different IPv4 /24s must NOT share a subnet key — if they \
8906 do, every IPv4 client on a dual-stack listener shares one bucket"
8907 );
8908
8909 let genuine: IpAddr = "2001:db8::1".parse().expect("must parse");
8911 assert!(
8912 subnet_key(&genuine, 24, 56).is_ipv6(),
8913 "a genuine IPv6 source must keep an IPv6 subnet key"
8914 );
8915 }
8916}
8917
8918#[cfg(test)]
8919mod initial_protected_bootstrap_tests {
8920 use std::thread;
8921
8922 use sozu_command::{
8923 channel::Channel,
8924 proto::command::{
8925 CountRequests, HealthCheckConfig, HealthCheckHeader, HealthCheckHeaderValue,
8926 InitialState, ProtectedHealthCheckOrder, ProtectedHealthCheckPhase,
8927 QueryProtectedHealthCheckSupport, SetProtectedHealthCheck, Status,
8928 },
8929 secret::{FingerprintKey, ProtectedHeader, SecretValue, Zeroizing},
8930 };
8931
8932 use super::*;
8933 use crate::testing::{ServerParts, prebuild_server};
8934
8935 const CLUSTER: &str = "bootstrap-cluster";
8936 const HEADER: &str = "X-Synthetic-Bootstrap";
8937 const SECRET: &[u8] = b"synthetic-bootstrap-secret";
8938 const DEADLINE: Duration = Duration::from_secs(20);
8939
8940 type Phase = ProtectedHealthCheckPhase;
8941
8942 fn request(id: &str, request_type: RequestType) -> WorkerRequest {
8943 WorkerRequest {
8944 id: id.to_owned(),
8945 content: request_type.into(),
8946 }
8947 }
8948
8949 fn capability(id: &str) -> WorkerRequest {
8950 request(
8951 id,
8952 RequestType::QueryProtectedHealthCheckSupport(QueryProtectedHealthCheckSupport {}),
8953 )
8954 }
8955
8956 fn status(id: &str) -> WorkerRequest {
8957 request(id, RequestType::Status(Status {}))
8958 }
8959
8960 fn foreign(id: &str) -> WorkerRequest {
8961 request(id, RequestType::CountRequests(CountRequests {}))
8962 }
8963
8964 fn payload(cluster: &str) -> SetProtectedHealthCheck {
8965 let key = FingerprintKey::from_parts(Zeroizing::new([7u8; 32]), [9u8; 16]);
8966 let mut carrier = HealthCheckHeaderValue {
8967 name: HEADER.to_owned(),
8968 value: SECRET.to_vec(),
8969 };
8970 let secret = SecretValue::take_from_ingress(&mut carrier).expect("a valid value");
8971 let header = ProtectedHeader::new(&key, HEADER.to_owned(), secret);
8972 SetProtectedHealthCheck {
8973 cluster_id: cluster.to_owned(),
8974 config: HealthCheckConfig {
8975 uri: "/".to_owned(),
8976 interval: 10,
8977 timeout: 5,
8978 healthy_threshold: 3,
8979 unhealthy_threshold: 3,
8980 headers: vec![HealthCheckHeader {
8981 name: HEADER.to_owned(),
8982 fingerprint: header.fingerprint().to_string(),
8983 }],
8984 ..Default::default()
8985 },
8986 values: vec![HealthCheckHeaderValue {
8987 name: HEADER.to_owned(),
8988 value: SECRET.to_vec(),
8989 }],
8990 }
8991 }
8992
8993 fn order(id: &str, phase: Phase, cluster: &str, version: u64) -> WorkerRequest {
8994 let prepare = (phase == Phase::ProtectedHealthCheckPrepare).then(|| payload(cluster));
8995 request(
8996 id,
8997 RequestType::ProtectedHealthCheckOrder(ProtectedHealthCheckOrder {
8998 phase: phase as i32,
8999 cluster_id: cluster.to_owned(),
9000 version,
9001 prepare,
9002 }),
9003 )
9004 }
9005
9006 fn prepare(id: &str, version: u64) -> WorkerRequest {
9007 order(id, Phase::ProtectedHealthCheckPrepare, CLUSTER, version)
9008 }
9009
9010 fn commit(id: &str, version: u64) -> WorkerRequest {
9011 order(id, Phase::ProtectedHealthCheckCommit, CLUSTER, version)
9012 }
9013
9014 fn release(id: &str, version: u64) -> WorkerRequest {
9015 order(id, Phase::ProtectedHealthCheckRelease, CLUSTER, version)
9016 }
9017
9018 fn abort(id: &str, version: u64) -> WorkerRequest {
9019 order(id, Phase::ProtectedHealthCheckAbort, CLUSTER, version)
9020 }
9021
9022 fn assert_reply(response: &WorkerResponse, phase: Phase, cluster: &str, version: u64) {
9023 assert_eq!(response.status, ResponseStatus::Ok as i32, "{response:?}");
9024 let Some(ContentType::ProtectedHealthCheckReply(reply)) = response
9025 .content
9026 .as_ref()
9027 .and_then(|content| content.content_type.as_ref())
9028 else {
9029 panic!("the answer must be a typed protected reply: {response:?}");
9030 };
9031 assert_eq!(reply.phase, phase as i32);
9032 assert_eq!(reply.cluster_id, cluster);
9033 assert_eq!(reply.version, version);
9034 }
9035
9036 fn assert_refused(response: &WorkerResponse, id: &str) {
9037 assert_eq!(response.id, id);
9038 assert_eq!(
9039 response.status,
9040 ResponseStatus::Failure as i32,
9041 "{response:?}"
9042 );
9043 assert!(response.content.is_none(), "{response:?}");
9044 }
9045
9046 fn assert_no_secret(responses: &[WorkerResponse]) {
9047 let shown = format!("{responses:?}");
9048 assert!(!shown.contains(HEADER), "{shown}");
9049 assert!(!shown.contains("synthetic-bootstrap-secret"), "{shown}");
9050 }
9051
9052 fn bootstrap(
9058 script: Vec<WorkerRequest>,
9059 ) -> (Server, Rc<RefCell<BackendMap>>, Vec<WorkerResponse>) {
9060 let ServerParts {
9061 event_loop,
9062 sessions,
9063 pool,
9064 backends,
9065 server_scm_socket,
9066 server_config,
9067 ..
9068 } = prebuild_server(16, 16384, true).expect("could not prebuild a test server");
9069 let (mut peer, proxy_channel) =
9070 Channel::generate(1000, 10000).expect("could not generate a test channel");
9071 let initial_state = InitialState {
9072 requests: vec![request(
9073 "initial",
9074 RequestType::AddCluster(Cluster {
9075 cluster_id: CLUSTER.to_owned(),
9076 ..Default::default()
9077 }),
9078 )],
9079 };
9080
9081 let driver = thread::spawn(move || {
9082 let mut answers = Vec::with_capacity(script.len());
9083 for message in script {
9084 peer.write_message(&message)
9085 .map_err(|error| format!("peer write: {error}"))?;
9086 answers.push(
9087 peer.read_message_blocking_timeout(Some(DEADLINE))
9088 .map_err(|error| format!("peer read: {error}"))?,
9089 );
9090 }
9091 Ok::<_, String>(answers)
9092 });
9093
9094 let held = backends.to_owned();
9095 let server = Server::new(
9096 event_loop,
9097 proxy_channel,
9098 server_scm_socket,
9099 sessions,
9100 pool,
9101 backends,
9102 None,
9103 None,
9104 None,
9105 server_config,
9106 Some(initial_state),
9107 true,
9108 )
9109 .expect("the constructor must return once the Status arrives");
9110 let answers = driver
9111 .join()
9112 .expect("the peer must not panic")
9113 .expect("every request must be answered");
9114 (server, held, answers)
9115 }
9116
9117 fn assert_holds_the_values(backends: &Rc<RefCell<BackendMap>>) {
9118 let backends = backends.borrow();
9119 let config = backends
9120 .health_check_configs
9121 .get(CLUSTER)
9122 .expect("the commit must have installed the metadata");
9123 assert_eq!(config.headers.len(), 1);
9124 assert_eq!(config.headers[0].name, HEADER);
9125 let set = backends
9126 .protected_health_check_headers
9127 .get(CLUSTER)
9128 .expect("the commit must have installed the values");
9129 assert_eq!(set.len(), 1);
9130 assert_eq!(set.headers()[0].name(), HEADER);
9131 assert_eq!(set.headers()[0].value().expose_secret_bytes(), SECRET);
9132 }
9133
9134 fn assert_holds_nothing(backends: &Rc<RefCell<BackendMap>>) {
9135 let backends = backends.borrow();
9136 assert!(
9137 !backends
9138 .protected_health_check_headers
9139 .contains_key(CLUSTER)
9140 );
9141 assert!(!backends.health_check_configs.contains_key(CLUSTER));
9142 }
9143
9144 fn finish(mut script: Vec<WorkerRequest>, version: u64) -> Vec<WorkerRequest> {
9146 script.extend([
9147 prepare("p", version),
9148 commit("c", version),
9149 release("r", version),
9150 status("s"),
9151 ]);
9152 script
9153 }
9154
9155 #[test]
9156 fn initial_protected_bootstrap_installs_the_values_before_the_constructor_returns() {
9157 let (_server, backends, answers) = bootstrap(vec![
9158 capability("cap"),
9159 prepare("prep", 1),
9160 commit("com", 1),
9161 release("rel", 1),
9162 status("st"),
9163 ]);
9164 assert_reply(&answers[0], Phase::ProtectedHealthCheckCapability, "", 0);
9165 assert_reply(&answers[1], Phase::ProtectedHealthCheckPrepare, CLUSTER, 1);
9166 assert_reply(&answers[2], Phase::ProtectedHealthCheckCommit, CLUSTER, 1);
9167 assert_reply(&answers[3], Phase::ProtectedHealthCheckRelease, CLUSTER, 1);
9168 assert_eq!(answers[4], WorkerResponse::ok("st"));
9169 assert_eq!(answers.len(), 5);
9170 assert_holds_the_values(&backends);
9171 assert_no_secret(&answers);
9172 }
9173
9174 #[test]
9175 fn initial_protected_bootstrap_still_accepts_a_direct_status() {
9176 let (_server, backends, answers) = bootstrap(vec![status("st")]);
9177 assert_eq!(answers, vec![WorkerResponse::ok("st")]);
9178 assert_holds_nothing(&backends);
9179 }
9180
9181 #[test]
9182 fn initial_protected_bootstrap_refuses_every_out_of_order_request() {
9183 let cases: Vec<(&str, Vec<WorkerRequest>)> = vec![
9186 (
9187 "prepare before the capability",
9188 finish(vec![prepare("bad", 1), capability("cap")], 1),
9189 ),
9190 (
9191 "commit before the capability",
9192 finish(vec![commit("bad", 1), capability("cap")], 1),
9193 ),
9194 (
9195 "commit before prepare",
9196 finish(vec![capability("cap"), commit("bad", 1)], 1),
9197 ),
9198 (
9199 "release before commit",
9200 finish(vec![capability("cap"), release("bad", 1)], 1),
9201 ),
9202 (
9203 "foreign request after the capability",
9204 finish(vec![capability("cap"), foreign("bad")], 1),
9205 ),
9206 (
9207 "foreign request before the capability",
9208 finish(vec![foreign("bad"), capability("cap")], 1),
9209 ),
9210 (
9211 "repeated capability",
9212 finish(vec![capability("cap"), capability("bad")], 1),
9213 ),
9214 (
9215 "abort without a started cycle",
9216 finish(vec![capability("cap"), abort("bad", 1)], 1),
9217 ),
9218 ("commit of the wrong version", {
9219 let mut script = vec![capability("cap"), prepare("prep", 1)];
9220 script.push(commit("bad", 2));
9221 script.extend([commit("com", 1), release("rel", 1), status("st")]);
9222 script
9223 }),
9224 ("commit of the wrong cluster", {
9225 let mut script = vec![capability("cap"), prepare("prep", 1)];
9226 script.push(order(
9227 "bad",
9228 Phase::ProtectedHealthCheckCommit,
9229 "other-cluster",
9230 1,
9231 ));
9232 script.extend([commit("com", 1), release("rel", 1), status("st")]);
9233 script
9234 }),
9235 ("second prepare while one is staged", {
9236 let mut script = vec![capability("cap"), prepare("prep", 1)];
9237 script.push(prepare("bad", 2));
9238 script.extend([commit("com", 1), release("rel", 1), status("st")]);
9239 script
9240 }),
9241 ("repeated commit", {
9242 let mut script = vec![capability("cap"), prepare("prep", 1), commit("com", 1)];
9243 script.push(commit("bad", 1));
9244 script.extend([release("rel", 1), status("st")]);
9245 script
9246 }),
9247 ("release of the wrong version", {
9248 let mut script = vec![capability("cap"), prepare("prep", 1), commit("com", 1)];
9249 script.push(release("bad", 2));
9250 script.extend([release("rel", 1), status("st")]);
9251 script
9252 }),
9253 ("status while a prepare is outstanding", {
9254 let mut script = vec![capability("cap"), prepare("prep", 1)];
9255 script.push(status("bad"));
9256 script.extend([commit("com", 1), release("rel", 1), status("st")]);
9257 script
9258 }),
9259 ("status while a commit is outstanding", {
9260 let mut script = vec![capability("cap"), prepare("prep", 1), commit("com", 1)];
9261 script.push(status("bad"));
9262 script.extend([release("rel", 1), status("st")]);
9263 script
9264 }),
9265 ("stale prepare after a released cycle", {
9266 let mut script = vec![
9267 capability("cap"),
9268 prepare("prep", 2),
9269 commit("com", 2),
9270 release("rel", 2),
9271 ];
9272 script.push(prepare("bad", 1));
9273 script.push(status("st"));
9274 script
9275 }),
9276 ];
9277 for (name, script) in cases {
9278 let refused: Vec<usize> = script
9279 .iter()
9280 .enumerate()
9281 .filter(|(_, message)| message.id == "bad")
9282 .map(|(index, _)| index)
9283 .collect();
9284 assert_eq!(refused.len(), 1, "{name}: exactly one refused request");
9285 let (_server, backends, answers) = bootstrap(script);
9286 assert_refused(&answers[refused[0]], "bad");
9287 assert!(
9288 answers
9289 .iter()
9290 .enumerate()
9291 .all(|(index, answer)| index == refused[0]
9292 || answer.status == ResponseStatus::Ok as i32),
9293 "{name}: only the refused request fails: {answers:?}"
9294 );
9295 assert_holds_the_values(&backends);
9296 assert_no_secret(&answers);
9297 }
9298 }
9299
9300 #[test]
9301 fn initial_protected_bootstrap_forwards_a_handler_refusal_and_stays_blocked() {
9302 let unknown = order(
9303 "bad",
9304 Phase::ProtectedHealthCheckPrepare,
9305 "unknown-cluster",
9306 1,
9307 );
9308 let (_server, backends, answers) = bootstrap(finish(vec![capability("cap"), unknown], 1));
9309 assert_refused(&answers[1], "bad");
9310 assert_eq!(
9311 answers[1].message,
9312 "the cluster is not known to this worker"
9313 );
9314 assert_holds_the_values(&backends);
9315 assert_no_secret(&answers);
9316 }
9317
9318 #[test]
9319 fn initial_protected_bootstrap_abort_drops_the_staged_change() {
9320 let (_server, backends, answers) = bootstrap(vec![
9321 capability("cap"),
9322 prepare("prep", 1),
9323 abort("abort", 1),
9324 status("st"),
9325 ]);
9326 assert_reply(&answers[2], Phase::ProtectedHealthCheckAbort, CLUSTER, 1);
9327 assert_eq!(answers[3], WorkerResponse::ok("st"));
9328 assert_holds_nothing(&backends);
9329 assert_no_secret(&answers);
9330 }
9331
9332 #[test]
9333 fn initial_protected_bootstrap_abort_restores_an_applied_change() {
9334 let (_server, backends, answers) = bootstrap(vec![
9335 capability("cap"),
9336 prepare("prep", 1),
9337 commit("com", 1),
9338 abort("abort", 1),
9339 status("st"),
9340 ]);
9341 assert_reply(&answers[3], Phase::ProtectedHealthCheckAbort, CLUSTER, 1);
9342 assert_holds_nothing(&backends);
9343 assert_no_secret(&answers);
9344 }
9345
9346 #[test]
9347 fn initial_protected_bootstrap_refuses_a_mismatched_abort_then_completes() {
9348 let mut script = vec![capability("cap"), prepare("prep", 1)];
9349 script.push(abort("bad", 9));
9350 script.extend([abort("abort", 1), prepare("again", 2)]);
9351 script.extend([commit("com", 2), release("rel", 2), status("st")]);
9352 let (_server, backends, answers) = bootstrap(script);
9353 assert_refused(&answers[2], "bad");
9354 assert_reply(&answers[3], Phase::ProtectedHealthCheckAbort, CLUSTER, 1);
9355 assert_reply(&answers[6], Phase::ProtectedHealthCheckRelease, CLUSTER, 2);
9356 assert_holds_the_values(&backends);
9357 assert_no_secret(&answers);
9358 }
9359
9360 #[test]
9361 fn initial_protected_bootstrap_returns_an_error_when_the_channel_closes() {
9362 let ServerParts {
9363 event_loop,
9364 sessions,
9365 pool,
9366 backends,
9367 server_scm_socket,
9368 server_config,
9369 ..
9370 } = prebuild_server(16, 16384, true).expect("could not prebuild a test server");
9371 let (peer, proxy_channel) =
9372 Channel::generate(1000, 10000).expect("could not generate a test channel");
9373 drop(peer);
9374 let result = Server::new(
9375 event_loop,
9376 proxy_channel,
9377 server_scm_socket,
9378 sessions,
9379 pool,
9380 backends,
9381 None,
9382 None,
9383 None,
9384 server_config,
9385 None,
9386 true,
9387 );
9388 let Err(error) = result else {
9389 panic!("a closed channel must not release the constructor");
9390 };
9391 assert!(!error.to_string().is_empty());
9392 }
9393
9394 #[test]
9395 fn initial_status_ack_failure_fails_the_constructor() {
9396 let ServerParts {
9397 event_loop,
9398 sessions,
9399 pool,
9400 backends,
9401 server_scm_socket,
9402 server_config,
9403 ..
9404 } = prebuild_server(16, 16384, true).expect("could not prebuild a test server");
9405 let (mut peer, proxy_channel) =
9406 Channel::generate(1000, 10000).expect("could not generate a test channel");
9407 peer.write_message(&status("st"))
9410 .expect("the peer must queue the Status");
9411 drop(peer);
9412 let result = Server::new(
9413 event_loop,
9414 proxy_channel,
9415 server_scm_socket,
9416 sessions,
9417 pool,
9418 backends,
9419 None,
9420 None,
9421 None,
9422 server_config,
9423 None,
9424 true,
9425 );
9426 let Err(error) = result else {
9427 panic!("a lost final ACK must not release the constructor");
9428 };
9429 assert!(
9430 matches!(error, ServerError::WriteInitialAnswer(_)),
9431 "{error:?}"
9432 );
9433 }
9434}