Skip to main content

plecto_control/manifest/
upstream.rs

1//! A named upstream the fast-path server forwards to (`[[upstream]]`, ADR 000013 / 000017 /
2//! 000035 / 000042), plus its load-balancing, health, circuit-breaker, and outlier-detection
3//! sub-config.
4
5use serde::{Deserialize, Serialize};
6
7/// A named upstream the fast-path server forwards a matched request to (ADR 000013 / 000017).
8/// One or more `addresses` (`host:port`, no scheme) are the upstream's instances; the fast path
9/// balances across the healthy ones per `lb_algorithm`. The forward leg is plain HTTP/1.1 unless
10/// `[upstream.tls]` is declared, which re-encrypts with rustls and lets ALPN negotiate h2 /
11/// http/1.1 (ADR 000042). Every upstream carries an active-health-check policy (`health`) —
12/// required, because instances start pessimistic (unhealthy) and only a passing probe puts one
13/// into rotation (ADR 000017).
14#[derive(Debug, Clone, Deserialize, schemars::JsonSchema, Serialize)]
15#[serde(deny_unknown_fields)]
16pub struct Upstream {
17    pub name: String,
18    /// Each backend instance's `host:port` (no scheme — `[upstream.tls]` decides the scheme for
19    /// the whole upstream, ADR 000042). Each entry is either a bare string (`"127.0.0.1:9000"`,
20    /// weight 1) or a weighted inline table (`{ address = "...", weight = N }`) for heterogeneous
21    /// instances (ADR 000035). Must be non-empty (validated at build). The two forms mix in one
22    /// list; a bare string and an explicit `weight = 1` are equivalent (same content hash).
23    pub addresses: Vec<AddressSpec>,
24    /// `[upstream.tls]` (ADR 000042): when present, the forward leg to every instance re-encrypts
25    /// with TLS and ALPN negotiates the protocol (h2 preferred, http/1.1 fallback). Absent = plain
26    /// HTTP/1.1 (the pre-000042 behaviour). There is no insecure/verify-off escape hatch —
27    /// verification failure keeps the instance out of rotation (fail-closed).
28    #[serde(default)]
29    pub tls: Option<UpstreamTls>,
30    /// Per-upstream load-balancing algorithm (ADR 000035): `round_robin` (default), `least_request`
31    /// (power-of-two-choices over per-instance active requests), or `maglev` (consistent hashing for
32    /// session affinity). The chosen algorithm selects an instance from the healthy set; default
33    /// `round_robin` keeps the pre-000035 behaviour with no per-request cost change.
34    #[serde(default)]
35    pub lb_algorithm: LbAlgorithm,
36    /// Maglev hash config (ADR 000035), `[upstream.hash]`. Required iff `lb_algorithm = "maglev"`
37    /// (a `maglev` upstream needs a hash key; any other algorithm must not set it). Absent otherwise.
38    #[serde(default)]
39    pub hash: Option<HashConfig>,
40    /// Active-health-check policy for this upstream's instances (ADR 000017).
41    pub health: HealthConfig,
42    /// How often (ms) hostname addresses are re-resolved and the endpoint set refreshed — the
43    /// standard periodic-DNS endpoint-discovery technique (the shape of nginx `resolve` / Envoy
44    /// STRICT_DNS): each A/AAAA record becomes a load-balancing endpoint with its own health;
45    /// a vanished record is dropped, a new one starts pessimistic (ADR 000017); a failed
46    /// resolution keeps the last-known-good set. Interval-based (getaddrinfo carries no TTL) —
47    /// pick a value at or below your DNS TTL. `0` disables (the default): hostnames still
48    /// resolve per connect, but the endpoint set stays as declared.
49    #[serde(default)]
50    pub resolve_interval_ms: u64,
51    /// End-to-end timeout (ms) for forwarding a request to this upstream (ADR 000019 / review
52    /// f000005 P2#4). Bounds time-to-response-headers; once headers arrive the body streams without
53    /// a deadline, so streaming responses are unaffected. Exceeding it fails closed with **504**.
54    /// Default 30000; **`0` disables** the timeout (for long-poll / streaming upstreams).
55    #[serde(default = "default_request_timeout_ms")]
56    pub request_timeout_ms: u64,
57    /// Maximum times the fast path re-sends this upstream's request to a DIFFERENT healthy instance
58    /// after a retryable failure (ADR 000023). A timeout is retried only for an idempotent method
59    /// (the upstream may have acted); a connect failure — the upstream never received the request —
60    /// is retried for any method. Only bodyless requests are retried (the opaque streamed body,
61    /// ADR 000013, can't be replayed). Default 1; **`0` disables** retry.
62    #[serde(default = "default_max_retries")]
63    pub max_retries: u64,
64    /// Overall request timeout (ms) bounding the WHOLE transaction — every attempt PLUS the backoff
65    /// between them (ADR 000031) — whereas `request_timeout_ms` bounds a single attempt (per-try).
66    /// This is the end-to-end deadline across retries; exceeding it fails closed **504**
67    /// (`x-plecto-fault: request-timeout`, distinct from the per-try `upstream-timeout`). Should be
68    /// `>= request_timeout_ms`; the runtime applies the tighter of the two regardless. Default `0` =
69    /// no overall bound (only the per-try timeout applies — the pre-000031 behaviour).
70    #[serde(default)]
71    pub overall_timeout_ms: u64,
72    /// Per-upstream circuit breaker (ADR 000028): bounds the load the fast path puts on this
73    /// upstream. Distinct from health — `health` ejects *failing* instances, this caps concurrent
74    /// work on *healthy* ones so a saturated backend sheds load fast instead of queueing unbounded.
75    /// Absent = unlimited (the default).
76    #[serde(default)]
77    pub circuit_breaker: CircuitBreaker,
78    /// Per-upstream outlier detection (ADR 000032): eject an instance from rotation when it
79    /// MISBEHAVES on live traffic (gateway-class 5xx), a third resilience axis distinct from active
80    /// health ("is it reachable?") and the circuit breaker ("is it saturated?"). Absent / threshold
81    /// `0` = disabled (the default).
82    #[serde(default)]
83    pub outlier_detection: OutlierDetection,
84}
85
86/// Upstream TLS re-encryption config (`[upstream.tls]`, ADR 000042). Presence of the section
87/// enables TLS to every instance of the upstream; server certificate verification is ALWAYS on
88/// (no insecure option — custom CA covers the self-signed / internal-CA case, fail-closed).
89#[derive(Debug, Clone, Default, Deserialize, schemars::JsonSchema, Serialize, PartialEq, Eq)]
90#[serde(deny_unknown_fields)]
91pub struct UpstreamTls {
92    /// Manifest-relative path to a PEM bundle of CA certificates that REPLACES the webpki
93    /// (Mozilla) roots for this upstream — the internal-CA / self-signed deployment shape.
94    /// `None` = verify against the webpki roots. Only the path rides the content hash, like
95    /// `[[tls]]` cert paths (ADR 000014).
96    #[serde(default)]
97    pub ca_path: Option<String>,
98    /// Verification-name override (ADR 000050): when set, every TLS leg to this upstream uses
99    /// `sni` for BOTH the SNI extension and certificate-name verification, instead of deriving it
100    /// from the connected address. Required to make an IP-literal `addresses` entry (or a
101    /// `resolve_interval_ms`-expanded one, ADR 000044) verify against a normal hostname
102    /// certificate: an IP endpoint sends no SNI and is verified against its bare IP by default,
103    /// which fails unless the certificate carries an IP SAN. `None` = derive from the address
104    /// (the pre-000050 behaviour). Only the name rides the content hash, like `ca_path`.
105    #[serde(default)]
106    pub sni: Option<String>,
107    /// Manifest-relative path to the PEM certificate chain Plecto PRESENTS to this upstream
108    /// (upstream mTLS, ADR 000078) — for backends that require client authentication. Must be
109    /// set together with `client_key_path`; one without the other fails the build closed.
110    /// Every TLS leg to this upstream presents it — forwarded requests AND health probes,
111    /// which share one connector (ADR 000042 decision 5). `None` = no client certificate.
112    #[serde(default)]
113    pub client_cert_path: Option<String>,
114    /// Manifest-relative path to the PEM private key for `client_cert_path`. The file must be
115    /// owner-only on unix (group/other-readable fails the build closed, the STEK file
116    /// discipline of ADR 000062 (d) applied to this slice's new key material).
117    #[serde(default)]
118    pub client_key_path: Option<String>,
119}
120
121/// Per-upstream load-balancing algorithm (ADR 000035). `round_robin` is the default and keeps the
122/// pre-000035 healthy-set rotation (ADR 000024); the others are opt-in. This selects an INSTANCE
123/// within a chosen upstream group — a layer below the route→group weighted split (ADR 000034).
124#[derive(
125    Debug, Clone, Copy, Default, Deserialize, schemars::JsonSchema, Serialize, PartialEq, Eq,
126)]
127#[serde(rename_all = "snake_case")]
128pub enum LbAlgorithm {
129    /// Healthy-set round-robin (ADR 000017 / 000024) — the default.
130    #[default]
131    RoundRobin,
132    /// Weighted least-request via power-of-two-choices: sample two healthy instances, forward to the
133    /// one with the smaller `(active_requests + 1) / weight` (ADR 000035).
134    LeastRequest,
135    /// Consistent hashing via a (weighted) Maglev lookup table: a request's hash key maps to a
136    /// stable instance for session affinity / cache locality (ADR 000035). Needs `[upstream.hash]`.
137    Maglev,
138}
139
140/// One instance of an upstream (ADR 000035): a bare `host:port` string (weight 1) or a weighted
141/// inline table `{ address = "host:port", weight = N }`. The bare form preserves the pre-000035
142/// `addresses = ["h:p", ...]` manifest verbatim. A custom `Serialize` canonicalises a `weight = 1`
143/// table back to the bare string, so an explicitly-written default weight does not change the
144/// content hash (the manifest determinism invariant — same spirit as an explicit `isolation =
145/// "untrusted"`).
146#[derive(Debug, Clone, Deserialize, schemars::JsonSchema, PartialEq, Eq)]
147#[serde(untagged)]
148pub enum AddressSpec {
149    /// A bare `host:port` — weight 1.
150    Bare(String),
151    /// A weighted instance `{ address, weight }`.
152    Weighted(WeightedAddress),
153}
154
155/// The weighted form of an [`AddressSpec`]: an instance `host:port` plus its integer `weight`
156/// (ADR 000035). Weight biases both the least-request comparison and the Maglev table share toward
157/// higher-capacity instances; `1` is the default, `0` is rejected at build (drain an instance by
158/// removing its address, not by zeroing its weight).
159#[derive(Debug, Clone, Deserialize, schemars::JsonSchema, Serialize, PartialEq, Eq)]
160#[serde(deny_unknown_fields)]
161pub struct WeightedAddress {
162    pub address: String,
163    #[serde(default = "default_instance_weight")]
164    pub weight: u32,
165}
166
167impl AddressSpec {
168    /// This instance's `host:port`.
169    pub fn address(&self) -> &str {
170        match self {
171            AddressSpec::Bare(a) => a,
172            AddressSpec::Weighted(w) => &w.address,
173        }
174    }
175
176    /// This instance's load-balancing weight (bare form is 1).
177    pub fn weight(&self) -> u32 {
178        match self {
179            AddressSpec::Bare(_) => default_instance_weight(),
180            AddressSpec::Weighted(w) => w.weight,
181        }
182    }
183}
184
185impl Serialize for AddressSpec {
186    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
187        // Canonicalise a default `weight = 1` (whether bare or written-out) to the bare-string form
188        // so it does not perturb the semantic content hash (manifest determinism invariant, like an
189        // explicit `isolation = "untrusted"`). A non-default weight serialises as the table.
190        match self {
191            AddressSpec::Bare(addr) => serializer.serialize_str(addr),
192            AddressSpec::Weighted(w) if w.weight == default_instance_weight() => {
193                serializer.serialize_str(&w.address)
194            }
195            AddressSpec::Weighted(w) => w.serialize(serializer),
196        }
197    }
198}
199
200fn default_instance_weight() -> u32 {
201    1
202}
203
204/// Upper bound on a per-instance weight (ADR 000035). Keeps the least-request cross-product and the
205/// Maglev populate (which interleaves a backend every `max_weight / weight` rounds) bounded; mirrors
206/// Google's documented per-instance weight range (0–1000) for weighted Maglev. Drain via address
207/// removal, not weight 0, so the floor is 1.
208pub(crate) const MAX_INSTANCE_WEIGHT: u32 = 1000;
209
210/// The Maglev consistent-hashing config (ADR 000035), `[upstream.hash]`. Only valid when
211/// `lb_algorithm = "maglev"`. Names the request attribute hashed for affinity and the lookup-table
212/// size.
213#[derive(Debug, Clone, Deserialize, schemars::JsonSchema, Serialize, PartialEq, Eq)]
214#[serde(deny_unknown_fields)]
215pub struct HashConfig {
216    /// Which request attribute is the hash key.
217    pub key: HashKeyKind,
218    /// The header name to hash, required iff `key = "header"` (matched case-insensitively). Ignored
219    /// for `source_ip`.
220    #[serde(default)]
221    pub header: Option<String>,
222    /// Maglev lookup-table size `M` — must be PRIME (the permutation's `skip` is coprime to `M` only
223    /// then) and `>= instance count`. Default 65537. Larger `M` reduces disruption on instance
224    /// change at the cost of `M × 2` bytes per upstream; validated prime / in range at build.
225    #[serde(default = "default_hash_table_size")]
226    pub table_size: u32,
227}
228
229/// The request attribute a Maglev upstream hashes for affinity (ADR 000035).
230#[derive(Debug, Clone, Copy, Deserialize, schemars::JsonSchema, Serialize, PartialEq, Eq)]
231#[serde(rename_all = "snake_case")]
232pub enum HashKeyKind {
233    /// The value of a named request header (requires `header`).
234    Header,
235    /// The connection peer's IP address (NOT a spoofable forwarding header).
236    SourceIp,
237}
238
239fn default_hash_table_size() -> u32 {
240    65537
241}
242
243/// Upper bound on the Maglev `table_size` (ADR 000035), matching Envoy's documented cap. Bounds the
244/// per-upstream table memory (`table_size × 2` bytes); the operator must opt into anything this
245/// large explicitly. The default (65537) is two orders of magnitude below it.
246pub(crate) const MAX_HASH_TABLE_SIZE: u32 = 5_000_011;
247
248/// Per-upstream outlier-detection policy (ADR 000032). A THIRD resilience axis: active health (ADR
249/// 000017) asks "is the instance reachable?", the circuit breaker (ADR 000028) "is the upstream
250/// saturated?", and this "is the instance misbehaving on live traffic?". An ejected instance leaves
251/// the rotation for a (backing-off) ejection window, independent of the health bit; a shed request
252/// (circuit breaker) never feeds it, only real upstream 5xx responses do.
253#[derive(Debug, Clone, Deserialize, schemars::JsonSchema, Serialize)]
254#[serde(deny_unknown_fields)]
255pub struct OutlierDetection {
256    /// Consecutive gateway-class 5xx (502/503/504) from live forwards to eject an instance; `0` =
257    /// disabled (the default). Counts only real responses — a circuit-breaker 503 or a per-try
258    /// timeout is not an instance-misbehaviour signal.
259    #[serde(default)]
260    pub consecutive_gateway_failures: u32,
261    /// Base ejection duration (ms): an ejected instance is out of rotation this long, doubling per
262    /// consecutive ejection up to a bounded cap, then auto-returns when the window expires (no probe
263    /// needed). Default 30000.
264    #[serde(default = "default_base_ejection_time_ms")]
265    pub base_ejection_time_ms: u64,
266    /// Max % of the pool that may be outlier-ejected at once. The rest stay in rotation even while
267    /// failing — fail-closed must not become a self-inflicted total outage (ejecting every instance).
268    /// Default 10.
269    #[serde(default = "default_max_ejection_percent")]
270    pub max_ejection_percent: u32,
271}
272
273impl Default for OutlierDetection {
274    fn default() -> Self {
275        Self {
276            consecutive_gateway_failures: 0,
277            base_ejection_time_ms: default_base_ejection_time_ms(),
278            max_ejection_percent: default_max_ejection_percent(),
279        }
280    }
281}
282
283fn default_base_ejection_time_ms() -> u64 {
284    30_000
285}
286
287fn default_max_ejection_percent() -> u32 {
288    10
289}
290
291/// Per-upstream circuit-breaker thresholds (ADR 000028). Overload protection that is deliberately
292/// SEPARATE from outlier detection / health (ADR 000017): health answers "is this instance up?",
293/// the breaker answers "is this upstream saturated?". A request rejected by the breaker is the
294/// upstream shedding load, not an instance failing — so it never demotes an instance.
295#[derive(Debug, Clone, Default, Deserialize, schemars::JsonSchema, Serialize)]
296#[serde(deny_unknown_fields)]
297pub struct CircuitBreaker {
298    /// Max concurrent in-flight requests the fast path will forward to this upstream at once. At the
299    /// cap a new request fails closed with **503** (`x-plecto-fault: circuit-open`) rather than
300    /// piling onto the backend. `0` = unlimited (the default). Counts one slot per request across
301    /// the whole retry sequence, released when the upstream response headers arrive (or it fails).
302    #[serde(default)]
303    pub max_requests: u32,
304}
305
306/// Active-health-check policy (ADR 000017). A background prober issues `GET {path}` to each
307/// instance every `interval_ms` — on the instance's own traffic port, unless `port` names a
308/// dedicated health-check port — and a 2xx within `timeout_ms` is a success, anything else
309/// (non-2xx, timeout, connect error) a failure. `unhealthy_threshold` consecutive failures eject a
310/// healthy instance from the rotation; `healthy_threshold` consecutive successes restore an ejected
311/// one. Instances start pessimistic (unhealthy); the FIRST successful probe alone promotes a
312/// never-yet-healthy instance (cold-start fast path), after which the full `healthy_threshold`
313/// governs re-entry. Only `path` is required; the rest default. `PartialEq` lets a reload detect a
314/// changed policy and re-probe the upstream's instances from scratch (ADR 000017 reconcile).
315#[derive(Debug, Clone, Deserialize, schemars::JsonSchema, Serialize, PartialEq)]
316#[serde(deny_unknown_fields)]
317pub struct HealthConfig {
318    /// The probe request path, e.g. `/healthz`.
319    pub path: String,
320    /// Probe period in milliseconds (default 2000).
321    #[serde(default = "default_interval_ms")]
322    pub interval_ms: u64,
323    /// Per-probe timeout in milliseconds (default 1000).
324    #[serde(default = "default_timeout_ms")]
325    pub timeout_ms: u64,
326    /// Consecutive successes to restore an ejected instance (default 2). The first-ever promotion
327    /// of a never-yet-healthy instance needs only one success, regardless of this value.
328    #[serde(default = "default_healthy_threshold")]
329    pub healthy_threshold: u32,
330    /// Consecutive failures to eject a healthy instance (default 3).
331    #[serde(default = "default_unhealthy_threshold")]
332    pub unhealthy_threshold: u32,
333    /// Dedicated port the probe connects to, distinct from the instance's traffic port (e.g. a
334    /// separate metrics/health listener on the same host). `None` (default) probes the traffic port
335    /// itself, the pre-existing behaviour.
336    #[serde(default)]
337    pub port: Option<u16>,
338}
339
340fn default_request_timeout_ms() -> u64 {
341    30_000
342}
343
344fn default_max_retries() -> u64 {
345    1
346}
347
348fn default_interval_ms() -> u64 {
349    2000
350}
351fn default_timeout_ms() -> u64 {
352    1000
353}
354fn default_healthy_threshold() -> u32 {
355    2
356}
357fn default_unhealthy_threshold() -> u32 {
358    3
359}
360
361#[cfg(test)]
362mod tests {
363    use super::*;
364    use crate::error::ControlError;
365    use crate::manifest::Manifest;
366
367    #[test]
368    fn upstream_circuit_breaker_defaults_unlimited_and_parses() {
369        // Absent `[upstream.circuit_breaker]` → unlimited (max_requests 0), the safe default.
370        let m = Manifest::from_toml(
371            r#"
372[[upstream]]
373name = "a"
374addresses = ["127.0.0.1:9000"]
375[upstream.health]
376path = "/healthz"
377"#,
378        )
379        .unwrap();
380        assert_eq!(
381            m.upstreams[0].circuit_breaker.max_requests, 0,
382            "an absent breaker is unlimited"
383        );
384
385        // Present → the cap is read.
386        let m2 = Manifest::from_toml(
387            r#"
388[[upstream]]
389name = "a"
390addresses = ["127.0.0.1:9000"]
391[upstream.health]
392path = "/healthz"
393[upstream.circuit_breaker]
394max_requests = 64
395"#,
396        )
397        .unwrap();
398        assert_eq!(m2.upstreams[0].circuit_breaker.max_requests, 64);
399    }
400
401    #[test]
402    fn upstream_tls_sni_defaults_absent_and_parses() {
403        // ADR 000050: absent `sni` = derive from the connected address (pre-000050 behaviour).
404        let without = Manifest::from_toml(
405            r#"
406[[upstream]]
407name = "x"
408addresses = ["127.0.0.1:9000"]
409[upstream.health]
410path = "/healthz"
411[upstream.tls]
412"#,
413        )
414        .unwrap();
415        assert_eq!(without.upstreams[0].tls.as_ref().unwrap().sni, None);
416
417        // Present → the override is read, and it flips the content hash (a real config change).
418        let with = Manifest::from_toml(
419            r#"
420[[upstream]]
421name = "x"
422addresses = ["127.0.0.1:9000"]
423[upstream.health]
424path = "/healthz"
425[upstream.tls]
426sni = "backend.internal"
427"#,
428        )
429        .unwrap();
430        assert_eq!(
431            with.upstreams[0].tls.as_ref().unwrap().sni.as_deref(),
432            Some("backend.internal")
433        );
434        assert_ne!(
435            without.content_hash().unwrap(),
436            with.content_hash().unwrap(),
437            "declaring sni must flip the content hash"
438        );
439    }
440
441    #[test]
442    fn upstream_requires_health() {
443        // health is mandatory for every upstream (ADR 000017): a missing `[upstream.health]`
444        // table is rejected, since instances start pessimistic and need a probe to enter rotation.
445        let parsed = Manifest::from_toml(
446            r#"
447[[upstream]]
448name = "x"
449addresses = ["127.0.0.1:9000"]
450"#,
451        );
452        assert!(
453            parsed.is_err(),
454            "an upstream without [upstream.health] is rejected"
455        );
456    }
457
458    #[test]
459    fn health_requires_path_but_defaults_the_rest() {
460        // `path` is required; thresholds/interval/timeout default.
461        let no_path = Manifest::from_toml(
462            r#"
463[[upstream]]
464name = "x"
465addresses = ["127.0.0.1:9000"]
466[upstream.health]
467interval_ms = 500
468"#,
469        );
470        assert!(no_path.is_err(), "health without a probe path is rejected");
471
472        // explicit overrides ride through
473        let m = Manifest::from_toml(
474            r#"
475[[upstream]]
476name = "x"
477addresses = ["127.0.0.1:9000"]
478[upstream.health]
479path = "/up"
480interval_ms = 250
481timeout_ms = 100
482healthy_threshold = 1
483unhealthy_threshold = 5
484"#,
485        )
486        .unwrap();
487        let h = &m.upstreams[0].health;
488        assert_eq!(h.interval_ms, 250);
489        assert_eq!(h.timeout_ms, 100);
490        assert_eq!(h.healthy_threshold, 1);
491        assert_eq!(h.unhealthy_threshold, 5);
492    }
493
494    #[test]
495    fn health_port_defaults_to_none_but_can_override_the_probe_port() {
496        // absent `port` = probe the instance's own traffic port (pre-existing behaviour).
497        let default_port = Manifest::from_toml(
498            r#"
499[[upstream]]
500name = "x"
501addresses = ["127.0.0.1:9000"]
502[upstream.health]
503path = "/healthz"
504"#,
505        )
506        .unwrap();
507        assert_eq!(default_port.upstreams[0].health.port, None);
508
509        // an explicit `port` names a dedicated health-check listener.
510        let overridden = Manifest::from_toml(
511            r#"
512[[upstream]]
513name = "x"
514addresses = ["127.0.0.1:9000"]
515[upstream.health]
516path = "/healthz"
517port = 9100
518"#,
519        )
520        .unwrap();
521        assert_eq!(overridden.upstreams[0].health.port, Some(9100));
522    }
523
524    #[test]
525    fn upstream_request_timeout_defaults_and_overrides() {
526        // ADR 000019 / review f000005 P2#4: an upstream gets a 30s default end-to-end timeout when
527        // unspecified; an explicit value (incl. 0 = disabled) rides through and flips the hash.
528        let defaulted = Manifest::from_toml(
529            r#"
530[[upstream]]
531name = "x"
532addresses = ["127.0.0.1:9000"]
533[upstream.health]
534path = "/healthz"
535"#,
536        )
537        .unwrap();
538        assert_eq!(
539            defaulted.upstreams[0].request_timeout_ms, 30_000,
540            "an unspecified upstream timeout defaults to 30s"
541        );
542
543        let overridden = Manifest::from_toml(
544            r#"
545[[upstream]]
546name = "x"
547addresses = ["127.0.0.1:9000"]
548request_timeout_ms = 250
549[upstream.health]
550path = "/healthz"
551"#,
552        )
553        .unwrap();
554        assert_eq!(overridden.upstreams[0].request_timeout_ms, 250);
555        // a timeout change is a real config change → the content hash must flip.
556        assert_ne!(
557            defaulted.content_hash().unwrap(),
558            overridden.content_hash().unwrap(),
559            "changing the upstream timeout must flip the config version"
560        );
561
562        // `0` is an explicit opt-out (long-poll / streaming upstream) and must parse, not error.
563        let disabled = Manifest::from_toml(
564            r#"
565[[upstream]]
566name = "x"
567addresses = ["127.0.0.1:9000"]
568request_timeout_ms = 0
569[upstream.health]
570path = "/healthz"
571"#,
572        )
573        .unwrap();
574        assert_eq!(
575            disabled.upstreams[0].request_timeout_ms, 0,
576            "request_timeout_ms = 0 disables the timeout and must parse"
577        );
578    }
579
580    #[test]
581    fn upstream_max_retries_defaults_and_overrides() {
582        // ADR 000023: an upstream gets 1 retry by default; an explicit value (incl. 0 = disabled)
583        // rides through and flips the content hash.
584        let defaulted = Manifest::from_toml(
585            r#"
586[[upstream]]
587name = "x"
588addresses = ["127.0.0.1:9000"]
589[upstream.health]
590path = "/healthz"
591"#,
592        )
593        .unwrap();
594        assert_eq!(
595            defaulted.upstreams[0].max_retries, 1,
596            "an unspecified upstream defaults to one retry"
597        );
598
599        let disabled = Manifest::from_toml(
600            r#"
601[[upstream]]
602name = "x"
603addresses = ["127.0.0.1:9000"]
604max_retries = 0
605[upstream.health]
606path = "/healthz"
607"#,
608        )
609        .unwrap();
610        assert_eq!(disabled.upstreams[0].max_retries, 0, "0 disables retry");
611        assert_ne!(
612            defaulted.content_hash().unwrap(),
613            disabled.content_hash().unwrap(),
614            "changing max_retries must flip the config version"
615        );
616    }
617
618    // ----- ADR 000035: lb_algorithm, per-instance weight, maglev hash config -----
619
620    fn upstream_toml(body: &str) -> Result<Manifest, ControlError> {
621        Manifest::from_toml(&format!(
622            "[[upstream]]\nname = \"u\"\n{body}\n[upstream.health]\npath = \"/healthz\"\n"
623        ))
624    }
625
626    #[test]
627    fn lb_algorithm_defaults_round_robin_and_parses() {
628        // Absent → round_robin (the pre-000035 default).
629        let m = upstream_toml("addresses = [\"a:1\"]").unwrap();
630        assert_eq!(m.upstreams[0].lb_algorithm, LbAlgorithm::RoundRobin);
631
632        let lr = upstream_toml("addresses = [\"a:1\"]\nlb_algorithm = \"least_request\"").unwrap();
633        assert_eq!(lr.upstreams[0].lb_algorithm, LbAlgorithm::LeastRequest);
634
635        // a non-default algorithm flips the content hash (it is part of config identity).
636        assert_ne!(
637            m.content_hash().unwrap(),
638            lr.content_hash().unwrap(),
639            "changing lb_algorithm must flip the config version"
640        );
641    }
642
643    #[test]
644    fn addresses_parse_bare_and_weighted_mixed() {
645        // A bare string and a weighted inline table coexist in one list (ADR 000035).
646        let m = upstream_toml(
647            "addresses = [\"a:1\", { address = \"b:2\", weight = 5 }, { address = \"c:3\" }]",
648        )
649        .unwrap();
650        let a = &m.upstreams[0].addresses;
651        assert_eq!(a[0].address(), "a:1");
652        assert_eq!(a[0].weight(), 1, "bare = weight 1");
653        assert_eq!(a[1].address(), "b:2");
654        assert_eq!(a[1].weight(), 5);
655        assert_eq!(a[2].address(), "c:3");
656        assert_eq!(a[2].weight(), 1, "weighted form defaults weight to 1");
657    }
658
659    #[test]
660    fn explicit_weight_one_hashes_like_a_bare_address() {
661        // The determinism invariant: an explicit `weight = 1` is representation noise vs a bare
662        // string, so the two must share a content hash (like an explicit `isolation = "untrusted"`).
663        let bare = upstream_toml("addresses = [\"a:1\"]").unwrap();
664        let explicit = upstream_toml("addresses = [{ address = \"a:1\", weight = 1 }]").unwrap();
665        assert_eq!(
666            bare.content_hash().unwrap(),
667            explicit.content_hash().unwrap(),
668            "an explicit default weight must not change the config version"
669        );
670
671        // …but a non-default weight DOES change it.
672        let weighted = upstream_toml("addresses = [{ address = \"a:1\", weight = 3 }]").unwrap();
673        assert_ne!(
674            bare.content_hash().unwrap(),
675            weighted.content_hash().unwrap()
676        );
677    }
678
679    #[test]
680    fn maglev_hash_config_parses() {
681        let m = upstream_toml(
682            "addresses = [\"a:1\", \"b:2\"]\nlb_algorithm = \"maglev\"\n[upstream.hash]\nkey = \"header\"\nheader = \"x-user\"\ntable_size = 1009",
683        )
684        .unwrap();
685        let up = &m.upstreams[0];
686        assert_eq!(up.lb_algorithm, LbAlgorithm::Maglev);
687        let hash = up.hash.as_ref().unwrap();
688        assert_eq!(hash.key, HashKeyKind::Header);
689        assert_eq!(hash.header.as_deref(), Some("x-user"));
690        assert_eq!(hash.table_size, 1009);
691        assert!(up.validate_lb().is_ok());
692    }
693
694    #[test]
695    fn hash_table_size_defaults_to_65537() {
696        let m = upstream_toml(
697            "addresses = [\"a:1\"]\nlb_algorithm = \"maglev\"\n[upstream.hash]\nkey = \"source_ip\"",
698        )
699        .unwrap();
700        assert_eq!(m.upstreams[0].hash.as_ref().unwrap().table_size, 65537);
701        assert_eq!(
702            m.upstreams[0].hash.as_ref().unwrap().key,
703            HashKeyKind::SourceIp
704        );
705    }
706
707    #[test]
708    fn validate_lb_rejects_bad_configs() {
709        // weight 0 (drain via address removal, not weight 0)
710        let w0 = upstream_toml("addresses = [{ address = \"a:1\", weight = 0 }]").unwrap();
711        assert!(w0.upstreams[0].validate_lb().is_err(), "weight 0 rejected");
712
713        // weight over the cap
714        let wbig = upstream_toml(&format!(
715            "addresses = [{{ address = \"a:1\", weight = {} }}]",
716            MAX_INSTANCE_WEIGHT + 1
717        ))
718        .unwrap();
719        assert!(
720            wbig.upstreams[0].validate_lb().is_err(),
721            "over-cap weight rejected"
722        );
723
724        // maglev without a hash block
725        let no_hash = upstream_toml("addresses = [\"a:1\"]\nlb_algorithm = \"maglev\"").unwrap();
726        assert!(
727            no_hash.upstreams[0].validate_lb().is_err(),
728            "maglev needs [upstream.hash]"
729        );
730
731        // hash block on a non-maglev algorithm
732        let stray_hash =
733            upstream_toml("addresses = [\"a:1\"]\n[upstream.hash]\nkey = \"source_ip\"").unwrap();
734        assert!(
735            stray_hash.upstreams[0].validate_lb().is_err(),
736            "[upstream.hash] only valid with maglev"
737        );
738
739        // header key with no header name
740        let no_header = upstream_toml(
741            "addresses = [\"a:1\"]\nlb_algorithm = \"maglev\"\n[upstream.hash]\nkey = \"header\"",
742        )
743        .unwrap();
744        assert!(
745            no_header.upstreams[0].validate_lb().is_err(),
746            "header key needs a name"
747        );
748
749        // non-prime table size
750        let nonprime = upstream_toml(
751            "addresses = [\"a:1\"]\nlb_algorithm = \"maglev\"\n[upstream.hash]\nkey = \"source_ip\"\ntable_size = 1000",
752        )
753        .unwrap();
754        assert!(
755            nonprime.upstreams[0].validate_lb().is_err(),
756            "table_size must be prime"
757        );
758
759        // table smaller than the instance count
760        let too_small = upstream_toml(
761            "addresses = [\"a:1\", \"b:2\", \"c:3\"]\nlb_algorithm = \"maglev\"\n[upstream.hash]\nkey = \"source_ip\"\ntable_size = 2",
762        )
763        .unwrap();
764        assert!(
765            too_small.upstreams[0].validate_lb().is_err(),
766            "M must be >= instance count"
767        );
768
769        // a valid maglev config passes
770        let ok = upstream_toml(
771            "addresses = [\"a:1\", \"b:2\"]\nlb_algorithm = \"maglev\"\n[upstream.hash]\nkey = \"source_ip\"\ntable_size = 97",
772        )
773        .unwrap();
774        assert!(ok.upstreams[0].validate_lb().is_ok());
775
776        // health interval/timeout of 0 must fail closed (busy-loop / permanent 503).
777        let interval0 = Manifest::from_toml(
778            r#"
779[[upstream]]
780name = "u"
781addresses = ["a:1"]
782[upstream.health]
783path = "/healthz"
784interval_ms = 0
785timeout_ms = 1000
786"#,
787        )
788        .unwrap();
789        assert!(
790            interval0.upstreams[0].validate_lb().is_err(),
791            "interval_ms = 0 rejected"
792        );
793        let timeout0 = Manifest::from_toml(
794            r#"
795[[upstream]]
796name = "u"
797addresses = ["a:1"]
798[upstream.health]
799path = "/healthz"
800interval_ms = 1000
801timeout_ms = 0
802"#,
803        )
804        .unwrap();
805        assert!(
806            timeout0.upstreams[0].validate_lb().is_err(),
807            "timeout_ms = 0 rejected"
808        );
809    }
810}