1use core::fmt;
9use core::num::{NonZeroU32, NonZeroU64};
10use core::str::FromStr;
11use std::collections::BTreeMap;
12
13use crate::ErrorClass;
14use serde::Deserialize;
15
16#[derive(Debug, Clone, PartialEq, Eq)]
19pub struct ParseError {
20 what: &'static str,
21 input: String,
22 note: Option<&'static str>,
26}
27
28impl ParseError {
29 fn new(what: &'static str, input: &str) -> Self {
30 Self {
31 what,
32 input: input.to_owned(),
33 note: None,
34 }
35 }
36
37 fn with_note(what: &'static str, input: &str, note: &'static str) -> Self {
38 Self {
39 what,
40 input: input.to_owned(),
41 note: Some(note),
42 }
43 }
44}
45
46impl fmt::Display for ParseError {
47 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
48 match self.note {
49 None => write!(f, "unparseable {} literal: {:?}", self.what, self.input),
50 Some(note) => write!(
51 f,
52 "invalid {} literal: {:?} — {note}",
53 self.what, self.input
54 ),
55 }
56 }
57}
58
59impl core::error::Error for ParseError {}
60
61#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Deserialize)]
63#[serde(try_from = "String")]
64pub struct DurationMs(pub u64);
65
66impl FromStr for DurationMs {
67 type Err = ParseError;
68
69 fn from_str(s: &str) -> Result<Self, Self::Err> {
70 let err = || ParseError::new("duration", s);
71 let s = s.trim();
72 let unit_at = s.find(|c: char| !c.is_ascii_digit()).ok_or_else(err)?;
73 let (number, unit) = s.split_at(unit_at);
74 let n: u64 = number.parse().map_err(|_| err())?;
75 let mult = match unit {
76 "ms" => 1,
77 "s" => 1_000,
78 "m" => 60_000,
79 "h" => 3_600_000,
80 _ => return Err(err()),
81 };
82 Ok(Self(n * mult))
83 }
84}
85
86impl TryFrom<String> for DurationMs {
87 type Error = ParseError;
88
89 fn try_from(s: String) -> Result<Self, Self::Error> {
90 s.parse()
91 }
92}
93
94#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
96#[serde(try_from = "String")]
97pub struct Rate {
98 pub limit: NonZeroU64,
99 pub window_ms: u64,
100}
101
102impl FromStr for Rate {
103 type Err = ParseError;
104
105 fn from_str(s: &str) -> Result<Self, Self::Err> {
106 let err = || ParseError::new("rate", s);
107 let (limit, window) = s.trim().split_once('/').ok_or_else(err)?;
108 let limit: NonZeroU64 = limit.trim().parse().map_err(|_| err())?;
109 let window_ms = match window.trim() {
110 "s" | "sec" => 1_000,
111 "min" => 60_000,
112 "h" | "hour" => 3_600_000,
113 _ => return Err(err()),
114 };
115 Ok(Self { limit, window_ms })
116 }
117}
118
119impl TryFrom<String> for Rate {
120 type Error = ParseError;
121
122 fn try_from(s: String) -> Result<Self, Self::Error> {
123 s.parse()
124 }
125}
126
127#[derive(Debug, Clone, PartialEq, Deserialize)]
136#[serde(try_from = "String")]
137pub struct Schedule {
138 pub segments: Vec<ScheduleSegment>,
141}
142
143#[derive(Debug, Clone, Copy, PartialEq)]
145pub struct ScheduleSegment {
146 pub primary: SchedulePrimary,
147 pub up_to_ms: Option<u64>,
149}
150
151#[derive(Debug, Clone, Copy, PartialEq)]
153pub enum SchedulePrimary {
154 Exp {
155 base_ms: u64,
156 factor: f64,
157 cap_ms: u64,
158 jitter: bool,
159 },
160 Fixed {
161 period_ms: u64,
162 },
163}
164
165impl SchedulePrimary {
166 #[expect(
169 clippy::cast_precision_loss,
170 clippy::cast_possible_truncation,
171 clippy::cast_sign_loss,
172 clippy::cast_possible_wrap,
173 reason = "backoff arithmetic: values are small and non-negative by construction"
174 )]
175 fn wait_ms(self, attempt: u32) -> u64 {
176 match self {
177 Self::Exp {
178 base_ms,
179 factor,
180 cap_ms,
181 ..
182 } => {
183 let wait = base_ms as f64 * factor.powi(attempt as i32 - 1);
184 wait.min(cap_ms as f64).round() as u64
185 }
186 Self::Fixed { period_ms } => period_ms,
187 }
188 }
189
190 fn jitter(self) -> bool {
191 matches!(self, Self::Exp { jitter: true, .. })
192 }
193}
194
195impl Default for Schedule {
196 fn default() -> Self {
198 Self {
199 segments: vec![ScheduleSegment {
200 primary: SchedulePrimary::Exp {
201 base_ms: 200,
202 factor: 2.0,
203 cap_ms: 30_000,
204 jitter: true,
205 },
206 up_to_ms: None,
207 }],
208 }
209 }
210}
211
212impl Schedule {
213 #[must_use]
216 pub fn wait_ms(&self, attempt: u32) -> u64 {
217 self.wait_and_jitter(attempt).0
218 }
219
220 #[must_use]
232 pub fn wait_and_jitter(&self, attempt: u32) -> (u64, bool) {
233 let attempt = attempt.max(1);
234 let last = self.segments.len() - 1;
235 let (mut i, mut a, mut e) = (0_usize, 1_u32, 0_u64);
236 let mut emitted = 0_u32;
237 loop {
238 let segment = self.segments[i];
239 let wait = segment.primary.wait_ms(a);
240 if i < last
243 && let Some(bound) = segment.up_to_ms
244 && e.saturating_add(wait) > bound
245 {
246 (i, a, e) = (i + 1, 1, 0);
247 continue;
248 }
249 emitted += 1;
250 if emitted == attempt {
251 return (wait, segment.primary.jitter());
252 }
253 a += 1;
254 e = e.saturating_add(wait);
255 }
256 }
257}
258
259impl FromStr for SchedulePrimary {
260 type Err = ParseError;
261
262 fn from_str(s: &str) -> Result<Self, Self::Err> {
263 let err = || ParseError::new("schedule", s);
264 let s = s.trim();
265 if let Some(inner) = s.strip_prefix("exp(").and_then(|r| r.strip_suffix(')')) {
266 let parts: Vec<&str> = inner.split(',').map(str::trim).collect();
267 let [base, factor, rest @ ..] = parts.as_slice() else {
268 return Err(err());
269 };
270 let base_ms = base.parse::<DurationMs>().map_err(|_| err())?.0;
271 let factor: f64 = factor
272 .strip_prefix('x')
273 .ok_or_else(err)?
274 .parse()
275 .map_err(|_| err())?;
276 let mut cap_ms = u64::MAX;
277 let mut jitter = false;
278 for part in rest {
279 if let Some(d) = part.strip_prefix("max ") {
280 cap_ms = d.parse::<DurationMs>().map_err(|_| err())?.0;
281 } else if *part == "jitter" {
282 jitter = true;
283 } else {
284 return Err(err());
285 }
286 }
287 Ok(Self::Exp {
288 base_ms,
289 factor,
290 cap_ms,
291 jitter,
292 })
293 } else if let Some(inner) = s.strip_prefix("fixed(").and_then(|r| r.strip_suffix(')')) {
294 let period_ms = inner.parse::<DurationMs>().map_err(|_| err())?.0;
295 Ok(Self::Fixed { period_ms })
296 } else {
297 Err(err())
298 }
299 }
300}
301
302impl FromStr for Schedule {
303 type Err = ParseError;
304
305 fn from_str(s: &str) -> Result<Self, Self::Err> {
306 let err = || ParseError::new("schedule", s);
307 let tokens: Vec<&str> = s.split_whitespace().collect();
314 if tokens.is_empty() {
315 return Err(err());
316 }
317 let mut segments = Vec::new();
318 for segment_tokens in tokens.split(|t| *t == "andThen") {
319 let (primary_tokens, up_to_ms) = match segment_tokens.iter().position(|t| *t == "upTo")
320 {
321 None => (segment_tokens, None),
322 Some(pos) => {
323 let [duration] = &segment_tokens[pos + 1..] else {
325 return Err(err());
326 };
327 let bound = duration.parse::<DurationMs>().map_err(|_| err())?.0;
328 (&segment_tokens[..pos], Some(bound))
329 }
330 };
331 if primary_tokens.is_empty() {
332 return Err(err());
333 }
334 let primary: SchedulePrimary = primary_tokens.join(" ").parse().map_err(|_| err())?;
335 segments.push(ScheduleSegment { primary, up_to_ms });
336 }
337 let last = segments.len() - 1;
341 if segments
342 .iter()
343 .enumerate()
344 .any(|(i, segment)| (i < last) != segment.up_to_ms.is_some())
345 {
346 return Err(ParseError::with_note(
347 "schedule",
348 s,
349 "`upTo` must bound every segment except the last, and never the last \
350 (an unbounded segment never hands off; a bounded tail would leave \
351 attempts without a wait — cap total retrying with `attempts`)",
352 ));
353 }
354 Ok(Self { segments })
355 }
356}
357
358impl TryFrom<String> for Schedule {
359 type Error = ParseError;
360
361 fn try_from(s: String) -> Result<Self, Self::Error> {
362 s.parse()
363 }
364}
365
366#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
369#[serde(try_from = "String")]
370pub enum Condition {
371 Conn,
372 Timeout,
373 Cancelled,
374 Other,
375 Class4xx,
376 Class5xx,
377 Status(u16),
378}
379
380impl Condition {
381 pub fn matches(self, class: ErrorClass, http_status: Option<u16>) -> bool {
382 match self {
383 Self::Conn => class == ErrorClass::Conn,
384 Self::Timeout => class == ErrorClass::Timeout,
385 Self::Cancelled => class == ErrorClass::Cancelled,
386 Self::Other => class == ErrorClass::Other,
387 Self::Class4xx => {
388 class == ErrorClass::Http && http_status.is_some_and(|s| (400..=499).contains(&s))
389 }
390 Self::Class5xx => {
391 class == ErrorClass::Http && http_status.is_some_and(|s| (500..=599).contains(&s))
392 }
393 Self::Status(want) => class == ErrorClass::Http && http_status == Some(want),
394 }
395 }
396}
397
398impl FromStr for Condition {
399 type Err = ParseError;
400
401 fn from_str(s: &str) -> Result<Self, Self::Err> {
402 match s {
403 "conn" => Ok(Self::Conn),
404 "timeout" => Ok(Self::Timeout),
405 "cancelled" => Ok(Self::Cancelled),
406 "other" => Ok(Self::Other),
407 "4xx" => Ok(Self::Class4xx),
408 "5xx" => Ok(Self::Class5xx),
409 exact if exact.len() == 3 && exact.bytes().all(|b| b.is_ascii_digit()) => {
413 let code: u16 = exact
414 .parse()
415 .map_err(|_| ParseError::new("retry condition", s))?;
416 if (100..=599).contains(&code) {
417 Ok(Self::Status(code))
418 } else {
419 Err(ParseError::new("retry condition", s))
420 }
421 }
422 _ => Err(ParseError::new("retry condition", s)),
423 }
424 }
425}
426
427impl TryFrom<String> for Condition {
428 type Error = ParseError;
429
430 fn try_from(s: String) -> Result<Self, Self::Error> {
431 s.parse()
432 }
433}
434
435#[derive(Debug, Clone, PartialEq, Deserialize)]
438#[serde(default, deny_unknown_fields)]
439pub struct RetryPolicy {
440 pub attempts: NonZeroU32,
441 pub schedule: Schedule,
442 pub on: Vec<Condition>,
443}
444
445impl RetryPolicy {
446 pub const DEFAULT_ATTEMPTS: NonZeroU32 = NonZeroU32::new(3).unwrap();
447
448 pub fn default_on() -> Vec<Condition> {
450 vec![
451 Condition::Conn,
452 Condition::Timeout,
453 Condition::Status(429),
454 Condition::Class5xx,
455 ]
456 }
457
458 pub fn is_retryable(&self, class: ErrorClass, http_status: Option<u16>) -> bool {
459 self.on.iter().any(|c| c.matches(class, http_status))
460 }
461}
462
463impl Default for RetryPolicy {
464 fn default() -> Self {
465 Self {
466 attempts: Self::DEFAULT_ATTEMPTS,
467 schedule: Schedule::default(),
468 on: Self::default_on(),
469 }
470 }
471}
472
473#[derive(Debug, Clone, PartialEq, Deserialize)]
488#[serde(try_from = "BreakerPolicyDe")]
489pub struct BreakerPolicy {
490 pub failures: Option<NonZeroU64>,
493 pub cooldown: DurationMs,
494 pub window: Option<DurationMs>,
495 pub failure_rate: Option<f64>,
496 pub min_calls: Option<NonZeroU32>,
497}
498
499#[derive(Debug, Clone, Copy, PartialEq)]
502pub enum BreakerMode {
503 Count { failures: NonZeroU64 },
505 Rate {
507 window: DurationMs,
508 failure_rate: f64,
509 min_calls: NonZeroU32,
510 },
511}
512
513impl BreakerPolicy {
514 pub const DEFAULT_FAILURES: NonZeroU64 = NonZeroU64::new(5).unwrap();
516 pub const DEFAULT_MIN_CALLS: NonZeroU32 = NonZeroU32::new(10).unwrap();
518
519 #[must_use]
524 pub fn mode(&self) -> BreakerMode {
525 match (self.failures, self.window, self.failure_rate) {
526 (None, Some(window), Some(failure_rate)) => BreakerMode::Rate {
527 window,
528 failure_rate,
529 min_calls: self.min_calls.unwrap_or(Self::DEFAULT_MIN_CALLS),
530 },
531 (failures, _, _) => BreakerMode::Count {
532 failures: failures.unwrap_or(Self::DEFAULT_FAILURES),
533 },
534 }
535 }
536
537 #[must_use]
541 pub fn has_inert_rate_knobs(&self) -> bool {
542 self.failures.is_some()
543 && (self.window.is_some() || self.failure_rate.is_some() || self.min_calls.is_some())
544 }
545}
546
547#[derive(Deserialize)]
550#[serde(default, deny_unknown_fields)]
551struct BreakerPolicyDe {
552 failures: Option<NonZeroU64>,
553 cooldown: DurationMs,
554 window: Option<DurationMs>,
555 #[serde(deserialize_with = "de_failure_rate")]
556 failure_rate: Option<f64>,
557 min_calls: Option<NonZeroU32>,
558}
559
560impl Default for BreakerPolicyDe {
561 fn default() -> Self {
562 Self {
563 failures: None,
564 cooldown: DurationMs(15_000),
565 window: None,
566 failure_rate: None,
567 min_calls: None,
568 }
569 }
570}
571
572impl TryFrom<BreakerPolicyDe> for BreakerPolicy {
573 type Error = String;
574
575 fn try_from(de: BreakerPolicyDe) -> Result<Self, Self::Error> {
576 let rate_pair = de.window.is_some() && de.failure_rate.is_some();
577 let any_rate_knob =
578 de.window.is_some() || de.failure_rate.is_some() || de.min_calls.is_some();
579 if de.failures.is_none() && any_rate_knob && !rate_pair {
580 return Err(String::from(
581 "breaker rate mode requires both `window` and `failure_rate` \
582 (count mode sets `failures` instead)",
583 ));
584 }
585 Ok(Self {
586 failures: de.failures,
587 cooldown: de.cooldown,
588 window: de.window,
589 failure_rate: de.failure_rate,
590 min_calls: de.min_calls,
591 })
592 }
593}
594
595fn de_failure_rate<'de, D>(deserializer: D) -> Result<Option<f64>, D::Error>
600where
601 D: serde::Deserializer<'de>,
602{
603 let value = Option::<f64>::deserialize(deserializer)?;
604 if let Some(rate) = value
605 && !(rate > 0.0 && rate <= 1.0)
606 {
607 return Err(serde::de::Error::custom(format!(
608 "breaker.failure_rate must be greater than 0 and at most 1 (got {rate})"
609 )));
610 }
611 Ok(value)
612}
613
614impl Default for BreakerPolicy {
615 fn default() -> Self {
616 Self {
617 failures: None,
618 cooldown: DurationMs(15_000),
619 window: None,
620 failure_rate: None,
621 min_calls: None,
622 }
623 }
624}
625
626#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Deserialize)]
627#[serde(rename_all = "snake_case")]
628pub enum CacheScope {
629 #[default]
630 Memory,
631 Persistent,
632}
633
634#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Deserialize)]
635#[serde(rename_all = "snake_case")]
636pub enum CacheMode {
637 #[default]
638 Always,
639 Dev,
641}
642
643#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Deserialize)]
644#[serde(rename_all = "snake_case")]
645pub enum CacheKeySource {
646 #[default]
647 Args,
648 Url,
649}
650
651#[derive(Debug, Clone, PartialEq, Default, Deserialize)]
653#[serde(default, deny_unknown_fields)]
654pub struct CachePolicy {
655 pub ttl: Option<DurationMs>,
656 pub scope: CacheScope,
657 pub mode: CacheMode,
658 pub key: CacheKeySource,
659}
660
661#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
663#[serde(deny_unknown_fields)]
664pub struct IdempotencyPolicy {
665 pub header: String,
666}
667
668#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
672#[serde(try_from = "PollUntilRaw")]
673pub struct PollUntil {
674 pub field: String,
675 pub terminal: Vec<String>,
676}
677
678#[derive(Deserialize)]
679#[serde(deny_unknown_fields)]
680struct PollUntilRaw {
681 field: String,
682 terminal: Vec<String>,
683}
684
685impl TryFrom<PollUntilRaw> for PollUntil {
686 type Error = ParseError;
687
688 fn try_from(raw: PollUntilRaw) -> Result<Self, Self::Error> {
689 if raw.field.is_empty() {
690 return Err(ParseError::new("poll until.field", "(empty)"));
691 }
692 if raw.terminal.is_empty() {
693 return Err(ParseError::new("poll until.terminal", "(empty array)"));
694 }
695 Ok(Self {
696 field: raw.field,
697 terminal: raw.terminal,
698 })
699 }
700}
701
702#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
708#[serde(try_from = "PollPolicyRaw")]
709pub struct PollPolicy {
710 pub interval: DurationMs,
711 pub deadline: DurationMs,
712 pub until: PollUntil,
713}
714
715#[derive(Deserialize)]
716#[serde(deny_unknown_fields)]
717struct PollPolicyRaw {
718 interval: DurationMs,
719 deadline: DurationMs,
720 until: PollUntil,
721}
722
723impl TryFrom<PollPolicyRaw> for PollPolicy {
724 type Error = ParseError;
725
726 fn try_from(raw: PollPolicyRaw) -> Result<Self, Self::Error> {
727 if raw.interval.0 == 0 {
728 return Err(ParseError::with_note(
729 "poll.interval",
730 "0",
731 "must be a nonzero duration",
732 ));
733 }
734 Ok(Self {
735 interval: raw.interval,
736 deadline: raw.deadline,
737 until: raw.until,
738 })
739 }
740}
741
742#[derive(Debug, Clone, PartialEq, Default, Deserialize)]
745#[serde(default, deny_unknown_fields)]
746pub struct TargetPolicy {
747 pub timeout: Option<DurationMs>,
748 pub retry: Option<RetryPolicy>,
749 pub breaker: Option<BreakerPolicy>,
750 pub rate: Option<Rate>,
751 pub cache: Option<CachePolicy>,
752 pub idempotency: Option<IdempotencyPolicy>,
753 pub poll: Option<PollPolicy>,
754 pub fallback: Option<Vec<String>>,
755 pub budget: Option<String>,
756}
757
758#[derive(Debug, Clone, PartialEq, Default, Deserialize)]
759#[serde(default, deny_unknown_fields)]
760pub struct Defaults {
761 pub outbound: Option<TargetPolicy>,
762 pub llm: Option<TargetPolicy>,
763}
764
765#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Deserialize)]
766#[serde(rename_all = "snake_case")]
767pub enum NondeterminismResponse {
768 #[default]
769 Fail,
770 Warn,
771 Branch,
772}
773
774#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Deserialize)]
777#[serde(rename_all = "snake_case")]
778pub enum OnBusy {
779 #[default]
780 Skip,
781 Wait,
782 Fail,
783}
784
785#[derive(Debug, Clone, PartialEq, Eq, Default, Deserialize)]
787#[serde(default, deny_unknown_fields)]
788pub struct FlowsPolicy {
789 pub entrypoints: Vec<String>,
790 pub on_nondeterminism: NondeterminismResponse,
791 pub on_busy: OnBusy,
792}
793
794#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
801#[serde(try_from = "String")]
802pub struct JournalLocation(pub String);
803
804impl FromStr for JournalLocation {
805 type Err = ParseError;
806
807 fn from_str(s: &str) -> Result<Self, Self::Err> {
808 let valid = s.strip_prefix("file:").is_some_and(|rest| !rest.is_empty())
809 || s.strip_prefix("postgres://")
810 .is_some_and(|rest| !rest.is_empty());
811 if valid {
812 Ok(Self(s.to_owned()))
813 } else {
814 Err(ParseError::new("journal location", s))
815 }
816 }
817}
818
819impl TryFrom<String> for JournalLocation {
820 type Error = ParseError;
821
822 fn try_from(s: String) -> Result<Self, Self::Error> {
823 s.parse()
824 }
825}
826
827#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
836#[serde(default, deny_unknown_fields)]
837pub struct TelemetryPolicy {
838 pub otlp_endpoint: Option<String>,
839 pub console: bool,
840}
841
842impl Default for TelemetryPolicy {
843 fn default() -> Self {
844 Self {
846 otlp_endpoint: None,
847 console: true,
848 }
849 }
850}
851
852#[derive(Debug, Clone, PartialEq, Default, Deserialize)]
861#[serde(default, deny_unknown_fields)]
862pub struct Policy {
863 pub defaults: Defaults,
864 pub target: BTreeMap<String, TargetPolicy>,
865 pub flows: Option<FlowsPolicy>,
866 pub journal: Option<JournalLocation>,
869 pub telemetry: Option<TelemetryPolicy>,
873}
874
875#[derive(Debug, Clone, Default)]
878pub struct ResolvedPolicy {
879 pub timeout: Option<DurationMs>,
880 pub retry: Option<RetryPolicy>,
881 pub breaker: Option<BreakerPolicy>,
882 pub rate: Option<Rate>,
883 pub cache: Option<CachePolicy>,
884 pub idempotency: Option<IdempotencyPolicy>,
889 pub poll: Option<PollPolicy>,
891}
892
893impl Policy {
894 pub fn resolve(&self, target: &str) -> ResolvedPolicy {
895 ResolvedPolicy {
896 timeout: self.layer(target, |t| t.timeout.as_ref()).copied(),
897 retry: self.layer(target, |t| t.retry.as_ref()).cloned(),
898 breaker: self.layer(target, |t| t.breaker.as_ref()).cloned(),
899 rate: self.layer(target, |t| t.rate.as_ref()).copied(),
900 cache: self.layer(target, |t| t.cache.as_ref()).cloned(),
901 idempotency: self.layer(target, |t| t.idempotency.as_ref()).cloned(),
902 poll: self.layer(target, |t| t.poll.as_ref()).cloned(),
903 }
904 }
905
906 fn layer<'a, T>(
907 &'a self,
908 target: &str,
909 pick: impl Fn(&'a TargetPolicy) -> Option<&'a T>,
910 ) -> Option<&'a T> {
911 if let Some(t) = self.target.get(target)
912 && let Some(v) = pick(t)
913 {
914 return Some(v);
915 }
916 if target.starts_with("llm:")
917 && let Some(llm) = self.defaults.llm.as_ref()
918 && let Some(v) = pick(llm)
919 {
920 return Some(v);
921 }
922 self.defaults.outbound.as_ref().and_then(pick)
923 }
924}
925
926#[cfg(test)]
927mod tests {
928 use super::*;
929 use serde_json::json;
930
931 #[test]
932 fn duration_literals() {
933 assert_eq!("200ms".parse(), Ok(DurationMs(200)));
934 assert_eq!("30s".parse(), Ok(DurationMs(30_000)));
935 assert_eq!("5m".parse(), Ok(DurationMs(300_000)));
936 assert_eq!("2h".parse(), Ok(DurationMs(7_200_000)));
937 assert!("30".parse::<DurationMs>().is_err());
938 assert!("30sec".parse::<DurationMs>().is_err());
939 assert!("-1s".parse::<DurationMs>().is_err());
940 }
941
942 #[test]
943 fn rate_literals() {
944 let rate: Rate = "90/s".parse().unwrap();
945 assert_eq!((rate.limit.get(), rate.window_ms), (90, 1_000));
946 let rate: Rate = "60/min".parse().unwrap();
947 assert_eq!((rate.limit.get(), rate.window_ms), (60, 60_000));
948 assert!("0/s".parse::<Rate>().is_err(), "zero limit unrepresentable");
949 assert!("10/day".parse::<Rate>().is_err());
950 }
951
952 #[test]
953 fn schedule_exp_waits_and_cap() {
954 let schedule: Schedule = "exp(1s, x2, max 4s)".parse().unwrap();
955 let waits: Vec<u64> = (1..=4).map(|n| schedule.wait_ms(n)).collect();
956 assert_eq!(waits, [1_000, 2_000, 4_000, 4_000]);
957 }
958
959 #[test]
960 fn schedule_fixed_and_rejections() {
961 assert_eq!(
962 "fixed(1s)".parse::<Schedule>(),
963 Ok(Schedule {
964 segments: vec![ScheduleSegment {
965 primary: SchedulePrimary::Fixed { period_ms: 1_000 },
966 up_to_ms: None,
967 }],
968 })
969 );
970 assert!("linear(1s)".parse::<Schedule>().is_err());
971 }
972
973 #[test]
974 fn schedule_composition_parses_the_spec_example() {
975 let schedule: Schedule = "exp(1s, x2, max 5m) upTo 10m andThen fixed(1m)"
977 .parse()
978 .unwrap();
979 assert_eq!(
980 schedule.segments,
981 vec![
982 ScheduleSegment {
983 primary: SchedulePrimary::Exp {
984 base_ms: 1_000,
985 factor: 2.0,
986 cap_ms: 300_000,
987 jitter: false,
988 },
989 up_to_ms: Some(600_000),
990 },
991 ScheduleSegment {
992 primary: SchedulePrimary::Fixed { period_ms: 60_000 },
993 up_to_ms: None,
994 },
995 ]
996 );
997 assert_eq!(
999 "exp(1s, x2, max 5m) upTo 10m andThen fixed(1m)".parse::<Schedule>(),
1000 Ok(schedule)
1001 );
1002 }
1003
1004 #[test]
1005 fn schedule_composition_hands_off_when_the_bound_would_be_overshot() {
1006 let schedule: Schedule = "exp(1s, x2) upTo 4s andThen fixed(500ms)".parse().unwrap();
1007 let waits: Vec<u64> = (1..=5).map(|n| schedule.wait_ms(n)).collect();
1008 assert_eq!(waits, [1_000, 2_000, 500, 500, 500]);
1010 }
1011
1012 #[test]
1013 fn schedule_composition_exact_fit_stays_and_cascade_skips() {
1014 let schedule: Schedule =
1015 "fixed(1s) upTo 3s andThen fixed(10s) upTo 5s andThen fixed(250ms)"
1016 .parse()
1017 .unwrap();
1018 let waits: Vec<u64> = (1..=6).map(|n| schedule.wait_ms(n)).collect();
1019 assert_eq!(waits, [1_000, 1_000, 1_000, 250, 250, 250]);
1023 }
1024
1025 #[test]
1026 fn schedule_composition_restarts_exp_and_tracks_jitter_per_segment() {
1027 let schedule: Schedule = "fixed(1s) upTo 2s andThen exp(100ms, x3, jitter)"
1028 .parse()
1029 .unwrap();
1030 let waits: Vec<u64> = (1..=5).map(|n| schedule.wait_ms(n)).collect();
1032 assert_eq!(waits, [1_000, 1_000, 100, 300, 900]);
1033 assert_eq!(schedule.wait_and_jitter(1), (1_000, false));
1035 assert_eq!(schedule.wait_and_jitter(3), (100, true));
1036 }
1037
1038 #[test]
1039 fn schedule_composition_shape_rule_rejections() {
1040 for degenerate in [
1045 "fixed(1s) andThen fixed(2s)",
1046 "exp(1s, x2, max 5m) upTo 10m",
1047 "fixed(1s) upTo 3s andThen fixed(2s) andThen fixed(4s)",
1048 "fixed(1s) upTo 3s andThen fixed(2s) upTo 5s",
1049 ] {
1050 let error = degenerate.parse::<Schedule>().unwrap_err();
1051 assert!(
1052 error.to_string().contains("upTo"),
1053 "{degenerate}: expected the shape-rule note, got {error}"
1054 );
1055 }
1056 for broken in [
1058 "fixed(1s) upTo",
1059 "upTo 3s andThen fixed(1s)",
1060 "fixed(1s) upTo 1s upTo 2s andThen fixed(1s)",
1061 "fixed(1s) andThen",
1062 "andThen fixed(1s)",
1063 "fixed(1s) upTo 3s fixed(2s)",
1064 ] {
1065 assert!(
1066 broken.parse::<Schedule>().is_err(),
1067 "{broken} must be rejected"
1068 );
1069 }
1070 }
1071
1072 #[test]
1073 fn condition_matching() {
1074 let on = RetryPolicy::default_on();
1075 let matches = |class, status| on.iter().any(|c| c.matches(class, status));
1076 assert!(matches(ErrorClass::Conn, None));
1077 assert!(matches(ErrorClass::Http, Some(429)));
1078 assert!(matches(ErrorClass::Http, Some(503)));
1079 assert!(!matches(ErrorClass::Http, Some(400)));
1080 assert!(!matches(ErrorClass::Cancelled, None));
1081 assert!("teapot".parse::<Condition>().is_err());
1082 assert_eq!("429".parse::<Condition>(), Ok(Condition::Status(429)));
1084 assert_eq!("100".parse::<Condition>(), Ok(Condition::Status(100)));
1085 assert_eq!("599".parse::<Condition>(), Ok(Condition::Status(599)));
1086 for bad in ["999", "099", "600", "000", "12", "1234", "1x9"] {
1087 assert!(bad.parse::<Condition>().is_err(), "{bad} must be rejected");
1088 }
1089 }
1090
1091 #[test]
1092 fn zero_attempts_is_unrepresentable() {
1093 let doc = json!({ "target": { "x": { "retry": { "attempts": 0 } } } });
1094 let err = serde_path_to_error::deserialize::<_, Policy>(&doc).unwrap_err();
1095 assert_eq!(err.path().to_string(), "target.x.retry.attempts");
1096 }
1097
1098 #[test]
1099 fn breaker_failure_rate_range_is_enforced() {
1100 let bad = |rate: serde_json::Value| {
1102 let doc = json!({ "target": { "x": { "breaker": { "failure_rate": rate } } } });
1103 serde_path_to_error::deserialize::<_, Policy>(&doc)
1104 };
1105 for rate in [json!(0.0), json!(-0.1), json!(1.5), json!(2.0)] {
1106 let err = bad(rate.clone()).unwrap_err();
1107 assert_eq!(
1108 err.path().to_string(),
1109 "target.x.breaker.failure_rate",
1110 "out-of-range failure_rate {rate} must fail at its path"
1111 );
1112 }
1113 for rate in [0.01_f64, 0.5, 1.0] {
1116 let doc = json!({
1117 "target": { "x": { "breaker": { "window": "30s", "failure_rate": rate } } }
1118 });
1119 let policy = serde_path_to_error::deserialize::<_, Policy>(&doc).unwrap();
1120 let breaker = policy.target["x"].breaker.as_ref().unwrap();
1121 assert_eq!(breaker.failure_rate, Some(rate));
1122 }
1123 }
1124
1125 #[test]
1126 fn breaker_mode_selection_follows_the_schema() {
1127 let breaker = |doc: serde_json::Value| -> BreakerPolicy {
1128 let doc = json!({ "target": { "x": { "breaker": doc } } });
1129 let policy: Policy = serde_path_to_error::deserialize(&doc).unwrap();
1130 policy.target["x"].breaker.clone().unwrap()
1131 };
1132
1133 assert_eq!(
1135 breaker(json!({})).mode(),
1136 BreakerMode::Count {
1137 failures: BreakerPolicy::DEFAULT_FAILURES
1138 }
1139 );
1140
1141 assert_eq!(
1143 breaker(json!({ "window": "30s", "failure_rate": 0.5 })).mode(),
1144 BreakerMode::Rate {
1145 window: DurationMs(30_000),
1146 failure_rate: 0.5,
1147 min_calls: BreakerPolicy::DEFAULT_MIN_CALLS,
1148 }
1149 );
1150 assert_eq!(
1151 breaker(json!({ "window": "10s", "failure_rate": 1.0, "min_calls": 4 })).mode(),
1152 BreakerMode::Rate {
1153 window: DurationMs(10_000),
1154 failure_rate: 1.0,
1155 min_calls: NonZeroU32::new(4).unwrap(),
1156 }
1157 );
1158
1159 let mixed = breaker(json!({ "failures": 3, "window": "30s", "failure_rate": 0.5 }));
1162 assert_eq!(
1163 mixed.mode(),
1164 BreakerMode::Count {
1165 failures: NonZeroU64::new(3).unwrap()
1166 }
1167 );
1168 assert!(mixed.has_inert_rate_knobs());
1169 assert!(!breaker(json!({ "failures": 3 })).has_inert_rate_knobs());
1170 }
1171
1172 #[test]
1173 fn half_configured_breaker_rate_mode_is_rejected() {
1174 for doc in [
1178 json!({ "window": "30s" }),
1179 json!({ "failure_rate": 0.5 }),
1180 json!({ "min_calls": 10 }),
1181 json!({ "window": "30s", "min_calls": 10 }),
1182 json!({ "failure_rate": 0.5, "min_calls": 10 }),
1183 ] {
1184 let policy = json!({ "target": { "x": { "breaker": doc } } });
1185 let err = serde_path_to_error::deserialize::<_, Policy>(&policy).unwrap_err();
1186 assert_eq!(err.path().to_string(), "target.x.breaker", "doc: {doc}");
1187 assert!(
1188 err.inner().to_string().contains("rate mode requires both"),
1189 "doc {doc}: got {}",
1190 err.inner()
1191 );
1192 }
1193 let policy =
1196 json!({ "target": { "x": { "breaker": { "failures": 3, "window": "30s" } } } });
1197 assert!(serde_path_to_error::deserialize::<_, Policy>(&policy).is_ok());
1198 }
1199
1200 #[test]
1201 fn unknown_key_is_rejected_with_its_path() {
1202 let doc = json!({ "target": { "api.stripe.com": { "retry": { "atempts": 10 } } } });
1206 let err = serde_path_to_error::deserialize::<_, Policy>(&doc).unwrap_err();
1207 assert!(
1208 err.inner().to_string().contains("atempts")
1209 || err.inner().to_string().contains("unknown field"),
1210 "expected an unknown-field error, got {}",
1211 err.inner()
1212 );
1213 }
1214
1215 #[test]
1216 fn unknown_top_level_and_layer_keys_are_rejected() {
1217 assert!(
1218 serde_path_to_error::deserialize::<_, Policy>(&json!({ "bogus_top": true })).is_err()
1219 );
1220 assert!(
1221 serde_path_to_error::deserialize::<_, Policy>(
1222 &json!({ "target": { "api.x": { "retrys": {} } } })
1223 )
1224 .is_err(),
1225 "a mistyped layer table must be rejected, not dropped"
1226 );
1227 }
1228
1229 #[test]
1230 fn journal_and_telemetry_parse_and_validate() {
1231 let doc = json!({
1232 "journal": "file:/srv/keel/journal.db",
1233 "telemetry": { "otlp_endpoint": "http://collector:4317" }
1234 });
1235 let policy: Policy = serde_path_to_error::deserialize(&doc).unwrap();
1236 assert_eq!(
1237 policy.journal.unwrap(),
1238 JournalLocation("file:/srv/keel/journal.db".to_owned())
1239 );
1240 let telemetry = policy.telemetry.unwrap();
1241 assert_eq!(
1242 telemetry.otlp_endpoint.as_deref(),
1243 Some("http://collector:4317")
1244 );
1245 assert!(telemetry.console, "schema default console = true");
1246
1247 let bad = json!({ "journal": "sqlite:/tmp/x.db" });
1249 assert!(serde_path_to_error::deserialize::<_, Policy>(&bad).is_err());
1250 }
1251
1252 #[test]
1253 fn idempotency_resolves_like_any_other_layer() {
1254 let doc = json!({
1258 "defaults": {
1259 "outbound": { "idempotency": { "header": "X-Idem" } },
1260 "llm": { "idempotency": { "header": "X-Llm-Idem" } }
1261 },
1262 "target": {
1263 "api.stripe.com": { "idempotency": { "header": "Idempotency-Key" } },
1264 "api.plain.example": { "timeout": "1s" }
1265 }
1266 });
1267 let policy: Policy = serde_path_to_error::deserialize(&doc).unwrap();
1268
1269 let stripe = policy.resolve("api.stripe.com");
1271 assert_eq!(
1272 stripe.idempotency.as_ref().map(|i| i.header.as_str()),
1273 Some("Idempotency-Key")
1274 );
1275 let llm = policy.resolve("llm:openai");
1277 assert_eq!(
1278 llm.idempotency.as_ref().map(|i| i.header.as_str()),
1279 Some("X-Llm-Idem")
1280 );
1281 let plain = policy.resolve("api.plain.example");
1282 assert_eq!(
1283 plain.idempotency.as_ref().map(|i| i.header.as_str()),
1284 Some("X-Idem")
1285 );
1286 let empty: Policy = serde_path_to_error::deserialize(&json!({})).unwrap();
1288 assert!(empty.resolve("api.stripe.com").idempotency.is_none());
1289 }
1290
1291 #[test]
1292 fn poll_policy_parses_and_resolves() {
1293 let policy: Policy = serde_json::from_value(serde_json::json!({
1294 "target": { "api.jobs.example": { "poll": {
1295 "interval": "10s", "deadline": "90s",
1296 "until": { "field": "status", "terminal": ["completed", "failed"] }
1297 } } }
1298 }))
1299 .expect("valid poll policy");
1300 let resolved = policy.resolve("api.jobs.example");
1301 let poll = resolved.poll.expect("poll resolved");
1302 assert_eq!(poll.interval.0, 10_000);
1303 assert_eq!(poll.deadline.0, 90_000);
1304 assert_eq!(poll.until.field, "status");
1305 assert_eq!(poll.until.terminal, vec!["completed", "failed"]);
1306 }
1307
1308 #[test]
1309 fn poll_rejects_empty_terminal_and_empty_field() {
1310 for bad in [
1311 serde_json::json!({ "interval": "10s", "deadline": "90s",
1312 "until": { "field": "status", "terminal": [] } }),
1313 serde_json::json!({ "interval": "10s", "deadline": "90s",
1314 "until": { "field": "", "terminal": ["done"] } }),
1315 serde_json::json!({ "interval": "10s",
1316 "until": { "field": "status", "terminal": ["done"] } }),
1317 ] {
1318 let doc = serde_json::json!({ "target": { "x": { "poll": bad } } });
1319 assert!(serde_json::from_value::<Policy>(doc).is_err());
1320 }
1321 }
1322
1323 #[test]
1324 fn poll_rejects_zero_interval() {
1325 let doc = serde_json::json!({ "target": { "x": { "poll": {
1326 "interval": "0ms", "deadline": "90s",
1327 "until": { "field": "status", "terminal": ["done"] }
1328 } } } });
1329 assert!(serde_json::from_value::<Policy>(doc).is_err());
1330
1331 let doc = serde_json::json!({ "target": { "x": { "poll": {
1332 "interval": "1ms", "deadline": "90s",
1333 "until": { "field": "status", "terminal": ["done"] }
1334 } } } });
1335 assert!(serde_json::from_value::<Policy>(doc).is_ok());
1336 }
1337
1338 #[test]
1339 fn layer_resolution_precedence() {
1340 let doc = json!({
1341 "defaults": {
1342 "outbound": { "retry": { "attempts": 3 }, "rate": "9/s" },
1343 "llm": { "retry": { "attempts": 6 } }
1344 },
1345 "target": { "llm:openai": { "cache": { "ttl": "10m" } } }
1346 });
1347 let policy: Policy = serde_path_to_error::deserialize(&doc).unwrap();
1348
1349 let llm = policy.resolve("llm:openai");
1352 assert_eq!(llm.cache.unwrap().ttl, Some(DurationMs(600_000)));
1353 assert_eq!(llm.retry.unwrap().attempts.get(), 6);
1354 assert_eq!(llm.rate.unwrap().limit.get(), 9);
1355
1356 let plain = policy.resolve("api.example.com");
1358 assert_eq!(plain.retry.unwrap().attempts.get(), 3);
1359 assert!(plain.cache.is_none());
1360 }
1361
1362 #[test]
1363 fn flows_on_busy_parses_with_skip_default() {
1364 let p: Policy = serde_json::from_value(serde_json::json!({
1365 "flows": { "entrypoints": ["cmd:autonomous-run"], "on_busy": "wait" }
1366 }))
1367 .unwrap();
1368 assert_eq!(p.flows.as_ref().unwrap().on_busy, OnBusy::Wait);
1369 let p: Policy = serde_json::from_value(serde_json::json!({ "flows": {} })).unwrap();
1370 assert_eq!(p.flows.unwrap().on_busy, OnBusy::Skip);
1371 }
1372}