sozu_command_lib/
response.rs

1use std::{cmp::Ordering, collections::BTreeMap, fmt, net::SocketAddr};
2
3use crate::{
4    proto::command::{
5        AddBackend, FilteredTimeSerie, Header, HeaderPredicate, HstsConfig, LoadBalancingParams,
6        PathRule, PathRuleKind, RequestHttpFrontend, RequestTcpFrontend, RequestUdpFrontend,
7        Response, ResponseContent, ResponseStatus, RulePosition, RunState, WorkerResponse,
8    },
9    state::ClusterId,
10};
11
12impl Response {
13    pub fn new(
14        status: ResponseStatus,
15        message: String,
16        content: Option<ResponseContent>,
17    ) -> Response {
18        Response {
19            status: status as i32,
20            message,
21            content,
22        }
23    }
24}
25
26/// An HTTP or HTTPS frontend, as used *within* Sōzu
27#[derive(Clone, PartialOrd, Ord, PartialEq, Eq, Hash, Serialize, Deserialize)]
28pub struct HttpFrontend {
29    /// Send a 401, DENY, if cluster_id is None
30    pub cluster_id: Option<ClusterId>,
31    pub address: SocketAddr,
32    pub hostname: String,
33    #[serde(default)]
34    #[serde(skip_serializing_if = "is_default_path_rule")]
35    pub path: PathRule,
36    #[serde(default)]
37    #[serde(skip_serializing_if = "Option::is_none")]
38    pub method: Option<String>,
39    #[serde(default)]
40    pub position: RulePosition,
41    pub tags: Option<BTreeMap<String, String>>,
42    /// Resolved frontend-level policy carried over from
43    /// [`RequestHttpFrontend`]. The router consults these to build a
44    /// `Route::Frontend(Rc<Frontend>)` when any are non-default,
45    /// otherwise falls back to the legacy `Route::ClusterId` /
46    /// `Route::Deny` shapes.
47    #[serde(default)]
48    #[serde(skip_serializing_if = "Option::is_none")]
49    pub redirect: Option<i32>,
50    #[serde(default)]
51    #[serde(skip_serializing_if = "Option::is_none")]
52    pub redirect_scheme: Option<i32>,
53    #[serde(default)]
54    #[serde(skip_serializing_if = "Option::is_none")]
55    pub redirect_template: Option<String>,
56    #[serde(default)]
57    #[serde(skip_serializing_if = "Option::is_none")]
58    pub rewrite_host: Option<String>,
59    #[serde(default)]
60    #[serde(skip_serializing_if = "Option::is_none")]
61    pub rewrite_path: Option<String>,
62    #[serde(default)]
63    #[serde(skip_serializing_if = "Option::is_none")]
64    pub rewrite_port: Option<u32>,
65    #[serde(default)]
66    #[serde(skip_serializing_if = "Option::is_none")]
67    pub required_auth: Option<bool>,
68    #[serde(default)]
69    #[serde(skip_serializing_if = "Vec::is_empty")]
70    pub headers: Vec<Header>,
71    /// Resolved per-frontend HSTS (RFC 6797) policy. `None` means inherit
72    /// the listener default at frontend-add time in the worker.
73    #[serde(default)]
74    #[serde(skip_serializing_if = "Option::is_none")]
75    pub hsts: Option<HstsConfig>,
76    /// Header predicates the request must all satisfy for this frontend to
77    /// match (sozu-proxy/sozu#1253). Part of the frontend's identity.
78    #[serde(default)]
79    #[serde(skip_serializing_if = "Vec::is_empty")]
80    pub header_predicates: Vec<HeaderPredicate>,
81}
82
83impl fmt::Debug for HttpFrontend {
84    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
85        let tags_count = self.tags.as_ref().map(BTreeMap::len);
86        let tags_len = self.tags.as_ref().map(|tags| {
87            tags.iter().fold(0usize, |total, (key, value)| {
88                total.saturating_add(key.len()).saturating_add(value.len())
89            })
90        });
91        let headers_len = self.headers.iter().fold(0usize, |total, header| {
92            total
93                .saturating_add(header.key.len())
94                .saturating_add(header.val.len())
95        });
96
97        f.debug_struct("HttpFrontend")
98            .field("cluster_id_len", &self.cluster_id.as_deref().map(str::len))
99            .field("address", &self.address)
100            .field("hostname_len", &self.hostname.len())
101            .field("path_kind", &self.path.kind)
102            .field("path_len", &self.path.value.len())
103            .field("method_len", &self.method.as_ref().map(String::len))
104            .field("position", &self.position)
105            .field("tags_count", &tags_count)
106            .field("tags_len", &tags_len)
107            .field("redirect", &self.redirect)
108            .field("required_auth", &self.required_auth)
109            .field("redirect_scheme", &self.redirect_scheme)
110            .field(
111                "redirect_template_len",
112                &self.redirect_template.as_ref().map(String::len),
113            )
114            .field(
115                "rewrite_host_len",
116                &self.rewrite_host.as_ref().map(String::len),
117            )
118            .field(
119                "rewrite_path_len",
120                &self.rewrite_path.as_ref().map(String::len),
121            )
122            .field("rewrite_port", &self.rewrite_port)
123            .field("headers_count", &self.headers.len())
124            .field("headers_len", &headers_len)
125            .field("hsts", &self.hsts)
126            .field("header_predicates_count", &self.header_predicates.len())
127            .finish_non_exhaustive()
128    }
129}
130
131impl From<HttpFrontend> for RequestHttpFrontend {
132    fn from(val: HttpFrontend) -> Self {
133        let source_address = val.address;
134        let source_hostname = val.hostname.clone();
135        let request_frontend = RequestHttpFrontend {
136            cluster_id: val.cluster_id.as_deref().map(str::to_owned),
137            address: val.address.into(),
138            hostname: val.hostname,
139            path: val.path,
140            method: val.method,
141            position: val.position.into(),
142            tags: val.tags.unwrap_or_default(),
143            redirect: val.redirect,
144            redirect_scheme: val.redirect_scheme,
145            redirect_template: val.redirect_template,
146            rewrite_host: val.rewrite_host,
147            rewrite_path: val.rewrite_path,
148            rewrite_port: val.rewrite_port,
149            required_auth: val.required_auth,
150            headers: val.headers,
151            hsts: val.hsts,
152            header_predicates: val.header_predicates,
153        };
154
155        // POST: the proto-encoded address decodes back to the source SocketAddr
156        // and the hostname is unchanged — the in-Sōzu → wire conversion must be
157        // routing-preserving (the SocketAddress ⇔ SocketAddr round-trip is
158        // exercised in request.rs; here we tie the frontend identity to it).
159        debug_assert_eq!(
160            SocketAddr::from(request_frontend.address),
161            source_address,
162            "frontend address must round-trip through the proto encoding"
163        );
164        debug_assert_eq!(
165            request_frontend.hostname, source_hostname,
166            "frontend hostname must survive the proto conversion"
167        );
168        request_frontend
169    }
170}
171
172impl From<Backend> for AddBackend {
173    fn from(val: Backend) -> Self {
174        let source_address = val.address;
175        let source_cluster_id = val.cluster_id.clone();
176        let source_backend_id = val.backend_id.clone();
177        let add_backend = AddBackend {
178            cluster_id: val.cluster_id.to_string(),
179            backend_id: val.backend_id,
180            address: val.address.into(),
181            sticky_id: val.sticky_id,
182            load_balancing_parameters: val.load_balancing_parameters,
183            backup: val.backup,
184            unix_path: val.unix_path,
185        };
186
187        // POST: backend identity (cluster + backend id) and the wire address
188        // are preserved — a backend that changed cluster/id/address here would
189        // be registered under the wrong key and never receive (or steal)
190        // traffic.
191        debug_assert_eq!(
192            *add_backend.cluster_id, *source_cluster_id,
193            "backend cluster_id must survive the proto conversion"
194        );
195        debug_assert_eq!(
196            add_backend.backend_id, source_backend_id,
197            "backend_id must survive the proto conversion"
198        );
199        debug_assert_eq!(
200            SocketAddr::from(add_backend.address),
201            source_address,
202            "backend address must round-trip through the proto encoding"
203        );
204        add_backend
205    }
206}
207
208impl PathRule {
209    pub fn prefix<S>(value: S) -> Self
210    where
211        S: ToString,
212    {
213        let rule = Self {
214            kind: PathRuleKind::Prefix.into(),
215            value: value.to_string(),
216        };
217        // POST: the encoded kind decodes back to the Prefix variant — the proto
218        // i32 must round-trip or the router would misclassify the match type.
219        debug_assert_eq!(
220            PathRuleKind::try_from(rule.kind),
221            Ok(PathRuleKind::Prefix),
222            "prefix() must encode a Prefix-kind rule"
223        );
224        rule
225    }
226
227    pub fn regex<S>(value: S) -> Self
228    where
229        S: ToString,
230    {
231        let rule = Self {
232            kind: PathRuleKind::Regex.into(),
233            value: value.to_string(),
234        };
235        debug_assert_eq!(
236            PathRuleKind::try_from(rule.kind),
237            Ok(PathRuleKind::Regex),
238            "regex() must encode a Regex-kind rule"
239        );
240        rule
241    }
242
243    pub fn equals<S>(value: S) -> Self
244    where
245        S: ToString,
246    {
247        let rule = Self {
248            kind: PathRuleKind::Equals.into(),
249            value: value.to_string(),
250        };
251        debug_assert_eq!(
252            PathRuleKind::try_from(rule.kind),
253            Ok(PathRuleKind::Equals),
254            "equals() must encode an Equals-kind rule"
255        );
256        rule
257    }
258
259    pub fn from_cli_options(
260        path_prefix: Option<String>,
261        path_regex: Option<String>,
262        path_equals: Option<String>,
263    ) -> Self {
264        // PRE: prefix takes precedence over regex, which takes precedence over
265        // equals. Capture which arms are populated so the post-condition can
266        // assert the precedence actually fired.
267        let had_prefix = path_prefix.is_some();
268        let had_regex = path_regex.is_some();
269        let rule = match (path_prefix, path_regex, path_equals) {
270            (Some(prefix), _, _) => PathRule {
271                kind: PathRuleKind::Prefix as i32,
272                value: prefix,
273            },
274            (None, Some(regex), _) => PathRule {
275                kind: PathRuleKind::Regex as i32,
276                value: regex,
277            },
278            (None, None, Some(equals)) => PathRule {
279                kind: PathRuleKind::Equals as i32,
280                value: equals,
281            },
282            _ => PathRule::default(),
283        };
284
285        // POST: a present prefix wins outright; absent a prefix, a present
286        // regex wins. The resolved kind must reflect that precedence so two
287        // simultaneously-set CLI flags can never silently pick the wrong rule.
288        debug_assert!(
289            !had_prefix || rule.kind == PathRuleKind::Prefix as i32,
290            "a path prefix must produce a Prefix rule regardless of other flags"
291        );
292        debug_assert!(
293            had_prefix || !had_regex || rule.kind == PathRuleKind::Regex as i32,
294            "absent a prefix, a regex must produce a Regex rule"
295        );
296        rule
297    }
298}
299
300pub fn is_default_path_rule(p: &PathRule) -> bool {
301    PathRuleKind::try_from(p.kind) == Ok(PathRuleKind::Prefix) && p.value.is_empty()
302}
303
304impl fmt::Display for PathRule {
305    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
306        match PathRuleKind::try_from(self.kind) {
307            Ok(PathRuleKind::Prefix) => write!(f, "prefix '{}'", self.value),
308            Ok(PathRuleKind::Regex) => write!(f, "regexp '{}'", self.value),
309            Ok(PathRuleKind::Equals) => write!(f, "equals '{}'", self.value),
310            Err(_) => write!(f, ""),
311        }
312    }
313}
314
315/// A TCP frontend, as used *within* Sōzu
316#[derive(Debug, Clone, PartialOrd, Ord, PartialEq, Eq, Hash, Serialize, Deserialize)]
317pub struct TcpFrontend {
318    pub cluster_id: ClusterId,
319    pub address: SocketAddr,
320    /// custom tags to identify the frontend in the access logs
321    pub tags: BTreeMap<String, String>,
322    /// SNI hostname to match against the TLS ClientHello (exact hostname or a
323    /// single leading `*.` wildcard label). `None` matches regardless of SNI.
324    #[serde(default)]
325    pub sni: Option<String>,
326    /// ALPN protocol names this frontend matches; empty is the catch-all for
327    /// its `sni` on this listener.
328    #[serde(default)]
329    pub alpn: Vec<String>,
330}
331
332impl From<TcpFrontend> for RequestTcpFrontend {
333    fn from(val: TcpFrontend) -> Self {
334        let source_address = val.address;
335        let source_cluster_id = val.cluster_id.clone();
336        let source_sni = val.sni.clone();
337        let source_alpn = val.alpn.clone();
338        let request_frontend = RequestTcpFrontend {
339            cluster_id: val.cluster_id.to_string(),
340            address: val.address.into(),
341            tags: val.tags,
342            sni: val.sni,
343            alpn: val.alpn,
344        };
345
346        // POST: cluster identity and the wire address are preserved across the
347        // proto conversion (same routing guarantee as the HTTP frontend path).
348        debug_assert_eq!(
349            *request_frontend.cluster_id, *source_cluster_id,
350            "TCP frontend cluster_id must survive the proto conversion"
351        );
352        debug_assert_eq!(
353            SocketAddr::from(request_frontend.address),
354            source_address,
355            "TCP frontend address must round-trip through the proto encoding"
356        );
357        // POST: SNI/ALPN routing identity is preserved across the proto
358        // conversion — losing either here would silently misroute or
359        // over-match traffic on this frontend's listener.
360        debug_assert_eq!(
361            request_frontend.sni, source_sni,
362            "TCP frontend sni must survive the proto conversion"
363        );
364        debug_assert_eq!(
365            request_frontend.alpn, source_alpn,
366            "TCP frontend alpn must survive the proto conversion"
367        );
368        request_frontend
369    }
370}
371
372/// A UDP frontend, as used *within* Sōzu
373#[derive(Debug, Clone, PartialOrd, Ord, PartialEq, Eq, Hash, Serialize, Deserialize)]
374pub struct UdpFrontend {
375    pub cluster_id: ClusterId,
376    pub address: SocketAddr,
377    /// custom tags to identify the frontend in the access logs
378    pub tags: BTreeMap<String, String>,
379}
380
381impl From<UdpFrontend> for RequestUdpFrontend {
382    fn from(val: UdpFrontend) -> Self {
383        RequestUdpFrontend {
384            cluster_id: val.cluster_id.to_string(),
385            address: val.address.into(),
386            tags: val.tags,
387        }
388    }
389}
390
391/// A backend, as used *within* Sōzu
392#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
393pub struct Backend {
394    pub cluster_id: ClusterId,
395    pub backend_id: String,
396    pub address: SocketAddr,
397    #[serde(default)]
398    #[serde(skip_serializing_if = "Option::is_none")]
399    pub sticky_id: Option<String>,
400    #[serde(default)]
401    #[serde(skip_serializing_if = "Option::is_none")]
402    pub load_balancing_parameters: Option<LoadBalancingParams>,
403    #[serde(default)]
404    #[serde(skip_serializing_if = "Option::is_none")]
405    pub backup: Option<bool>,
406    /// UNIX domain socket dialed instead of `address`, see
407    /// `AddBackend::unix_path` (`command/src/command.proto`).
408    #[serde(default)]
409    #[serde(skip_serializing_if = "Option::is_none")]
410    pub unix_path: Option<String>,
411}
412
413impl Ord for Backend {
414    fn cmp(&self, o: &Backend) -> Ordering {
415        // INV: Equal can only be returned when every keyed field compares
416        // Equal — the tuple of field orderings is the source of truth, and the
417        // `.then(...)` fold must not collapse two distinct backends to Equal
418        // (which would let one silently evict the other from a BTree key set).
419        // Computed inline (no recursion into `cmp`) so it stays cheap.
420        let fields_all_equal = self.cluster_id == o.cluster_id
421            && self.backend_id == o.backend_id
422            && self.sticky_id == o.sticky_id
423            && self.load_balancing_parameters == o.load_balancing_parameters
424            && self.backup == o.backup
425            && self.unix_path == o.unix_path
426            && self.address == o.address;
427
428        let ordering = self
429            .cluster_id
430            .cmp(&o.cluster_id)
431            .then(self.backend_id.cmp(&o.backend_id))
432            .then(self.sticky_id.cmp(&o.sticky_id))
433            .then(
434                self.load_balancing_parameters
435                    .cmp(&o.load_balancing_parameters),
436            )
437            .then(self.backup.cmp(&o.backup))
438            .then(self.unix_path.cmp(&o.unix_path))
439            .then(socketaddr_cmp(&self.address, &o.address));
440
441        debug_assert_eq!(
442            ordering == Ordering::Equal,
443            fields_all_equal,
444            "Backend::cmp returns Equal iff every keyed field is equal"
445        );
446        ordering
447    }
448}
449
450impl PartialOrd for Backend {
451    fn partial_cmp(&self, other: &Backend) -> Option<Ordering> {
452        Some(self.cmp(other))
453    }
454}
455
456impl Backend {
457    pub fn to_add_backend(self) -> AddBackend {
458        let source_address = self.address;
459        let source_backend_id = self.backend_id.clone();
460        let add_backend = AddBackend {
461            cluster_id: self.cluster_id.to_string(),
462            address: self.address.into(),
463            sticky_id: self.sticky_id,
464            backend_id: self.backend_id,
465            load_balancing_parameters: self.load_balancing_parameters,
466            backup: self.backup,
467            unix_path: self.unix_path,
468        };
469
470        // POST: identity (backend id) and wire address are preserved — same
471        // routing guarantee as the `From<Backend>` path, kept in lockstep.
472        debug_assert_eq!(
473            add_backend.backend_id, source_backend_id,
474            "backend_id must survive to_add_backend"
475        );
476        debug_assert_eq!(
477            SocketAddr::from(add_backend.address),
478            source_address,
479            "backend address must round-trip through to_add_backend"
480        );
481        add_backend
482    }
483}
484
485impl fmt::Display for RunState {
486    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
487        write!(f, "{self:?}")
488    }
489}
490
491pub type MessageId = String;
492
493impl WorkerResponse {
494    pub fn ok<T>(id: T) -> Self
495    where
496        T: ToString,
497    {
498        Self {
499            id: id.to_string(),
500            message: String::new(),
501            status: ResponseStatus::Ok.into(),
502            content: None,
503        }
504    }
505
506    pub fn ok_with_content<T>(id: T, content: ResponseContent) -> Self
507    where
508        T: ToString,
509    {
510        Self {
511            id: id.to_string(),
512            status: ResponseStatus::Ok.into(),
513            message: String::new(),
514            content: Some(content),
515        }
516    }
517
518    pub fn error<T, U>(id: T, error: U) -> Self
519    where
520        T: ToString,
521        U: ToString,
522    {
523        Self {
524            id: id.to_string(),
525            message: error.to_string(),
526            status: ResponseStatus::Failure.into(),
527            content: None,
528        }
529    }
530
531    pub fn processing<T>(id: T) -> Self
532    where
533        T: ToString,
534    {
535        Self {
536            id: id.to_string(),
537            message: String::new(),
538            status: ResponseStatus::Processing.into(),
539            content: None,
540        }
541    }
542
543    pub fn with_status<T>(id: T, status: ResponseStatus) -> Self
544    where
545        T: ToString,
546    {
547        Self {
548            id: id.to_string(),
549            message: String::new(),
550            status: status.into(),
551            content: None,
552        }
553    }
554
555    pub fn is_failure(&self) -> bool {
556        self.status == ResponseStatus::Failure as i32
557    }
558}
559
560impl fmt::Display for WorkerResponse {
561    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
562        write!(f, "{}-{:?}", self.id, self.status)
563    }
564}
565
566impl fmt::Display for FilteredTimeSerie {
567    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
568        // `last_minute` / `last_hour` are prost `repeated uint32`, so they are
569        // `Vec<u32>` and not `[u32; 60]`: six fixed 10-wide windows indexed a
570        // series that holds fewer samples out of range. `chunks(10)` renders
571        // the exact same six windows for a full series and simply stops early
572        // for a short one.
573        writeln!(f, "FilteredTimeSerie {{")?;
574        writeln!(f, "last_second: {},", self.last_second)?;
575        writeln!(f, "last_minute:")?;
576        for window in self.last_minute.chunks(10) {
577            writeln!(f, "{window:?}")?;
578        }
579        writeln!(f, "last_hour:")?;
580        for window in self.last_hour.chunks(10) {
581            writeln!(f, "{window:?}")?;
582        }
583        write!(f, "}}")
584    }
585}
586
587fn socketaddr_cmp(a: &SocketAddr, b: &SocketAddr) -> Ordering {
588    let ordering = a.ip().cmp(&b.ip()).then(a.port().cmp(&b.port()));
589    // INV: two socket addresses compare Equal iff both IP and port match —
590    // the `.then` fold must not declare distinct (ip, port) pairs equal, which
591    // would make two different backends indistinguishable to the BTree key.
592    debug_assert_eq!(
593        ordering == Ordering::Equal,
594        a.ip() == b.ip() && a.port() == b.port(),
595        "socketaddr_cmp is Equal iff ip and port both match"
596    );
597    ordering
598}
599
600#[cfg(test)]
601mod tests {
602    //! `FilteredTimeSerie::last_minute` / `last_hour` are prost
603    //! `repeated uint32` (`command.proto`), i.e. `Vec<u32>` and NOT
604    //! `[u32; 60]` — nothing in the type or the wire format guarantees the 60
605    //! samples the `Display` impl used to slice with six fixed 10-wide windows.
606    //!
607    //! To SEE THESE RED (regression proof): restore the pre-fix body of
608    //! `FilteredTimeSerie`'s [`std::fmt::Display`] impl, the six
609    //! `&self.last_minute[0..10] … [50..60]` / `&self.last_hour[..]` fixed
610    //! windows — the short-series expectation below then panics with
611    //! `range end index 10 out of range for slice of length 3`.
612    use crate::proto::command::FilteredTimeSerie;
613
614    #[test]
615    fn a_series_shorter_than_sixty_samples_renders_instead_of_panicking() {
616        let serie = FilteredTimeSerie {
617            last_second: 42,
618            last_minute: vec![1, 2, 3],
619            last_hour: vec![],
620        };
621
622        let rendered = serie.to_string();
623
624        assert!(
625            rendered.contains("last_second: 42"),
626            "the rendering must keep its header, got {rendered:?}"
627        );
628        assert!(
629            rendered.contains("[1, 2, 3]"),
630            "a partial minute window must render the samples it has, got {rendered:?}"
631        );
632    }
633
634    #[test]
635    fn a_full_series_still_renders_six_windows_per_serie() {
636        // The layout the six fixed windows produced must not drift: one header
637        // line, one label per serie, six 10-sample windows each, one closing
638        // brace.
639        let serie = FilteredTimeSerie {
640            last_second: 7,
641            last_minute: (0..60).collect(),
642            last_hour: (0..60).collect(),
643        };
644
645        let rendered = serie.to_string();
646        let lines: Vec<&str> = rendered.lines().collect();
647
648        assert_eq!(
649            lines.len(),
650            17,
651            "1 header + 1 label + 6 windows + 1 label + 6 windows + 1 brace, got {rendered:?}"
652        );
653        assert_eq!(lines[2], "last_minute:");
654        assert_eq!(lines[3], "[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]");
655        assert_eq!(lines[8], "[50, 51, 52, 53, 54, 55, 56, 57, 58, 59]");
656        assert_eq!(lines[9], "last_hour:");
657        assert_eq!(lines[16], "}");
658    }
659}
sozu_command_lib/response.rs

sozu_command_lib/
response.rs
