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 tcp: Rc<RefCell<tcp::TcpProxy>>,
1405 udp: Rc<RefCell<udp::UdpProxy>>,
1406 zombie_check_interval: Duration,
1407}
1408
1409impl Server {
1410 pub fn try_new_from_config(
1411 worker_to_main_channel: ProxyChannel,
1412 worker_to_main_scm: ScmSocket,
1413 config: ServerConfig,
1414 initial_state: InitialState,
1415 expects_initial_status: bool,
1416 ) -> Result<Self, ServerError> {
1417 let event_loop = Poll::new().map_err(ServerError::CreatePoll)?;
1418 if let Some(cap) = config.basic_auth_max_credential_bytes {
1425 crate::protocol::mux::auth::set_max_decoded_credential_bytes(cap as usize);
1426 }
1427 #[cfg(all(target_os = "linux", feature = "splice"))]
1434 if let Some(cap) = config.splice_pipe_capacity_bytes {
1435 crate::splice::set_pipe_capacity(cap as usize);
1436 }
1437 let pool = Rc::new(RefCell::new(Pool::with_capacity(
1438 config.min_buffers as usize,
1439 config.max_buffers as usize,
1440 config.buffer_size as usize,
1441 )));
1442 let backends = Rc::new(RefCell::new(BackendMap::new()));
1443
1444 let sessions: Rc<RefCell<SessionManager>> = SessionManager::new(
1449 Slab::with_capacity(config.slab_capacity() as usize),
1450 config.max_connections as usize,
1451 config
1452 .max_connections_per_ip
1453 .unwrap_or(sozu_command::config::DEFAULT_MAX_CONNECTIONS_PER_IP),
1454 config
1455 .retry_after
1456 .unwrap_or(sozu_command::config::DEFAULT_RETRY_AFTER),
1457 config
1458 .max_connections_per_subnet
1459 .unwrap_or(sozu_command::config::DEFAULT_MAX_CONNECTIONS_PER_SUBNET),
1460 config
1461 .subnet_ipv4_prefix
1462 .unwrap_or(sozu_command::config::DEFAULT_SUBNET_IPV4_PREFIX),
1463 config
1464 .subnet_ipv6_prefix
1465 .unwrap_or(sozu_command::config::DEFAULT_SUBNET_IPV6_PREFIX),
1466 );
1467 sessions.borrow_mut().max_connection_attempts = match config.max_connection_attempts {
1471 Some(attempts)
1472 if sozu_command::config::validate_max_connection_attempts(attempts).is_ok() =>
1473 {
1474 attempts
1475 }
1476 Some(attempts) => {
1477 error!(
1478 "max_connection_attempts = {} is out of 1..=255, keeping {}",
1479 attempts,
1480 sozu_command::config::DEFAULT_MAX_CONNECTION_ATTEMPTS
1481 );
1482 sozu_command::config::DEFAULT_MAX_CONNECTION_ATTEMPTS
1483 }
1484 None => sozu_command::config::DEFAULT_MAX_CONNECTION_ATTEMPTS,
1485 };
1486 {
1487 let mut s = sessions.borrow_mut();
1488 let entry = s.slab.vacant_entry();
1489 trace!("taking token {:?} for channel", SessionToken(entry.key()));
1490 entry.insert(Rc::new(RefCell::new(ListenSession {
1491 protocol: Protocol::Channel,
1492 })));
1493 }
1494 {
1495 let mut s = sessions.borrow_mut();
1496 let entry = s.slab.vacant_entry();
1497 trace!("taking token {:?} for metrics", SessionToken(entry.key()));
1498 entry.insert(Rc::new(RefCell::new(ListenSession {
1499 protocol: Protocol::Timer,
1500 })));
1501 }
1502 {
1503 let mut s = sessions.borrow_mut();
1504 let entry = s.slab.vacant_entry();
1505 trace!("taking token {:?} for metrics", SessionToken(entry.key()));
1506 entry.insert(Rc::new(RefCell::new(ListenSession {
1507 protocol: Protocol::Metrics,
1508 })));
1509 }
1510
1511 Server::new(
1512 event_loop,
1513 worker_to_main_channel,
1514 worker_to_main_scm,
1515 sessions,
1516 pool,
1517 backends,
1518 None,
1519 None,
1520 None,
1521 config,
1522 Some(initial_state),
1523 expects_initial_status,
1524 )
1525 }
1526
1527 #[allow(clippy::too_many_arguments)]
1528 pub fn new(
1529 poll: Poll,
1530 mut channel: ProxyChannel,
1531 scm: ScmSocket,
1532 sessions: Rc<RefCell<SessionManager>>,
1533 pool: Rc<RefCell<Pool>>,
1534 backends: Rc<RefCell<BackendMap>>,
1535 http: Option<http::HttpProxy>,
1536 https: Option<https::HttpsProxy>,
1537 tcp: Option<tcp::TcpProxy>,
1538 server_config: ServerConfig,
1539 initial_state: Option<InitialState>,
1540 expects_initial_status: bool,
1541 ) -> Result<Self, ServerError> {
1542 FEATURES.with(|_features| {
1543 });
1545
1546 poll.registry()
1547 .register(
1548 &mut channel,
1549 Token(0),
1550 Interest::READABLE | Interest::WRITABLE,
1551 )
1552 .map_err(ServerError::RegisterChannel)?;
1553
1554 METRICS.with(|metrics| {
1555 if let Some(sock) = (*metrics.borrow_mut()).socket_mut() {
1556 poll.registry()
1557 .register(sock, Token(2), Interest::WRITABLE)
1558 .expect("should register the metrics socket");
1559 }
1560 });
1561
1562 let base_sessions_count = sessions.borrow().slab.len();
1563
1564 let http = Rc::new(RefCell::new(match http {
1565 Some(http) => http,
1566 None => {
1567 let registry = poll
1568 .registry()
1569 .try_clone()
1570 .map_err(ServerError::CloneRegistry)?;
1571
1572 http::HttpProxy::new(registry, sessions.clone(), pool.clone(), backends.clone())
1573 }
1574 }));
1575
1576 let https = Rc::new(RefCell::new(match https {
1577 Some(https) => https,
1578 None => {
1579 let registry = poll
1580 .registry()
1581 .try_clone()
1582 .map_err(ServerError::CloneRegistry)?;
1583
1584 https::HttpsProxy::new(registry, sessions.clone(), pool.clone(), backends.clone())
1585 }
1586 }));
1587
1588 let tcp = Rc::new(RefCell::new(match tcp {
1589 Some(tcp) => tcp,
1590 None => {
1591 let registry = poll
1592 .registry()
1593 .try_clone()
1594 .map_err(ServerError::CloneRegistry)?;
1595
1596 tcp::TcpProxy::new(registry, sessions.clone(), pool.clone(), backends.clone())
1597 }
1598 }));
1599
1600 let udp = Rc::new(RefCell::new({
1604 let registry = poll
1605 .registry()
1606 .try_clone()
1607 .map_err(ServerError::CloneRegistry)?;
1608
1609 udp::UdpProxy::new(
1610 registry,
1611 sessions.clone(),
1612 pool.clone(),
1613 backends.clone(),
1614 server_config.max_connections as usize,
1615 server_config.buffer_size as usize,
1616 )
1617 }));
1618
1619 let mut server = Server {
1620 accept_queue_timeout: Duration::from_secs(u64::from(
1621 server_config.accept_queue_timeout,
1622 )),
1623 accept_queue: VecDeque::new(),
1624 evict_on_queue_full: server_config.evict_on_queue_full.unwrap_or(false),
1625 accept_ready: HashSet::new(),
1626 backends,
1627 base_sessions_count,
1628 channel,
1629 config_state: ConfigState::new(),
1630 current_poll_errors: 0,
1631 health_checker: HealthChecker::new(),
1632 http,
1633 https,
1634 last_sessions_len: 0, last_shutting_down_message: None,
1636 last_zombie_check: Instant::now(), loop_start: Instant::now(), started_at: Instant::now(), last_saturation_tick: Instant::now(), max_poll_errors: 10000, pool,
1642 poll_timeout: Some(Duration::from_millis(1000)), poll,
1644 scm_listeners: None,
1645 scm,
1646 sessions,
1647 should_poll_at: None,
1648 shutting_down: None,
1649 closing_sessions: Vec::new(),
1650 protected_health_check: ProtectedWorker::default(),
1651 tcp,
1652 udp,
1653 zombie_check_interval: Duration::from_secs(u64::from(
1654 server_config.zombie_check_interval,
1655 )),
1656 };
1657
1658 info!("will try to receive listeners");
1679 server
1680 .scm
1681 .set_blocking(true)
1682 .map_err(|scm_err| ServerError::ScmSocket {
1683 msg: "Could not set the scm socket to blocking".to_string(),
1684 scm_err,
1685 })?;
1686 let listeners =
1687 server
1688 .scm
1689 .receive_listeners()
1690 .map_err(|scm_err| ServerError::ScmSocket {
1691 msg: "could not receive listeners from the scm socket".to_string(),
1692 scm_err,
1693 })?;
1694 server
1695 .scm
1696 .set_blocking(false)
1697 .map_err(|scm_err| ServerError::ScmSocket {
1698 msg: "Could not set the scm socket to unblocking".to_string(),
1699 scm_err,
1700 })?;
1701 info!("received listeners: {:?}", listeners);
1702 server.scm_listeners = Some(listeners);
1703
1704 if let Some(state) = initial_state {
1706 for request in state.requests {
1707 trace!("generating initial config request: {:#?}", request);
1708 server.notify_proxys(request);
1709 }
1710
1711 QUEUE.with(|queue| {
1713 (*queue.borrow_mut()).clear();
1714 });
1715 }
1716
1717 server.report_unclaimed_inherited_sockets();
1718
1719 if expects_initial_status {
1720 server.block_channel();
1724 server.await_initial_status()?;
1725 server.unblock_channel();
1726 }
1727
1728 Ok(server)
1729 }
1730
1731 fn await_initial_status(&mut self) -> Result<(), ServerError> {
1743 let mut stage = BootstrapStage::Start;
1744 loop {
1745 let WorkerRequest { id, mut content } = self
1746 .channel
1747 .read_message()
1748 .map_err(ServerError::ReadInitialMessage)?;
1749 debug!("got initial message {}", id);
1751
1752 let Some(request_type) = content.request_type.as_mut() else {
1753 self.answer_initial(worker_response_error(id, REFUSED_INITIAL_REQUEST))?;
1754 continue;
1755 };
1756
1757 if matches!(request_type, RequestType::Status(_)) {
1758 if !stage.accepts_status() {
1759 self.answer_initial(worker_response_error(id, REFUSED_INITIAL_REQUEST))?;
1760 continue;
1761 }
1762 self.answer_initial(WorkerResponse::ok(id))?;
1763 return Ok(());
1764 }
1765
1766 let Some(step) = stage.step(request_type) else {
1767 self.answer_initial(worker_response_error(id, REFUSED_INITIAL_REQUEST))?;
1768 continue;
1769 };
1770 let response = self.protected_health_check.handle(
1771 &id,
1772 request_type,
1773 &mut self.config_state,
1774 &mut self.backends.borrow_mut(),
1775 );
1776 match response {
1777 Some(response) if step.is_answered_by(&response) => {
1778 stage = step.next;
1779 self.answer_initial(response)?;
1780 }
1781 Some(response) if response.status != ResponseStatus::Ok as i32 => {
1783 self.answer_initial(response)?;
1784 }
1785 _ => {
1786 self.answer_initial(worker_response_error(id, REFUSED_INITIAL_REQUEST))?;
1787 }
1788 }
1789 }
1790 }
1791
1792 fn answer_initial(&mut self, response: WorkerResponse) -> Result<(), ServerError> {
1793 self.channel
1794 .write_message(&response)
1795 .map_err(ServerError::WriteInitialAnswer)
1796 }
1797
1798 pub fn run(&mut self) {
1800 let mut events = Events::with_capacity(1024); let mut event_owners: Vec<Option<Rc<RefCell<dyn ProxySession>>>> =
1805 Vec::with_capacity(events.capacity());
1806 self.last_sessions_len = self.sessions.borrow().slab.len();
1807
1808 self.last_zombie_check = Instant::now();
1809 self.loop_start = Instant::now();
1810
1811 loop {
1812 self.check_for_poll_errors();
1813
1814 let timeout = self.reset_loop_time_and_get_timeout();
1815
1816 match self.poll.poll(&mut events, timeout) {
1817 Ok(_) => self.current_poll_errors = 0,
1818 Err(error) => {
1819 error!("Error while polling events: {:?}", error);
1820 self.current_poll_errors += 1;
1821 continue;
1822 }
1823 }
1824
1825 let after_epoll = Instant::now();
1826 time!(
1827 names::event_loop::EPOLL_TIME,
1828 (after_epoll - self.loop_start).as_millis()
1829 );
1830 self.loop_start = after_epoll;
1831
1832 tick_metric_detail_leases(after_epoll);
1834
1835 self.send_queue();
1836
1837 self.snapshot_event_owners(events.iter().map(|event| event.token()), &mut event_owners);
1838 for (event, owner) in events.iter().zip(event_owners.drain(..)) {
1839 match event.token() {
1840 Token(0) => {
1842 if event.is_error() {
1843 error!("error reading from command channel");
1844 continue;
1845 }
1846 if event.is_read_closed() || event.is_write_closed() {
1847 error!("command channel was closed");
1848 return;
1849 }
1850 let ready = Ready::from(event);
1851 self.channel.handle_events(ready);
1852
1853 loop {
1855 QUEUE.with(|queue| {
1856 if !(*queue.borrow()).is_empty() {
1857 self.channel.interest.insert(Ready::WRITABLE);
1858 }
1859 });
1860
1861 if self.channel.readiness() == Ready::EMPTY {
1864 break;
1865 }
1866
1867 if self.read_channel_messages_and_notify() {
1869 return;
1870 }
1871
1872 QUEUE.with(|queue| {
1873 if !(*queue.borrow()).is_empty() {
1874 self.channel.interest.insert(Ready::WRITABLE);
1875 }
1876 });
1877
1878 self.send_queue();
1879 }
1880 }
1881 Token(1) => {
1883 while let Some(t) = TIMER.with(|timer| timer.borrow_mut().poll()) {
1884 self.timeout(t);
1885 }
1886 }
1887 Token(2) => METRICS.with(|metrics| {
1889 (*metrics.borrow_mut()).writable();
1890 }),
1891 token if self.health_checker.owns_token(token) => {
1894 self.health_checker.ready(token);
1895 }
1896 token if self.udp.borrow().health_owns_token(token) => {
1897 self.udp.borrow_mut().health_ready(token);
1898 }
1899 token => self.ready_if_still_owned(token, Ready::from(event), owner),
1900 }
1901 }
1902
1903 if let Some(t) = self.should_poll_at.as_ref()
1904 && *t <= Instant::now()
1905 {
1906 while let Some(t) = TIMER.with(|timer| timer.borrow_mut().poll()) {
1907 self.timeout(t);
1909 }
1910 }
1911 self.handle_remaining_readiness();
1912 self.create_sessions();
1913
1914 self.should_poll_at = TIMER.with(|timer| timer.borrow().next_poll_date());
1915
1916 self.zombie_check();
1917 self.health_checker
1918 .poll(&self.backends, self.poll.registry());
1919 self.udp.borrow_mut().health_poll();
1923
1924 {
1937 let sessions = self.sessions.borrow();
1938 let nb_connections = sessions.nb_connections;
1939 let max_connections = sessions.max_connections;
1940 let slab_len = sessions.slab.len();
1941 let slab_capacity = sessions.slab.capacity();
1942 let accept_threshold = sessions.accept_slab_threshold();
1943
1944 gauge!(names::client::CONNECTIONS, nb_connections);
1945 gauge!(names::client::CONNECTIONS_MAX, max_connections);
1946 if let Some(percent) = (nb_connections * 100).checked_div(max_connections) {
1947 gauge!(names::client::CONNECTIONS_PERCENT, percent);
1948 }
1949
1950 gauge!(names::slab::ENTRIES, slab_len);
1951 gauge!(names::slab::CAPACITY, slab_capacity);
1952 if let Some(percent) = (slab_len * 100).checked_div(slab_capacity) {
1953 gauge!(names::slab::USAGE_PERCENT, percent);
1954 }
1955 if let Some(percent) = (slab_len * 100).checked_div(accept_threshold) {
1956 gauge!(names::slab::ACCEPT_THRESHOLD_PERCENT, percent);
1957 }
1958 }
1959 {
1965 let pool = self.pool.borrow();
1966 let used = pool.inner.used();
1967 let capacity = pool.inner.capacity();
1968 gauge!(names::buffer::IN_USE, used);
1969 gauge!(names::buffer::CAPACITY, capacity);
1970 if let Some(percent) = (used * 100).checked_div(capacity) {
1971 gauge!(names::buffer::USAGE_PERCENT, percent);
1972 }
1973 }
1974 let now = Instant::now();
1982 if now.duration_since(self.last_saturation_tick) >= ACCEPT_SATURATION_TICK {
1983 if !self.sessions.borrow().can_accept {
1984 incr!(names::accept_queue::SATURATED_SECONDS);
1985 }
1986 self.last_saturation_tick = now;
1987 }
1988 gauge!(
1992 names::process::UPTIME_SECONDS,
1993 self.started_at.elapsed().as_secs() as usize
1994 );
1995 gauge!(
2000 names::server::LIVE,
2001 if self.shutting_down.is_some() { 0 } else { 1 }
2002 );
2003 METRICS.with(|metrics| {
2004 (*metrics.borrow_mut()).send_data();
2005 });
2006
2007 if self.shutting_down.is_some() && self.shut_down_sessions() {
2008 return;
2009 }
2010 if !self.closing_sessions.is_empty() {
2011 self.drive_closing_sessions();
2012 }
2013 }
2014 }
2015
2016 fn check_for_poll_errors(&mut self) {
2017 if self.current_poll_errors >= self.max_poll_errors {
2018 error!(
2019 "Something is going very wrong. Last {} poll() calls failed, crashing..",
2020 self.current_poll_errors
2021 );
2022 panic!(
2023 "poll() calls failed {} times in a row",
2024 self.current_poll_errors
2025 );
2026 }
2027 }
2028
2029 fn reset_loop_time_and_get_timeout(&mut self) -> Option<Duration> {
2030 let now = Instant::now();
2031 time!(
2032 names::event_loop::EVENT_LOOP_TIME,
2033 (now - self.loop_start).as_millis()
2034 );
2035
2036 let mut timeout = match self.should_poll_at.as_ref() {
2037 None => self.poll_timeout,
2038 Some(i) => {
2039 if *i <= now {
2040 self.poll_timeout
2041 } else {
2042 let dur = *i - now;
2043 match self.poll_timeout {
2044 None => Some(dur),
2045 Some(t) => {
2046 if t < dur {
2047 Some(t)
2048 } else {
2049 Some(dur)
2050 }
2051 }
2052 }
2053 }
2054 }
2055 };
2056
2057 if self.shutting_down.is_some() || !self.closing_sessions.is_empty() {
2061 let shutdown_tick = Duration::from_millis(100);
2062 timeout = match timeout {
2063 None => Some(shutdown_tick),
2064 Some(current) => Some(current.min(shutdown_tick)),
2065 };
2066 }
2067
2068 self.loop_start = now;
2069 timeout
2070 }
2071
2072 fn read_channel_messages_and_notify(&mut self) -> bool {
2074 if !self.channel.readiness().is_readable() {
2075 return false;
2076 }
2077
2078 if let Err(e) = self.channel.readable() {
2079 error!("error reading from channel: {:?}", e);
2080 }
2081
2082 loop {
2083 let request = self.channel.read_message();
2084 debug!("Received request {:?}", request);
2085 match request {
2086 Ok(request) => match request.content.request_type {
2087 Some(RequestType::HardStop(_)) => {
2088 let req_id = request.id.clone();
2089 self.notify(request);
2090 if let Err(e) = self.channel.write_message(&WorkerResponse::ok(req_id)) {
2091 error!("Could not send ok response to the main process: {}", e);
2092 }
2093 if let Err(e) = self.channel.run() {
2094 error!("Error while running the server channel: {}", e);
2095 }
2096 return true;
2097 }
2098 Some(RequestType::SoftStop(_)) => {
2099 self.shutting_down = Some(request.id.clone());
2100 self.last_sessions_len = self.sessions.borrow().slab.len();
2101 self.notify(request);
2102 }
2103 Some(RequestType::ReturnListenSockets(_)) => {
2104 info!("received ReturnListenSockets order");
2105 match self.return_listen_sockets() {
2106 Ok(_) => push_queue(WorkerResponse::ok(request.id)),
2107 Err(error) => push_queue(worker_response_error(
2108 request.id,
2109 format!("Could not send listeners on scm socket: {error:?}"),
2110 )),
2111 }
2112 }
2113 _ => self.notify(request),
2114 },
2115 Err(_) => {
2117 if (self.channel.interest & self.channel.readiness).is_readable() {
2119 if let Err(e) = self.channel.readable() {
2120 error!("error reading from channel: {:?}", e);
2121 }
2122 continue;
2123 }
2124 break;
2125 }
2126 }
2127 }
2128 false
2129 }
2130
2131 fn zombie_check(&mut self) {
2133 let now = Instant::now();
2134 if now - self.last_zombie_check < self.zombie_check_interval {
2135 return;
2136 }
2137 info!("zombie check");
2138 debug_assert!(
2141 now >= self.last_zombie_check,
2142 "zombie-check timestamp must never move backwards"
2143 );
2144 self.last_zombie_check = now;
2145
2146 let mut zombie_tokens = HashSet::new();
2147
2148 for (_index, session) in self
2150 .sessions
2151 .borrow_mut()
2152 .slab
2153 .iter_mut()
2154 .filter(|(_, c)| now - c.borrow().last_event() > self.zombie_check_interval)
2155 {
2156 let session_token = session.borrow().frontend_token();
2157 if !zombie_tokens.contains(&session_token) {
2158 session.borrow().print_session();
2159 zombie_tokens.insert(session_token);
2160 }
2161 }
2162
2163 debug_assert!(
2168 !self.sessions.borrow().slab.iter().any(|(_, session)| {
2169 let s = session.borrow();
2170 zombie_tokens.contains(&s.frontend_token())
2171 && matches!(
2172 s.protocol(),
2173 Protocol::HTTPListen | Protocol::HTTPSListen | Protocol::TCPListen
2174 )
2175 }),
2176 "zombie reaping must never target a listener session"
2177 );
2178
2179 let zombie_count = zombie_tokens.len() as i64;
2180 count!(names::misc::ZOMBIES, zombie_count);
2181
2182 let remaining_count = self.shut_down_sessions_by_frontend_tokens(zombie_tokens);
2183 info!(
2184 "removing {} zombies ({} remaining entries after close)",
2185 zombie_count, remaining_count
2186 );
2187 }
2188
2189 fn shut_down_sessions_by_frontend_tokens(&self, tokens: HashSet<Token>) -> usize {
2192 if tokens.is_empty() {
2193 return 0;
2194 }
2195
2196 for token in &tokens {
2198 if self.sessions.borrow().slab.contains(token.0) {
2199 let slab_before = self.sessions.borrow().slab.len();
2200 let session = { self.sessions.borrow_mut().slab.remove(token.0) };
2201 session.borrow_mut().close();
2202 self.sessions.borrow_mut().decr();
2203 debug_assert!(
2208 !self.sessions.borrow().slab.contains(token.0),
2209 "removed token must be absent from the slab"
2210 );
2211 debug_assert!(
2212 self.sessions.borrow().slab.len() < slab_before,
2213 "removing a present session must free at least its own slab slot"
2214 );
2215 }
2216 }
2217
2218 let mut dangling_entries = HashSet::new();
2220 for (entry_key, session) in &self.sessions.borrow().slab {
2221 if tokens.contains(&session.borrow().frontend_token()) {
2222 dangling_entries.insert(entry_key);
2223 }
2224 }
2225
2226 let mut dangling_entries_count = 0;
2228 for entry_key in dangling_entries {
2229 let mut sessions = self.sessions.borrow_mut();
2230 if sessions.slab.contains(entry_key) {
2231 sessions.slab.remove(entry_key);
2232 dangling_entries_count += 1;
2233 }
2234 }
2235 debug_assert!(
2239 !self
2240 .sessions
2241 .borrow()
2242 .slab
2243 .iter()
2244 .any(|(_, session)| tokens.contains(&session.borrow().frontend_token())),
2245 "no slab entry may reference a closed frontend token after teardown"
2246 );
2247 dangling_entries_count
2248 }
2249
2250 fn shut_down_sessions(&mut self) -> bool {
2252 let sessions_count = self.sessions.borrow().slab.len();
2253 let mut sessions_to_shut_down = HashSet::new();
2254
2255 for (_key, session) in &self.sessions.borrow().slab {
2256 let mut session = session.borrow_mut();
2257 if session.shutting_down() {
2258 debug!(
2259 "Server killing session from shutting_down: token={:?}, protocol={:?}",
2260 session.frontend_token(),
2261 session.protocol()
2262 );
2263 sessions_to_shut_down.insert(Token(session.frontend_token().0));
2264 }
2265 }
2266 let _ = self.shut_down_sessions_by_frontend_tokens(sessions_to_shut_down);
2267
2268 let new_sessions_count = self.sessions.borrow().slab.len();
2269
2270 if new_sessions_count < sessions_count {
2271 let now = Instant::now();
2272 if let Some(last) = self.last_shutting_down_message
2273 && (now - last) > Duration::from_secs(5)
2274 {
2275 info!(
2276 "closed {} sessions, {} sessions left, base_sessions_count = {}",
2277 sessions_count - new_sessions_count,
2278 new_sessions_count,
2279 self.base_sessions_count
2280 );
2281 }
2282 self.last_shutting_down_message = Some(now);
2283 }
2284
2285 if new_sessions_count <= self.base_sessions_count {
2286 info!("last session stopped, shutting down!");
2287 if let Err(e) = self.channel.run() {
2288 error!("Error while running the server channel: {}", e);
2289 }
2290 let id = self
2292 .shutting_down
2293 .take()
2294 .expect("should have shut down correctly"); debug!("Responding OK to main process for request {}", id);
2297
2298 let proxy_response = WorkerResponse::ok(id);
2299 if let Err(e) = self.channel.write_message(&proxy_response) {
2300 error!("Could not write response to the main process: {}", e);
2301 }
2302 if let Err(e) = self.channel.run() {
2303 error!("Error while running the server channel: {}", e);
2304 }
2305 return true;
2306 }
2307
2308 if new_sessions_count < self.last_sessions_len {
2309 info!(
2310 "shutting down, {} slab elements remaining (base: {})",
2311 new_sessions_count - self.base_sessions_count,
2312 self.base_sessions_count
2313 );
2314 self.last_sessions_len = new_sessions_count;
2315 }
2316
2317 false
2318 }
2319
2320 fn close_sessions(&mut self, target: &SessionTarget) -> ClosedSessions {
2326 let mut targeted: HashMap<Token, Rc<RefCell<dyn ProxySession>>> = HashMap::new();
2329 for (_key, session) in &self.sessions.borrow().slab {
2330 let borrowed = session.borrow();
2331 if borrowed.targets(target) {
2332 targeted
2333 .entry(borrowed.frontend_token())
2334 .or_insert_with(|| session.clone());
2335 }
2336 }
2337
2338 let mut to_close = HashSet::new();
2339 let mut draining = 0u64;
2340 for (token, session) in targeted {
2341 if session.borrow_mut().shutting_down() {
2342 to_close.insert(token);
2343 } else {
2344 draining += 1;
2345 let already_closing = self.closing_sessions.iter().any(|(_, pending)| {
2346 pending.upgrade().is_some_and(|p| Rc::ptr_eq(&p, &session))
2347 });
2348 if !already_closing {
2349 self.closing_sessions.push((token, Rc::downgrade(&session)));
2350 }
2351 }
2352 }
2353 let closed = to_close.len() as u64;
2354 self.shut_down_sessions_by_frontend_tokens(to_close);
2355
2356 debug_assert!(
2358 self.closing_sessions.len() as u64 >= draining,
2359 "every draining session must be watched by drive_closing_sessions"
2360 );
2361 ClosedSessions { closed, draining }
2362 }
2363
2364 fn drive_closing_sessions(&mut self) {
2368 let pending = std::mem::take(&mut self.closing_sessions);
2369 let mut to_close = HashSet::new();
2370 for (token, weak) in pending {
2371 let Some(session) = weak.upgrade() else {
2372 continue;
2373 };
2374 let still_there = self
2376 .sessions
2377 .borrow()
2378 .slab
2379 .get(token.0)
2380 .is_some_and(|entry| Rc::ptr_eq(entry, &session));
2381 if !still_there {
2382 continue;
2383 }
2384 if session.borrow_mut().shutting_down() {
2385 to_close.insert(token);
2386 } else {
2387 self.closing_sessions.push((token, weak));
2388 }
2389 }
2390 debug_assert!(
2392 self.closing_sessions
2393 .iter()
2394 .all(|(token, _)| !to_close.contains(token)),
2395 "a session closed this tick must not stay watched"
2396 );
2397 self.shut_down_sessions_by_frontend_tokens(to_close);
2398 }
2399
2400 fn kill_session(&self, session: Rc<RefCell<dyn ProxySession>>) {
2401 let token = session.borrow().frontend_token();
2402 let _ = self.shut_down_sessions_by_frontend_tokens(HashSet::from([token]));
2403 }
2404
2405 fn send_queue(&mut self) {
2406 if self.channel.readiness.is_writable() {
2407 QUEUE.with(|q| {
2408 let mut queue = q.borrow_mut();
2409 loop {
2410 if let Some(resp) = queue.pop_front() {
2411 debug!("Sending response {:?}", resp);
2412 if let Err(e) = self.channel.write_message(&resp) {
2413 error!("Could not write message {} on the channel: {}", resp, e);
2414 queue.push_front(resp);
2415 }
2416 }
2417
2418 if self.channel.back_buf.available_data() > 0
2419 && let Err(e) = self.channel.writable()
2420 {
2421 error!("error writing to channel: {:?}", e);
2422 }
2423
2424 if !self.channel.readiness.is_writable() {
2425 break;
2426 }
2427
2428 if self.channel.back_buf.available_data() == 0 && queue.is_empty() {
2429 break;
2430 }
2431 }
2432 });
2433 }
2434 }
2435
2436 fn notify(&mut self, message: WorkerRequest) {
2437 let now = std::time::Instant::now();
2443 tick_metric_detail_leases(now);
2444 match &message.content.request_type {
2445 Some(RequestType::ConfigureMetrics(configuration)) => {
2446 match MetricsConfiguration::try_from(*configuration) {
2447 Ok(metrics_config) => {
2448 METRICS.with(|metrics| {
2449 (*metrics.borrow_mut()).configure(&metrics_config);
2450 push_queue(WorkerResponse::ok(message.id));
2451 });
2452 }
2453 Err(e) => {
2454 error!("Error configuring metrics: {}", e);
2455 push_queue(WorkerResponse::error(message.id, e));
2456 }
2457 }
2458 return;
2459 }
2460 Some(RequestType::QueryMetrics(query_metrics_options)) => {
2461 METRICS.with(|metrics| {
2462 match (*metrics.borrow_mut()).query(query_metrics_options) {
2463 Ok(c) => push_queue(WorkerResponse::ok_with_content(message.id, c)),
2464 Err(e) => {
2465 error!("Error querying metrics: {}", e);
2466 push_queue(WorkerResponse::error(message.id, e))
2467 }
2468 }
2469 });
2470 return;
2471 }
2472 Some(RequestType::SetMetricDetail(req)) => {
2480 let presented_binding = crate::metrics::PeerBinding {
2487 pid: req.peer_pid,
2488 session_ulid: req.peer_session_ulid.as_deref().and_then(|s| {
2494 rusty_ulid::Ulid::from_str(s)
2495 .map(u128::from)
2496 .ok()
2497 .or_else(|| u128::from_str_radix(s.trim_start_matches("0x"), 16).ok())
2498 }),
2499 };
2500 if req.clear.unwrap_or(false) {
2501 if req.client_id.len() > crate::metrics::LEASE_CLIENT_ID_MAX_BYTES {
2512 let msg = format!(
2513 "SetMetricDetail: clear client_id length {} exceeds {} bytes",
2514 req.client_id.len(),
2515 crate::metrics::LEASE_CLIENT_ID_MAX_BYTES,
2516 );
2517 error!("{}", msg);
2518 push_queue(WorkerResponse::error(message.id.clone(), msg));
2519 return;
2520 }
2521 let (outcome, effective_after, configured_after, lease_count_after) = METRICS
2529 .with(|metrics| {
2530 let mut m = metrics.borrow_mut();
2531 let outcome = m.lease_clear(&req.client_id, presented_binding);
2532 (
2533 outcome,
2534 m.detail_effective(),
2535 m.detail_configured(),
2536 m.lease_count(),
2537 )
2538 });
2539 match outcome {
2540 crate::metrics::LeaseClearOutcome::Cleared { previous_effective } => {
2541 push_metric_detail_transition(
2542 previous_effective,
2543 effective_after,
2544 "lease_clear",
2545 Some(req.client_id.clone()),
2546 );
2547 push_queue(WorkerResponse::ok_with_content(
2548 message.id.clone(),
2549 worker_metric_detail_status_content(
2550 configured_after,
2551 effective_after,
2552 previous_effective,
2553 lease_count_after,
2554 ),
2555 ));
2556 }
2557 crate::metrics::LeaseClearOutcome::NotFound => {
2558 push_queue(WorkerResponse::ok_with_content(
2562 message.id.clone(),
2563 worker_metric_detail_status_content(
2564 configured_after,
2565 effective_after,
2566 effective_after,
2567 lease_count_after,
2568 ),
2569 ));
2570 }
2571 crate::metrics::LeaseClearOutcome::Unauthorized => {
2572 let msg = "SetMetricDetail: clear refused (peer \
2583 binding does not match the apply-time owner)"
2584 .to_owned();
2585 error!("{}", msg);
2586 push_queue(WorkerResponse::error(message.id.clone(), msg));
2587 }
2588 }
2589 return;
2590 }
2591 let detail_proto = match req.detail {
2592 Some(d) => d,
2593 None => {
2594 let msg = "SetMetricDetail without `detail` and without `clear`".to_owned();
2601 error!("{}", msg);
2602 push_queue(WorkerResponse::error(message.id.clone(), msg));
2603 return;
2604 }
2605 };
2606 let detail_enum = match MetricDetail::try_from(detail_proto) {
2607 Ok(d) => d,
2608 Err(e) => {
2609 let msg =
2610 format!("SetMetricDetail: invalid MetricDetail variant {detail_proto}");
2611 error!("{}: {}", msg, e);
2612 push_queue(WorkerResponse::error(message.id.clone(), msg));
2613 return;
2614 }
2615 };
2616 let level = MetricDetailLevel::from(detail_enum);
2617 if let Some(t) = req.ttl_seconds
2626 && u64::from(t) > crate::metrics::LEASE_TTL_MAX.as_secs()
2627 {
2628 let msg = format!(
2629 "SetMetricDetail: ttl_seconds={t} exceeds LEASE_TTL_MAX={}",
2630 crate::metrics::LEASE_TTL_MAX.as_secs()
2631 );
2632 error!("{}", msg);
2633 push_queue(WorkerResponse::error(message.id.clone(), msg));
2634 return;
2635 }
2636 let ttl_seconds = req.ttl_seconds.filter(|&t| t > 0).unwrap_or_else(|| {
2637 u32::try_from(crate::metrics::LEASE_TTL_DEFAULT.as_secs()).unwrap_or(60)
2644 });
2645 let ttl = std::time::Duration::from_secs(ttl_seconds.into());
2646 let (outcome, configured_after, lease_count_after) = METRICS.with(|metrics| {
2647 let mut m = metrics.borrow_mut();
2648 let outcome =
2649 m.lease_apply(req.client_id.clone(), level, ttl, presented_binding, now);
2650 (outcome, m.detail_configured(), m.lease_count())
2651 });
2652 match outcome {
2653 crate::metrics::LeaseApplyOutcome::Applied {
2654 previous_effective,
2655 new_effective,
2656 } => {
2657 push_metric_detail_transition(
2658 previous_effective,
2659 new_effective,
2660 "lease_apply",
2661 Some(req.client_id.clone()),
2662 );
2663 push_queue(WorkerResponse::ok_with_content(
2664 message.id.clone(),
2665 worker_metric_detail_status_content(
2666 configured_after,
2667 new_effective,
2668 previous_effective,
2669 lease_count_after,
2670 ),
2671 ));
2672 }
2673 crate::metrics::LeaseApplyOutcome::ClientIdTooLong => {
2674 let msg = format!(
2675 "SetMetricDetail: client_id length {} exceeds {} bytes",
2676 req.client_id.len(),
2677 crate::metrics::LEASE_CLIENT_ID_MAX_BYTES,
2678 );
2679 error!("{}", msg);
2680 push_queue(WorkerResponse::error(message.id.clone(), msg));
2681 }
2682 crate::metrics::LeaseApplyOutcome::TableFull => {
2683 let msg = format!(
2690 "SetMetricDetail: lease table at capacity ({} entries); reject new \
2691 apply — operators must retry after an active lease expires or is \
2692 cleared",
2693 crate::metrics::LEASE_TABLE_CAP,
2694 );
2695 error!("{}", msg);
2696 push_queue(WorkerResponse::error(message.id.clone(), msg));
2697 }
2698 crate::metrics::LeaseApplyOutcome::TtlOutOfRange => {
2699 let msg = format!(
2705 "SetMetricDetail: ttl exceeds LEASE_TTL_MAX={} (internal contract \
2706 violation: dispatch gate should have rejected)",
2707 crate::metrics::LEASE_TTL_MAX.as_secs(),
2708 );
2709 error!("{}", msg);
2710 push_queue(WorkerResponse::error(message.id.clone(), msg));
2711 }
2712 crate::metrics::LeaseApplyOutcome::Unauthorized => {
2713 let msg = "SetMetricDetail: renewal refused (peer binding does not \
2722 match the apply-time owner)"
2723 .to_owned();
2724 error!("{}", msg);
2725 push_queue(WorkerResponse::error(message.id.clone(), msg));
2726 }
2727 }
2728 return;
2729 }
2730 Some(RequestType::Logging(logging_filter)) => {
2731 info!(
2732 "{} changing logging filter to {}",
2733 message.id, logging_filter
2734 );
2735 let (directives, _errors) = logging::parse_logging_spec(logging_filter);
2737 logging::LOGGER.with(|logger| {
2738 logger.borrow_mut().set_directives(directives);
2739 });
2740 push_queue(WorkerResponse::ok(message.id));
2741 return;
2742 }
2743 Some(RequestType::QueryClustersHashes(_)) => {
2744 push_queue(WorkerResponse::ok_with_content(
2745 message.id.clone(),
2746 ContentType::ClusterHashes(ClusterHashes {
2747 map: self
2748 .config_state
2749 .hash_state()
2750 .into_iter()
2751 .map(|(cluster_id, hash)| (cluster_id.to_string(), hash))
2752 .collect(),
2753 })
2754 .into(),
2755 ));
2756 return;
2757 }
2758 Some(RequestType::QueryClusterById(cluster_id)) => {
2759 push_queue(WorkerResponse::ok_with_content(
2760 message.id.clone(),
2761 ContentType::Clusters(ClusterInformations {
2762 vec: self
2763 .config_state
2764 .cluster_state(cluster_id)
2765 .map_or(vec![], |ci| vec![ci]),
2766 })
2767 .into(),
2768 ));
2769 }
2770 Some(RequestType::SetMaxConnectionsPerIp(limit)) => {
2771 let mut sessions = self.sessions.borrow_mut();
2772 let previous = sessions.max_connections_per_ip;
2773 sessions.max_connections_per_ip = *limit;
2774 if *limit == 0 {
2779 sessions.clear_cluster_ip_tracking();
2780 }
2781 info!(
2782 "{} updated global max_connections_per_ip from {} to {}",
2783 message.id, previous, limit
2784 );
2785 push_queue(WorkerResponse::ok(message.id));
2786 return;
2787 }
2788 Some(RequestType::QueryMaxConnectionsPerIp(_)) => {
2789 let limit = self.sessions.borrow().max_connections_per_ip;
2790 push_queue(WorkerResponse::ok_with_content(
2791 message.id,
2792 ContentType::MaxConnectionsPerIpLimit(
2793 sozu_command::proto::command::MaxConnectionsPerIpLimit { limit },
2794 )
2795 .into(),
2796 ));
2797 return;
2798 }
2799 Some(RequestType::SetMaxConnectionsPerSubnet(limit)) => {
2800 let mut sessions = self.sessions.borrow_mut();
2801 let previous = sessions.max_connections_per_subnet;
2802 sessions.max_connections_per_subnet = *limit;
2803 if *limit == 0 {
2809 sessions.clear_cluster_subnet_tracking();
2810 }
2811 info!(
2812 "{} updated global max_connections_per_subnet from {} to {}",
2813 message.id, previous, limit
2814 );
2815 push_queue(WorkerResponse::ok(message.id));
2816 return;
2817 }
2818 Some(RequestType::SetMaxConnectionAttempts(attempts)) => {
2819 if let Err(reason) =
2820 sozu_command::config::validate_max_connection_attempts(*attempts)
2821 {
2822 push_queue(WorkerResponse::error(message.id, reason));
2823 return;
2824 }
2825 let mut sessions = self.sessions.borrow_mut();
2826 let previous = sessions.max_connection_attempts;
2827 sessions.max_connection_attempts = *attempts;
2828 info!(
2829 "{} updated global max_connection_attempts from {} to {}",
2830 message.id, previous, attempts
2831 );
2832 push_queue(WorkerResponse::ok(message.id));
2833 return;
2834 }
2835 Some(RequestType::CloseSessions(request)) => {
2836 let target = SessionTarget::from(request);
2837 let closed = self.close_sessions(&target);
2838 info!(
2839 "{} CloseSessions on {:?}: {} sessions closed, {} draining",
2840 message.id, target, closed.closed, closed.draining
2841 );
2842 push_queue(WorkerResponse::ok_with_content(
2843 message.id,
2844 ContentType::ClosedSessions(closed).into(),
2845 ));
2846 return;
2847 }
2848 Some(RequestType::QueryMaxConnectionsPerSubnet(_)) => {
2849 let sessions = self.sessions.borrow();
2850 let limit = sessions.max_connections_per_subnet;
2851 let ipv4_prefix = sessions.subnet_ipv4_prefix;
2855 let ipv6_prefix = sessions.subnet_ipv6_prefix;
2856 drop(sessions);
2857 push_queue(WorkerResponse::ok_with_content(
2858 message.id,
2859 ContentType::MaxConnectionsPerSubnetLimit(
2860 sozu_command::proto::command::MaxConnectionsPerSubnetLimit {
2861 limit,
2862 ipv4_prefix,
2863 ipv6_prefix,
2864 },
2865 )
2866 .into(),
2867 ));
2868 return;
2869 }
2870 Some(RequestType::QueryClustersByDomain(domain)) => {
2871 let cluster_ids = self
2872 .config_state
2873 .get_cluster_ids_by_domain(domain.hostname.clone(), domain.path.clone());
2874 let vec = cluster_ids
2875 .iter()
2876 .filter_map(|cluster_id| self.config_state.cluster_state(cluster_id))
2877 .collect();
2878
2879 push_queue(WorkerResponse::ok_with_content(
2880 message.id.clone(),
2881 ContentType::Clusters(ClusterInformations { vec }).into(),
2882 ));
2883 return;
2884 }
2885 Some(RequestType::QueryCertificatesFromWorkers(filters))
2896 if filters.fingerprint.is_some() =>
2897 {
2898 let certs = self.config_state.get_certificates(filters.clone());
2899 let response = if !certs.is_empty() {
2900 WorkerResponse::ok_with_content(
2901 message.id.clone(),
2902 ContentType::CertificatesWithFingerprints(CertificatesWithFingerprints {
2903 certs,
2904 })
2905 .into(),
2906 )
2907 } else {
2908 worker_response_error(
2909 message.id.clone(),
2910 "Could not find certificate for this fingerprint",
2911 )
2912 };
2913 push_queue(response);
2914 return;
2915 }
2916 _other_request => {}
2919 }
2920 self.notify_proxys(message);
2921 }
2922
2923 pub fn notify_proxys(&mut self, mut request: WorkerRequest) {
2924 if matches!(
2928 request.content.request_type,
2929 Some(RequestType::SetProtectedHealthCheck(_))
2930 ) {
2931 push_queue(worker_response_error(
2932 request.id,
2933 "SetProtectedHealthCheck is not a worker order",
2934 ));
2935 return;
2936 }
2937 if let Some(request_type) = request.content.request_type.as_mut()
2938 && let Some(response) = self.protected_health_check.handle(
2939 &request.id,
2940 request_type,
2941 &mut self.config_state,
2942 &mut self.backends.borrow_mut(),
2943 )
2944 {
2945 push_queue(response);
2946 return;
2947 }
2948
2949 if let Some(reason) = request
2955 .content
2956 .request_type
2957 .as_ref()
2958 .and_then(|request_type| {
2959 self.config_state
2960 .health_check_port_unix_conflict(request_type)
2961 })
2962 {
2963 push_queue(worker_response_error(request.id, reason));
2964 return;
2965 }
2966
2967 let cluster_objects = match &request.content.request_type {
2971 Some(RequestType::RemoveCluster(cluster_id)) => {
2972 self.config_state.cluster_removal_requests(cluster_id)
2973 }
2974 _ => Vec::new(),
2975 };
2976
2977 let rollback_candidate = request
2981 .content
2982 .request_type
2983 .as_ref()
2984 .and_then(|request_type| self.config_state.rollback_for(request_type));
2985 let mut rollback = None;
2986 if let Err(e) = self.config_state.dispatch(&request.content) {
2987 if self
2988 .config_state
2989 .removed_with_its_cluster(&request.content, &e)
2990 {
2991 debug!("{} already removed with its cluster: {}", request.id, e);
2997 push_queue(WorkerResponse::ok(request.id));
2998 return;
2999 }
3000 if is_frontend_removal(&request.content)
3001 && matches!(e, StateError::NotFound { .. } | StateError::NoChange)
3002 {
3003 push_queue(WorkerResponse::error(
3013 request.id,
3014 format!("Could not execute order on config state: {e}"),
3015 ));
3016 return;
3017 }
3018 if matches!(e, StateError::ClusterNotFound { .. }) {
3019 push_queue(WorkerResponse::error(
3025 request.id,
3026 format!("Could not execute order on config state: {e}"),
3027 ));
3028 return;
3029 }
3030 error!("Could not execute order on config state: {}", e);
3031 } else {
3032 rollback = rollback_candidate;
3033 }
3034
3035 let req_id = request.id.clone();
3036
3037 match request.content.request_type {
3038 Some(RequestType::AddCluster(ref cluster)) => {
3039 if let Some(hc) = cluster.health_check.as_ref()
3047 && let Err(reason) = sozu_command::config::validate_health_check_config(hc)
3048 {
3049 push_queue(worker_response_error(req_id, reason));
3050 return;
3051 }
3052 if let Err(reason) = sozu_command::config::validate_shuffle_sharding(
3055 cluster.shard_percent,
3056 cluster.shard_min_backends,
3057 cluster.shard_mode,
3058 ) {
3059 push_queue(worker_response_error(req_id, reason));
3060 return;
3061 }
3062 if let Some(Err(reason)) = cluster
3063 .max_connection_attempts
3064 .map(sozu_command::config::validate_max_connection_attempts)
3065 {
3066 push_queue(worker_response_error(req_id, reason));
3067 return;
3068 }
3069 self.protected_health_check
3070 .forget_cluster(&cluster.cluster_id);
3071 self.add_cluster(cluster);
3072 if cluster.health_check.is_none() {
3076 self.health_checker.remove_cluster(&cluster.cluster_id);
3077 }
3078 METRICS.with(|metrics| {
3083 (*metrics.borrow_mut()).add_cluster(&cluster.cluster_id);
3084 });
3085 }
3087 Some(RequestType::RemoveCluster(ref cluster_id)) => {
3088 self.protected_health_check.forget_cluster(cluster_id);
3089 self.remove_cluster_objects(&req_id, cluster_id, cluster_objects);
3090 self.remove_health_check_state(cluster_id);
3091 METRICS.with(|metrics| {
3092 (*metrics.borrow_mut()).remove_cluster(cluster_id);
3093 });
3094 }
3096 Some(RequestType::SetHealthCheck(ref set)) => {
3097 if let Err(reason) = sozu_command::config::validate_health_check_config(&set.config)
3098 {
3099 push_queue(worker_response_error(req_id, reason));
3100 return;
3101 }
3102 self.protected_health_check.forget_cluster(&set.cluster_id);
3103 self.backends
3104 .borrow_mut()
3105 .set_health_check_config(&set.cluster_id, Some(set.config.to_owned()));
3106 push_queue(WorkerResponse::ok(req_id));
3107 return;
3108 }
3109 Some(RequestType::RemoveHealthCheck(ref cluster_id)) => {
3110 self.remove_health_check_state(cluster_id);
3111 push_queue(WorkerResponse::ok(req_id));
3112 return;
3113 }
3114 Some(RequestType::AddBackend(ref backend))
3115 | Some(RequestType::AddUnixBackend(ref backend)) => {
3116 push_queue(self.add_backend(&req_id, backend));
3117 return;
3118 }
3119 Some(RequestType::RemoveBackend(ref remove_backend)) => {
3120 push_queue(self.remove_backend(&req_id, remove_backend));
3121 return;
3122 }
3123 _ => {}
3124 };
3125
3126 let proxy_destinations = request.content.get_destinations();
3127 let mut notify_response = None;
3128 if proxy_destinations.to_http_proxy {
3129 notify_response = Some(self.http.borrow_mut().notify(request.clone()));
3130 }
3131 if proxy_destinations.to_https_proxy {
3132 let http_proxy_response = self.https.borrow_mut().notify(request.clone());
3133 if http_proxy_response.is_failure() || notify_response.is_none() {
3134 notify_response = Some(http_proxy_response);
3135 }
3136 }
3137 if proxy_destinations.to_tcp_proxy {
3138 let tcp_proxy_response = self.tcp.borrow_mut().notify(request.clone());
3139 if tcp_proxy_response.is_failure() || notify_response.is_none() {
3140 notify_response = Some(tcp_proxy_response);
3141 }
3142 }
3143 if proxy_destinations.to_udp_proxy {
3144 let udp_proxy_response = self.udp.borrow_mut().notify(request.clone());
3145 if udp_proxy_response.is_failure() || notify_response.is_none() {
3146 notify_response = Some(udp_proxy_response);
3147 }
3148 }
3149 if let Some(response) = notify_response {
3150 self.finish_proxy_order(&mut rollback, response);
3151 }
3152
3153 match request.content.request_type {
3154 Some(RequestType::AddHttpListener(listener)) => {
3156 let response = self.notify_add_http_listener(&req_id, listener);
3157 self.finish_proxy_order(&mut rollback, response);
3158 }
3159 Some(
3160 RequestType::AddHttpsListener(listener)
3161 | RequestType::AddHttpsListenerWithClientAuth(listener),
3162 ) => {
3163 let response = self.notify_add_https_listener(&req_id, listener);
3164 self.finish_proxy_order(&mut rollback, response);
3165 }
3166 Some(RequestType::AddTcpListener(listener)) => {
3167 let response = self.notify_add_tcp_listener(&req_id, listener);
3168 self.finish_proxy_order(&mut rollback, response);
3169 }
3170 Some(RequestType::AddUdpListener(listener)) => {
3171 let response = self.notify_add_udp_listener(&req_id, listener);
3172 self.finish_proxy_order(&mut rollback, response);
3173 }
3174 Some(RequestType::UpdateHttpListener(patch)) => {
3175 let response = self.notify_update_http_listener(&req_id, patch);
3176 self.finish_proxy_order(&mut rollback, response);
3177 }
3178 Some(
3179 RequestType::UpdateHttpsListener(patch)
3180 | RequestType::UpdateHttpsListenerWithClientAuth(patch),
3181 ) => {
3182 let response = self.notify_update_https_listener(&req_id, patch);
3183 self.finish_proxy_order(&mut rollback, response);
3184 }
3185 Some(RequestType::UpdateTcpListener(patch)) => {
3186 let response = self.notify_update_tcp_listener(&req_id, patch);
3187 self.finish_proxy_order(&mut rollback, response);
3188 }
3189 Some(RequestType::UpdateUdpListener(patch)) => {
3190 let response = self.notify_update_udp_listener(&req_id, patch);
3191 self.finish_proxy_order(&mut rollback, response);
3192 }
3193 Some(RequestType::RemoveListener(ref remove)) => {
3194 debug!("{} remove {:?} listener {:?}", req_id, remove.proxy, remove);
3195 debug_assert!(
3198 self.base_sessions_count > 0,
3199 "removing a listener with base_sessions_count == 0 would underflow"
3200 );
3201 self.base_sessions_count -= 1;
3202 let listener_type = ListenerType::try_from(remove.proxy);
3203 let address = remove.listener_key();
3211 let listen_token = match listener_type {
3212 Ok(ListenerType::Http) => self.http.borrow().listener_token(&address),
3213 Ok(ListenerType::Https) => self.https.borrow().listener_token(&address),
3214 Ok(ListenerType::Tcp) => self.tcp.borrow().listener_token(&address),
3215 Ok(ListenerType::Udp) => self.udp.borrow().listener_token(&address),
3216 Err(_) => None,
3217 };
3218 let response = match listener_type {
3219 Ok(ListenerType::Http) => self.http.borrow_mut().notify(request),
3220 Ok(ListenerType::Https) => self.https.borrow_mut().notify(request),
3221 Ok(ListenerType::Tcp) => self.tcp.borrow_mut().notify(request),
3222 Ok(ListenerType::Udp) => self.udp.borrow_mut().notify(request),
3223 Err(_) => WorkerResponse::error(req_id, "Wrong variant ListenerType"),
3224 };
3225 if let Some(token) = listen_token {
3226 self.sessions.borrow_mut().slab.try_remove(token.0);
3227 self.accept_ready.remove(&ListenToken(token.0));
3230 info!("released listen token {:?}", token);
3231 }
3232 self.finish_proxy_order(&mut rollback, response);
3233 }
3234 Some(RequestType::ActivateListener(ref activate)) => {
3235 let response = self.notify_activate_listener(&req_id, activate);
3236 self.finish_proxy_order(&mut rollback, response);
3237 }
3238 Some(RequestType::DeactivateListener(ref deactivate)) => {
3239 let response = self.notify_deactivate_listener(&req_id, deactivate);
3240 self.finish_proxy_order(&mut rollback, response);
3241 }
3242 _other_request => {}
3243 };
3244 }
3245
3246 fn finish_proxy_order(&mut self, rollback: &mut Option<Request>, response: WorkerResponse) {
3249 if response.is_failure()
3250 && let Some(rollback) = rollback.take()
3251 {
3252 debug_assert!(
3253 rollback.request_type.is_some(),
3254 "a ConfigState rollback must carry one request"
3255 );
3256 if let Err(error) = self.config_state.dispatch(&rollback) {
3257 error!(
3258 "Could not roll back worker config state after a proxy refused an order: {}",
3259 error
3260 );
3261 }
3262 debug_assert!(
3263 rollback.request_type.is_some(),
3264 "dispatching a rollback must not consume its request type"
3265 );
3266 }
3267
3268 push_queue(response);
3269 }
3270
3271 fn add_cluster(&mut self, cluster: &Cluster) {
3272 let mut backends = self.backends.borrow_mut();
3273 backends.set_load_balancing_policy_for_cluster(
3274 &cluster.cluster_id,
3275 LoadBalancingAlgorithms::try_from(cluster.load_balancing).unwrap_or_default(),
3276 cluster
3277 .load_metric
3278 .and_then(|n| LoadMetric::try_from(n).ok()),
3279 );
3280 backends.set_shuffle_sharding_for_cluster(
3281 &cluster.cluster_id,
3282 ShuffleSharding::from_cluster(
3283 cluster.shard_percent,
3284 cluster.shard_min_backends,
3285 cluster.shard_mode,
3286 ),
3287 );
3288 backends.set_health_check_config(&cluster.cluster_id, cluster.health_check.to_owned());
3289 backends.set_cluster_http2(&cluster.cluster_id, cluster.http2.unwrap_or(false));
3290 }
3291
3292 fn add_backend(&mut self, req_id: &str, add_backend: &AddBackend) -> WorkerResponse {
3293 if let Some(Err(reason)) = add_backend
3294 .unix_path
3295 .as_deref()
3296 .map(sozu_command::config::parse_backend_unix_path)
3297 {
3298 return WorkerResponse::error(req_id, format!("invalid unix_path: {reason}"));
3299 }
3300 let new_backend = Backend::new(
3301 &add_backend.backend_id,
3302 add_backend.address.into(),
3303 add_backend.sticky_id.clone(),
3304 add_backend.load_balancing_parameters,
3305 add_backend.backup,
3306 )
3307 .with_unix_path(add_backend.unix_path.clone());
3308 self.backends
3309 .borrow_mut()
3310 .add_backend(&add_backend.cluster_id, new_backend);
3311
3312 WorkerResponse::ok(req_id)
3313 }
3314
3315 fn remove_cluster_objects(&mut self, req_id: &str, cluster_id: &str, requests: Vec<Request>) {
3327 for request in requests {
3328 let result = match request.request_type {
3329 Some(
3330 RequestType::RemoveHttpFrontend(front)
3331 | RequestType::RemoveHttpFrontendWithHeaderPredicates(front),
3332 ) => self.http.borrow_mut().remove_http_frontend(front),
3333 Some(
3334 RequestType::RemoveHttpsFrontend(front)
3335 | RequestType::RemoveHttpsFrontendWithHeaderPredicates(front),
3336 ) => self
3337 .https
3338 .borrow_mut()
3339 .remove_https_frontend(front)
3340 .map(|_| ()),
3341 Some(RequestType::RemoveTcpFrontend(front)) => {
3342 self.tcp.borrow_mut().remove_tcp_front(front)
3343 }
3344 Some(RequestType::RemoveUdpFrontend(front)) => {
3345 self.udp.borrow_mut().remove_udp_front(front)
3346 }
3347 Some(RequestType::RemoveBackend(backend)) => {
3348 let response = self.remove_backend(req_id, &backend);
3349 debug_assert_eq!(
3350 response.status,
3351 ResponseStatus::Ok as i32,
3352 "Server::remove_backend always answers ok"
3353 );
3354 Ok(())
3355 }
3356 other => {
3357 debug_assert!(
3358 false,
3359 "cluster_removal_requests only returns frontend and backend removals, got {other:?}"
3360 );
3361 Ok(())
3362 }
3363 };
3364 if let Err(error) = result {
3365 warn!(
3366 "{} removing cluster {}: could not remove one of its objects: {}",
3367 req_id, cluster_id, error
3368 );
3369 }
3370 }
3371 self.backends.borrow_mut().remove_cluster(cluster_id);
3372 debug_assert!(
3373 !self.backends.borrow().backends.contains_key(cluster_id),
3374 "a removed cluster must leave no backend list behind"
3375 );
3376 }
3377
3378 fn remove_health_check_state(&mut self, cluster_id: &str) {
3383 self.protected_health_check.forget_cluster(cluster_id);
3384 self.health_checker.remove_cluster(cluster_id);
3385 self.backends
3386 .borrow_mut()
3387 .set_health_check_config(cluster_id, None);
3388 }
3389
3390 fn remove_backend(&mut self, req_id: &str, backend: &RemoveBackend) -> WorkerResponse {
3391 let address = backend.address.into();
3392 let id_still_live = {
3399 let mut backends = self.backends.borrow_mut();
3400 backends.remove_backend(&backend.cluster_id, &backend.backend_id, &address);
3401 backends.has_backend_id(&backend.cluster_id, &backend.backend_id)
3402 };
3403 if !id_still_live {
3404 METRICS.with(|metrics| {
3405 (*metrics.borrow_mut()).remove_backend(&backend.cluster_id, &backend.backend_id);
3406 });
3407 }
3408
3409 WorkerResponse::ok(req_id)
3410 }
3411
3412 fn notify_add_http_listener(
3413 &mut self,
3414 req_id: &str,
3415 listener: HttpListenerConfig,
3416 ) -> WorkerResponse {
3417 debug!("{} add http listener {:?}", req_id, listener);
3418
3419 let address: SocketAddr = listener.address.into();
3420 let retained_frontends = self
3421 .config_state
3422 .http_fronts
3423 .values()
3424 .filter(|frontend| frontend.address == address)
3425 .cloned()
3426 .map(Into::into)
3427 .collect();
3428
3429 if self.sessions.borrow().at_capacity() {
3430 return worker_response_error(req_id, "session list is full, cannot add a listener");
3431 }
3432
3433 let mut session_manager = self.sessions.borrow_mut();
3434 let slab_before = session_manager.slab.len();
3436 debug_assert!(
3437 !session_manager
3438 .slab
3439 .contains(session_manager.slab.vacant_key()),
3440 "the next vacant slab key must be free before insertion"
3441 );
3442 let entry = session_manager.slab.vacant_entry();
3443 let token = Token(entry.key());
3444
3445 match self.http.borrow_mut().add_listener_with_retained_frontends(
3446 listener,
3447 token,
3448 retained_frontends,
3449 ) {
3450 Ok(_token) => {
3451 entry.insert(Rc::new(RefCell::new(ListenSession {
3452 protocol: Protocol::HTTPListen,
3453 })));
3454 debug_assert!(
3457 session_manager.slab.contains(token.0),
3458 "listener insert must occupy the token's slab key"
3459 );
3460 debug_assert_eq!(
3461 session_manager.slab.len(),
3462 slab_before + 1,
3463 "adding a listener must occupy exactly one slab slot"
3464 );
3465 self.base_sessions_count += 1;
3466 WorkerResponse::ok(req_id)
3467 }
3468 Err(e) => worker_response_error(req_id, format!("Could not add HTTP listener: {e}")),
3469 }
3470 }
3471
3472 fn notify_add_https_listener(
3473 &mut self,
3474 req_id: &str,
3475 listener: HttpsListenerConfig,
3476 ) -> WorkerResponse {
3477 debug!("{} add https listener {:?}", req_id, listener);
3478
3479 let wire_address = listener.address;
3480 let address: SocketAddr = wire_address.into();
3481 let retained_frontends = self
3482 .config_state
3483 .https_fronts
3484 .values()
3485 .filter(|frontend| frontend.address == address)
3486 .cloned()
3487 .map(Into::into)
3488 .collect();
3489 let retained_certificates = self
3490 .config_state
3491 .certificate_add_requests_for_address(address);
3492
3493 if self.sessions.borrow().at_capacity() {
3494 return worker_response_error(req_id, "session list is full, cannot add a listener");
3495 }
3496
3497 let mut session_manager = self.sessions.borrow_mut();
3498 let slab_before = session_manager.slab.len();
3499 debug_assert!(
3500 !session_manager
3501 .slab
3502 .contains(session_manager.slab.vacant_key()),
3503 "the next vacant slab key must be free before insertion"
3504 );
3505 let entry = session_manager.slab.vacant_entry();
3506 let token = Token(entry.key());
3507
3508 match self.https.borrow_mut().add_listener_with_retained_state(
3509 listener.clone(),
3510 token,
3511 retained_frontends,
3512 retained_certificates,
3513 ) {
3514 Ok(_token) => {
3515 entry.insert(Rc::new(RefCell::new(ListenSession {
3516 protocol: Protocol::HTTPSListen,
3517 })));
3518 debug_assert!(
3519 session_manager.slab.contains(token.0),
3520 "listener insert must occupy the token's slab key"
3521 );
3522 debug_assert_eq!(
3523 session_manager.slab.len(),
3524 slab_before + 1,
3525 "adding a listener must occupy exactly one slab slot"
3526 );
3527 self.base_sessions_count += 1;
3528 WorkerResponse::ok(req_id)
3529 }
3530 Err(e) => worker_response_error(req_id, format!("Could not add HTTPS listener: {e}")),
3531 }
3532 }
3533
3534 fn notify_add_tcp_listener(
3535 &mut self,
3536 req_id: &str,
3537 listener: CommandTcpListener,
3538 ) -> WorkerResponse {
3539 debug!("{} add tcp listener {:?}", req_id, listener);
3540
3541 let address: SocketAddr = listener.address.into();
3542 let retained_frontends = self
3543 .config_state
3544 .tcp_fronts
3545 .values()
3546 .flatten()
3547 .filter(|frontend| frontend.address == address)
3548 .cloned()
3549 .map(Into::into)
3550 .collect();
3551
3552 if self.sessions.borrow().at_capacity() {
3553 return worker_response_error(req_id, "session list is full, cannot add a listener");
3554 }
3555
3556 let mut session_manager = self.sessions.borrow_mut();
3557 let slab_before = session_manager.slab.len();
3558 debug_assert!(
3559 !session_manager
3560 .slab
3561 .contains(session_manager.slab.vacant_key()),
3562 "the next vacant slab key must be free before insertion"
3563 );
3564 let entry = session_manager.slab.vacant_entry();
3565 let token = Token(entry.key());
3566
3567 match self.tcp.borrow_mut().add_listener_with_retained_frontends(
3568 listener,
3569 token,
3570 retained_frontends,
3571 ) {
3572 Ok(_token) => {
3573 entry.insert(Rc::new(RefCell::new(ListenSession {
3574 protocol: Protocol::TCPListen,
3575 })));
3576 debug_assert!(
3577 session_manager.slab.contains(token.0),
3578 "listener insert must occupy the token's slab key"
3579 );
3580 debug_assert_eq!(
3581 session_manager.slab.len(),
3582 slab_before + 1,
3583 "adding a listener must occupy exactly one slab slot"
3584 );
3585 self.base_sessions_count += 1;
3586 WorkerResponse::ok(req_id)
3587 }
3588 Err(e) => worker_response_error(req_id, format!("Could not add TCP listener: {e}")),
3589 }
3590 }
3591
3592 fn notify_add_udp_listener(
3593 &mut self,
3594 req_id: &str,
3595 listener: CommandUdpListener,
3596 ) -> WorkerResponse {
3597 debug!("{} add udp listener {:?}", req_id, listener);
3598
3599 let address: SocketAddr = listener.address.into();
3600 let retained_frontends = self
3601 .config_state
3602 .udp_fronts
3603 .values()
3604 .flatten()
3605 .filter(|frontend| frontend.address == address)
3606 .cloned()
3607 .map(Into::into)
3608 .collect();
3609
3610 if self.sessions.borrow().at_capacity() {
3611 return worker_response_error(req_id, "session list is full, cannot add a listener");
3612 }
3613
3614 let mut session_manager = self.sessions.borrow_mut();
3615 let entry = session_manager.slab.vacant_entry();
3616 let token = Token(entry.key());
3617
3618 match self.udp.borrow_mut().add_listener_with_retained_frontends(
3619 listener,
3620 token,
3621 retained_frontends,
3622 ) {
3623 Ok(_token) => {
3624 entry.insert(Rc::new(RefCell::new(ListenSession {
3625 protocol: Protocol::UDPListen,
3626 })));
3627 self.base_sessions_count += 1;
3628 WorkerResponse::ok(req_id)
3629 }
3630 Err(e) => worker_response_error(req_id, format!("Could not add UDP listener: {e}")),
3631 }
3632 }
3633
3634 fn notify_update_udp_listener(
3635 &mut self,
3636 req_id: &str,
3637 patch: UpdateUdpListenerConfig,
3638 ) -> WorkerResponse {
3639 debug!("{} update udp listener {:?}", req_id, patch.address);
3640 match self.udp.borrow_mut().update_listener(patch) {
3641 Ok(()) => WorkerResponse::ok(req_id),
3642 Err(e) => worker_response_error(req_id, format!("Could not update UDP listener: {e}")),
3643 }
3644 }
3645
3646 fn notify_update_http_listener(
3647 &mut self,
3648 req_id: &str,
3649 patch: UpdateHttpListenerConfig,
3650 ) -> WorkerResponse {
3651 debug!("{} update http listener {:?}", req_id, patch.address);
3652 match self.http.borrow_mut().update_listener(patch) {
3653 Ok(()) => WorkerResponse::ok(req_id),
3654 Err(e) => worker_response_error(req_id, format!("Could not update HTTP listener: {e}")),
3655 }
3656 }
3657
3658 fn notify_update_https_listener(
3659 &mut self,
3660 req_id: &str,
3661 patch: UpdateHttpsListenerConfig,
3662 ) -> WorkerResponse {
3663 debug!("{} update https listener {:?}", req_id, patch.address);
3664 match self.https.borrow_mut().update_listener(patch) {
3665 Ok(()) => WorkerResponse::ok(req_id),
3666 Err(e) => {
3667 worker_response_error(req_id, format!("Could not update HTTPS listener: {e}"))
3668 }
3669 }
3670 }
3671
3672 fn notify_update_tcp_listener(
3673 &mut self,
3674 req_id: &str,
3675 patch: UpdateTcpListenerConfig,
3676 ) -> WorkerResponse {
3677 debug!("{} update tcp listener {:?}", req_id, patch.address);
3678 match self.tcp.borrow_mut().update_listener(patch) {
3679 Ok(()) => WorkerResponse::ok(req_id),
3680 Err(e) => worker_response_error(req_id, format!("Could not update TCP listener: {e}")),
3681 }
3682 }
3683
3684 fn reserve_listen_token(&mut self, token: Token, protocol: Protocol) {
3719 let mut sessions = self.sessions.borrow_mut();
3720 match sessions.slab.get_mut(token.0) {
3721 Some(slot) => {
3722 *slot = Rc::new(RefCell::new(ListenSession { protocol }));
3723 info!("reserved listen token {:?} for {:?}", token, protocol);
3724 }
3725 None => error!(
3729 "listen token {:?} ({:?}) has no slab slot to reserve; the listener would not be reachable if reactivated",
3730 token, protocol
3731 ),
3732 }
3733 }
3734
3735 fn report_unclaimed_inherited_sockets(&self) {
3755 let Some(scm_listeners) = self.scm_listeners.as_ref() else {
3756 return;
3757 };
3758 let leftovers: [(&str, &Vec<(ListenerKey, RawFd)>); 4] = [
3759 ("HTTP", &scm_listeners.http),
3760 ("HTTPS", &scm_listeners.tls),
3761 ("TCP", &scm_listeners.tcp),
3762 ("UDP", &scm_listeners.udp),
3763 ];
3764 for (label, table) in leftovers {
3765 for (address, fd) in table {
3766 let has_listener = match label {
3767 "HTTP" => self.http.borrow().listener_token(address).is_some(),
3768 "HTTPS" => self.https.borrow().listener_token(address).is_some(),
3769 "TCP" => self.tcp.borrow().listener_token(address).is_some(),
3770 _ => self.udp.borrow().listener_token(address).is_some(),
3771 };
3772 if has_listener {
3773 info!(
3774 "inherited {} listening socket for {} (fd {}) is held for an \
3775 ActivateListener this worker has not processed yet",
3776 label, address, fd
3777 );
3778 } else {
3779 warn!(
3780 "inherited {} listening socket for {} (fd {}) is held with no listener \
3781 at that address: until an AddListener plus ActivateListener for {} \
3782 adopts it, it is expected to keep taking a share of new connections \
3783 that nothing accepts (inferred from SO_REUSEPORT semantics, not \
3784 measured here)",
3785 label, address, fd, address
3786 );
3787 }
3788 }
3789 }
3790 }
3791
3792 fn notify_activate_listener(
3793 &mut self,
3794 req_id: &str,
3795 activate: &ActivateListener,
3796 ) -> WorkerResponse {
3797 use crate::InheritedSocketFate;
3798
3799 debug!(
3800 "{} activate {:?} listener {:?}",
3801 req_id, activate.proxy, activate
3802 );
3803
3804 let address = activate.listener_key();
3805
3806 match ListenerType::try_from(activate.proxy) {
3807 Ok(ListenerType::Http) => {
3808 let listener = match self.http.borrow().inherited_socket_fate(&address) {
3809 InheritedSocketFate::Adopted => self
3810 .scm_listeners
3811 .as_mut()
3812 .and_then(|listeners| listeners.get_http(&address))
3813 .map(|fd| unsafe { MioTcpListener::from_raw_fd(fd) }),
3818 InheritedSocketFate::Refused => {
3819 discard_inherited_socket::<MioTcpListener>(
3820 self.scm_listeners
3821 .as_mut()
3822 .and_then(|listeners| listeners.get_http(&address)),
3823 &address,
3824 "HTTP",
3825 );
3826 None
3827 }
3828 InheritedSocketFate::Unclaimed => None,
3831 };
3832
3833 let activated_token = self.http.borrow_mut().activate_listener(&address, listener);
3834 match activated_token {
3835 Ok(token) => {
3836 self.accept(ListenToken(token.0), Protocol::HTTPListen);
3837 WorkerResponse::ok(req_id)
3838 }
3839 Err(activate_error) => worker_response_error(
3840 req_id,
3841 format!("Could not activate HTTP listener: {activate_error}"),
3842 ),
3843 }
3844 }
3845 Ok(ListenerType::Https) => {
3846 let listener = match self.https.borrow().inherited_socket_fate(&address) {
3847 InheritedSocketFate::Adopted => self
3848 .scm_listeners
3849 .as_mut()
3850 .and_then(|listeners| listeners.get_https(&address))
3851 .map(|fd| unsafe { MioTcpListener::from_raw_fd(fd) }),
3856 InheritedSocketFate::Refused => {
3857 discard_inherited_socket::<MioTcpListener>(
3858 self.scm_listeners
3859 .as_mut()
3860 .and_then(|listeners| listeners.get_https(&address)),
3861 &address,
3862 "HTTPS",
3863 );
3864 None
3865 }
3866 InheritedSocketFate::Unclaimed => None,
3869 };
3870
3871 let activated_token = self
3872 .https
3873 .borrow_mut()
3874 .activate_listener(&address, listener);
3875 match activated_token {
3876 Ok(token) => {
3877 self.accept(ListenToken(token.0), Protocol::HTTPSListen);
3878 WorkerResponse::ok(req_id)
3879 }
3880 Err(activate_error) => worker_response_error(
3881 req_id,
3882 format!("Could not activate HTTPS listener: {activate_error}"),
3883 ),
3884 }
3885 }
3886 Ok(ListenerType::Tcp) => {
3887 let listener = match self.tcp.borrow().inherited_socket_fate(&address) {
3888 InheritedSocketFate::Adopted => self
3889 .scm_listeners
3890 .as_mut()
3891 .and_then(|listeners| listeners.get_tcp(&address))
3892 .map(|fd| unsafe { MioTcpListener::from_raw_fd(fd) }),
3897 InheritedSocketFate::Refused => {
3898 discard_inherited_socket::<MioTcpListener>(
3899 self.scm_listeners
3900 .as_mut()
3901 .and_then(|listeners| listeners.get_tcp(&address)),
3902 &address,
3903 "TCP",
3904 );
3905 None
3906 }
3907 InheritedSocketFate::Unclaimed => None,
3910 };
3911
3912 let listener_token = self.tcp.borrow_mut().activate_listener(&address, listener);
3913 match listener_token {
3914 Ok(token) => {
3915 self.accept(ListenToken(token.0), Protocol::TCPListen);
3916 WorkerResponse::ok(req_id)
3917 }
3918 Err(activate_error) => worker_response_error(
3919 req_id,
3920 format!("Could not activate TCP listener: {activate_error}"),
3921 ),
3922 }
3923 }
3924 Ok(ListenerType::Udp) => {
3925 let socket = match self.udp.borrow().inherited_socket_fate(&address) {
3926 InheritedSocketFate::Adopted => self
3927 .scm_listeners
3928 .as_mut()
3929 .and_then(|listeners| listeners.get_udp(&address))
3930 .map(|fd| unsafe { MioUdpSocket::from_raw_fd(fd) }),
3938 InheritedSocketFate::Refused => {
3939 discard_inherited_socket::<MioUdpSocket>(
3940 self.scm_listeners
3941 .as_mut()
3942 .and_then(|listeners| listeners.get_udp(&address)),
3943 &address,
3944 "UDP",
3945 );
3946 None
3947 }
3948 InheritedSocketFate::Unclaimed => None,
3951 };
3952
3953 let activated_token = self.udp.borrow_mut().activate_listener(&address, socket);
3954 match activated_token {
3955 Ok(token) => {
3956 let session_installed = self.udp.borrow().has_listener_session(token)
4006 && self.sessions.borrow().slab.contains(token.0);
4007 if !session_installed {
4008 let session = match self.udp.borrow_mut().build_session(token) {
4015 Some(session) => session,
4016 None => {
4017 return worker_response_error(
4018 req_id,
4019 format!(
4020 "Could not build the UDP listener session for {address}"
4021 ),
4022 );
4023 }
4024 };
4025 let mut sessions = self.sessions.borrow_mut();
4026 match sessions.slab.get_mut(token.0) {
4027 Some(slot) => *slot = session,
4028 None => {
4029 return worker_response_error(
4030 req_id,
4031 format!(
4032 "UDP listener {address} has no session slot at {token:?}"
4033 ),
4034 );
4035 }
4036 }
4037 }
4038 WorkerResponse::ok(req_id)
4039 }
4040 Err(activate_error) => worker_response_error(
4041 req_id,
4042 format!("Could not activate UDP listener: {activate_error}"),
4043 ),
4044 }
4045 }
4046 Err(_) => worker_response_error(req_id, "Wrong variant for ListenerType on request"),
4047 }
4048 }
4049
4050 fn notify_deactivate_listener(
4051 &mut self,
4052 req_id: &str,
4053 deactivate: &DeactivateListener,
4054 ) -> WorkerResponse {
4055 debug!(
4056 "{} deactivate {:?} listener {:?}",
4057 req_id, deactivate.proxy, deactivate
4058 );
4059
4060 let address = deactivate.listener_key();
4061
4062 match ListenerType::try_from(deactivate.proxy) {
4063 Ok(ListenerType::Http) => {
4064 let (token, mut listener) =
4065 match self.http.borrow_mut().give_back_listener(&address) {
4066 Ok((token, listener)) => (token, listener),
4067 Err(e) => {
4068 return worker_response_error(
4069 req_id,
4070 format!("Couldn't deactivate HTTP listener at {address}: {e}"),
4071 );
4072 }
4073 };
4074
4075 if let Err(e) = self.poll.registry().deregister(&mut listener) {
4076 error!(
4077 "error deregistering HTTP listen socket({:?}): {:?}",
4078 deactivate, e
4079 );
4080 }
4081
4082 self.reserve_listen_token(token, Protocol::HTTPListen);
4087 self.accept_ready.remove(&ListenToken(token.0));
4093
4094 if deactivate.to_scm {
4095 self.unblock_scm_socket();
4096 let listeners = Listeners {
4097 http: vec![(address.clone(), listener.as_raw_fd())],
4098 tls: vec![],
4099 tcp: vec![],
4100 udp: vec![],
4101 };
4102 info!("sending HTTP listener: {:?}", listeners);
4103 let res = self.scm.send_listeners(&listeners);
4104
4105 self.block_scm_socket();
4106
4107 info!("sent HTTP listener: {:?}", res);
4108 }
4109 WorkerResponse::ok(req_id)
4110 }
4111 Ok(ListenerType::Https) => {
4112 let (token, mut listener) =
4113 match self.https.borrow_mut().give_back_listener(&address) {
4114 Ok((token, listener)) => (token, listener),
4115 Err(e) => {
4116 return worker_response_error(
4117 req_id,
4118 format!("Couldn't deactivate HTTPS listener at {address}: {e}",),
4119 );
4120 }
4121 };
4122 if let Err(e) = self.poll.registry().deregister(&mut listener) {
4123 error!(
4124 "error deregistering HTTPS listen socket({:?}): {:?}",
4125 deactivate, e
4126 );
4127 }
4128 self.reserve_listen_token(token, Protocol::HTTPSListen);
4131 self.accept_ready.remove(&ListenToken(token.0));
4132
4133 if deactivate.to_scm {
4134 self.unblock_scm_socket();
4135 let listeners = Listeners {
4136 http: vec![],
4137 tls: vec![(address.clone(), listener.as_raw_fd())],
4138 tcp: vec![],
4139 udp: vec![],
4140 };
4141 info!("sending HTTPS listener: {:?}", listeners);
4142 let res = self.scm.send_listeners(&listeners);
4143
4144 self.block_scm_socket();
4145
4146 info!("sent HTTPS listener: {:?}", res);
4147 }
4148 WorkerResponse::ok(req_id)
4149 }
4150 Ok(ListenerType::Tcp) => {
4151 let (token, mut listener) = match self.tcp.borrow_mut().give_back_listener(&address)
4152 {
4153 Ok((token, listener)) => (token, listener),
4154 Err(e) => {
4155 return worker_response_error(
4156 req_id,
4157 format!("Could not deactivate TCP listener at {address}: {e}"),
4158 );
4159 }
4160 };
4161
4162 if let Err(e) = self.poll.registry().deregister(&mut listener) {
4163 error!(
4164 "error deregistering TCP listen socket({:?}): {:?}",
4165 deactivate, e
4166 );
4167 }
4168 self.reserve_listen_token(token, Protocol::TCPListen);
4171 self.accept_ready.remove(&ListenToken(token.0));
4172
4173 if deactivate.to_scm {
4174 self.unblock_scm_socket();
4175 let listeners = Listeners {
4176 http: vec![],
4177 tls: vec![],
4178 tcp: vec![(address.clone(), listener.as_raw_fd())],
4179 udp: vec![],
4180 };
4181 info!("sending TCP listener: {:?}", listeners);
4182 let res = self.scm.send_listeners(&listeners);
4183
4184 self.block_scm_socket();
4185
4186 info!("sent TCP listener: {:?}", res);
4187 }
4188 WorkerResponse::ok(req_id)
4189 }
4190 Ok(ListenerType::Udp) => {
4191 let (token, mut listener) = match self.udp.borrow_mut().give_back_listener(&address)
4192 {
4193 Ok((token, listener)) => (token, listener),
4194 Err(e) => {
4195 return worker_response_error(
4196 req_id,
4197 format!("Could not deactivate UDP listener at {address}: {e}"),
4198 );
4199 }
4200 };
4201
4202 if let Err(e) = self.poll.registry().deregister(&mut listener) {
4203 error!(
4204 "error deregistering UDP listen socket({:?}): {:?}",
4205 deactivate, e
4206 );
4207 }
4208 self.reserve_listen_token(token, Protocol::UDPListen);
4213 self.accept_ready.remove(&ListenToken(token.0));
4217
4218 if deactivate.to_scm {
4219 self.unblock_scm_socket();
4220 let listeners = Listeners {
4221 http: vec![],
4222 tls: vec![],
4223 tcp: vec![],
4224 udp: vec![(address.clone(), listener.as_raw_fd())],
4225 };
4226 info!("sending UDP listener: {:?}", listeners);
4227 let res = self.scm.send_listeners(&listeners);
4228
4229 self.block_scm_socket();
4230
4231 info!("sent UDP listener: {:?}", res);
4232 }
4233 WorkerResponse::ok(req_id)
4234 }
4235 Err(_) => worker_response_error(req_id, "Wrong variant for ListenerType on request"),
4236 }
4237 }
4238
4239 pub fn return_listen_sockets(&mut self) -> Result<(), ScmSocketError> {
4241 self.unblock_scm_socket();
4242
4243 let mut http_listeners = self.http.borrow_mut().give_back_listeners();
4244 for &mut (_, ref mut sock) in http_listeners.iter_mut() {
4245 if let Err(e) = self.poll.registry().deregister(sock) {
4246 error!(
4247 "error deregistering HTTP listen socket({:?}): {:?}",
4248 sock, e
4249 );
4250 }
4251 }
4252
4253 let mut https_listeners = self.https.borrow_mut().give_back_listeners();
4254 for &mut (_, ref mut sock) in https_listeners.iter_mut() {
4255 if let Err(e) = self.poll.registry().deregister(sock) {
4256 error!(
4257 "error deregistering HTTPS listen socket({:?}): {:?}",
4258 sock, e
4259 );
4260 }
4261 }
4262
4263 let mut tcp_listeners = self.tcp.borrow_mut().give_back_listeners();
4264 for &mut (_, ref mut sock) in tcp_listeners.iter_mut() {
4265 if let Err(e) = self.poll.registry().deregister(sock) {
4266 error!("error deregistering TCP listen socket({:?}): {:?}", sock, e);
4267 }
4268 }
4269
4270 let mut udp_listeners = self.udp.borrow_mut().give_back_listeners();
4271 for &mut (_, ref mut sock) in udp_listeners.iter_mut() {
4272 if let Err(e) = self.poll.registry().deregister(sock) {
4273 error!("error deregistering UDP listen socket({:?}): {:?}", sock, e);
4274 }
4275 }
4276
4277 let listeners = Listeners {
4279 http: http_listeners
4280 .iter()
4281 .map(|(key, listener)| (key.clone(), listener.as_raw_fd()))
4282 .collect(),
4283 tls: https_listeners
4284 .iter()
4285 .map(|(key, listener)| (key.clone(), listener.as_raw_fd()))
4286 .collect(),
4287 tcp: tcp_listeners
4288 .iter()
4289 .map(|(key, listener)| (key.clone(), listener.as_raw_fd()))
4290 .collect(),
4291 udp: udp_listeners
4292 .iter()
4293 .map(|(key, listener)| (key.clone(), listener.as_raw_fd()))
4294 .collect(),
4295 };
4296 debug_assert_eq!(
4300 listeners.http.len(),
4301 http_listeners.len(),
4302 "every HTTP listener must be collected exactly once"
4303 );
4304 debug_assert_eq!(
4305 listeners.tls.len(),
4306 https_listeners.len(),
4307 "every HTTPS listener must be collected exactly once"
4308 );
4309 debug_assert_eq!(
4310 listeners.tcp.len(),
4311 tcp_listeners.len(),
4312 "every TCP listener must be collected exactly once"
4313 );
4314 info!("sending default listeners: {:?}", listeners);
4315 let res = self.scm.send_listeners(&listeners);
4316
4317 self.block_scm_socket();
4318
4319 info!("sent default listeners: {:?}", res);
4320 res
4321 }
4322
4323 fn block_scm_socket(&mut self) {
4324 if let Err(e) = self.scm.set_blocking(true) {
4325 error!("Could not block scm socket: {}", e);
4326 }
4327 }
4328
4329 fn unblock_scm_socket(&mut self) {
4330 if let Err(e) = self.scm.set_blocking(false) {
4331 error!("Could not unblock scm socket: {}", e);
4332 }
4333 }
4334
4335 pub fn to_session(&self, token: Token) -> SessionToken {
4336 SessionToken(token.0)
4337 }
4338
4339 pub fn from_session(&self, token: SessionToken) -> Token {
4340 Token(token.0)
4341 }
4342
4343 pub fn accept(&mut self, token: ListenToken, protocol: Protocol) {
4344 let (proto_key, accepted_protocol) = match protocol {
4357 Protocol::TCPListen => ("listener.accepted.tcp", Protocol::TCPListen),
4358 Protocol::HTTPListen => ("listener.accepted.http", Protocol::HTTPListen),
4359 Protocol::HTTPSListen => ("listener.accepted.https", Protocol::HTTPSListen),
4360 other => {
4361 warn!(
4362 "accept() called with non-listen protocol {:?} on token {:?}; skipping",
4363 other, token
4364 );
4365 return;
4366 }
4367 };
4368
4369 debug_assert!(
4372 matches!(
4373 accepted_protocol,
4374 Protocol::TCPListen | Protocol::HTTPListen | Protocol::HTTPSListen
4375 ),
4376 "accept dispatch must run with a listen protocol only"
4377 );
4378
4379 loop {
4380 let result = match accepted_protocol {
4381 Protocol::TCPListen => self.tcp.borrow_mut().accept(token),
4382 Protocol::HTTPListen => self.http.borrow_mut().accept(token),
4383 Protocol::HTTPSListen => self.https.borrow_mut().accept(token),
4384 other => unreachable!(
4388 "accept dispatch reached non-listen protocol {:?} after outer guard",
4389 other
4390 ),
4391 };
4392 match result {
4393 Ok((sock, peer)) => {
4394 incr!(names::listener::ACCEPTED_TOTAL);
4398 incr!(proto_key);
4399 incr!(per_source_bucket(&peer));
4400 let queue_before = self.accept_queue.len();
4401 self.accept_queue.push_back((
4402 sock,
4403 token,
4404 accepted_protocol,
4405 Instant::now(),
4406 peer,
4407 ));
4408 debug_assert_eq!(
4410 self.accept_queue.len(),
4411 queue_before + 1,
4412 "each accepted socket must enqueue exactly one entry"
4413 );
4414 }
4415 Err(AcceptError::WouldBlock) => {
4416 self.accept_ready.remove(&token);
4417 break;
4418 }
4419 Err(other) => {
4420 error!(
4421 "error accepting {:?} sockets: {:?}",
4422 accepted_protocol, other
4423 );
4424 self.accept_ready.remove(&token);
4425 break;
4426 }
4427 }
4428 }
4429
4430 gauge!(names::accept_queue::CONNECTIONS, self.accept_queue.len());
4431 }
4432
4433 pub fn create_sessions(&mut self) {
4434 while let Some((sock, token, protocol, timestamp, peer)) = self.accept_queue.pop_back() {
4435 let wait_time = Instant::now() - timestamp;
4436 time!(names::accept_queue::WAIT_TIME, wait_time.as_millis());
4437 if wait_time > self.accept_queue_timeout {
4438 incr!(names::accept_queue::TIMEOUT);
4439 continue;
4440 }
4441
4442 if !self.sessions.borrow_mut().check_limits() {
4443 if !self.evict_on_queue_full {
4449 incr!(names::listener::CONNECTION_CAPPED);
4450 break;
4451 }
4452
4453 if self.shutting_down.is_some() {
4457 incr!(names::listener::CONNECTION_CAPPED);
4458 break;
4459 }
4460
4461 let to_evict = (self.sessions.borrow().max_connections / 100).max(1);
4470 let evicted = self.evict_least_active_sessions(to_evict);
4471 if evicted == 0 {
4472 warn!("evict_on_queue_full enabled but no candidate sessions to evict");
4478 incr!(names::listener::CONNECTION_CAPPED);
4479 break;
4480 }
4481
4482 count!(names::sessions::EVICTED, evicted as i64);
4483 warn!(
4484 "evicted {} least recently active sessions to make room",
4485 evicted
4486 );
4487
4488 if !self.sessions.borrow_mut().check_limits() {
4489 incr!(names::listener::CONNECTION_CAPPED);
4490 break;
4491 }
4492 }
4493
4494 debug_assert!(
4501 matches!(
4502 protocol,
4503 Protocol::TCPListen | Protocol::HTTPListen | Protocol::HTTPSListen
4504 ),
4505 "accept queue must only hold listen protocols, got {protocol:?}"
4506 );
4507 match protocol {
4508 Protocol::TCPListen => {
4509 let proxy = self.tcp.clone();
4510 if self
4511 .tcp
4512 .borrow_mut()
4513 .create_session(sock, peer, token, wait_time, proxy)
4514 .is_err()
4515 {
4516 break;
4517 }
4518 }
4519 Protocol::HTTPListen => {
4520 let proxy = self.http.clone();
4521 if self
4522 .http
4523 .borrow_mut()
4524 .create_session(sock, peer, token, wait_time, proxy)
4525 .is_err()
4526 {
4527 break;
4528 }
4529 }
4530 Protocol::HTTPSListen => {
4531 if self
4532 .https
4533 .borrow_mut()
4534 .create_session(sock, peer, token, wait_time, self.https.clone())
4535 .is_err()
4536 {
4537 break;
4538 }
4539 }
4540 _ => panic!("should not call accept() on a HTTP, HTTPS or TCP session"),
4541 };
4542 let nb_before = self.sessions.borrow().nb_connections;
4543 self.sessions.borrow_mut().incr();
4544 debug_assert_eq!(
4546 self.sessions.borrow().nb_connections,
4547 nb_before + 1,
4548 "create_sessions must account exactly one new connection per created session"
4549 );
4550 }
4551
4552 gauge!(names::accept_queue::CONNECTIONS, self.accept_queue.len());
4553 }
4554
4555 fn snapshot_event_owners(
4558 &self,
4559 tokens: impl Iterator<Item = Token>,
4560 owners: &mut Vec<Option<Rc<RefCell<dyn ProxySession>>>>,
4561 ) {
4562 debug_assert!(
4563 owners.is_empty(),
4564 "the previous turn must have consumed every event owner"
4565 );
4566 let sessions = self.sessions.borrow();
4567 owners.extend(tokens.map(|token| sessions.slab.get(token.0).cloned()));
4568 }
4569
4570 fn ready_if_still_owned(
4592 &mut self,
4593 token: Token,
4594 events: Ready,
4595 owner: Option<Rc<RefCell<dyn ProxySession>>>,
4596 ) {
4597 let still_owned = owner.is_some_and(|owner| {
4598 self.sessions
4599 .borrow()
4600 .slab
4601 .get(token.0)
4602 .is_some_and(|current| Rc::ptr_eq(current, &owner))
4603 });
4604 if !still_owned {
4605 trace!(
4606 "PROXY\t{:?} dropping events {:?} from a socket whose session closed earlier in this epoll result",
4607 token, events
4608 );
4609 return;
4610 }
4611 self.ready(token, events);
4612 }
4613
4614 pub fn ready(&mut self, token: Token, events: Ready) {
4615 trace!("PROXY\t{:?} got events: {:?}", token, events);
4616
4617 let session_token = token.0;
4618 if self.sessions.borrow().slab.contains(session_token) {
4619 let protocol = self.sessions.borrow().slab[session_token]
4621 .borrow()
4622 .protocol();
4623 match protocol {
4632 Protocol::HTTPListen | Protocol::HTTPSListen | Protocol::TCPListen => {
4633 if events.is_readable() {
4635 self.accept_ready.insert(ListenToken(token.0));
4636 if self.sessions.borrow().can_accept {
4637 self.accept(ListenToken(token.0), protocol);
4638 }
4639 return;
4640 }
4641
4642 if events.is_writable() {
4643 error!(
4644 "received writable for listener {:?}, this should not happen",
4645 token
4646 );
4647 return;
4648 }
4649
4650 if events.is_hup() {
4651 error!("should not happen: server {:?} closed", token);
4652 return;
4653 }
4654
4655 unreachable!();
4656 }
4657 _ => {}
4658 }
4659
4660 let session = self.sessions.borrow_mut().slab[session_token].clone();
4661 session.borrow_mut().update_readiness(token, events);
4662 if session.borrow_mut().ready(session.clone()) {
4663 debug!(
4664 "Server killing session from ready: token={:?}, protocol={:?}, events={:?}",
4665 token, protocol, events
4666 );
4667 self.kill_session(session);
4668 }
4669 }
4670 }
4671
4672 pub fn timeout(&mut self, token: Token) {
4673 trace!("PROXY\t{:?} got timeout", token);
4674
4675 let session_token = token.0;
4676 if self.sessions.borrow().slab.contains(session_token) {
4677 let session = self.sessions.borrow_mut().slab[session_token].clone();
4678 if session.borrow_mut().timeout(token) {
4679 debug!(
4680 "Server killing session from timeout: token={:?}, protocol={:?}",
4681 token,
4682 session.borrow().protocol()
4683 );
4684 self.kill_session(session);
4685 } else if session.borrow().needs_ready_pass() {
4686 if session.borrow_mut().ready(session.clone()) {
4691 debug!(
4692 "Server killing session from ready after timeout: token={:?}, protocol={:?}",
4693 token,
4694 session.borrow().protocol()
4695 );
4696 self.kill_session(session);
4697 }
4698 }
4699 }
4700 }
4701
4702 pub fn handle_remaining_readiness(&mut self) {
4703 if self.sessions.borrow().can_accept && !self.accept_ready.is_empty() {
4706 while let Some(token) = self
4707 .accept_ready
4708 .iter()
4709 .next()
4710 .map(|token| ListenToken(token.0))
4711 {
4712 let protocol = self
4729 .sessions
4730 .borrow()
4731 .slab
4732 .get(token.0)
4733 .map(|session| session.borrow().protocol());
4734 let Some(protocol) = protocol else {
4735 error!(
4736 "accept_ready holds listen token {:?}, which no longer has a session; dropping it",
4737 token
4738 );
4739 self.accept_ready.remove(&token);
4740 continue;
4741 };
4742 self.accept(token, protocol);
4743 if !self.sessions.borrow().can_accept || self.accept_ready.is_empty() {
4744 break;
4745 }
4746 }
4747 }
4748 }
4749 fn block_channel(&mut self) {
4750 if let Err(e) = self.channel.blocking() {
4751 error!("Could not block channel: {}", e);
4752 }
4753 }
4754 fn unblock_channel(&mut self) {
4755 if let Err(e) = self.channel.nonblocking() {
4756 error!("Could not block channel: {}", e);
4757 }
4758 }
4759
4760 fn evict_least_active_sessions(&self, count: usize) -> usize {
4772 if count == 0 {
4773 return 0;
4774 }
4775
4776 let tokens = {
4777 let sessions = self.sessions.borrow();
4778 let mut candidates: Vec<(Token, Instant)> = sessions
4779 .slab
4780 .iter()
4781 .filter(|(_, session)| {
4782 !matches!(
4783 session.borrow().protocol(),
4784 Protocol::HTTPListen
4785 | Protocol::HTTPSListen
4786 | Protocol::TCPListen
4787 | Protocol::UDPListen
4788 | Protocol::Channel
4789 | Protocol::Metrics
4790 | Protocol::Timer
4791 )
4792 })
4793 .map(|(_, session)| {
4794 let s = session.borrow();
4795 (s.frontend_token(), s.last_event())
4796 })
4797 .collect();
4798
4799 if candidates.is_empty() {
4802 return 0;
4803 }
4804
4805 let pivot = count.min(candidates.len()) - 1;
4806 candidates.select_nth_unstable_by_key(pivot, |&(_, last_event)| last_event);
4807
4808 candidates[..=pivot]
4809 .iter()
4810 .map(|&(token, _)| token)
4811 .collect::<HashSet<Token>>()
4812 };
4813
4814 let evicted = tokens.len();
4815 self.shut_down_sessions_by_frontend_tokens(tokens);
4816 evicted
4817 }
4818}
4819
4820fn discard_inherited_socket<T: FromRawFd>(
4840 fd: Option<RawFd>,
4841 address: &ListenerKey,
4842 protocol: &str,
4843) {
4844 let Some(fd) = fd else {
4845 return;
4846 };
4847 warn!(
4848 "closing the inherited {} listening socket for {}: this worker already has an active \
4849 listener there, so nothing will adopt this descriptor unless that listener is \
4850 deactivated and reactivated, and until then it is expected to keep taking a share of \
4851 new connections that nothing accepts (inferred from SO_REUSEPORT semantics, not \
4852 measured here)",
4853 protocol, address
4854 );
4855 drop(unsafe { T::from_raw_fd(fd) });
4859}
4860
4861const REFUSED_INITIAL_REQUEST: &str = "unexpected request before the initial status";
4863
4864fn worker_response_error<S: ToString, T: ToString>(request_id: S, error: T) -> WorkerResponse {
4867 error!(
4868 "error on request {}, {}",
4869 request_id.to_string(),
4870 error.to_string()
4871 );
4872 WorkerResponse::error(request_id, error)
4873}
4874
4875pub struct ListenSession {
4876 pub protocol: Protocol,
4877}
4878
4879impl ProxySession for ListenSession {
4880 fn last_event(&self) -> Instant {
4881 Instant::now()
4882 }
4883
4884 fn print_session(&self) {}
4885
4886 fn frontend_token(&self) -> Token {
4887 Token(0)
4888 }
4889
4890 fn protocol(&self) -> Protocol {
4891 self.protocol
4892 }
4893
4894 fn ready(&mut self, _session: Rc<RefCell<dyn ProxySession>>) -> SessionIsToBeClosed {
4895 false
4896 }
4897
4898 fn shutting_down(&mut self) -> SessionIsToBeClosed {
4899 false
4900 }
4901
4902 fn update_readiness(&mut self, _token: Token, _events: Ready) {}
4903
4904 fn close(&mut self) {}
4905
4906 fn timeout(&mut self, _token: Token) -> SessionIsToBeClosed {
4907 error!(
4908 "called ProxySession::timeout(token={:?}, time) on ListenSession {{ protocol: {:?} }}",
4909 _token, self.protocol
4910 );
4911 false
4912 }
4913}
4914
4915#[cfg(test)]
4916mod accept_telemetry_tests {
4917 use super::*;
4918
4919 #[test]
4922 fn per_source_bucket_collapses_ipv4_slash24() {
4923 let a: SocketAddr = "203.0.113.5:1234".parse().unwrap();
4924 let b: SocketAddr = "203.0.113.250:9999".parse().unwrap();
4925 assert_eq!(
4926 per_source_bucket(&a),
4927 per_source_bucket(&b),
4928 "addresses in the same /24 must land in the same bucket"
4929 );
4930 }
4931
4932 #[test]
4934 fn per_source_bucket_collapses_ipv6_slash48() {
4935 let a: SocketAddr = "[2001:db8:1234::1]:443".parse().unwrap();
4936 let b: SocketAddr = "[2001:db8:1234:abcd::ffff]:8443".parse().unwrap();
4937 assert_eq!(
4938 per_source_bucket(&a),
4939 per_source_bucket(&b),
4940 "addresses in the same /48 must land in the same bucket"
4941 );
4942 }
4943
4944 #[test]
4947 fn per_source_bucket_keys_are_bounded() {
4948 assert_eq!(PER_SOURCE_BUCKET_KEYS.len(), PER_SOURCE_BUCKETS);
4949 for (i, key) in PER_SOURCE_BUCKET_KEYS.iter().enumerate() {
4950 let expected = format!("client.connect.per_source.bucket_{i:03}");
4951 assert_eq!(*key, expected.as_str());
4952 }
4953 }
4954
4955 #[test]
4958 fn per_source_bucket_distributes_distinct_subnets() {
4959 let mut hits = std::collections::HashSet::new();
4960 for i in 0..200u8 {
4961 let addr: SocketAddr = format!("10.0.{i}.42:80").parse().unwrap();
4962 hits.insert(per_source_bucket(&addr));
4963 }
4964 assert!(
4968 hits.len() >= 100,
4969 "expected at least 100 distinct buckets across 200 /24s, got {}",
4970 hits.len()
4971 );
4972 }
4973}
4974
4975#[cfg(test)]
4976mod state_error_log_tests {
4977 use sozu_command::state::StateError;
4978
4979 #[test]
4980 fn worker_state_dispatch_error_sink_bounds_the_reason() {
4981 const REASON_SECRET: &str = "WORKER_STATE_ERROR_REASON_SECRET_SENTINEL";
4982
4983 let reason = format!("{REASON_SECRET}{}", "x".repeat(4096));
4984 let reason_len = reason.len();
4985 let output = crate::capture_test_logs(move || {
4986 let error = StateError::InvalidTcpFrontend {
4987 address: "127.0.0.1:443"
4988 .parse()
4989 .expect("test TCP frontend address must parse"),
4990 reason,
4991 };
4992 error!("Could not execute order on config state: {}", error);
4993 });
4994
4995 assert!(
4996 !output.contains(REASON_SECRET),
4997 "worker state-error sink leaked the reason: {output}"
4998 );
4999 assert!(
5000 output.contains(&format!("reason_bytes={reason_len}")),
5001 "worker state-error sink omitted bounded reason metadata: {output}"
5002 );
5003 assert!(
5004 output.len() <= 512,
5005 "worker state-error sink is not bounded: {} bytes",
5006 output.len()
5007 );
5008 }
5009}
5010
5011#[cfg(test)]
5012mod eviction_tests {
5013 use std::collections::HashSet;
5014 use std::time::{Duration, Instant};
5015
5016 use mio::Token;
5017
5018 #[test]
5022 fn select_nth_finds_oldest_sessions() {
5023 let now = Instant::now();
5024 let mut candidates = [
5025 (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)), ];
5031
5032 let count = 2;
5033 let pivot = count.min(candidates.len()) - 1;
5034 candidates.select_nth_unstable_by_key(pivot, |&(_, last_event)| last_event);
5035
5036 let selected: HashSet<Token> = candidates[..=pivot]
5037 .iter()
5038 .map(|&(token, _)| token)
5039 .collect();
5040
5041 assert_eq!(selected.len(), 2);
5042 assert!(
5043 selected.contains(&Token(2)),
5044 "should contain 50s-old session"
5045 );
5046 assert!(
5047 selected.contains(&Token(4)),
5048 "should contain 30s-old session"
5049 );
5050 }
5051
5052 #[test]
5056 fn select_nth_with_count_exceeding_candidates() {
5057 let now = Instant::now();
5058 let mut candidates = [(Token(1), now - Duration::from_secs(10))];
5059
5060 let count = 5;
5061 let pivot = count.min(candidates.len()) - 1;
5062 candidates.select_nth_unstable_by_key(pivot, |&(_, last_event)| last_event);
5063
5064 let selected: HashSet<Token> = candidates[..=pivot]
5065 .iter()
5066 .map(|&(token, _)| token)
5067 .collect();
5068
5069 assert_eq!(selected.len(), 1);
5070 assert!(selected.contains(&Token(1)));
5071 }
5072}
5073
5074#[cfg(test)]
5075mod accept_ready_tests {
5076 use sozu_command::{
5077 config::ListenerBuilder,
5078 proto::command::{DeactivateListener, ListenerType, SocketAddress},
5079 };
5080
5081 use super::*;
5082 use crate::testing::{ServerParts, prebuild_server, provide_port};
5083
5084 pub(super) fn server_with_tcp_listener() -> (Server, Token, SocketAddress) {
5089 let ServerParts {
5090 event_loop,
5091 registry,
5092 sessions,
5093 pool,
5094 backends,
5095 server_scm_socket,
5096 server_config,
5097 ..
5098 } = prebuild_server(16, 16384, true).expect("could not prebuild a test server");
5103 let (_command_channel, proxy_channel) =
5104 Channel::generate(1000, 10000).expect("could not generate a test channel");
5105
5106 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
5107 let listener_config = ListenerBuilder::new_tcp(address)
5108 .to_tcp(None)
5109 .expect("could not build a TcpListenerConfig for the test");
5110
5111 let listen_token = {
5112 let mut session_manager = sessions.borrow_mut();
5113 let entry = session_manager.slab.vacant_entry();
5114 let token = Token(entry.key());
5115 entry.insert(Rc::new(RefCell::new(ListenSession {
5116 protocol: Protocol::TCPListen,
5117 })));
5118 token
5119 };
5120
5121 let mut tcp_proxy =
5122 tcp::TcpProxy::new(registry, sessions.clone(), pool.clone(), backends.clone());
5123 tcp_proxy
5124 .add_listener(listener_config, listen_token)
5125 .expect("could not add the test TCP listener");
5126 tcp_proxy
5127 .activate_listener(&address.into(), None)
5128 .expect("could not activate the test TCP listener");
5129
5130 let server = Server::new(
5131 event_loop,
5132 proxy_channel,
5133 server_scm_socket,
5134 sessions,
5135 pool,
5136 backends,
5137 None,
5138 None,
5139 Some(tcp_proxy),
5140 server_config,
5141 None,
5142 false,
5143 )
5144 .expect("could not build the test server");
5145
5146 (server, listen_token, address)
5149 }
5150
5151 #[test]
5163 fn deactivating_a_listener_drops_its_pending_accept_token() {
5164 let (mut server, listen_token, address) = server_with_tcp_listener();
5165
5166 server.accept_ready.insert(ListenToken(listen_token.0));
5168
5169 let response = server.notify_deactivate_listener(
5170 "test-deactivate",
5171 &DeactivateListener {
5172 interface: None,
5173 address,
5174 proxy: ListenerType::Tcp as i32,
5175 to_scm: false,
5176 },
5177 );
5178 assert_eq!(
5179 response.status,
5180 ResponseStatus::Ok as i32,
5181 "deactivating an activated TCP listener must succeed: {response:?}"
5182 );
5183 assert!(
5184 server.sessions.borrow().slab.contains(listen_token.0),
5185 "deactivating a listener must KEEP its slab slot reserved — see \
5186 `reserve_listen_token`; the accept purge below is what stops the \
5187 deferred pass from accepting on a socket-less listener"
5188 );
5189 assert!(
5190 !server.accept_ready.contains(&ListenToken(listen_token.0)),
5191 "a deactivated listener must not stay queued for a deferred accept"
5192 );
5193
5194 server.handle_remaining_readiness();
5196 }
5197
5198 #[test]
5209 fn a_stale_accept_token_is_dropped_instead_of_panicking() {
5210 let (mut server, listen_token, _address) = server_with_tcp_listener();
5211
5212 server.sessions.borrow_mut().slab.remove(listen_token.0);
5213 server.accept_ready.insert(ListenToken(listen_token.0));
5214
5215 server.handle_remaining_readiness();
5216
5217 assert!(
5218 server.accept_ready.is_empty(),
5219 "a listen token with no session must be dropped from accept_ready"
5220 );
5221 }
5222}
5223#[cfg(test)]
5228mod event_owner_tests {
5229 use super::*;
5230
5231 struct ClosingSession {
5234 sessions: Rc<RefCell<SessionManager>>,
5235 frontend: Token,
5236 backend: Token,
5237 }
5238
5239 impl ProxySession for ClosingSession {
5240 fn protocol(&self) -> Protocol {
5241 Protocol::TCP
5242 }
5243 fn ready(&mut self, _session: Rc<RefCell<dyn ProxySession>>) -> SessionIsToBeClosed {
5244 true
5245 }
5246 fn shutting_down(&mut self) -> SessionIsToBeClosed {
5247 true
5248 }
5249 fn update_readiness(&mut self, _token: Token, _events: Ready) {}
5250 fn close(&mut self) {
5251 self.sessions.borrow_mut().slab.try_remove(self.backend.0);
5252 }
5253 fn timeout(&mut self, _token: Token) -> SessionIsToBeClosed {
5254 false
5255 }
5256 fn last_event(&self) -> Instant {
5257 Instant::now()
5258 }
5259 fn print_session(&self) {}
5260 fn frontend_token(&self) -> Token {
5261 self.frontend
5262 }
5263 }
5264
5265 struct DialingSession {
5269 sessions: Rc<RefCell<SessionManager>>,
5270 frontend: Token,
5271 backend: Option<Token>,
5272 delivered: Vec<Token>,
5273 }
5274
5275 impl ProxySession for DialingSession {
5276 fn protocol(&self) -> Protocol {
5277 Protocol::TCP
5278 }
5279 fn ready(&mut self, session: Rc<RefCell<dyn ProxySession>>) -> SessionIsToBeClosed {
5280 if self.backend.is_none() {
5281 self.backend = Some(Token(self.sessions.borrow_mut().slab.insert(session)));
5282 }
5283 false
5284 }
5285 fn shutting_down(&mut self) -> SessionIsToBeClosed {
5286 true
5287 }
5288 fn update_readiness(&mut self, token: Token, _events: Ready) {
5289 self.delivered.push(token);
5290 }
5291 fn close(&mut self) {}
5292 fn timeout(&mut self, _token: Token) -> SessionIsToBeClosed {
5293 false
5294 }
5295 fn last_event(&self) -> Instant {
5296 Instant::now()
5297 }
5298 fn print_session(&self) {}
5299 fn frontend_token(&self) -> Token {
5300 self.frontend
5301 }
5302 }
5303
5304 #[test]
5312 fn an_event_of_a_closed_session_is_not_delivered_to_the_session_reusing_its_token() {
5313 let (mut server, _listen_token, _address) =
5314 super::accept_ready_tests::server_with_tcp_listener();
5315 let sessions = server.sessions.clone();
5316
5317 let closing = Rc::new(RefCell::new(ClosingSession {
5318 sessions: sessions.clone(),
5319 frontend: Token(0),
5320 backend: Token(0),
5321 }));
5322 let dialing = Rc::new(RefCell::new(DialingSession {
5323 sessions: sessions.clone(),
5324 frontend: Token(0),
5325 backend: None,
5326 delivered: Vec::new(),
5327 }));
5328 {
5329 let mut sessions = sessions.borrow_mut();
5330 let closing_session: Rc<RefCell<dyn ProxySession>> = closing.clone();
5331 let frontend = Token(sessions.slab.insert(closing_session.clone()));
5332 let backend = Token(sessions.slab.insert(closing_session));
5333 let mut closing = closing.borrow_mut();
5334 closing.frontend = frontend;
5335 closing.backend = backend;
5336 let dialing_session: Rc<RefCell<dyn ProxySession>> = dialing.clone();
5337 dialing.borrow_mut().frontend = Token(sessions.slab.insert(dialing_session));
5338 sessions.incr();
5339 sessions.incr();
5340 }
5341 let closed_frontend = closing.borrow().frontend;
5342 let closed_backend = closing.borrow().backend;
5343 let dialing_frontend = dialing.borrow().frontend;
5344
5345 let batch = [
5346 (closed_frontend, Ready::READABLE | Ready::HUP),
5347 (dialing_frontend, Ready::READABLE | Ready::WRITABLE),
5348 (
5349 closed_backend,
5350 Ready::READABLE | Ready::WRITABLE | Ready::HUP,
5351 ),
5352 ];
5353 let mut owners = Vec::new();
5354 server.snapshot_event_owners(batch.iter().map(|(token, _)| *token), &mut owners);
5355 for ((token, events), owner) in batch.into_iter().zip(owners.drain(..)) {
5356 server.ready_if_still_owned(token, events, owner);
5357 }
5358
5359 assert_eq!(
5360 dialing.borrow().backend,
5361 Some(closed_backend),
5362 "precondition: the dial must reuse the key the closed session freed in this batch"
5363 );
5364 assert!(
5365 !sessions.borrow().slab.contains(closed_frontend.0),
5366 "the closed session must have left the slab"
5367 );
5368 assert_eq!(
5369 dialing.borrow().delivered,
5370 vec![dialing_frontend],
5371 "the closed backend's HUP must not reach the session that reused its token"
5372 );
5373 }
5374}
5375#[cfg(test)]
5379mod closing_sessions_tests {
5380 use std::cell::Cell;
5381
5382 use super::*;
5383
5384 struct ScriptedSession {
5387 frontend: Token,
5388 can_close_now: bool,
5389 closed: Rc<Cell<bool>>,
5390 }
5391
5392 impl ProxySession for ScriptedSession {
5393 fn protocol(&self) -> Protocol {
5394 Protocol::HTTP
5395 }
5396 fn ready(&mut self, _session: Rc<RefCell<dyn ProxySession>>) -> SessionIsToBeClosed {
5397 false
5398 }
5399 fn shutting_down(&mut self) -> SessionIsToBeClosed {
5400 self.can_close_now
5401 }
5402 fn update_readiness(&mut self, _token: Token, _events: Ready) {}
5403 fn close(&mut self) {
5404 self.closed.set(true);
5405 }
5406 fn timeout(&mut self, _token: Token) -> SessionIsToBeClosed {
5407 false
5408 }
5409 fn last_event(&self) -> Instant {
5410 Instant::now()
5411 }
5412 fn print_session(&self) {}
5413 fn frontend_token(&self) -> Token {
5414 self.frontend
5415 }
5416 }
5417
5418 fn insert_scripted(
5419 server: &Server,
5420 can_close_now: bool,
5421 ) -> (Token, Rc<RefCell<dyn ProxySession>>, Rc<Cell<bool>>) {
5422 let closed = Rc::new(Cell::new(false));
5423 let mut sessions = server.sessions.borrow_mut();
5424 let entry = sessions.slab.vacant_entry();
5425 let token = Token(entry.key());
5426 let session: Rc<RefCell<dyn ProxySession>> = Rc::new(RefCell::new(ScriptedSession {
5427 frontend: token,
5428 can_close_now,
5429 closed: closed.clone(),
5430 }));
5431 entry.insert(session.clone());
5432 sessions.incr();
5433 (token, session, closed)
5434 }
5435
5436 #[test]
5444 fn a_draining_sessions_reused_slab_key_is_not_closed() {
5445 let (mut server, _listen_token, _address) =
5446 super::accept_ready_tests::server_with_tcp_listener();
5447
5448 let (token, draining, draining_closed) = insert_scripted(&server, true);
5449 server
5450 .closing_sessions
5451 .push((token, Rc::downgrade(&draining)));
5452
5453 server.sessions.borrow_mut().slab.remove(token.0);
5455 let (reused, unrelated, unrelated_closed) = insert_scripted(&server, true);
5456 assert_eq!(reused, token, "precondition: the slab must reuse the key");
5457 assert!(
5458 !Rc::ptr_eq(&draining, &unrelated),
5459 "precondition: the two sessions are distinct"
5460 );
5461
5462 server.drive_closing_sessions();
5463
5464 assert!(
5465 server.sessions.borrow().slab.contains(token.0),
5466 "the unrelated session that reused the key must stay in the slab"
5467 );
5468 assert!(
5469 !unrelated_closed.get() && !draining_closed.get(),
5470 "neither session may be closed by the stale watch"
5471 );
5472 assert!(
5473 server.closing_sessions.is_empty(),
5474 "the stale watch must be dropped, not kept"
5475 );
5476 }
5477
5478 #[test]
5481 fn a_draining_session_still_in_its_slot_is_closed_when_ready() {
5482 let (mut server, _listen_token, _address) =
5483 super::accept_ready_tests::server_with_tcp_listener();
5484
5485 let (token, session, closed) = insert_scripted(&server, true);
5486 server
5487 .closing_sessions
5488 .push((token, Rc::downgrade(&session)));
5489
5490 server.drive_closing_sessions();
5491
5492 assert!(
5493 closed.get(),
5494 "a still-owned closable session must be closed"
5495 );
5496 assert!(!server.sessions.borrow().slab.contains(token.0));
5497 assert!(server.closing_sessions.is_empty());
5498 }
5499}
5500#[cfg(test)]
5505mod listener_lifecycle_tests {
5506 use std::net::UdpSocket as StdUdpSocket;
5507
5508 use sozu_command::{
5509 config::ListenerBuilder,
5510 proto::command::{
5511 ActivateListener, DeactivateListener, ListenerType, LoadBalancingParams,
5512 RemoveListener, RequestUdpFrontend, SocketAddress, filtered_metrics,
5513 },
5514 };
5515
5516 use super::accept_ready_tests::server_with_tcp_listener;
5517 use super::*;
5518 use crate::{
5519 metrics::{METRICS, names},
5520 testing::{ServerParts, prebuild_server, provide_port},
5521 };
5522
5523 pub(super) fn bare_server() -> Server {
5528 let ServerParts {
5529 event_loop,
5530 sessions,
5531 pool,
5532 backends,
5533 server_scm_socket,
5534 server_config,
5535 ..
5536 } = prebuild_server(16, 16384, true).expect("could not prebuild a test server");
5537 let (_command_channel, proxy_channel) =
5538 Channel::generate(1000, 10000).expect("could not generate a test channel");
5539
5540 Server::new(
5541 event_loop,
5542 proxy_channel,
5543 server_scm_socket,
5544 sessions,
5545 pool,
5546 backends,
5547 None,
5548 None,
5549 None,
5550 server_config,
5551 None,
5552 false,
5553 )
5554 .expect("could not build the test server")
5555 }
5556
5557 fn add_udp_listener(server: &mut Server, address: SocketAddress) -> Token {
5558 let listener_config = ListenerBuilder::new_udp(address)
5559 .to_udp(None)
5560 .expect("could not build a UdpListenerConfig for the test");
5561 let response = server.notify_add_udp_listener("test-add-udp", listener_config);
5562 assert_eq!(
5563 response.status,
5564 ResponseStatus::Ok as i32,
5565 "adding a UDP listener must succeed: {response:?}"
5566 );
5567 server
5568 .udp
5569 .borrow()
5570 .listener_token(&address.into())
5571 .expect("the added UDP listener must own a token")
5572 }
5573
5574 fn activate(server: &mut Server, address: SocketAddress, proxy: ListenerType) {
5575 let response = server.notify_activate_listener(
5576 "test-activate",
5577 &ActivateListener {
5578 interface: None,
5579 address,
5580 proxy: proxy as i32,
5581 from_scm: false,
5582 },
5583 );
5584 assert_eq!(
5585 response.status,
5586 ResponseStatus::Ok as i32,
5587 "activating a {proxy:?} listener must succeed: {response:?}"
5588 );
5589 }
5590
5591 fn deactivate(server: &mut Server, address: SocketAddress, proxy: ListenerType) {
5592 let response = server.notify_deactivate_listener(
5593 "test-deactivate",
5594 &DeactivateListener {
5595 interface: None,
5596 address,
5597 proxy: proxy as i32,
5598 to_scm: false,
5599 },
5600 );
5601 assert_eq!(
5602 response.status,
5603 ResponseStatus::Ok as i32,
5604 "deactivating a {proxy:?} listener must succeed: {response:?}"
5605 );
5606 }
5607
5608 fn slot_state(server: &Server, token: Token) -> Option<(Protocol, bool)> {
5614 let sessions = server.sessions.borrow();
5615 let session = sessions.slab.get(token.0)?;
5616 let session = session.borrow();
5617 Some((session.protocol(), session.frontend_token() == Token(0)))
5618 }
5619
5620 #[test]
5641 fn a_reactivated_tcp_listener_is_reachable_from_the_event_loop() {
5642 let (mut server, listen_token, address) = server_with_tcp_listener();
5643
5644 deactivate(&mut server, address, ListenerType::Tcp);
5645 activate(&mut server, address, ListenerType::Tcp);
5646
5647 server.sessions.borrow_mut().can_accept = false;
5648 server.ready(listen_token, Ready::READABLE);
5649
5650 assert!(
5651 server.accept_ready.contains(&ListenToken(listen_token.0)),
5652 "a reactivated listener must still be dispatched by `ready()`"
5653 );
5654 }
5655
5656 #[test]
5668 fn a_reactivated_udp_listener_gets_its_session_back() {
5669 let mut server = bare_server();
5670 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
5671 let listen_token = add_udp_listener(&mut server, address);
5672
5673 assert_eq!(
5674 slot_state(&server, listen_token),
5675 Some((Protocol::UDPListen, true)),
5676 "an added, not-yet-activated UDP listener holds the placeholder"
5677 );
5678
5679 activate(&mut server, address, ListenerType::Udp);
5680 assert_eq!(
5681 slot_state(&server, listen_token),
5682 Some((Protocol::UDPListen, false)),
5683 "activating must install the real UdpListenerSession"
5684 );
5685
5686 deactivate(&mut server, address, ListenerType::Udp);
5687 activate(&mut server, address, ListenerType::Udp);
5688
5689 assert_eq!(
5690 slot_state(&server, listen_token),
5691 Some((Protocol::UDPListen, false)),
5692 "reactivating must reinstall the real UdpListenerSession"
5693 );
5694 }
5695
5696 #[test]
5720 fn a_deactivated_udp_listeners_key_is_not_handed_to_another_session() {
5721 let mut server = bare_server();
5722 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
5723 let listen_token = add_udp_listener(&mut server, address);
5724 activate(&mut server, address, ListenerType::Udp);
5725 deactivate(&mut server, address, ListenerType::Udp);
5726
5727 let interloper = {
5730 let mut sessions = server.sessions.borrow_mut();
5731 let entry = sessions.slab.vacant_entry();
5732 let key = entry.key();
5733 entry.insert(Rc::new(RefCell::new(ListenSession {
5734 protocol: Protocol::Channel,
5735 })));
5736 Token(key)
5737 };
5738 assert_ne!(
5739 interloper, listen_token,
5740 "a deactivated listener's key must not be handed to another session"
5741 );
5742
5743 activate(&mut server, address, ListenerType::Udp);
5744
5745 assert_eq!(
5746 slot_state(&server, interloper),
5747 Some((Protocol::Channel, true)),
5748 "reactivating a listener must not overwrite another session's slot"
5749 );
5750 }
5751
5752 const UDP_FLOW_CLUSTER: &str = "udp-duplicate-activate";
5754
5755 fn slab_session(server: &Server, token: Token) -> Rc<RefCell<dyn ProxySession>> {
5761 server
5762 .sessions
5763 .borrow()
5764 .slab
5765 .get(token.0)
5766 .expect("the listener must own a slab slot")
5767 .clone()
5768 }
5769
5770 fn active_flows_gauge() -> i64 {
5774 METRICS.with(|metrics| {
5775 metrics
5776 .borrow_mut()
5777 .dump_local_proxy_metrics()
5778 .get(names::udp::ACTIVE_FLOWS)
5779 .and_then(|metric| match metric.inner {
5780 Some(filtered_metrics::Inner::Gauge(value)) => Some(value as i64),
5781 _ => None,
5782 })
5783 .unwrap_or(0)
5784 })
5785 }
5786
5787 fn udp_worker_with_backend(
5793 server: &mut Server,
5794 address: SocketAddress,
5795 ) -> (StdUdpSocket, StdUdpSocket) {
5796 server.notify_proxys(WorkerRequest {
5798 id: "test-add-udp-cluster".to_owned(),
5799 content: RequestType::AddCluster(Cluster {
5800 cluster_id: UDP_FLOW_CLUSTER.to_owned(),
5801 ..Default::default()
5802 })
5803 .into(),
5804 });
5805 server.notify_proxys(WorkerRequest {
5806 id: "test-add-udp-front".to_owned(),
5807 content: RequestType::AddUdpFrontend(RequestUdpFrontend {
5808 cluster_id: UDP_FLOW_CLUSTER.to_owned(),
5809 address,
5810 tags: Default::default(),
5811 })
5812 .into(),
5813 });
5814
5815 let backend_socket =
5816 StdUdpSocket::bind("127.0.0.1:0").expect("could not bind the test UDP backend");
5817 backend_socket
5818 .set_read_timeout(Some(Duration::from_millis(500)))
5819 .expect("could not set the backend read timeout");
5820 let backend_address = backend_socket
5821 .local_addr()
5822 .expect("the test UDP backend must have a local address");
5823
5824 server.notify_proxys(WorkerRequest {
5825 id: "test-add-udp-backend".to_owned(),
5826 content: RequestType::AddBackend(AddBackend {
5827 cluster_id: UDP_FLOW_CLUSTER.to_owned(),
5828 backend_id: format!("{UDP_FLOW_CLUSTER}-0"),
5829 address: backend_address.into(),
5830 load_balancing_parameters: Some(LoadBalancingParams::default()),
5831 sticky_id: None,
5832 backup: None,
5833 unix_path: None,
5834 })
5835 .into(),
5836 });
5837
5838 let client_socket =
5839 StdUdpSocket::bind("127.0.0.1:0").expect("could not bind the test UDP client");
5840 (client_socket, backend_socket)
5841 }
5842
5843 fn forward_one(
5849 server: &mut Server,
5850 listen_token: Token,
5851 listen_address: SocketAddress,
5852 client: &StdUdpSocket,
5853 backend: &StdUdpSocket,
5854 payload: &[u8],
5855 ) -> Option<Vec<u8>> {
5856 let target: std::net::SocketAddr = listen_address.into();
5857 client
5858 .send_to(payload, target)
5859 .expect("the test client must reach the UDP listener");
5860 server.ready(listen_token, Ready::READABLE);
5861
5862 let mut received = [0u8; 64];
5863 backend
5864 .recv_from(&mut received)
5865 .ok()
5866 .map(|(len, _)| received[..len].to_vec())
5867 }
5868
5869 #[test]
5884 fn a_duplicate_udp_activate_keeps_the_live_listener_session() {
5885 let mut server = bare_server();
5886 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
5887 let listen_token = add_udp_listener(&mut server, address);
5888 activate(&mut server, address, ListenerType::Udp);
5889
5890 let installed = slab_session(&server, listen_token);
5891
5892 let response = server.notify_activate_listener(
5893 "test-activate-twice",
5894 &ActivateListener {
5895 interface: None,
5896 address,
5897 proxy: ListenerType::Udp as i32,
5898 from_scm: false,
5899 },
5900 );
5901 assert_eq!(
5908 response.status,
5909 ResponseStatus::Ok as i32,
5910 "a repeated ActivateListener on an active UDP listener must answer ok: {response:?}"
5911 );
5912 assert_eq!(
5913 slot_state(&server, listen_token),
5914 Some((Protocol::UDPListen, false)),
5915 "the real UdpListenerSession must still be the one in the slab"
5916 );
5917 assert!(
5918 Rc::ptr_eq(&installed, &slab_session(&server, listen_token)),
5919 "a repeated ActivateListener must not replace the live UDP listener session"
5920 );
5921 }
5922
5923 #[test]
5935 fn in_flight_udp_flows_survive_a_duplicate_udp_activate() {
5936 let mut server = bare_server();
5937 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
5938 let listen_token = add_udp_listener(&mut server, address);
5939 activate(&mut server, address, ListenerType::Udp);
5940 let (client, backend) = udp_worker_with_backend(&mut server, address);
5941
5942 assert_eq!(
5943 forward_one(
5944 &mut server,
5945 listen_token,
5946 address,
5947 &client,
5948 &backend,
5949 b"before"
5950 )
5951 .as_deref(),
5952 Some(&b"before"[..]),
5953 "the flow must be established before the duplicate activate"
5954 );
5955 assert_eq!(
5956 active_flows_gauge(),
5957 1,
5958 "one established flow must be on the gauge"
5959 );
5960
5961 activate(&mut server, address, ListenerType::Udp);
5962
5963 assert_eq!(
5968 active_flows_gauge(),
5969 1,
5970 "a repeated ActivateListener must neither open nor evict a flow"
5971 );
5972 assert_eq!(
5973 forward_one(
5974 &mut server,
5975 listen_token,
5976 address,
5977 &client,
5978 &backend,
5979 b"after"
5980 )
5981 .as_deref(),
5982 Some(&b"after"[..]),
5983 "an in-flight UDP flow must survive a repeated ActivateListener"
5984 );
5985 }
5986
5987 #[test]
5998 fn a_duplicate_udp_activate_strands_no_upstream_slab_slot() {
5999 let mut server = bare_server();
6000 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
6001 let listen_token = add_udp_listener(&mut server, address);
6002 activate(&mut server, address, ListenerType::Udp);
6003 let (client, backend) = udp_worker_with_backend(&mut server, address);
6004
6005 let baseline = server.sessions.borrow().slab.len();
6008
6009 assert!(
6010 forward_one(
6011 &mut server,
6012 listen_token,
6013 address,
6014 &client,
6015 &backend,
6016 b"flow"
6017 )
6018 .is_some(),
6019 "the flow must be established before the duplicate activate"
6020 );
6021 assert_eq!(
6022 server.sessions.borrow().slab.len(),
6023 baseline + 1,
6024 "an established flow owns exactly one upstream slab slot"
6025 );
6026
6027 activate(&mut server, address, ListenerType::Udp);
6028 deactivate(&mut server, address, ListenerType::Udp);
6029
6030 assert_eq!(
6031 server.sessions.borrow().slab.len(),
6032 baseline,
6033 "tearing the listener down must release every per-flow slab slot"
6034 );
6035 assert_eq!(
6036 active_flows_gauge(),
6037 0,
6038 "tearing the listener down must return the active-flows gauge to zero"
6039 );
6040 }
6041
6042 #[test]
6058 fn replaying_activate_listener_is_a_clean_no_op_for_an_active_udp_listener() {
6059 let mut server = bare_server();
6060 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
6061 let listener_config = ListenerBuilder::new_udp(address)
6062 .to_udp(None)
6063 .expect("could not build a UdpListenerConfig for the test");
6064
6065 server.notify_proxys(WorkerRequest {
6068 id: "test-add-udp".to_owned(),
6069 content: RequestType::AddUdpListener(listener_config).into(),
6070 });
6071 server.notify_proxys(WorkerRequest {
6072 id: "test-activate".to_owned(),
6073 content: RequestType::ActivateListener(ActivateListener {
6074 interface: None,
6075 address,
6076 proxy: ListenerType::Udp as i32,
6077 from_scm: false,
6078 })
6079 .into(),
6080 });
6081 let listen_token = server
6082 .udp
6083 .borrow()
6084 .listener_token(&address.into())
6085 .expect("the added UDP listener must own a token");
6086 let installed = slab_session(&server, listen_token);
6087
6088 let replayed: Vec<ActivateListener> = server
6089 .config_state
6090 .generate_activate_requests()
6091 .into_iter()
6092 .filter_map(|request| match request.request_type {
6093 Some(RequestType::ActivateListener(activate)) => Some(activate),
6094 _ => None,
6095 })
6096 .collect();
6097 assert_eq!(
6098 replayed.len(),
6099 1,
6100 "an active UDP listener must be replayed exactly once: {replayed:?}"
6101 );
6102
6103 for activate in &replayed {
6104 let response = server.notify_activate_listener("test-replay", activate);
6105 assert_eq!(
6106 response.status,
6107 ResponseStatus::Ok as i32,
6108 "a replayed ActivateListener must answer ok: {response:?}"
6109 );
6110 }
6111
6112 assert_eq!(
6113 slot_state(&server, listen_token),
6114 Some((Protocol::UDPListen, false)),
6115 "the replay must leave the real UdpListenerSession in the slab"
6116 );
6117 assert!(
6118 Rc::ptr_eq(&installed, &slab_session(&server, listen_token)),
6119 "replaying ActivateListener must leave the live UDP listener session in place"
6120 );
6121 }
6122
6123 #[test]
6142 fn removing_a_listener_releases_its_reserved_slab_slot() {
6143 let mut server = bare_server();
6144 let baseline = server.sessions.borrow().slab.len();
6145 let base_sessions_count = server.base_sessions_count;
6146
6147 for cycle in 0..3 {
6148 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
6149 let listen_token = add_udp_listener(&mut server, address);
6150 activate(&mut server, address, ListenerType::Udp);
6151 if cycle % 2 == 0 {
6152 deactivate(&mut server, address, ListenerType::Udp);
6153 }
6154
6155 server.notify_proxys(WorkerRequest {
6156 id: "test-remove".to_owned(),
6157 content: RequestType::RemoveListener(RemoveListener {
6158 interface: None,
6159 address,
6160 proxy: ListenerType::Udp as i32,
6161 })
6162 .into(),
6163 });
6164
6165 assert!(
6166 !server.sessions.borrow().slab.contains(listen_token.0),
6167 "cycle {cycle}: removing a listener must release its reserved slot"
6168 );
6169 assert_eq!(
6170 server.sessions.borrow().slab.len(),
6171 baseline,
6172 "cycle {cycle}: the slab must return to its baseline occupancy"
6173 );
6174 assert_eq!(
6175 server.base_sessions_count, base_sessions_count,
6176 "cycle {cycle}: base_sessions_count must track the reserved slots"
6177 );
6178 }
6179 }
6180
6181 #[test]
6192 fn udp_flow_limits_come_from_the_cluster_not_the_global_defaults() {
6193 use std::{net::SocketAddr, time::Instant};
6194
6195 use crate::protocol::udp::{
6196 BackendId, BackendSource, ManagerInput, MetricEvent, Output, UdpManager,
6197 };
6198
6199 struct OneBackend;
6200 impl BackendSource for OneBackend {
6201 fn select(
6202 &mut self,
6203 _cluster: &str,
6204 _key: Option<u64>,
6205 _now: Instant,
6206 ) -> Option<(BackendId, SocketAddr)> {
6207 Some(("b".to_owned(), SocketAddr::from(([127, 0, 0, 1], 5353))))
6208 }
6209 }
6210 fn admit(manager: &RefCell<UdpManager>, port: u16) -> bool {
6213 let mut manager = manager.borrow_mut();
6214 manager.handle_input(
6215 ManagerInput::ClientDatagram {
6216 src: SocketAddr::from(([192, 0, 2, 1], port)),
6217 payload: b"q",
6218 backends: &mut OneBackend,
6219 },
6220 Instant::now(),
6221 );
6222 let mut opened = false;
6223 let mut shed = false;
6224 while let Some(out) = manager.poll_output() {
6225 opened |= matches!(out, Output::OpenUpstream { .. });
6226 shed |= matches!(out, Output::Metric(MetricEvent::FlowShedSourceLimit));
6227 }
6228 assert_ne!(opened, shed, "a datagram is either admitted or shed");
6229 opened
6230 }
6231 fn add_cluster(server: &mut Server, max_connections_per_ip: Option<u64>) {
6232 server.notify_proxys(WorkerRequest {
6233 id: "test-add-cluster".to_owned(),
6234 content: RequestType::AddCluster(Cluster {
6235 cluster_id: "udp-limits".to_owned(),
6236 max_connections_per_ip,
6237 ..Default::default()
6238 })
6239 .into(),
6240 });
6241 }
6242
6243 let mut server = bare_server();
6244 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
6245 let token = add_udp_listener(&mut server, address);
6246 server.notify(WorkerRequest {
6247 id: "test-set-per-ip".to_owned(),
6248 content: RequestType::SetMaxConnectionsPerIp(1).into(),
6249 });
6250 server.notify(WorkerRequest {
6251 id: "test-set-per-subnet".to_owned(),
6252 content: RequestType::SetMaxConnectionsPerSubnet(1).into(),
6253 });
6254 add_cluster(&mut server, None);
6255 server.notify_proxys(WorkerRequest {
6256 id: "test-add-udp-front".to_owned(),
6257 content: RequestType::AddUdpFrontend(RequestUdpFrontend {
6258 cluster_id: "udp-limits".to_owned(),
6259 address,
6260 tags: Default::default(),
6261 })
6262 .into(),
6263 });
6264 let manager = server
6265 .udp
6266 .borrow()
6267 .manager(token)
6268 .expect("the UDP listener must own a flow manager");
6269
6270 {
6271 let manager = manager.borrow();
6272 let cfg = manager.cluster_config();
6273 assert_eq!(
6274 cfg.max_flows_per_ip, 0,
6275 "the global per-IP default must not apply to UDP"
6276 );
6277 assert_eq!(
6278 cfg.max_flows_per_subnet, 0,
6279 "the global per-subnet default must not apply to UDP"
6280 );
6281 }
6282 assert!(admit(&manager, 1000));
6283 assert!(
6284 admit(&manager, 1001),
6285 "a global limit of 1 must not shed a second UDP flow"
6286 );
6287
6288 add_cluster(&mut server, Some(2));
6289 assert_eq!(manager.borrow().cluster_config().max_flows_per_ip, 2);
6290 assert!(
6291 !admit(&manager, 1002),
6292 "the cluster's own limit of 2 counts the two open flows and sheds a third"
6293 );
6294 }
6295}
6296
6297#[cfg(test)]
6298mod add_cluster_validation_tests {
6299 use sozu_command::proto::command::{LoadBalancingParams, ShardMode};
6300
6301 use super::listener_lifecycle_tests::bare_server;
6302 use super::*;
6303 use crate::{backends::ShardOutcome, load_balancing::affinity_key_from_ip};
6304
6305 const CLUSTER: &str = "sharded";
6306 const ADD_CLUSTER_ID: &str = "test-add-invalid-shard";
6307
6308 #[test]
6318 fn an_add_cluster_with_an_invalid_shard_percent_is_refused_and_unsharded() {
6319 let mut server = bare_server();
6320 server.notify_proxys(WorkerRequest {
6324 id: "test-add-unsharded-cluster".to_owned(),
6325 content: RequestType::AddCluster(Cluster {
6326 cluster_id: CLUSTER.to_owned(),
6327 load_balancing: LoadBalancingAlgorithms::RoundRobin as i32,
6328 ..Default::default()
6329 })
6330 .into(),
6331 });
6332 server.notify_proxys(WorkerRequest {
6333 id: ADD_CLUSTER_ID.to_owned(),
6334 content: RequestType::AddCluster(Cluster {
6335 cluster_id: CLUSTER.to_owned(),
6336 load_balancing: LoadBalancingAlgorithms::RoundRobin as i32,
6337 shard_percent: Some(0),
6340 shard_min_backends: Some(2),
6341 shard_mode: Some(ShardMode::Strict as i32),
6342 ..Default::default()
6343 })
6344 .into(),
6345 });
6346 let response = QUEUE
6347 .with(|queue| {
6348 queue
6349 .borrow()
6350 .iter()
6351 .find(|response| response.id == ADD_CLUSTER_ID)
6352 .cloned()
6353 })
6354 .expect("the worker must answer the AddCluster");
6355 assert_eq!(
6356 response.status,
6357 ResponseStatus::Failure as i32,
6358 "an invalid shard_percent must be refused: {response:?}"
6359 );
6360 assert_eq!(response.message, "shard_percent must lie in 1..=100");
6361
6362 for index in 0..4u16 {
6363 server.notify_proxys(WorkerRequest {
6364 id: format!("test-add-backend-{index}"),
6365 content: RequestType::AddBackend(AddBackend {
6366 cluster_id: CLUSTER.to_owned(),
6367 backend_id: format!("{CLUSTER}-{index}"),
6368 address: SocketAddr::from(([127, 0, 0, 1], 23_000 + index)).into(),
6369 load_balancing_parameters: Some(LoadBalancingParams::default()),
6370 sticky_id: None,
6371 backup: None,
6372 unix_path: None,
6373 })
6374 .into(),
6375 });
6376 }
6377 let key = affinity_key_from_ip(IpAddr::V4(Ipv4Addr::LOCALHOST));
6378 let mut backends = server.backends.borrow_mut();
6379 let list = backends
6380 .backends
6381 .get_mut(CLUSTER)
6382 .expect("the backends must have created the cluster's list");
6383 assert_eq!(
6384 list.select_with_key(Some(key), Instant::now()).1,
6385 ShardOutcome::Unsharded,
6386 "a refused AddCluster must not arm a shard"
6387 );
6388 }
6389}
6390
6391#[cfg(test)]
6395mod health_check_port_unix_tests {
6396 use sozu_command::{
6397 config::HEALTH_CHECK_PORT_UNIX_REFUSED,
6398 proto::command::{HealthCheckConfig, HealthCheckMode, SetHealthCheck, SocketAddress},
6399 };
6400
6401 use super::listener_lifecycle_tests::bare_server;
6402 use super::*;
6403
6404 const CLUSTER: &str = "port-unix-cluster";
6405
6406 fn probe(port: Option<u32>) -> HealthCheckConfig {
6407 HealthCheckConfig {
6408 interval: 10,
6409 timeout: 5,
6410 healthy_threshold: 3,
6411 unhealthy_threshold: 3,
6412 mode: HealthCheckMode::Tcp as i32,
6413 port,
6414 ..Default::default()
6415 }
6416 }
6417
6418 fn cluster(health_check: Option<HealthCheckConfig>) -> RequestType {
6419 RequestType::AddCluster(Cluster {
6420 cluster_id: CLUSTER.to_owned(),
6421 health_check,
6422 ..Default::default()
6423 })
6424 }
6425
6426 fn unix_backend() -> RequestType {
6427 RequestType::AddUnixBackend(AddBackend {
6428 cluster_id: CLUSTER.to_owned(),
6429 backend_id: format!("{CLUSTER}-0"),
6430 address: SocketAddress::new_v4(127, 0, 0, 1, 1),
6431 unix_path: Some("/run/app.sock".to_owned()),
6432 ..Default::default()
6433 })
6434 }
6435
6436 fn send(server: &mut Server, request: RequestType) -> Vec<WorkerResponse> {
6437 server.notify_proxys(WorkerRequest {
6438 id: "test-request".to_owned(),
6439 content: request.into(),
6440 });
6441 QUEUE.with(|queue| {
6442 queue
6443 .borrow_mut()
6444 .drain(..)
6445 .filter(|response| response.id != "EVENT")
6446 .collect()
6447 })
6448 }
6449
6450 fn snapshot(server: &Server) -> (String, Vec<*const RefCell<Backend>>, Vec<String>, usize) {
6454 let backends = server.backends.borrow();
6455 (
6456 format!(
6457 "{:?}{:?}",
6458 server.config_state.clusters, server.config_state.backends
6459 ),
6460 backends
6461 .backends
6462 .get(CLUSTER)
6463 .map(|list| list.backends.iter().map(Rc::as_ptr).collect())
6464 .unwrap_or_default(),
6465 backends
6466 .health_check_configs
6467 .values()
6468 .map(|config| format!("{config:?}"))
6469 .collect(),
6470 server.health_checker.in_flight_len(),
6471 )
6472 }
6473
6474 fn assert_refused(server: &mut Server, request: RequestType) {
6475 let before = snapshot(server);
6476 let responses = send(server, request);
6477 assert_eq!(responses.len(), 1, "one answer expected: {responses:?}");
6478 assert_eq!(responses[0].status, ResponseStatus::Failure as i32);
6479 assert!(
6480 responses[0]
6481 .message
6482 .contains(HEALTH_CHECK_PORT_UNIX_REFUSED),
6483 "the answer names the refusal: {responses:?}"
6484 );
6485 assert_eq!(before, snapshot(server), "a refusal changes nothing");
6486 }
6487
6488 #[test]
6489 fn a_probe_port_already_set_refuses_a_unix_backend() {
6490 let mut server = bare_server();
6491 send(&mut server, cluster(Some(probe(Some(8081)))));
6492 assert_refused(&mut server, unix_backend());
6493 assert!(snapshot(&server).1.is_empty(), "no backend was added");
6494 }
6495
6496 #[test]
6497 fn a_unix_backend_already_there_refuses_a_probe_port() {
6498 let mut server = bare_server();
6499 send(&mut server, cluster(Some(probe(None))));
6500 send(&mut server, unix_backend());
6501 assert_eq!(snapshot(&server).1.len(), 1, "test setup: the unix backend");
6502
6503 assert_refused(
6504 &mut server,
6505 RequestType::SetHealthCheck(SetHealthCheck {
6506 cluster_id: CLUSTER.to_owned(),
6507 config: probe(Some(8081)),
6508 }),
6509 );
6510 assert_refused(&mut server, cluster(Some(probe(Some(8081)))));
6511 assert_eq!(
6512 server.backends.borrow().health_check_configs[CLUSTER].port,
6513 None,
6514 "the previous probe config stays"
6515 );
6516 }
6517}
6518
6519#[cfg(test)]
6522mod remove_health_check_tests {
6523 use sozu_command::proto::command::{HealthCheckConfig, LoadBalancingParams};
6524
6525 use super::listener_lifecycle_tests::bare_server;
6526 use super::*;
6527
6528 const CLUSTER: &str = "health-checked";
6529
6530 fn health_check() -> HealthCheckConfig {
6531 HealthCheckConfig {
6532 uri: "/livez".to_owned(),
6533 interval: 1,
6534 timeout: 1,
6535 healthy_threshold: 1,
6536 unhealthy_threshold: 1,
6537 ..Default::default()
6538 }
6539 }
6540
6541 fn add_cluster(server: &mut Server, health_check: Option<HealthCheckConfig>) {
6542 server.notify_proxys(WorkerRequest {
6543 id: "test-add-cluster".to_owned(),
6544 content: RequestType::AddCluster(Cluster {
6545 cluster_id: CLUSTER.to_owned(),
6546 health_check,
6547 ..Default::default()
6548 })
6549 .into(),
6550 });
6551 }
6552
6553 fn server_with_down_backend() -> Server {
6556 let mut server = bare_server();
6557 add_cluster(&mut server, Some(health_check()));
6558 server.notify_proxys(WorkerRequest {
6559 id: "test-add-backend".to_owned(),
6560 content: RequestType::AddBackend(AddBackend {
6561 cluster_id: CLUSTER.to_owned(),
6562 backend_id: format!("{CLUSTER}-0"),
6563 address: SocketAddr::from(([127, 0, 0, 1], 23_100)).into(),
6564 load_balancing_parameters: Some(LoadBalancingParams::default()),
6565 sticky_id: None,
6566 backup: None,
6567 unix_path: None,
6568 })
6569 .into(),
6570 });
6571 server.backends.borrow().backends[CLUSTER].backends[0]
6572 .borrow_mut()
6573 .health
6574 .record_failure(1);
6575 assert!(!backend_is_healthy(&server), "test setup: backend DOWN");
6576 server
6577 }
6578
6579 fn backend_is_healthy(server: &Server) -> bool {
6580 server.backends.borrow().backends[CLUSTER].backends[0]
6581 .borrow()
6582 .health
6583 .is_healthy()
6584 }
6585
6586 #[test]
6590 fn remove_health_check_resets_a_backend_marked_down() {
6591 let mut server = server_with_down_backend();
6592 server.notify_proxys(WorkerRequest {
6593 id: "test-remove-health-check".to_owned(),
6594 content: RequestType::RemoveHealthCheck(CLUSTER.to_owned()).into(),
6595 });
6596 assert!(
6597 !server
6598 .backends
6599 .borrow()
6600 .health_check_configs
6601 .contains_key(CLUSTER),
6602 "RemoveHealthCheck must forget the cluster's configuration"
6603 );
6604 assert!(
6605 backend_is_healthy(&server),
6606 "RemoveHealthCheck must reset a backend a probe marked DOWN"
6607 );
6608 }
6609
6610 #[test]
6616 fn add_cluster_without_health_check_resets_a_backend_marked_down() {
6617 let mut server = server_with_down_backend();
6618 add_cluster(&mut server, None);
6619 assert!(
6620 !server
6621 .backends
6622 .borrow()
6623 .health_check_configs
6624 .contains_key(CLUSTER),
6625 "an AddCluster without a health check must forget the configuration"
6626 );
6627 assert!(
6628 backend_is_healthy(&server),
6629 "an AddCluster without a health check must reset a backend marked DOWN"
6630 );
6631 }
6632
6633 #[test]
6640 fn an_idempotent_reload_keeps_a_backend_marked_down_and_its_counters() {
6641 let ported = HealthCheckConfig {
6642 port: Some(8081),
6643 ..health_check()
6644 };
6645 let mut server = bare_server();
6646 add_cluster(&mut server, Some(ported.clone()));
6647 let add_backend = || WorkerRequest {
6648 id: "test-add-backend".to_owned(),
6649 content: RequestType::AddBackend(AddBackend {
6650 cluster_id: CLUSTER.to_owned(),
6651 backend_id: format!("{CLUSTER}-0"),
6652 address: SocketAddr::from(([127, 0, 0, 1], 23_100)).into(),
6653 load_balancing_parameters: Some(LoadBalancingParams::default()),
6654 sticky_id: None,
6655 backup: None,
6656 unix_path: None,
6657 })
6658 .into(),
6659 };
6660 server.notify_proxys(add_backend());
6661 {
6662 let backends = server.backends.borrow();
6663 let mut backend = backends.backends[CLUSTER].backends[0].borrow_mut();
6664 for _ in 0..3 {
6667 backend.health.record_failure(3);
6668 }
6669 }
6670 let before = server.backends.borrow().backends[CLUSTER].backends[0]
6671 .borrow()
6672 .health
6673 .clone();
6674 assert!(!before.is_healthy(), "test setup: backend DOWN");
6675 assert_eq!(before.consecutive_failures, 3, "test setup: counters");
6676
6677 add_cluster(&mut server, Some(ported.clone()));
6679 server.notify_proxys(add_backend());
6680
6681 let after = server.backends.borrow().backends[CLUSTER].backends[0]
6682 .borrow()
6683 .health
6684 .clone();
6685 assert_eq!(after, before, "a reload must not reset the runtime health");
6686 assert_eq!(
6687 server.backends.borrow().health_check_configs[CLUSTER],
6688 ported,
6689 "the health check configuration stays"
6690 );
6691 }
6692}
6693
6694#[cfg(test)]
6701mod remove_cluster_cascade_tests {
6702 use sozu_command::{
6703 config::ListenerBuilder,
6704 proto::command::{
6705 LoadBalancingParams, PathRule, RequestHttpFrontend, RequestTcpFrontend,
6706 RequestUdpFrontend, RulePosition, SocketAddress,
6707 },
6708 };
6709
6710 use super::listener_lifecycle_tests::bare_server;
6711 use super::*;
6712 use crate::{L7ListenerHandler, protocol::http::parser::Method, testing::provide_port};
6713
6714 const REMOVED: &str = "cascade_removed";
6715 const KEPT: &str = "cascade_kept";
6716
6717 fn send(server: &mut Server, id: &str, request: RequestType) -> Vec<WorkerResponse> {
6721 server.notify_proxys(WorkerRequest {
6722 id: id.to_owned(),
6723 content: request.into(),
6724 });
6725 QUEUE.with(|queue| {
6726 queue
6727 .borrow_mut()
6728 .drain(..)
6729 .filter(|response| response.id != "EVENT")
6730 .collect()
6731 })
6732 }
6733
6734 fn http_front(cluster_id: Option<&str>, hostname: &str, path: &str) -> RequestHttpFrontend {
6735 RequestHttpFrontend {
6736 cluster_id: cluster_id.map(str::to_owned),
6737 hostname: hostname.to_owned(),
6738 path: PathRule::prefix(path.to_owned()),
6739 address: SocketAddress::new_v4(127, 0, 0, 1, 0),
6740 ..Default::default()
6741 }
6742 }
6743
6744 #[test]
6749 fn a_proxy_rejection_rolls_back_reversible_worker_state() {
6750 let cluster_id = "worker-rollback";
6751 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
6752 let http = http_front(Some(cluster_id), "rollback.test", "/");
6753
6754 let requests = [
6755 RequestType::AddHttpFrontend(RequestHttpFrontend {
6756 address,
6757 ..http.clone()
6758 }),
6759 RequestType::AddHttpsFrontend(RequestHttpFrontend { address, ..http }),
6760 RequestType::AddTcpFrontend(RequestTcpFrontend {
6761 cluster_id: cluster_id.to_owned(),
6762 address,
6763 ..Default::default()
6764 }),
6765 RequestType::AddUdpFrontend(RequestUdpFrontend {
6766 cluster_id: cluster_id.to_owned(),
6767 address,
6768 ..Default::default()
6769 }),
6770 ];
6771
6772 for (index, request) in requests.into_iter().enumerate() {
6773 let mut server = bare_server();
6774 for response in send(
6775 &mut server,
6776 &format!("cluster-{index}"),
6777 RequestType::AddCluster(Cluster {
6778 cluster_id: cluster_id.to_owned(),
6779 ..Default::default()
6780 }),
6781 ) {
6782 assert_eq!(response.status, ResponseStatus::Ok as i32);
6783 }
6784
6785 let responses = send(&mut server, &format!("frontend-{index}"), request);
6786 assert_eq!(
6787 responses
6788 .iter()
6789 .map(|response| response.status)
6790 .collect::<Vec<_>>(),
6791 vec![ResponseStatus::Failure as i32],
6792 "a frontend with no listener must be refused: {responses:?}"
6793 );
6794
6795 let cluster = server
6796 .config_state
6797 .cluster_state(cluster_id)
6798 .expect("the setup cluster must stay in worker state");
6799 assert!(
6800 cluster.http_frontends.is_empty()
6801 && cluster.https_frontends.is_empty()
6802 && cluster.tcp_frontends.is_empty()
6803 && cluster.udp_frontends.is_empty(),
6804 "a refused frontend must not remain in worker ConfigState: {cluster:?}"
6805 );
6806 }
6807 }
6808
6809 #[test]
6814 fn a_refused_frontend_removal_restores_the_exact_stored_frontend() {
6815 let cluster_id = "worker-remove-rollback";
6816 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
6817 let tags = std::collections::BTreeMap::from([("owner".to_owned(), "preserved".to_owned())]);
6818 let mut http = http_front(Some(cluster_id), "remove.test", "/api");
6819 http.address = address;
6820 http.method = Some("POST".to_owned());
6821 http.position = RulePosition::Post.into();
6822 http.tags = tags.clone();
6823 http.rewrite_host = Some("internal.test".to_owned());
6824
6825 let mut http_selector = http.clone();
6826 http_selector.position = RulePosition::Tree.into();
6827 http_selector.tags.clear();
6828 http_selector.rewrite_host = None;
6829
6830 let mut https = http.clone();
6831 https.hostname = "secure-remove.test".to_owned();
6832 let mut https_selector = https.clone();
6833 https_selector.position = RulePosition::Tree.into();
6834 https_selector.tags.clear();
6835 https_selector.rewrite_host = None;
6836
6837 let tcp = RequestTcpFrontend {
6838 cluster_id: cluster_id.to_owned(),
6839 address,
6840 tags: tags.clone(),
6841 sni: Some("REMOVE.TEST".to_owned()),
6842 alpn: vec!["http/1.1".to_owned(), "h2".to_owned()],
6843 };
6844 let tcp_selector = RequestTcpFrontend {
6845 tags: Default::default(),
6846 sni: Some("remove.test".to_owned()),
6847 alpn: vec!["h2".to_owned(), "http/1.1".to_owned()],
6848 ..tcp.clone()
6849 };
6850
6851 let udp = RequestUdpFrontend {
6852 cluster_id: cluster_id.to_owned(),
6853 address,
6854 tags,
6855 };
6856
6857 let cases = [
6858 (
6859 "http",
6860 RequestType::AddHttpFrontend(http),
6861 RequestType::RemoveHttpFrontend(http_selector),
6862 ),
6863 (
6864 "https",
6865 RequestType::AddHttpsFrontend(https),
6866 RequestType::RemoveHttpsFrontend(https_selector),
6867 ),
6868 (
6869 "tcp",
6870 RequestType::AddTcpFrontend(tcp),
6871 RequestType::RemoveTcpFrontend(tcp_selector),
6872 ),
6873 (
6874 "udp",
6875 RequestType::AddUdpFrontend(udp.clone()),
6876 RequestType::RemoveUdpFrontend(udp),
6877 ),
6878 ];
6879
6880 for (protocol, add, remove) in cases {
6881 let mut server = bare_server();
6882 server
6883 .config_state
6884 .dispatch(
6885 &RequestType::AddCluster(Cluster {
6886 cluster_id: cluster_id.to_owned(),
6887 ..Default::default()
6888 })
6889 .into(),
6890 )
6891 .expect("the setup cluster must enter worker state");
6892 server
6893 .config_state
6894 .dispatch(&add.into())
6895 .unwrap_or_else(|error| {
6896 panic!("the {protocol} setup frontend must be stored: {error}")
6897 });
6898 let before = server
6899 .config_state
6900 .cluster_state(cluster_id)
6901 .expect("the setup cluster must expose its frontend");
6902
6903 let responses = send(&mut server, &format!("remove-{protocol}"), remove);
6904 assert_eq!(
6905 responses
6906 .iter()
6907 .map(|response| response.status)
6908 .collect::<Vec<_>>(),
6909 vec![ResponseStatus::Failure as i32],
6910 "a frontend removal with no runtime listener must be refused: {responses:?}"
6911 );
6912 assert_eq!(
6913 server.config_state.cluster_state(cluster_id),
6914 Some(before),
6915 "a refused {protocol} removal must restore the complete stored frontend"
6916 );
6917 }
6918 }
6919
6920 #[test]
6925 fn a_refused_duplicate_frontend_add_preserves_the_preexisting_frontend() {
6926 let cluster_id = "worker-add-rollback";
6927 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
6928 let tags =
6929 std::collections::BTreeMap::from([("owner".to_owned(), "preexisting".to_owned())]);
6930 let mut http = http_front(Some(cluster_id), "duplicate.test", "/api");
6931 http.address = address;
6932 http.method = Some("PATCH".to_owned());
6933 http.position = RulePosition::Pre.into();
6934 http.tags = tags.clone();
6935
6936 let mut https = http.clone();
6937 https.hostname = "secure-duplicate.test".to_owned();
6938 let tcp = RequestTcpFrontend {
6939 cluster_id: cluster_id.to_owned(),
6940 address,
6941 tags: tags.clone(),
6942 sni: Some("duplicate.test".to_owned()),
6943 alpn: vec!["h2".to_owned()],
6944 };
6945 let udp = RequestUdpFrontend {
6946 cluster_id: cluster_id.to_owned(),
6947 address,
6948 tags,
6949 };
6950
6951 let cases = [
6952 ("http", RequestType::AddHttpFrontend(http)),
6953 ("https", RequestType::AddHttpsFrontend(https)),
6954 ("tcp", RequestType::AddTcpFrontend(tcp)),
6955 ("udp", RequestType::AddUdpFrontend(udp)),
6956 ];
6957
6958 for (protocol, add) in cases {
6959 let mut server = bare_server();
6960 server
6961 .config_state
6962 .dispatch(
6963 &RequestType::AddCluster(Cluster {
6964 cluster_id: cluster_id.to_owned(),
6965 ..Default::default()
6966 })
6967 .into(),
6968 )
6969 .expect("the setup cluster must enter worker state");
6970 server
6971 .config_state
6972 .dispatch(&add.clone().into())
6973 .unwrap_or_else(|error| {
6974 panic!("the {protocol} setup frontend must be stored: {error}")
6975 });
6976 let before = server
6977 .config_state
6978 .cluster_state(cluster_id)
6979 .expect("the setup cluster must expose its frontend");
6980
6981 let responses = send(&mut server, &format!("duplicate-{protocol}"), add);
6982 assert_eq!(
6983 responses
6984 .iter()
6985 .map(|response| response.status)
6986 .collect::<Vec<_>>(),
6987 vec![ResponseStatus::Failure as i32],
6988 "a duplicate frontend with no runtime listener must be refused: {responses:?}"
6989 );
6990 assert_eq!(
6991 server.config_state.cluster_state(cluster_id),
6992 Some(before),
6993 "a refused duplicate {protocol} add must preserve the preexisting frontend"
6994 );
6995 }
6996 }
6997
6998 #[test]
7002 fn a_refused_duplicate_listener_add_preserves_the_preexisting_listener() {
7003 let address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7004 let listener = ListenerBuilder::new_http(address)
7005 .to_http(None)
7006 .expect("the test HTTP listener configuration must build");
7007 let mut server = bare_server();
7008 server
7009 .config_state
7010 .dispatch(&RequestType::AddHttpListener(listener.clone()).into())
7011 .expect("the setup listener must enter worker state");
7012 let before = server.config_state.http_listeners.clone();
7013
7014 let threshold = server.sessions.borrow().accept_slab_threshold();
7017 while server.sessions.borrow().slab.len() < threshold {
7018 server
7019 .sessions
7020 .borrow_mut()
7021 .slab
7022 .insert(Rc::new(RefCell::new(ListenSession {
7023 protocol: Protocol::HTTPListen,
7024 })));
7025 }
7026 assert!(server.sessions.borrow().at_capacity());
7027
7028 let responses = send(
7029 &mut server,
7030 "duplicate-listener",
7031 RequestType::AddHttpListener(listener),
7032 );
7033 assert_eq!(
7034 responses
7035 .iter()
7036 .map(|response| response.status)
7037 .collect::<Vec<_>>(),
7038 vec![ResponseStatus::Failure as i32],
7039 "the full session slab must refuse the duplicate listener: {responses:?}"
7040 );
7041 assert_eq!(
7042 server.config_state.http_listeners, before,
7043 "a refused duplicate listener add must preserve the preexisting listener"
7044 );
7045 }
7046
7047 fn backend(cluster_id: &str, port: u16) -> AddBackend {
7048 AddBackend {
7049 cluster_id: cluster_id.to_owned(),
7050 backend_id: format!("{cluster_id}-0"),
7051 address: SocketAddress::new_v4(127, 0, 0, 1, port),
7052 load_balancing_parameters: Some(LoadBalancingParams::default()),
7053 ..Default::default()
7054 }
7055 }
7056
7057 #[test]
7068 fn remove_then_readd_same_cluster_id_preserves_and_releases_live_admission_leases() {
7069 let mut server = bare_server();
7070 let cluster_id = ClusterId::from(REMOVED);
7071 let source: IpAddr = "203.0.113.9".parse().expect("test address must parse");
7072 let subnet = server.sessions.borrow().subnet_key(&source);
7073 let old_token = Token(60_001);
7074 let new_token = Token(60_002);
7075
7076 for response in send(
7077 &mut server,
7078 "add-old-incarnation",
7079 RequestType::AddCluster(Cluster {
7080 cluster_id: REMOVED.to_owned(),
7081 max_connections_per_ip: Some(2),
7082 max_connections_per_subnet: Some(2),
7083 ..Default::default()
7084 }),
7085 ) {
7086 assert_eq!(response.status, ResponseStatus::Ok as i32);
7087 }
7088 server.sessions.borrow_mut().track_cluster_connection(
7089 old_token,
7090 cluster_id.clone(),
7091 source,
7092 Some(2),
7093 );
7094
7095 for response in send(
7096 &mut server,
7097 "remove-old-incarnation",
7098 RequestType::RemoveCluster(REMOVED.to_owned()),
7099 ) {
7100 assert_eq!(response.status, ResponseStatus::Ok as i32);
7101 }
7102 for response in send(
7103 &mut server,
7104 "add-new-incarnation",
7105 RequestType::AddCluster(Cluster {
7106 cluster_id: REMOVED.to_owned(),
7107 max_connections_per_ip: Some(2),
7108 max_connections_per_subnet: Some(2),
7109 ..Default::default()
7110 }),
7111 ) {
7112 assert_eq!(response.status, ResponseStatus::Ok as i32);
7113 }
7114
7115 {
7116 let mut sessions = server.sessions.borrow_mut();
7117 assert_eq!(sessions.connections_per_cluster_ip[REMOVED][&source], 1);
7118 assert_eq!(sessions.connections_per_cluster_subnet[REMOVED][&subnet], 1);
7119 sessions.track_cluster_connection(new_token, cluster_id, source, Some(2));
7120 assert_eq!(sessions.connections_per_cluster_ip[REMOVED][&source], 2);
7121 assert_eq!(sessions.connections_per_cluster_subnet[REMOVED][&subnet], 2);
7122
7123 sessions.untrack_all_cluster_ip(old_token);
7124 assert_eq!(sessions.connections_per_cluster_ip[REMOVED][&source], 1);
7125 assert_eq!(sessions.connections_per_cluster_subnet[REMOVED][&subnet], 1);
7126 assert!(sessions.cluster_ip_tracks.contains_key(&new_token));
7127 assert!(sessions.cluster_subnet_tracks.contains_key(&new_token));
7128
7129 sessions.untrack_all_cluster_ip(new_token);
7130 assert!(!sessions.connections_per_cluster_ip.contains_key(REMOVED));
7131 assert!(
7132 !sessions
7133 .connections_per_cluster_subnet
7134 .contains_key(REMOVED)
7135 );
7136 }
7137 }
7138
7139 fn http_route(
7143 server: &Server,
7144 address: SocketAddress,
7145 hostname: &str,
7146 path: &str,
7147 ) -> Option<Option<String>> {
7148 let http = server.http.borrow();
7149 let token = http
7150 .listener_token(&SocketAddr::from(address).into())
7151 .expect("the test HTTP listener must own a token");
7152 let listener = http
7153 .get_listener(&token)
7154 .expect("the test HTTP listener must exist");
7155 listener
7156 .borrow()
7157 .frontend_from_request(hostname, path, &Method::Get, &crate::router::NoHeaders)
7158 .ok()
7159 .map(|route| route.cluster_id.map(|id| id.to_string()))
7160 }
7161
7162 #[test]
7169 fn remove_cluster_removes_its_routes_and_backends_from_the_worker() {
7170 let mut server = bare_server();
7171 let http_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7172 let tcp_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7173
7174 let mut removed_front = http_front(Some(REMOVED), "removed.test", "/");
7175 removed_front.address = http_address;
7176 removed_front.position = RulePosition::Pre.into();
7179 let mut deny_front = http_front(None, "removed.test", "/deny");
7180 deny_front.address = http_address;
7181 let mut kept_front = http_front(Some(KEPT), "kept.test", "/");
7182 kept_front.address = http_address;
7183 let removed_tcp_front = RequestTcpFrontend {
7184 cluster_id: REMOVED.to_owned(),
7185 address: tcp_address,
7186 ..Default::default()
7187 };
7188
7189 let setup = vec![
7190 RequestType::AddHttpListener(
7191 ListenerBuilder::new_http(http_address)
7192 .to_http(None)
7193 .expect("could not build the test HTTP listener config"),
7194 ),
7195 RequestType::AddTcpListener(
7196 ListenerBuilder::new_tcp(tcp_address)
7197 .to_tcp(None)
7198 .expect("could not build the test TCP listener config"),
7199 ),
7200 RequestType::AddCluster(Cluster {
7201 cluster_id: REMOVED.to_owned(),
7202 ..Default::default()
7203 }),
7204 RequestType::AddCluster(Cluster {
7205 cluster_id: KEPT.to_owned(),
7206 ..Default::default()
7207 }),
7208 RequestType::AddHttpFrontend(removed_front.clone()),
7209 RequestType::AddHttpFrontend(deny_front),
7210 RequestType::AddHttpFrontend(kept_front),
7211 RequestType::AddTcpFrontend(removed_tcp_front.clone()),
7212 RequestType::AddBackend(backend(REMOVED, 1026)),
7213 RequestType::AddBackend(backend(KEPT, 1027)),
7214 ];
7215 for (index, request) in setup.into_iter().enumerate() {
7216 for response in send(&mut server, &format!("setup-{index}"), request) {
7217 assert_eq!(
7218 response.status,
7219 ResponseStatus::Ok as i32,
7220 "setup must succeed: {response:?}"
7221 );
7222 }
7223 }
7224 assert_eq!(
7225 http_route(&server, http_address, "removed.test", "/"),
7226 Some(Some(REMOVED.to_owned())),
7227 "premise: the removed cluster's hostname routes to it"
7228 );
7229
7230 let responses = send(
7231 &mut server,
7232 "remove-cluster",
7233 RequestType::RemoveCluster(REMOVED.to_owned()),
7234 );
7235 assert_eq!(
7236 responses.iter().map(|r| r.id.as_str()).collect::<Vec<_>>(),
7237 vec!["remove-cluster"],
7238 "the worker must answer the RemoveCluster exactly once: {responses:?}"
7239 );
7240 assert_eq!(
7241 responses[0].status,
7242 ResponseStatus::Ok as i32,
7243 "RemoveCluster must succeed: {responses:?}"
7244 );
7245
7246 assert_eq!(
7247 http_route(&server, http_address, "removed.test", "/"),
7248 None,
7249 "the removed cluster's hostname must no longer route"
7250 );
7251 assert_eq!(
7252 http_route(&server, http_address, "removed.test", "/deny"),
7253 Some(None),
7254 "a route that names no cluster must survive the removal of a cluster on its hostname"
7255 );
7256 assert_eq!(
7257 http_route(&server, http_address, "kept.test", "/"),
7258 Some(Some(KEPT.to_owned())),
7259 "another cluster's route must stay"
7260 );
7261 let backends = server.backends.borrow();
7262 assert!(
7263 !backends.backends.contains_key(REMOVED),
7264 "the removed cluster's backends must be gone from the BackendMap"
7265 );
7266 assert!(
7267 backends.backends.contains_key(KEPT),
7268 "another cluster's backends must stay"
7269 );
7270 drop(backends);
7271
7272 for (index, request) in [
7273 RequestType::RemoveHttpFrontend(removed_front),
7274 RequestType::RemoveTcpFrontend(removed_tcp_front),
7275 RequestType::RemoveBackend(RemoveBackend {
7276 cluster_id: REMOVED.to_owned(),
7277 backend_id: format!("{REMOVED}-0"),
7278 address: SocketAddress::new_v4(127, 0, 0, 1, 1026),
7279 }),
7280 ]
7281 .into_iter()
7282 .enumerate()
7283 {
7284 let responses = send(&mut server, &format!("follow-up-{index}"), request);
7285 assert!(
7286 !responses.is_empty()
7287 && responses
7288 .iter()
7289 .all(|r| r.status == ResponseStatus::Ok as i32),
7290 "a removal the cascade already performed must answer ok: {responses:?}"
7291 );
7292 }
7293 }
7294
7295 #[test]
7299 fn a_backend_of_an_unknown_cluster_is_refused_by_the_worker() {
7300 const ORPHAN: &str = "orphan";
7301 let mut server = bare_server();
7302 let assert_refused = |server: &mut Server, id: &str, port: u16| {
7303 let responses = send(server, id, RequestType::AddBackend(backend(ORPHAN, port)));
7304 assert!(
7305 !responses.is_empty()
7306 && responses
7307 .iter()
7308 .all(|r| r.status == ResponseStatus::Failure as i32),
7309 "a backend of an unknown cluster must be refused: {responses:?}"
7310 );
7311 assert!(
7312 !server.backends.borrow().backends.contains_key(ORPHAN),
7313 "a refused backend must not reach the BackendMap"
7314 );
7315 };
7316
7317 assert_refused(&mut server, "never-added", 1026);
7318
7319 for (index, request) in [
7320 RequestType::AddCluster(Cluster {
7321 cluster_id: ORPHAN.to_owned(),
7322 ..Default::default()
7323 }),
7324 RequestType::AddBackend(backend(ORPHAN, 1026)),
7325 RequestType::RemoveCluster(ORPHAN.to_owned()),
7326 ]
7327 .into_iter()
7328 .enumerate()
7329 {
7330 for response in send(&mut server, &format!("step-{index}"), request) {
7331 assert_eq!(
7332 response.status,
7333 ResponseStatus::Ok as i32,
7334 "step must succeed: {response:?}"
7335 );
7336 }
7337 }
7338
7339 assert_refused(&mut server, "removed", 1027);
7340 }
7341
7342 #[test]
7347 fn a_stale_frontend_removal_never_removes_another_clusters_route() {
7348 let http_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7349 let route = |cluster_id: &str| {
7350 let mut front = http_front(Some(cluster_id), "reused.test", "/");
7351 front.address = http_address;
7352 front
7353 };
7354 let cluster = |cluster_id: &str| {
7355 RequestType::AddCluster(Cluster {
7356 cluster_id: cluster_id.to_owned(),
7357 ..Default::default()
7358 })
7359 };
7360 let listener = || {
7361 RequestType::AddHttpListener(
7362 ListenerBuilder::new_http(http_address)
7363 .to_http(None)
7364 .expect("could not build the test HTTP listener config"),
7365 )
7366 };
7367
7368 for (a_is_gone, setup, expected_status) in [
7369 (
7370 true,
7371 vec![
7372 listener(),
7373 cluster("stale_a"),
7374 RequestType::AddHttpFrontend(route("stale_a")),
7375 RequestType::RemoveCluster("stale_a".to_owned()),
7376 cluster("stale_b"),
7377 RequestType::AddHttpFrontend(route("stale_b")),
7378 ],
7379 ResponseStatus::Ok,
7380 ),
7381 (
7382 false,
7383 vec![
7384 listener(),
7385 cluster("stale_a"),
7386 cluster("stale_b"),
7387 RequestType::AddHttpFrontend(route("stale_b")),
7388 ],
7389 ResponseStatus::Failure,
7390 ),
7391 ] {
7392 let mut server = bare_server();
7393 for (index, request) in setup.into_iter().enumerate() {
7394 for response in send(&mut server, &format!("setup-{index}"), request) {
7395 assert_eq!(
7396 response.status,
7397 ResponseStatus::Ok as i32,
7398 "setup must succeed: {response:?}"
7399 );
7400 }
7401 }
7402 assert_eq!(
7403 http_route(&server, http_address, "reused.test", "/"),
7404 Some(Some("stale_b".to_owned())),
7405 "premise: cluster B owns the route"
7406 );
7407
7408 let responses = send(
7409 &mut server,
7410 "stale-removal",
7411 RequestType::RemoveHttpFrontend(route("stale_a")),
7412 );
7413 assert_eq!(
7414 responses.iter().map(|r| r.status).collect::<Vec<_>>(),
7415 vec![expected_status as i32],
7416 "A gone: {a_is_gone}, answers: {responses:?}"
7417 );
7418 assert_eq!(
7419 http_route(&server, http_address, "reused.test", "/"),
7420 Some(Some("stale_b".to_owned())),
7421 "cluster B's route must survive a stale removal naming cluster A (A gone: {a_is_gone})"
7422 );
7423 }
7424 }
7425
7426 #[test]
7432 fn a_stale_tcp_or_udp_frontend_removal_never_unmaps_another_clusters_listener() {
7433 let cluster = |cluster_id: &str| {
7434 RequestType::AddCluster(Cluster {
7435 cluster_id: cluster_id.to_owned(),
7436 ..Default::default()
7437 })
7438 };
7439 for udp in [false, true] {
7440 let x = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7441 let y = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7442 let listener = |address: SocketAddress| {
7443 if udp {
7444 RequestType::AddUdpListener(
7445 ListenerBuilder::new_udp(address)
7446 .to_udp(None)
7447 .expect("could not build the test UDP listener config"),
7448 )
7449 } else {
7450 RequestType::AddTcpListener(
7451 ListenerBuilder::new_tcp(address)
7452 .to_tcp(None)
7453 .expect("could not build the test TCP listener config"),
7454 )
7455 }
7456 };
7457 let front = |cluster_id: &str, address: SocketAddress| {
7458 if udp {
7459 (
7460 RequestType::AddUdpFrontend(RequestUdpFrontend {
7461 cluster_id: cluster_id.to_owned(),
7462 address,
7463 ..Default::default()
7464 }),
7465 RequestType::RemoveUdpFrontend(RequestUdpFrontend {
7466 cluster_id: cluster_id.to_owned(),
7467 address,
7468 ..Default::default()
7469 }),
7470 )
7471 } else {
7472 (
7473 RequestType::AddTcpFrontend(RequestTcpFrontend {
7474 cluster_id: cluster_id.to_owned(),
7475 address,
7476 ..Default::default()
7477 }),
7478 RequestType::RemoveTcpFrontend(RequestTcpFrontend {
7479 cluster_id: cluster_id.to_owned(),
7480 address,
7481 ..Default::default()
7482 }),
7483 )
7484 }
7485 };
7486 let owner = |server: &Server| {
7487 let key = SocketAddr::from(x).into();
7488 if udp {
7489 server.udp.borrow().listener_cluster(&key)
7490 } else {
7491 server.tcp.borrow().listener_cluster(&key)
7492 }
7493 .map(|id| id.to_string())
7494 };
7495
7496 let mut server = bare_server();
7497 for (index, request) in [
7498 listener(x),
7499 listener(y),
7500 cluster("stale_a"),
7501 front("stale_a", x).0,
7502 RequestType::RemoveCluster("stale_a".to_owned()),
7503 cluster("stale_b"),
7504 front("stale_b", x).0,
7505 cluster("stale_a"),
7506 front("stale_a", y).0,
7507 ]
7508 .into_iter()
7509 .enumerate()
7510 {
7511 for response in send(&mut server, &format!("setup-{index}"), request) {
7512 assert_eq!(
7513 response.status,
7514 ResponseStatus::Ok as i32,
7515 "setup must succeed (udp: {udp}): {response:?}"
7516 );
7517 }
7518 }
7519 assert_eq!(
7520 owner(&server).as_deref(),
7521 Some("stale_b"),
7522 "premise: cluster B owns the listener on X (udp: {udp})"
7523 );
7524
7525 let responses = send(&mut server, "stale-removal", front("stale_a", x).1);
7526 assert_eq!(
7527 responses.iter().map(|r| r.status).collect::<Vec<_>>(),
7528 vec![ResponseStatus::Failure as i32],
7529 "the stale removal must be refused (udp: {udp}): {responses:?}"
7530 );
7531 assert_eq!(
7532 owner(&server).as_deref(),
7533 Some("stale_b"),
7534 "cluster B must keep the listener on X after a stale removal naming A (udp: {udp})"
7535 );
7536 }
7537 }
7538}
7539
7540#[cfg(test)]
7541mod scm_listener_handoff_tests {
7542 use std::{
7543 net::{TcpStream as StdTcpStream, UdpSocket as StdUdpSocket},
7544 os::fd::{IntoRawFd, RawFd},
7545 };
7546
7547 use sozu_command::{config::ListenerBuilder, proto::command::SocketAddress};
7548
7549 use super::*;
7550 use crate::{
7551 ProxyError,
7552 socket::{server_bind, udp_bind},
7553 testing::{ServerParts, prebuild_server, provide_port},
7554 };
7555
7556 fn state_with_four_active_listeners(
7559 http_address: SocketAddress,
7560 https_address: SocketAddress,
7561 tcp_address: SocketAddress,
7562 udp_address: SocketAddress,
7563 ) -> ConfigState {
7564 let mut state = ConfigState::new();
7565 let adds: Vec<(Request, SocketAddress, ListenerType)> = vec![
7566 (
7567 RequestType::AddHttpListener(
7568 ListenerBuilder::new_http(http_address)
7569 .to_http(None)
7570 .expect("could not build the test HTTP listener config"),
7571 )
7572 .into(),
7573 http_address,
7574 ListenerType::Http,
7575 ),
7576 (
7577 RequestType::AddHttpsListener(
7578 ListenerBuilder::new_https(https_address)
7579 .to_tls(None)
7580 .expect("could not build the test HTTPS listener config"),
7581 )
7582 .into(),
7583 https_address,
7584 ListenerType::Https,
7585 ),
7586 (
7587 RequestType::AddTcpListener(
7588 ListenerBuilder::new_tcp(tcp_address)
7589 .to_tcp(None)
7590 .expect("could not build the test TCP listener config"),
7591 )
7592 .into(),
7593 tcp_address,
7594 ListenerType::Tcp,
7595 ),
7596 (
7597 RequestType::AddUdpListener(
7598 ListenerBuilder::new_udp(udp_address)
7599 .to_udp(None)
7600 .expect("could not build the test UDP listener config"),
7601 )
7602 .into(),
7603 udp_address,
7604 ListenerType::Udp,
7605 ),
7606 ];
7607 for (add, address, proxy) in adds {
7608 state.dispatch(&add).expect("could not add the listener");
7609 state
7610 .dispatch(
7611 &RequestType::ActivateListener(ActivateListener {
7612 interface: None,
7613 address,
7614 proxy: proxy.into(),
7615 from_scm: false,
7616 })
7617 .into(),
7618 )
7619 .expect("could not mark the listener active");
7620 }
7621 state
7622 }
7623
7624 fn inherited_table<'a>(server: &'a Server, label: &str) -> &'a Vec<(ListenerKey, RawFd)> {
7626 let scm_listeners = server
7627 .scm_listeners
7628 .as_ref()
7629 .expect("scm_listeners must be present");
7630 match label {
7631 "HTTP" => &scm_listeners.http,
7632 "HTTPS" => &scm_listeners.tls,
7633 "TCP" => &scm_listeners.tcp,
7634 "UDP" => &scm_listeners.udp,
7635 other => panic!("unknown protocol label {other}"),
7636 }
7637 }
7638
7639 fn descriptor_is_socket_on_port(fd: RawFd, port: u16) -> bool {
7649 let mut storage: libc::sockaddr_storage = unsafe { std::mem::zeroed() };
7650 let mut length = std::mem::size_of::<libc::sockaddr_storage>() as libc::socklen_t;
7651 let result = unsafe {
7655 libc::getsockname(
7656 fd,
7657 (&raw mut storage).cast::<libc::sockaddr>(),
7658 &raw mut length,
7659 )
7660 };
7661 if result != 0 || storage.ss_family != libc::AF_INET as libc::sa_family_t {
7662 return false;
7663 }
7664 let inet = unsafe { &*(&raw const storage).cast::<libc::sockaddr_in>() };
7667 u16::from_be(inet.sin_port) == port
7668 }
7669
7670 const QUEUED_DATAGRAM: &[u8] = b"queued-before-the-handoff";
7672
7673 #[test]
7703 fn inherited_listener_sockets_are_adopted_by_the_initial_activation() {
7704 let ServerParts {
7705 event_loop,
7706 sessions,
7707 pool,
7708 backends,
7709 client_scm_socket,
7710 server_scm_socket,
7711 server_config,
7712 ..
7713 } = prebuild_server(16, 16384, false).expect("could not prebuild a test server");
7714 let (_command_channel, proxy_channel) =
7715 Channel::generate(1000, 10000).expect("could not generate a test channel");
7716
7717 let http_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7718 let https_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7719 let tcp_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7720 let udp_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7721
7722 let http_listener = server_bind(http_address.into(), None)
7725 .expect("could not bind the inherited HTTP socket");
7726 let https_listener = server_bind(https_address.into(), None)
7727 .expect("could not bind the inherited HTTPS socket");
7728 let tcp_listener =
7729 server_bind(tcp_address.into(), None).expect("could not bind the inherited TCP socket");
7730 let udp_socket =
7731 udp_bind(udp_address.into(), None).expect("could not bind the inherited UDP socket");
7732
7733 let http_client = StdTcpStream::connect::<SocketAddr>(http_address.into())
7738 .expect("could not queue a connection on the inherited HTTP socket");
7739 let https_client = StdTcpStream::connect::<SocketAddr>(https_address.into())
7740 .expect("could not queue a connection on the inherited HTTPS socket");
7741 let tcp_client = StdTcpStream::connect::<SocketAddr>(tcp_address.into())
7742 .expect("could not queue a connection on the inherited TCP socket");
7743 let datagram_sender =
7744 StdUdpSocket::bind("127.0.0.1:0").expect("could not bind the test datagram sender");
7745 datagram_sender
7746 .send_to(QUEUED_DATAGRAM, SocketAddr::from(udp_address))
7747 .expect("could not queue a datagram on the inherited UDP socket");
7748
7749 let listeners = Listeners {
7752 http: vec![(http_address.into(), http_listener.into_raw_fd())],
7753 tls: vec![(https_address.into(), https_listener.into_raw_fd())],
7754 tcp: vec![(tcp_address.into(), tcp_listener.into_raw_fd())],
7755 udp: vec![(udp_address.into(), udp_socket.into_raw_fd())],
7756 };
7757 client_scm_socket
7758 .send_listeners(&listeners)
7759 .expect("could not send the inherited listeners");
7760 listeners.close();
7761
7762 let initial_state =
7768 state_with_four_active_listeners(http_address, https_address, tcp_address, udp_address)
7769 .produce_initial_state();
7770 assert_eq!(
7771 initial_state.requests.len(),
7772 8,
7773 "the generator must emit one Add and one Activate per active listener"
7774 );
7775
7776 let server = Server::new(
7777 event_loop,
7778 proxy_channel,
7779 server_scm_socket,
7780 sessions,
7781 pool,
7782 backends,
7783 None,
7784 None,
7785 None,
7786 server_config,
7787 Some(initial_state),
7788 false,
7789 )
7790 .expect("could not build the test server");
7791
7792 let accepted: Vec<Protocol> = server
7793 .accept_queue
7794 .iter()
7795 .map(|(_, _, protocol, _, _)| *protocol)
7796 .collect();
7797 assert!(
7798 accepted.contains(&Protocol::HTTPListen),
7799 "the connection queued on the inherited HTTP socket must survive the hand-off, got {accepted:?}"
7800 );
7801 assert!(
7802 accepted.contains(&Protocol::HTTPSListen),
7803 "the connection queued on the inherited HTTPS socket must survive the hand-off, got {accepted:?}"
7804 );
7805 assert!(
7806 accepted.contains(&Protocol::TCPListen),
7807 "the connection queued on the inherited TCP socket must survive the hand-off, got {accepted:?}"
7808 );
7809
7810 for (protocol, client) in [
7818 (Protocol::HTTPListen, &http_client),
7819 (Protocol::HTTPSListen, &https_client),
7820 (Protocol::TCPListen, &tcp_client),
7821 ] {
7822 let (sock, _, _, _, peer) = server
7823 .accept_queue
7824 .iter()
7825 .find(|(_, _, queued, _, _)| *queued == protocol)
7826 .expect("the queued connection was asserted above");
7827 assert_eq!(
7828 socket2::SockRef::from(sock).tcp_nodelay().ok(),
7829 Some(true),
7830 "a {protocol:?} connection queued before the hand-off must carry TCP_NODELAY"
7831 );
7832 assert_eq!(
7833 *peer,
7834 client.local_addr().expect("the test client has an address"),
7835 "the {protocol:?} peer address must be the client's"
7836 );
7837 }
7838
7839 let (_, adopted_udp_socket) = server
7840 .udp
7841 .borrow_mut()
7842 .give_back_listeners()
7843 .pop()
7844 .expect("the UDP listener must hold a socket after activation");
7845 let mut buffer = [0u8; 64];
7846 let received = adopted_udp_socket.recv_from(&mut buffer);
7847 match received {
7848 Ok((length, _)) => assert_eq!(
7849 &buffer[..length],
7850 QUEUED_DATAGRAM,
7851 "the datagram queued on the inherited UDP socket must survive the hand-off"
7852 ),
7853 Err(error) => panic!(
7854 "the datagram queued on the inherited UDP socket must survive the hand-off, got {error:?}"
7855 ),
7856 }
7857 }
7858
7859 #[test]
7883 fn a_repeated_activation_closes_the_descriptor_it_cannot_adopt_now() {
7884 let ServerParts {
7885 event_loop,
7886 sessions,
7887 pool,
7888 backends,
7889 server_scm_socket,
7890 server_config,
7891 ..
7892 } = prebuild_server(16, 16384, true).expect("could not prebuild a test server");
7893 let (_command_channel, proxy_channel) =
7894 Channel::generate(1000, 10000).expect("could not generate a test channel");
7895
7896 let http_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7897 let https_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7898 let tcp_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7899 let udp_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
7900
7901 let initial_state =
7904 state_with_four_active_listeners(http_address, https_address, tcp_address, udp_address)
7905 .produce_initial_state();
7906 let mut server = Server::new(
7907 event_loop,
7908 proxy_channel,
7909 server_scm_socket,
7910 sessions,
7911 pool,
7912 backends,
7913 None,
7914 None,
7915 None,
7916 server_config,
7917 Some(initial_state),
7918 false,
7919 )
7920 .expect("could not build the test server");
7921
7922 let http_fd = server_bind(http_address.into(), None)
7924 .expect("could not bind the spare HTTP socket")
7925 .into_raw_fd();
7926 let https_fd = server_bind(https_address.into(), None)
7927 .expect("could not bind the spare HTTPS socket")
7928 .into_raw_fd();
7929 let tcp_fd = server_bind(tcp_address.into(), None)
7930 .expect("could not bind the spare TCP socket")
7931 .into_raw_fd();
7932 let udp_fd = udp_bind(udp_address.into(), None)
7933 .expect("could not bind the spare UDP socket")
7934 .into_raw_fd();
7935 {
7936 let scm_listeners = server
7937 .scm_listeners
7938 .as_mut()
7939 .expect("Server::new must have populated scm_listeners");
7940 scm_listeners.http.push((http_address.into(), http_fd));
7941 scm_listeners.tls.push((https_address.into(), https_fd));
7942 scm_listeners.tcp.push((tcp_address.into(), tcp_fd));
7943 scm_listeners.udp.push((udp_address.into(), udp_fd));
7944 }
7945
7946 for (address, proxy) in [
7947 (http_address, ListenerType::Http),
7948 (https_address, ListenerType::Https),
7949 (tcp_address, ListenerType::Tcp),
7950 (udp_address, ListenerType::Udp),
7951 ] {
7952 let response = server.notify_activate_listener(
7953 "test-repeat",
7954 &ActivateListener {
7955 interface: None,
7956 address,
7957 proxy: proxy.into(),
7958 from_scm: false,
7959 },
7960 );
7961 assert_eq!(
7962 response.status,
7963 ResponseStatus::Ok as i32,
7964 "a repeated {proxy:?} activation must still answer ok: {response:?}"
7965 );
7966 }
7967
7968 let scm_listeners = server
7969 .scm_listeners
7970 .as_ref()
7971 .expect("scm_listeners must still be present");
7972 for (label, table, fd, port) in [
7973 (
7974 "HTTP",
7975 &scm_listeners.http,
7976 http_fd,
7977 http_address.port as u16,
7978 ),
7979 (
7980 "HTTPS",
7981 &scm_listeners.tls,
7982 https_fd,
7983 https_address.port as u16,
7984 ),
7985 ("TCP", &scm_listeners.tcp, tcp_fd, tcp_address.port as u16),
7986 ("UDP", &scm_listeners.udp, udp_fd, udp_address.port as u16),
7987 ] {
7988 assert!(
7989 !table.iter().any(|(_, entry)| *entry == fd),
7990 "the inherited {label} descriptor this activation cannot adopt must leave the SCM table"
7991 );
7992 assert!(
7993 !descriptor_is_socket_on_port(fd, port),
7994 "the inherited {label} descriptor this activation cannot adopt must be closed, \
7995 not left bound in the address's SO_REUSEPORT group, where it is expected — \
7996 inferred, not measured here — to go on taking a share of new connections"
7997 );
7998 }
7999 }
8000
8001 #[test]
8018 fn an_inherited_descriptor_with_no_listener_yet_is_adopted_by_a_later_activation() {
8019 let ServerParts {
8020 event_loop,
8021 sessions,
8022 pool,
8023 backends,
8024 server_scm_socket,
8025 server_config,
8026 ..
8027 } = prebuild_server(16, 16384, true).expect("could not prebuild a test server");
8028 let (_command_channel, proxy_channel) =
8029 Channel::generate(1000, 10000).expect("could not generate a test channel");
8030
8031 let http_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8032 let https_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8033 let tcp_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8034 let udp_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8035
8036 let mut server = Server::new(
8037 event_loop,
8038 proxy_channel,
8039 server_scm_socket,
8040 sessions,
8041 pool,
8042 backends,
8043 None,
8044 None,
8045 None,
8046 server_config,
8047 None,
8048 false,
8049 )
8050 .expect("could not build the test server");
8051
8052 let http_listener = server_bind(http_address.into(), None)
8056 .expect("could not bind the inherited HTTP socket");
8057 let https_listener = server_bind(https_address.into(), None)
8058 .expect("could not bind the inherited HTTPS socket");
8059 let tcp_listener =
8060 server_bind(tcp_address.into(), None).expect("could not bind the inherited TCP socket");
8061 let udp_socket =
8062 udp_bind(udp_address.into(), None).expect("could not bind the inherited UDP socket");
8063 let _http_client = StdTcpStream::connect::<SocketAddr>(http_address.into())
8064 .expect("could not queue a connection on the inherited HTTP socket");
8065 let _https_client = StdTcpStream::connect::<SocketAddr>(https_address.into())
8066 .expect("could not queue a connection on the inherited HTTPS socket");
8067 let _tcp_client = StdTcpStream::connect::<SocketAddr>(tcp_address.into())
8068 .expect("could not queue a connection on the inherited TCP socket");
8069 let (http_fd, https_fd, tcp_fd, udp_fd) = (
8070 http_listener.into_raw_fd(),
8071 https_listener.into_raw_fd(),
8072 tcp_listener.into_raw_fd(),
8073 udp_socket.into_raw_fd(),
8074 );
8075 {
8076 let scm_listeners = server
8077 .scm_listeners
8078 .as_mut()
8079 .expect("Server::new must have populated scm_listeners");
8080 scm_listeners.http.push((http_address.into(), http_fd));
8081 scm_listeners.tls.push((https_address.into(), https_fd));
8082 scm_listeners.tcp.push((tcp_address.into(), tcp_fd));
8083 scm_listeners.udp.push((udp_address.into(), udp_fd));
8084 }
8085
8086 let add_requests =
8087 state_with_four_active_listeners(http_address, https_address, tcp_address, udp_address);
8088 let cases = [
8089 ("HTTP", http_address, ListenerType::Http, http_fd),
8090 ("HTTPS", https_address, ListenerType::Https, https_fd),
8091 ("TCP", tcp_address, ListenerType::Tcp, tcp_fd),
8092 ("UDP", udp_address, ListenerType::Udp, udp_fd),
8093 ];
8094
8095 for (label, address, proxy, fd) in cases {
8098 let response = server.notify_activate_listener(
8099 "test-early",
8100 &ActivateListener {
8101 interface: None,
8102 address,
8103 proxy: proxy.into(),
8104 from_scm: false,
8105 },
8106 );
8107 assert_ne!(
8108 response.status,
8109 ResponseStatus::Ok as i32,
8110 "activating {label} with no listener must fail: {response:?}"
8111 );
8112 assert!(
8113 inherited_table(&server, label)
8114 .iter()
8115 .any(|(_, e)| *e == fd),
8116 "the {label} descriptor must stay in the SCM table while no listener owns \
8117 its address"
8118 );
8119 assert!(
8120 descriptor_is_socket_on_port(fd, address.port as u16),
8121 "the {label} descriptor must stay open while no listener owns its address"
8122 );
8123 }
8124
8125 for request in add_requests.produce_initial_state().requests {
8128 server.notify_proxys(request);
8129 }
8130 for (label, _address, _proxy, fd) in cases {
8131 assert!(
8132 !inherited_table(&server, label)
8133 .iter()
8134 .any(|(_, e)| *e == fd),
8135 "the retained {label} descriptor must be adopted by the later activation"
8136 );
8137 }
8138 let accepted: Vec<Protocol> = server
8139 .accept_queue
8140 .iter()
8141 .map(|(_, _, protocol, _, _)| *protocol)
8142 .collect();
8143 for expected in [
8144 Protocol::HTTPListen,
8145 Protocol::HTTPSListen,
8146 Protocol::TCPListen,
8147 ] {
8148 assert!(
8149 accepted.contains(&expected),
8150 "the connection queued on the retained {expected:?} descriptor must be served \
8151 by the later activation, got {accepted:?}"
8152 );
8153 }
8154 }
8155
8156 #[test]
8181 fn a_socket_parked_by_a_failed_registration_is_not_mistaken_for_a_live_one() {
8182 let ServerParts {
8183 registry,
8184 sessions,
8185 pool,
8186 backends,
8187 ..
8188 } = prebuild_server(16, 16384, true).expect("could not prebuild a test server");
8189
8190 let token_for = |protocol: Protocol| {
8191 let mut manager = sessions.borrow_mut();
8192 let entry = manager.slab.vacant_entry();
8193 let token = Token(entry.key());
8194 entry.insert(Rc::new(RefCell::new(ListenSession { protocol })));
8195 token
8196 };
8197 let pre_registered = |address: SocketAddress| {
8200 let mut socket = server_bind(address.into(), None)
8201 .expect("could not bind the test listening socket");
8202 registry
8203 .register(&mut socket, Token(usize::MAX - 2), Interest::READABLE)
8204 .expect("could not pre-register the test socket");
8205 socket
8206 };
8207
8208 let http_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8209 let mut http_proxy = http::HttpProxy::new(
8210 registry.try_clone().expect("could not clone the registry"),
8211 sessions.clone(),
8212 pool.clone(),
8213 backends.clone(),
8214 );
8215 http_proxy
8216 .add_listener(
8217 ListenerBuilder::new_http(http_address)
8218 .to_http(None)
8219 .expect("could not build the test HTTP listener config"),
8220 token_for(Protocol::HTTPListen),
8221 )
8222 .expect("could not add the test HTTP listener");
8223 assert!(
8224 http_proxy
8225 .activate_listener(&http_address.into(), Some(pre_registered(http_address)))
8226 .is_err(),
8227 "the pre-registered descriptor must make the HTTP registration fail"
8228 );
8229 assert!(
8230 matches!(
8231 http_proxy.give_back_listener(&http_address.into()),
8232 Err(ProxyError::UnactivatedListener)
8233 ),
8234 "an HTTP listener holding only a parked socket must not answer as activated"
8235 );
8236 assert!(
8237 http_proxy.soft_stop().is_ok(),
8238 "a parked socket must not fail an HTTP soft stop"
8239 );
8240
8241 let https_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8242 let mut https_proxy = https::HttpsProxy::new(
8243 registry.try_clone().expect("could not clone the registry"),
8244 sessions.clone(),
8245 pool.clone(),
8246 backends.clone(),
8247 );
8248 https_proxy
8249 .add_listener(
8250 ListenerBuilder::new_https(https_address)
8251 .to_tls(None)
8252 .expect("could not build the test HTTPS listener config"),
8253 token_for(Protocol::HTTPSListen),
8254 )
8255 .expect("could not add the test HTTPS listener");
8256 assert!(
8257 https_proxy
8258 .activate_listener(&https_address.into(), Some(pre_registered(https_address)))
8259 .is_err(),
8260 "the pre-registered descriptor must make the HTTPS registration fail"
8261 );
8262 assert!(
8263 matches!(
8264 https_proxy.give_back_listener(&https_address.into()),
8265 Err(ProxyError::UnactivatedListener)
8266 ),
8267 "an HTTPS listener holding only a parked socket must not answer as activated"
8268 );
8269 assert!(
8270 https_proxy.hard_stop().is_ok(),
8271 "a parked socket must not fail an HTTPS hard stop"
8272 );
8273
8274 let tcp_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8275 let mut tcp_proxy = tcp::TcpProxy::new(
8276 registry.try_clone().expect("could not clone the registry"),
8277 sessions.clone(),
8278 pool.clone(),
8279 backends.clone(),
8280 );
8281 tcp_proxy
8282 .add_listener(
8283 ListenerBuilder::new_tcp(tcp_address)
8284 .to_tcp(None)
8285 .expect("could not build the test TCP listener config"),
8286 token_for(Protocol::TCPListen),
8287 )
8288 .expect("could not add the test TCP listener");
8289 assert!(
8290 tcp_proxy
8291 .activate_listener(&tcp_address.into(), Some(pre_registered(tcp_address)))
8292 .is_err(),
8293 "the pre-registered descriptor must make the TCP registration fail"
8294 );
8295 assert!(
8296 matches!(
8297 tcp_proxy.give_back_listener(&tcp_address.into()),
8298 Err(ProxyError::UnactivatedListener)
8299 ),
8300 "a TCP listener holding only a parked socket must not answer as activated"
8301 );
8302
8303 let udp_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8304 let mut udp_proxy = udp::UdpProxy::new(
8305 registry.try_clone().expect("could not clone the registry"),
8306 sessions.clone(),
8307 pool.clone(),
8308 backends.clone(),
8309 16,
8310 16384,
8311 );
8312 udp_proxy
8313 .add_listener(
8314 ListenerBuilder::new_udp(udp_address)
8315 .to_udp(None)
8316 .expect("could not build the test UDP listener config"),
8317 token_for(Protocol::UDP),
8318 )
8319 .expect("could not add the test UDP listener");
8320 let mut udp_socket =
8321 udp_bind(udp_address.into(), None).expect("could not bind the test UDP socket");
8322 registry
8323 .register(&mut udp_socket, Token(usize::MAX - 2), Interest::READABLE)
8324 .expect("could not pre-register the test UDP socket");
8325 assert!(
8326 udp_proxy
8327 .activate_listener(&udp_address.into(), Some(udp_socket))
8328 .is_err(),
8329 "the pre-registered descriptor must make the UDP registration fail"
8330 );
8331 assert!(
8332 matches!(
8333 udp_proxy.give_back_listener(&udp_address.into()),
8334 Err(ProxyError::UnactivatedListener)
8335 ),
8336 "a UDP listener holding only a parked socket must not answer as activated"
8337 );
8338 }
8339
8340 #[test]
8354 fn a_failed_registration_does_not_close_the_socket_it_was_handed() {
8355 let ServerParts {
8356 registry,
8357 sessions,
8358 pool,
8359 backends,
8360 ..
8361 } = prebuild_server(16, 16384, true).expect("could not prebuild a test server");
8362
8363 let token_for = |protocol: Protocol| {
8364 let mut manager = sessions.borrow_mut();
8365 let entry = manager.slab.vacant_entry();
8366 let token = Token(entry.key());
8367 entry.insert(Rc::new(RefCell::new(ListenSession { protocol })));
8368 token
8369 };
8370
8371 let handed_out = |address: SocketAddress| {
8374 let mut socket = server_bind(address.into(), None)
8375 .expect("could not bind the test listening socket");
8376 registry
8377 .register(&mut socket, Token(usize::MAX - 1), Interest::READABLE)
8378 .expect("could not pre-register the test socket");
8379 let fd = socket.as_raw_fd();
8380 (socket, fd)
8381 };
8382
8383 let http_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8384 let token = token_for(Protocol::HTTPListen);
8385 let mut http_proxy = http::HttpProxy::new(
8386 registry.try_clone().expect("could not clone the registry"),
8387 sessions.clone(),
8388 pool.clone(),
8389 backends.clone(),
8390 );
8391 http_proxy
8392 .add_listener(
8393 ListenerBuilder::new_http(http_address)
8394 .to_http(None)
8395 .expect("could not build the test HTTP listener config"),
8396 token,
8397 )
8398 .expect("could not add the test HTTP listener");
8399 let (socket, http_fd) = handed_out(http_address);
8400 let result = http_proxy.activate_listener(&http_address.into(), Some(socket));
8401 assert!(
8402 result.is_err(),
8403 "the pre-registered descriptor must make the HTTP registration fail"
8404 );
8405 assert!(
8406 descriptor_is_socket_on_port(http_fd, http_address.port as u16),
8407 "a failed HTTP registration must not close the socket it was handed"
8408 );
8409
8410 let tcp_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8411 let token = token_for(Protocol::TCPListen);
8412 let mut tcp_proxy = tcp::TcpProxy::new(
8413 registry.try_clone().expect("could not clone the registry"),
8414 sessions.clone(),
8415 pool.clone(),
8416 backends.clone(),
8417 );
8418 tcp_proxy
8419 .add_listener(
8420 ListenerBuilder::new_tcp(tcp_address)
8421 .to_tcp(None)
8422 .expect("could not build the test TCP listener config"),
8423 token,
8424 )
8425 .expect("could not add the test TCP listener");
8426 let (socket, tcp_fd) = handed_out(tcp_address);
8427 let result = tcp_proxy.activate_listener(&tcp_address.into(), Some(socket));
8428 assert!(
8429 result.is_err(),
8430 "the pre-registered descriptor must make the TCP registration fail"
8431 );
8432 assert!(
8433 descriptor_is_socket_on_port(tcp_fd, tcp_address.port as u16),
8434 "a failed TCP registration must not close the socket it was handed"
8435 );
8436
8437 let https_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8438 let token = token_for(Protocol::HTTPSListen);
8439 let mut https_proxy = https::HttpsProxy::new(
8440 registry.try_clone().expect("could not clone the registry"),
8441 sessions.clone(),
8442 pool.clone(),
8443 backends.clone(),
8444 );
8445 https_proxy
8446 .add_listener(
8447 ListenerBuilder::new_https(https_address)
8448 .to_tls(None)
8449 .expect("could not build the test HTTPS listener config"),
8450 token,
8451 )
8452 .expect("could not add the test HTTPS listener");
8453 let (socket, https_fd) = handed_out(https_address);
8454 let result = https_proxy.activate_listener(&https_address.into(), Some(socket));
8455 assert!(
8456 result.is_err(),
8457 "the pre-registered descriptor must make the HTTPS registration fail"
8458 );
8459 assert!(
8460 descriptor_is_socket_on_port(https_fd, https_address.port as u16),
8461 "a failed HTTPS registration must not close the socket it was handed"
8462 );
8463
8464 let udp_address = SocketAddress::new_v4(127, 0, 0, 1, provide_port());
8465 let token = token_for(Protocol::UDP);
8466 let mut udp_proxy = udp::UdpProxy::new(
8467 registry.try_clone().expect("could not clone the registry"),
8468 sessions.clone(),
8469 pool.clone(),
8470 backends.clone(),
8471 16,
8472 16384,
8473 );
8474 udp_proxy
8475 .add_listener(
8476 ListenerBuilder::new_udp(udp_address)
8477 .to_udp(None)
8478 .expect("could not build the test UDP listener config"),
8479 token,
8480 )
8481 .expect("could not add the test UDP listener");
8482 let mut socket =
8483 udp_bind(udp_address.into(), None).expect("could not bind the test UDP socket");
8484 registry
8485 .register(&mut socket, Token(usize::MAX - 1), Interest::READABLE)
8486 .expect("could not pre-register the test UDP socket");
8487 let udp_fd = socket.as_raw_fd();
8488 let result = udp_proxy.activate_listener(&udp_address.into(), Some(socket));
8489 assert!(
8490 result.is_err(),
8491 "the pre-registered descriptor must make the UDP registration fail"
8492 );
8493 assert!(
8494 descriptor_is_socket_on_port(udp_fd, udp_address.port as u16),
8495 "a failed UDP registration must not close the socket it was handed"
8496 );
8497 }
8498}
8499
8500#[cfg(test)]
8501mod tests {
8502 use super::{ClusterId, SessionManager, mask_ipv4, mask_ipv6, subnet_key};
8503 use mio::Token;
8504 use slab::Slab;
8505 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
8506
8507 #[test]
8514 fn a_cluster_override_dropping_to_zero_after_a_track_is_a_no_op() {
8515 let manager = SessionManager::new(Slab::with_capacity(8), 8, 0, 0, 0, 24, 56);
8516 let token = Token(1);
8517 let cluster = ClusterId::from("api");
8518 let ip: IpAddr = "203.0.113.9".parse().expect("test address must parse");
8519
8520 manager
8522 .borrow_mut()
8523 .track_cluster_subnet(token, cluster.clone(), ip, Some(5));
8524 assert!(
8525 !manager.borrow().subnet_tracking_is_empty(),
8526 "premise: the positive override must have recorded a slot"
8527 );
8528
8529 manager
8533 .borrow_mut()
8534 .track_cluster_subnet(token, cluster.clone(), ip, Some(0));
8535
8536 manager.borrow_mut().untrack_all_cluster_ip(token);
8538 assert!(
8539 manager.borrow().subnet_tracking_is_empty(),
8540 "teardown must drain the slot even though the cap was disabled in between"
8541 );
8542 }
8543
8544 #[test]
8549 fn masking_handles_both_shift_overflowing_prefix_bounds() {
8550 let v4: Ipv4Addr = "203.0.113.45".parse().expect("test address must parse");
8551 assert_eq!(
8552 mask_ipv4(v4, 32),
8553 v4,
8554 "a /32 must be the address itself, not a wrapped-shift artefact"
8555 );
8556 assert_eq!(
8557 mask_ipv4(v4, 0),
8558 Ipv4Addr::UNSPECIFIED,
8559 "a /0 must mask every bit away"
8560 );
8561
8562 let v6: Ipv6Addr = "2001:db8:dead:beef::1"
8563 .parse()
8564 .expect("test address must parse");
8565 assert_eq!(
8566 mask_ipv6(v6, 128),
8567 v6,
8568 "a /128 must be the address itself, not a wrapped-shift artefact"
8569 );
8570 assert_eq!(
8571 mask_ipv6(v6, 0),
8572 Ipv6Addr::UNSPECIFIED,
8573 "a /0 must mask every bit away"
8574 );
8575
8576 assert_eq!(
8579 mask_ipv4(v4, 33),
8580 v4,
8581 "an over-wide IPv4 prefix must clamp to the address, not panic or wrap"
8582 );
8583 assert_eq!(
8584 mask_ipv6(v6, 129),
8585 v6,
8586 "an over-wide IPv6 prefix must clamp to the address, not panic or wrap"
8587 );
8588 }
8589
8590 #[test]
8593 fn masking_groups_addresses_at_the_recommended_prefixes() {
8594 let parse_v4 = |s: &str| -> Ipv4Addr { s.parse().expect("test address must parse") };
8595 assert_eq!(
8596 mask_ipv4(parse_v4("203.0.113.1"), 24),
8597 mask_ipv4(parse_v4("203.0.113.254"), 24),
8598 "two hosts in one /24 must share a subnet key"
8599 );
8600 assert_ne!(
8601 mask_ipv4(parse_v4("203.0.113.1"), 24),
8602 mask_ipv4(parse_v4("203.0.114.1"), 24),
8603 "hosts in different /24s must NOT share a subnet key"
8604 );
8605
8606 let parse_v6 = |s: &str| -> Ipv6Addr { s.parse().expect("test address must parse") };
8607 assert_eq!(
8608 mask_ipv6(parse_v6("2001:db8:abcd:0000::1"), 56),
8609 mask_ipv6(parse_v6("2001:db8:abcd:00ff::9999"), 56),
8610 "two addresses in one /56 must share a subnet key"
8611 );
8612 assert_ne!(
8613 mask_ipv6(parse_v6("2001:db8:abcd:0000::1"), 56),
8614 mask_ipv6(parse_v6("2001:db8:abcd:0100::1"), 56),
8615 "addresses in different /56s must NOT share a subnet key"
8616 );
8617
8618 assert_eq!(
8621 mask_ipv6(parse_v6("2001:db8:abcd:1234::1"), 56),
8622 mask_ipv6(parse_v6("2001:db8:abcd:1234:ffff:ffff:ffff:ffff"), 56),
8623 "every address a /64 holder can mint must collapse to one /56 key — that \
8624 collapse IS the mitigation for sozu-proxy/sozu#1270"
8625 );
8626 }
8627
8628 #[test]
8633 fn an_ipv4_mapped_source_is_charged_to_the_ipv4_prefix() {
8634 let mapped: IpAddr = "::ffff:203.0.113.7".parse().expect("must parse");
8635 let native: IpAddr = "203.0.113.7".parse().expect("must parse");
8636 assert_eq!(
8637 subnet_key(&mapped, 24, 56),
8638 subnet_key(&native, 24, 56),
8639 "a mapped IPv4 source must key identically to the same address arriving natively"
8640 );
8641 assert!(
8642 subnet_key(&mapped, 24, 56).is_ipv4(),
8643 "the key for a mapped source must be an IPv4 key"
8644 );
8645
8646 let other: IpAddr = "::ffff:198.51.100.7".parse().expect("must parse");
8650 assert_ne!(
8651 subnet_key(&mapped, 24, 56),
8652 subnet_key(&other, 24, 56),
8653 "mapped sources from different IPv4 /24s must NOT share a subnet key — if they \
8654 do, every IPv4 client on a dual-stack listener shares one bucket"
8655 );
8656
8657 let genuine: IpAddr = "2001:db8::1".parse().expect("must parse");
8659 assert!(
8660 subnet_key(&genuine, 24, 56).is_ipv6(),
8661 "a genuine IPv6 source must keep an IPv6 subnet key"
8662 );
8663 }
8664}
8665
8666#[cfg(test)]
8667mod initial_protected_bootstrap_tests {
8668 use std::thread;
8669
8670 use sozu_command::{
8671 channel::Channel,
8672 proto::command::{
8673 CountRequests, HealthCheckConfig, HealthCheckHeader, HealthCheckHeaderValue,
8674 InitialState, ProtectedHealthCheckOrder, ProtectedHealthCheckPhase,
8675 QueryProtectedHealthCheckSupport, SetProtectedHealthCheck, Status,
8676 },
8677 secret::{FingerprintKey, ProtectedHeader, SecretValue, Zeroizing},
8678 };
8679
8680 use super::*;
8681 use crate::testing::{ServerParts, prebuild_server};
8682
8683 const CLUSTER: &str = "bootstrap-cluster";
8684 const HEADER: &str = "X-Synthetic-Bootstrap";
8685 const SECRET: &[u8] = b"synthetic-bootstrap-secret";
8686 const DEADLINE: Duration = Duration::from_secs(20);
8687
8688 type Phase = ProtectedHealthCheckPhase;
8689
8690 fn request(id: &str, request_type: RequestType) -> WorkerRequest {
8691 WorkerRequest {
8692 id: id.to_owned(),
8693 content: request_type.into(),
8694 }
8695 }
8696
8697 fn capability(id: &str) -> WorkerRequest {
8698 request(
8699 id,
8700 RequestType::QueryProtectedHealthCheckSupport(QueryProtectedHealthCheckSupport {}),
8701 )
8702 }
8703
8704 fn status(id: &str) -> WorkerRequest {
8705 request(id, RequestType::Status(Status {}))
8706 }
8707
8708 fn foreign(id: &str) -> WorkerRequest {
8709 request(id, RequestType::CountRequests(CountRequests {}))
8710 }
8711
8712 fn payload(cluster: &str) -> SetProtectedHealthCheck {
8713 let key = FingerprintKey::from_parts(Zeroizing::new([7u8; 32]), [9u8; 16]);
8714 let mut carrier = HealthCheckHeaderValue {
8715 name: HEADER.to_owned(),
8716 value: SECRET.to_vec(),
8717 };
8718 let secret = SecretValue::take_from_ingress(&mut carrier).expect("a valid value");
8719 let header = ProtectedHeader::new(&key, HEADER.to_owned(), secret);
8720 SetProtectedHealthCheck {
8721 cluster_id: cluster.to_owned(),
8722 config: HealthCheckConfig {
8723 uri: "/".to_owned(),
8724 interval: 10,
8725 timeout: 5,
8726 healthy_threshold: 3,
8727 unhealthy_threshold: 3,
8728 headers: vec![HealthCheckHeader {
8729 name: HEADER.to_owned(),
8730 fingerprint: header.fingerprint().to_string(),
8731 }],
8732 ..Default::default()
8733 },
8734 values: vec![HealthCheckHeaderValue {
8735 name: HEADER.to_owned(),
8736 value: SECRET.to_vec(),
8737 }],
8738 }
8739 }
8740
8741 fn order(id: &str, phase: Phase, cluster: &str, version: u64) -> WorkerRequest {
8742 let prepare = (phase == Phase::ProtectedHealthCheckPrepare).then(|| payload(cluster));
8743 request(
8744 id,
8745 RequestType::ProtectedHealthCheckOrder(ProtectedHealthCheckOrder {
8746 phase: phase as i32,
8747 cluster_id: cluster.to_owned(),
8748 version,
8749 prepare,
8750 }),
8751 )
8752 }
8753
8754 fn prepare(id: &str, version: u64) -> WorkerRequest {
8755 order(id, Phase::ProtectedHealthCheckPrepare, CLUSTER, version)
8756 }
8757
8758 fn commit(id: &str, version: u64) -> WorkerRequest {
8759 order(id, Phase::ProtectedHealthCheckCommit, CLUSTER, version)
8760 }
8761
8762 fn release(id: &str, version: u64) -> WorkerRequest {
8763 order(id, Phase::ProtectedHealthCheckRelease, CLUSTER, version)
8764 }
8765
8766 fn abort(id: &str, version: u64) -> WorkerRequest {
8767 order(id, Phase::ProtectedHealthCheckAbort, CLUSTER, version)
8768 }
8769
8770 fn assert_reply(response: &WorkerResponse, phase: Phase, cluster: &str, version: u64) {
8771 assert_eq!(response.status, ResponseStatus::Ok as i32, "{response:?}");
8772 let Some(ContentType::ProtectedHealthCheckReply(reply)) = response
8773 .content
8774 .as_ref()
8775 .and_then(|content| content.content_type.as_ref())
8776 else {
8777 panic!("the answer must be a typed protected reply: {response:?}");
8778 };
8779 assert_eq!(reply.phase, phase as i32);
8780 assert_eq!(reply.cluster_id, cluster);
8781 assert_eq!(reply.version, version);
8782 }
8783
8784 fn assert_refused(response: &WorkerResponse, id: &str) {
8785 assert_eq!(response.id, id);
8786 assert_eq!(
8787 response.status,
8788 ResponseStatus::Failure as i32,
8789 "{response:?}"
8790 );
8791 assert!(response.content.is_none(), "{response:?}");
8792 }
8793
8794 fn assert_no_secret(responses: &[WorkerResponse]) {
8795 let shown = format!("{responses:?}");
8796 assert!(!shown.contains(HEADER), "{shown}");
8797 assert!(!shown.contains("synthetic-bootstrap-secret"), "{shown}");
8798 }
8799
8800 fn bootstrap(
8806 script: Vec<WorkerRequest>,
8807 ) -> (Server, Rc<RefCell<BackendMap>>, Vec<WorkerResponse>) {
8808 let ServerParts {
8809 event_loop,
8810 sessions,
8811 pool,
8812 backends,
8813 server_scm_socket,
8814 server_config,
8815 ..
8816 } = prebuild_server(16, 16384, true).expect("could not prebuild a test server");
8817 let (mut peer, proxy_channel) =
8818 Channel::generate(1000, 10000).expect("could not generate a test channel");
8819 let initial_state = InitialState {
8820 requests: vec![request(
8821 "initial",
8822 RequestType::AddCluster(Cluster {
8823 cluster_id: CLUSTER.to_owned(),
8824 ..Default::default()
8825 }),
8826 )],
8827 };
8828
8829 let driver = thread::spawn(move || {
8830 let mut answers = Vec::with_capacity(script.len());
8831 for message in script {
8832 peer.write_message(&message)
8833 .map_err(|error| format!("peer write: {error}"))?;
8834 answers.push(
8835 peer.read_message_blocking_timeout(Some(DEADLINE))
8836 .map_err(|error| format!("peer read: {error}"))?,
8837 );
8838 }
8839 Ok::<_, String>(answers)
8840 });
8841
8842 let held = backends.to_owned();
8843 let server = Server::new(
8844 event_loop,
8845 proxy_channel,
8846 server_scm_socket,
8847 sessions,
8848 pool,
8849 backends,
8850 None,
8851 None,
8852 None,
8853 server_config,
8854 Some(initial_state),
8855 true,
8856 )
8857 .expect("the constructor must return once the Status arrives");
8858 let answers = driver
8859 .join()
8860 .expect("the peer must not panic")
8861 .expect("every request must be answered");
8862 (server, held, answers)
8863 }
8864
8865 fn assert_holds_the_values(backends: &Rc<RefCell<BackendMap>>) {
8866 let backends = backends.borrow();
8867 let config = backends
8868 .health_check_configs
8869 .get(CLUSTER)
8870 .expect("the commit must have installed the metadata");
8871 assert_eq!(config.headers.len(), 1);
8872 assert_eq!(config.headers[0].name, HEADER);
8873 let set = backends
8874 .protected_health_check_headers
8875 .get(CLUSTER)
8876 .expect("the commit must have installed the values");
8877 assert_eq!(set.len(), 1);
8878 assert_eq!(set.headers()[0].name(), HEADER);
8879 assert_eq!(set.headers()[0].value().expose_secret_bytes(), SECRET);
8880 }
8881
8882 fn assert_holds_nothing(backends: &Rc<RefCell<BackendMap>>) {
8883 let backends = backends.borrow();
8884 assert!(
8885 !backends
8886 .protected_health_check_headers
8887 .contains_key(CLUSTER)
8888 );
8889 assert!(!backends.health_check_configs.contains_key(CLUSTER));
8890 }
8891
8892 fn finish(mut script: Vec<WorkerRequest>, version: u64) -> Vec<WorkerRequest> {
8894 script.extend([
8895 prepare("p", version),
8896 commit("c", version),
8897 release("r", version),
8898 status("s"),
8899 ]);
8900 script
8901 }
8902
8903 #[test]
8904 fn initial_protected_bootstrap_installs_the_values_before_the_constructor_returns() {
8905 let (_server, backends, answers) = bootstrap(vec![
8906 capability("cap"),
8907 prepare("prep", 1),
8908 commit("com", 1),
8909 release("rel", 1),
8910 status("st"),
8911 ]);
8912 assert_reply(&answers[0], Phase::ProtectedHealthCheckCapability, "", 0);
8913 assert_reply(&answers[1], Phase::ProtectedHealthCheckPrepare, CLUSTER, 1);
8914 assert_reply(&answers[2], Phase::ProtectedHealthCheckCommit, CLUSTER, 1);
8915 assert_reply(&answers[3], Phase::ProtectedHealthCheckRelease, CLUSTER, 1);
8916 assert_eq!(answers[4], WorkerResponse::ok("st"));
8917 assert_eq!(answers.len(), 5);
8918 assert_holds_the_values(&backends);
8919 assert_no_secret(&answers);
8920 }
8921
8922 #[test]
8923 fn initial_protected_bootstrap_still_accepts_a_direct_status() {
8924 let (_server, backends, answers) = bootstrap(vec![status("st")]);
8925 assert_eq!(answers, vec![WorkerResponse::ok("st")]);
8926 assert_holds_nothing(&backends);
8927 }
8928
8929 #[test]
8930 fn initial_protected_bootstrap_refuses_every_out_of_order_request() {
8931 let cases: Vec<(&str, Vec<WorkerRequest>)> = vec![
8934 (
8935 "prepare before the capability",
8936 finish(vec![prepare("bad", 1), capability("cap")], 1),
8937 ),
8938 (
8939 "commit before the capability",
8940 finish(vec![commit("bad", 1), capability("cap")], 1),
8941 ),
8942 (
8943 "commit before prepare",
8944 finish(vec![capability("cap"), commit("bad", 1)], 1),
8945 ),
8946 (
8947 "release before commit",
8948 finish(vec![capability("cap"), release("bad", 1)], 1),
8949 ),
8950 (
8951 "foreign request after the capability",
8952 finish(vec![capability("cap"), foreign("bad")], 1),
8953 ),
8954 (
8955 "foreign request before the capability",
8956 finish(vec![foreign("bad"), capability("cap")], 1),
8957 ),
8958 (
8959 "repeated capability",
8960 finish(vec![capability("cap"), capability("bad")], 1),
8961 ),
8962 (
8963 "abort without a started cycle",
8964 finish(vec![capability("cap"), abort("bad", 1)], 1),
8965 ),
8966 ("commit of the wrong version", {
8967 let mut script = vec![capability("cap"), prepare("prep", 1)];
8968 script.push(commit("bad", 2));
8969 script.extend([commit("com", 1), release("rel", 1), status("st")]);
8970 script
8971 }),
8972 ("commit of the wrong cluster", {
8973 let mut script = vec![capability("cap"), prepare("prep", 1)];
8974 script.push(order(
8975 "bad",
8976 Phase::ProtectedHealthCheckCommit,
8977 "other-cluster",
8978 1,
8979 ));
8980 script.extend([commit("com", 1), release("rel", 1), status("st")]);
8981 script
8982 }),
8983 ("second prepare while one is staged", {
8984 let mut script = vec![capability("cap"), prepare("prep", 1)];
8985 script.push(prepare("bad", 2));
8986 script.extend([commit("com", 1), release("rel", 1), status("st")]);
8987 script
8988 }),
8989 ("repeated commit", {
8990 let mut script = vec![capability("cap"), prepare("prep", 1), commit("com", 1)];
8991 script.push(commit("bad", 1));
8992 script.extend([release("rel", 1), status("st")]);
8993 script
8994 }),
8995 ("release of the wrong version", {
8996 let mut script = vec![capability("cap"), prepare("prep", 1), commit("com", 1)];
8997 script.push(release("bad", 2));
8998 script.extend([release("rel", 1), status("st")]);
8999 script
9000 }),
9001 ("status while a prepare is outstanding", {
9002 let mut script = vec![capability("cap"), prepare("prep", 1)];
9003 script.push(status("bad"));
9004 script.extend([commit("com", 1), release("rel", 1), status("st")]);
9005 script
9006 }),
9007 ("status while a commit is outstanding", {
9008 let mut script = vec![capability("cap"), prepare("prep", 1), commit("com", 1)];
9009 script.push(status("bad"));
9010 script.extend([release("rel", 1), status("st")]);
9011 script
9012 }),
9013 ("stale prepare after a released cycle", {
9014 let mut script = vec![
9015 capability("cap"),
9016 prepare("prep", 2),
9017 commit("com", 2),
9018 release("rel", 2),
9019 ];
9020 script.push(prepare("bad", 1));
9021 script.push(status("st"));
9022 script
9023 }),
9024 ];
9025 for (name, script) in cases {
9026 let refused: Vec<usize> = script
9027 .iter()
9028 .enumerate()
9029 .filter(|(_, message)| message.id == "bad")
9030 .map(|(index, _)| index)
9031 .collect();
9032 assert_eq!(refused.len(), 1, "{name}: exactly one refused request");
9033 let (_server, backends, answers) = bootstrap(script);
9034 assert_refused(&answers[refused[0]], "bad");
9035 assert!(
9036 answers
9037 .iter()
9038 .enumerate()
9039 .all(|(index, answer)| index == refused[0]
9040 || answer.status == ResponseStatus::Ok as i32),
9041 "{name}: only the refused request fails: {answers:?}"
9042 );
9043 assert_holds_the_values(&backends);
9044 assert_no_secret(&answers);
9045 }
9046 }
9047
9048 #[test]
9049 fn initial_protected_bootstrap_forwards_a_handler_refusal_and_stays_blocked() {
9050 let unknown = order(
9051 "bad",
9052 Phase::ProtectedHealthCheckPrepare,
9053 "unknown-cluster",
9054 1,
9055 );
9056 let (_server, backends, answers) = bootstrap(finish(vec![capability("cap"), unknown], 1));
9057 assert_refused(&answers[1], "bad");
9058 assert_eq!(
9059 answers[1].message,
9060 "the cluster is not known to this worker"
9061 );
9062 assert_holds_the_values(&backends);
9063 assert_no_secret(&answers);
9064 }
9065
9066 #[test]
9067 fn initial_protected_bootstrap_abort_drops_the_staged_change() {
9068 let (_server, backends, answers) = bootstrap(vec![
9069 capability("cap"),
9070 prepare("prep", 1),
9071 abort("abort", 1),
9072 status("st"),
9073 ]);
9074 assert_reply(&answers[2], Phase::ProtectedHealthCheckAbort, CLUSTER, 1);
9075 assert_eq!(answers[3], WorkerResponse::ok("st"));
9076 assert_holds_nothing(&backends);
9077 assert_no_secret(&answers);
9078 }
9079
9080 #[test]
9081 fn initial_protected_bootstrap_abort_restores_an_applied_change() {
9082 let (_server, backends, answers) = bootstrap(vec![
9083 capability("cap"),
9084 prepare("prep", 1),
9085 commit("com", 1),
9086 abort("abort", 1),
9087 status("st"),
9088 ]);
9089 assert_reply(&answers[3], Phase::ProtectedHealthCheckAbort, CLUSTER, 1);
9090 assert_holds_nothing(&backends);
9091 assert_no_secret(&answers);
9092 }
9093
9094 #[test]
9095 fn initial_protected_bootstrap_refuses_a_mismatched_abort_then_completes() {
9096 let mut script = vec![capability("cap"), prepare("prep", 1)];
9097 script.push(abort("bad", 9));
9098 script.extend([abort("abort", 1), prepare("again", 2)]);
9099 script.extend([commit("com", 2), release("rel", 2), status("st")]);
9100 let (_server, backends, answers) = bootstrap(script);
9101 assert_refused(&answers[2], "bad");
9102 assert_reply(&answers[3], Phase::ProtectedHealthCheckAbort, CLUSTER, 1);
9103 assert_reply(&answers[6], Phase::ProtectedHealthCheckRelease, CLUSTER, 2);
9104 assert_holds_the_values(&backends);
9105 assert_no_secret(&answers);
9106 }
9107
9108 #[test]
9109 fn initial_protected_bootstrap_returns_an_error_when_the_channel_closes() {
9110 let ServerParts {
9111 event_loop,
9112 sessions,
9113 pool,
9114 backends,
9115 server_scm_socket,
9116 server_config,
9117 ..
9118 } = prebuild_server(16, 16384, true).expect("could not prebuild a test server");
9119 let (peer, proxy_channel) =
9120 Channel::generate(1000, 10000).expect("could not generate a test channel");
9121 drop(peer);
9122 let result = Server::new(
9123 event_loop,
9124 proxy_channel,
9125 server_scm_socket,
9126 sessions,
9127 pool,
9128 backends,
9129 None,
9130 None,
9131 None,
9132 server_config,
9133 None,
9134 true,
9135 );
9136 let Err(error) = result else {
9137 panic!("a closed channel must not release the constructor");
9138 };
9139 assert!(!error.to_string().is_empty());
9140 }
9141
9142 #[test]
9143 fn initial_status_ack_failure_fails_the_constructor() {
9144 let ServerParts {
9145 event_loop,
9146 sessions,
9147 pool,
9148 backends,
9149 server_scm_socket,
9150 server_config,
9151 ..
9152 } = prebuild_server(16, 16384, true).expect("could not prebuild a test server");
9153 let (mut peer, proxy_channel) =
9154 Channel::generate(1000, 10000).expect("could not generate a test channel");
9155 peer.write_message(&status("st"))
9158 .expect("the peer must queue the Status");
9159 drop(peer);
9160 let result = Server::new(
9161 event_loop,
9162 proxy_channel,
9163 server_scm_socket,
9164 sessions,
9165 pool,
9166 backends,
9167 None,
9168 None,
9169 None,
9170 server_config,
9171 None,
9172 true,
9173 );
9174 let Err(error) = result else {
9175 panic!("a lost final ACK must not release the constructor");
9176 };
9177 assert!(
9178 matches!(error, ServerError::WriteInitialAnswer(_)),
9179 "{error:?}"
9180 );
9181 }
9182}