1mod round;
20
21use std::num::NonZeroU64;
22
23use kernel::account_recovery::{AccountStep, RoundEnd};
24use kernel::event::record::{FiredAction, WatchdogFired};
25use kernel::model::AccountRoster;
26use kernel::{AxCode, AxError, Payload, Retries, RunId, ServerLabel, StallVerdict, TimeMs};
27
28use round::Round;
29
30#[derive(Debug, Default)]
34pub struct Watchdog {
35 corrections: u32,
36 provider_failures: u32,
37 streak: u32,
43 retries: Retries,
44 jitter_seed: u64,
47 round: Option<Round>,
49}
50
51#[derive(Debug, Clone, PartialEq, Eq)]
54pub enum Disposal {
55 Proceed,
56 CorrectiveSteer {
57 text: String,
58 },
59 BackOff {
65 until: TimeMs,
66 code: AxCode,
67 subject: String,
68 },
69 Switch {
73 to: ServerLabel,
74 code: AxCode,
75 subject: String,
76 },
77 Freeze {
78 reason: FreezeReason,
79 },
80}
81
82impl Disposal {
83 fn backing_off(until: TimeMs, failure: &AxError) -> Disposal {
85 Disposal::BackOff {
86 until,
87 code: *failure.code(),
88 subject: failure.subject().to_owned(),
89 }
90 }
91}
92
93#[derive(Debug, Clone, Copy, PartialEq, Eq)]
94pub enum FreezeReason {
95 Stall,
96 ProviderRefused,
97 AccountsExhausted,
101}
102
103impl FreezeReason {
104 fn as_str(self) -> &'static str {
105 match self {
106 FreezeReason::Stall => "stall",
107 FreezeReason::ProviderRefused => "provider_refused",
108 FreezeReason::AccountsExhausted => "accounts_exhausted",
109 }
110 }
111}
112
113impl Watchdog {
114 #[must_use]
115 pub fn new(retries: Retries, run: RunId) -> Watchdog {
116 Watchdog {
117 retries,
118 jitter_seed: seed_of(run),
119 ..Watchdog::default()
120 }
121 }
122
123 pub fn on_stall(&mut self, verdict: &StallVerdict) -> Disposal {
126 match verdict {
127 StallVerdict::Ok => Disposal::Proceed,
128 StallVerdict::Stall { repeats } => {
129 if self.corrections == 0 {
130 self.corrections = 1;
131 Disposal::CorrectiveSteer {
132 text: format!(
133 "You have repeated the same call {repeats} times in a row. \
134 Change the approach or report why the goal cannot be met."
135 ),
136 }
137 } else {
138 Disposal::Freeze {
139 reason: FreezeReason::Stall,
140 }
141 }
142 }
143 }
144 }
145
146 pub fn open_round(
159 &mut self,
160 roster: Option<AccountRoster>,
161 ) -> Result<Option<ServerLabel>, AxError> {
162 let round = Round::open(roster, self.retries)?;
163 let current = round.accounts.current().cloned();
164 self.round = Some(round);
165 Ok(current)
166 }
167
168 #[must_use]
171 pub fn round_open(&self) -> bool {
172 self.round.is_some()
173 }
174
175 pub fn on_provider_failure(&mut self, failure: &AxError, now: TimeMs) -> Disposal {
200 self.provider_failures = self.provider_failures.saturating_add(1);
201 self.streak = self.streak.saturating_add(1);
202 let wait = wait_ms(self.jitter_seed, self.streak, failure);
203 let not_before = TimeMs::new(now.value().saturating_add(wait));
204 let mut round = match self
205 .round
206 .take()
207 .map_or_else(|| Round::open(None, self.retries), Ok)
208 {
209 Ok(round) => round,
210 Err(_exhausted) => {
213 return Disposal::Freeze {
214 reason: FreezeReason::AccountsExhausted,
215 };
216 }
217 };
218 round.note(failure);
219 let step = round.accounts.next(failure);
220 let disposal = match &step {
221 AccountStep::Resend => Disposal::backing_off(not_before, failure),
222 AccountStep::Switch { to } => {
223 self.streak = 0;
224 Disposal::Switch {
225 to: to.clone(),
226 code: *failure.code(),
227 subject: failure.subject().to_owned(),
228 }
229 }
230 AccountStep::Stop {
231 why: RoundEnd::Refused | RoundEnd::Unknown | RoundEnd::Cap,
232 } => Disposal::Freeze {
233 reason: FreezeReason::ProviderRefused,
234 },
235 AccountStep::Stop {
236 why: RoundEnd::Exhausted,
237 } => Disposal::Freeze {
238 reason: FreezeReason::AccountsExhausted,
239 },
240 };
241 round.accounts = round.accounts.apply(&step);
242 self.round = Some(round);
243 disposal
244 }
245
246 pub fn on_provider_answered(&mut self) {
249 self.streak = 0;
250 self.round = None;
251 }
252
253 #[must_use]
258 pub fn admits_repair(&self) -> bool {
259 self.round
260 .as_ref()
261 .is_none_or(|round| round.accounts.admits_repair())
262 }
263
264 pub fn repaired(&mut self) {
267 self.round = self.round.take().map(Round::repaired);
268 }
269
270 #[must_use]
275 pub fn exhausted(&self) -> AxError {
276 self.round
277 .as_ref()
278 .map_or_else(round::nothing_listed, Round::exhausted)
279 }
280
281 pub fn fired_payload(&self, disposal: &Disposal) -> Result<Payload, AxError> {
288 let action = match disposal {
289 Disposal::Proceed => {
290 return Err(AxError::failure(
291 AxCode::InvalidArgs,
292 "encode watchdog_fired",
293 "Proceed does not fire",
294 )
295 .with_recovery(
296 "report this against runtime::watchdog: `Proceed` is the verdict \
297 that records nothing, and the caller asked it for a payload",
298 ));
299 }
300 Disposal::CorrectiveSteer { text } => FiredAction::Steer { text: text.clone() },
301 Disposal::BackOff {
302 until,
303 code,
304 subject,
305 } => FiredAction::BackOff {
306 until_ms: until.value(),
307 code: code.as_str().to_owned(),
308 subject: subject.clone(),
309 },
310 Disposal::Switch { to, code, subject } => FiredAction::Switch {
311 to: to.clone(),
312 code: code.as_str().to_owned(),
313 subject: subject.clone(),
314 },
315 Disposal::Freeze { reason } => FiredAction::Freeze {
316 reason: reason.as_str().to_owned(),
317 },
318 };
319 Payload::of(&WatchdogFired {
320 action,
321 corrections: self.corrections,
322 provider_failures: self.provider_failures,
323 })
324 }
325}
326
327#[must_use]
336pub fn backoff_ms(run: RunId, failures_in_a_row: u32, failure: &AxError) -> u64 {
337 wait_ms(seed_of(run), failures_in_a_row, failure)
338}
339
340fn wait_ms(seed: u64, streak: u32, failure: &AxError) -> u64 {
349 const FIRST_MS: u64 = 500;
350 const CEILING_MS: u64 = 60_000;
351 let doublings = streak.saturating_sub(1).min(16);
352 let base = FIRST_MS
353 .checked_shl(doublings)
354 .map_or(CEILING_MS, |wait| wait.min(CEILING_MS));
355 let spread = NonZeroU64::MIN.saturating_add(base >> 1);
356 let own = base.saturating_add(mixed(seed ^ u64::from(streak)) % spread);
357 failure.retry_after_ms().map_or(own, |told| own.max(told))
358}
359
360fn seed_of(run: RunId) -> u64 {
364 run.as_bytes()
365 .iter()
366 .fold(0xcbf2_9ce4_8422_2325, |hash, byte| {
367 (hash ^ u64::from(*byte)).wrapping_mul(0x0000_0100_0000_01b3)
368 })
369}
370
371fn mixed(word: u64) -> u64 {
374 let word = (word ^ (word >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
375 let word = (word ^ (word >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb);
376 word ^ (word >> 31)
377}
378
379#[cfg(test)]
380#[allow(
381 clippy::unwrap_used,
382 clippy::expect_used,
383 clippy::panic,
384 clippy::indexing_slicing,
385 clippy::wildcard_enum_match_arm,
386 clippy::let_underscore_must_use,
387 clippy::let_underscore_untyped,
388 clippy::arithmetic_side_effects,
389 reason = "test code"
390)]
391mod tests {
392 use super::*;
393 use kernel::ActionFingerprint;
394 use kernel::stall::observe;
395
396 const BASES: [u64; 9] = [
398 500, 1_000, 2_000, 4_000, 8_000, 16_000, 32_000, 60_000, 60_000,
399 ];
400
401 fn run(tail: u8) -> RunId {
404 let mut bytes = [
405 0x01, 0x93, 0x2a, 0x7c, 0x10, 0x00, 0x70, 0x00, 0x80, 0, 0, 0, 0, 0, 0, 0,
406 ];
407 bytes[15] = tail;
408 RunId::from_bytes(bytes)
409 }
410
411 fn waits(dog: &mut Watchdog, count: usize) -> Vec<u64> {
414 (0..count)
415 .map(
416 |_| match dog.on_provider_failure(&provider_error(true), TimeMs::new(0)) {
417 Disposal::BackOff { until, .. } => until.value(),
418 other => panic!("a retriable failure under its ceiling backs off: {other:?}"),
419 },
420 )
421 .collect()
422 }
423
424 fn within_schedule(waits: &[u64]) -> bool {
425 waits
426 .iter()
427 .zip(BASES)
428 .all(|(wait, base)| (base..=base + base / 2).contains(wait))
429 }
430
431 #[test]
435 fn each_run_waits_its_own_jittered_schedule() {
436 let schedule = |tail: u8| waits(&mut Watchdog::new(Retries::UntilHalted, run(tail)), 9);
437 let (first, second) = (schedule(1), schedule(2));
438 assert!(
439 within_schedule(&first) && within_schedule(&second),
440 "{first:?} {second:?}"
441 );
442 assert_eq!(first, schedule(1), "the same run waits the same schedule");
443 assert_ne!(first, second, "two runs do not ask again in step");
444 }
445
446 #[test]
447 fn disposal_is_graded_steer_first_freeze_second() {
448 let mut dog = Watchdog::new(Retries::UntilHalted, run(1));
449 let same = ActionFingerprint::derive(b"exec identical");
450 let sample = vec![same, same, same];
451 let verdict = observe(&sample);
452 let first = dog.on_stall(&verdict);
453 match &first {
454 Disposal::CorrectiveSteer { text } => {
455 assert!(text.contains("repeated the same call 3 times"))
456 }
457 other => panic!("first hit must steer, got {other:?}"),
458 }
459 let second = dog.on_stall(&verdict);
460 assert_eq!(
461 second,
462 Disposal::Freeze {
463 reason: FreezeReason::Stall
464 }
465 );
466 let payload = serde_json::to_value(dog.fired_payload(&second).unwrap()).unwrap();
468 assert_eq!(payload["action"], "freeze");
469 assert_eq!(payload["reason"], "stall");
470 assert_eq!(
473 dog.fired_payload(&Disposal::Proceed)
474 .unwrap_err()
475 .recovery(),
476 "report this against runtime::watchdog: `Proceed` is the verdict that records nothing, and the caller asked it for a payload"
477 );
478 }
479
480 #[test]
481 fn ok_verdicts_never_dispose() {
482 let mut dog = Watchdog::new(Retries::UntilHalted, run(1));
483 assert_eq!(dog.on_stall(&StallVerdict::Ok), Disposal::Proceed);
484 assert_eq!(dog.on_stall(&StallVerdict::Ok), Disposal::Proceed);
485 }
486
487 fn provider_error(retriable: bool) -> AxError {
488 let draft = AxError::failure(AxCode::Provider, "call the model", "the provider said no");
489 let draft = if retriable { draft.retriable() } else { draft };
490 draft.with_recovery("dispatch again, or attach a second endpoint")
491 }
492
493 #[test]
494 fn a_failure_the_provider_will_repeat_stops_after_one() {
495 let mut dog = Watchdog::new(Retries::UntilHalted, run(1));
496 assert_eq!(
497 dog.on_provider_failure(&provider_error(false), TimeMs::new(9_000)),
498 Disposal::Freeze {
499 reason: FreezeReason::ProviderRefused
500 },
501 "a request the provider rejected on its shape buys the same rejection again"
502 );
503 }
504
505 #[test]
506 fn an_answer_ends_the_streak_the_backoff_counts() {
507 let mut dog = Watchdog::new(Retries::AtMost(2), run(1));
508 let before = waits(&mut dog, 2);
509 assert!(within_schedule(&before), "{before:?}");
510 dog.on_provider_answered();
511 assert_eq!(
512 waits(&mut dog, 2),
513 before,
514 "an outage after a recovery starts from the first wait, under a fresh ceiling"
515 );
516 }
517
518 #[test]
519 fn a_retriable_failure_backs_off_and_never_freezes_by_itself() {
520 let mut dog = Watchdog::new(Retries::UntilHalted, run(1));
521 let waits: Vec<u64> = (0..64u64)
522 .map(
523 |_| match dog.on_provider_failure(&provider_error(true), TimeMs::new(1_000)) {
524 Disposal::BackOff { until, .. } => until.value().saturating_sub(1_000),
525 other => {
526 panic!("only Halt stops a run that is waiting out a provider: {other:?}")
527 }
528 },
529 )
530 .collect();
531 assert!(
532 within_schedule(&waits),
533 "each failure in a row doubles the wait, up to a minute: {waits:?}"
534 );
535 assert!(
536 waits[9..]
537 .iter()
538 .all(|wait| (60_000..=90_000).contains(wait))
539 );
540 let payload = serde_json::to_value(
541 dog.fired_payload(&Disposal::BackOff {
542 until: TimeMs::new(1_000),
543 code: AxCode::Provider,
544 subject: "the provider said no".to_owned(),
545 })
546 .unwrap(),
547 )
548 .unwrap();
549 assert_eq!(payload["action"], "back_off");
550 assert_eq!(payload["until_ms"], 1_000);
551 assert_eq!(payload["provider_failures"], 64);
552 assert_eq!(
553 payload["subject"], "the provider said no",
554 "a line saying it backed off says what it backed off from"
555 );
556 }
557
558 fn waits_of_fresh_run() -> Vec<u64> {
559 waits(&mut Watchdog::new(Retries::UntilHalted, run(1)), 2)
560 }
561
562 use kernel::ProviderFailureKind;
563 use kernel::account_recovery::AccountRetries;
564 use kernel::model::{AccountRoster, RosterEntry};
565
566 fn label(name: &str) -> ServerLabel {
567 ServerLabel::parse(name).unwrap()
568 }
569
570 fn roster(accounts: &[(&str, bool)], current: &str, retries: AccountRetries) -> AccountRoster {
573 AccountRoster {
574 provider: "house".to_owned(),
575 accounts: accounts
576 .iter()
577 .map(|(name, usable)| RosterEntry {
578 account: label(name),
579 usable: *usable,
580 })
581 .collect(),
582 current: label(current),
583 retries,
584 }
585 }
586
587 fn rejected_key() -> AxError {
589 AxError::provider(
590 ProviderFailureKind::Refused { status: 401 },
591 "call the model",
592 "401 Unauthorized",
593 )
594 .account_unusable()
595 .with_recovery("file another key")
596 }
597
598 fn busy() -> AxError {
600 AxError::provider(
601 ProviderFailureKind::Refused { status: 429 },
602 "call the model",
603 "429 Too Many Requests",
604 )
605 .retriable()
606 .with_recovery("wait, then ask again")
607 }
608
609 fn lost() -> AxError {
611 AxError::provider(ProviderFailureKind::Cut, "call the model", "stream cut")
612 .effect_unknown()
613 .with_recovery("ask again")
614 }
615
616 fn switched_to(name: &str, failure: &AxError) -> Disposal {
617 Disposal::Switch {
618 to: label(name),
619 code: *failure.code(),
620 subject: failure.subject().to_owned(),
621 }
622 }
623
624 #[test]
628 fn a_round_opens_on_the_current_account_or_the_first_redeemable_one() {
629 let mut dog = Watchdog::new(Retries::UntilHalted, run(1));
630 let opened = [
631 roster(&[("a", true), ("b", true)], "b", AccountRetries::Two),
632 roster(&[("a", false), ("b", true)], "a", AccountRetries::Two),
633 ]
634 .map(|listed| {
635 let account = dog.open_round(Some(listed)).unwrap();
636 dog.on_provider_answered();
637 account
638 });
639 assert_eq!(opened, [Some(label("b")), Some(label("b"))]);
640 assert_eq!(
641 dog.open_round(Some(roster(&[("solo", true)], "solo", AccountRetries::Two)))
642 .unwrap(),
643 None,
644 "one account names nothing to select"
645 );
646 }
647
648 #[test]
653 fn several_accounts_switch_on_a_rejected_key_and_after_the_busy_budget() {
654 let mut dog = Watchdog::new(Retries::UntilHalted, run(1));
655 dog.open_round(Some(roster(
656 &[("a", true), ("b", true), ("c", true)],
657 "a",
658 AccountRetries::One,
659 )))
660 .unwrap();
661 let first = dog.on_provider_failure(&rejected_key(), TimeMs::new(0));
662 assert_eq!(first, switched_to("b", &rejected_key()));
663 let payload = serde_json::to_value(dog.fired_payload(&first).unwrap()).unwrap();
664 assert_eq!(
665 (&payload["action"], &payload["to"], &payload["code"]),
666 (
667 &serde_json::json!("switch"),
668 &serde_json::json!("b"),
669 &serde_json::json!("E_PROVIDER")
670 )
671 );
672 let waits = waits(&mut dog, 1);
673 assert!(
674 within_schedule(&waits),
675 "the new account starts from the first wait: {waits:?}"
676 );
677 assert_eq!(
678 dog.on_provider_failure(&busy(), TimeMs::new(0)),
679 switched_to("c", &busy()),
680 "one resend spent, the next account takes the request"
681 );
682 }
683
684 #[test]
688 fn a_round_with_no_account_left_freezes_as_exhausted() {
689 let mut dog = Watchdog::new(Retries::UntilHalted, run(1));
690 dog.open_round(Some(roster(
691 &[("a", true), ("b", true), ("c", false)],
692 "a",
693 AccountRetries::Two,
694 )))
695 .unwrap();
696 assert_eq!(
697 dog.on_provider_failure(&rejected_key(), TimeMs::new(0)),
698 switched_to("b", &rejected_key())
699 );
700 assert_eq!(
701 dog.on_provider_failure(&rejected_key(), TimeMs::new(0)),
702 Disposal::Freeze {
703 reason: FreezeReason::AccountsExhausted
704 }
705 );
706 let exhausted = dog.exhausted();
707 assert_eq!(
708 (*exhausted.code(), exhausted.subject()),
709 (
710 AxCode::ProviderAccountsExhausted,
711 "house: a E_PROVIDER Refused { status: 401 }; \
712 b E_PROVIDER Refused { status: 401 }; c not redeemable"
713 )
714 );
715 let none = Watchdog::new(Retries::UntilHalted, run(1)).open_round(Some(roster(
716 &[("a", false), ("b", false)],
717 "a",
718 AccountRetries::Two,
719 )));
720 assert_eq!(
721 none.map_err(|err| (*err.code(), err.subject().to_owned())),
722 Err((
723 AxCode::ProviderAccountsExhausted,
724 "house: a not redeemable; b not redeemable".to_owned()
725 ))
726 );
727 }
728
729 #[test]
733 fn an_unknown_effect_never_switches_the_account() {
734 let mut dog = Watchdog::new(Retries::UntilHalted, run(1));
735 dog.open_round(Some(roster(
736 &[("a", true), ("b", true)],
737 "a",
738 AccountRetries::One,
739 )))
740 .unwrap();
741 assert!(matches!(
742 dog.on_provider_failure(&lost(), TimeMs::new(0)),
743 Disposal::BackOff { .. }
744 ));
745 assert_eq!(
746 dog.on_provider_failure(&lost(), TimeMs::new(0)),
747 Disposal::Freeze {
748 reason: FreezeReason::ProviderRefused
749 }
750 );
751 }
752
753 #[test]
758 fn a_repair_is_admitted_within_the_account_budget_and_counted() {
759 let mut dog = Watchdog::new(Retries::UntilHalted, run(1));
760 assert!(dog.admits_repair(), "no round: the repair goes out");
761 dog.open_round(Some(roster(
762 &[("a", true), ("b", true)],
763 "a",
764 AccountRetries::One,
765 )))
766 .unwrap();
767 assert!(dog.admits_repair());
768 dog.repaired();
769 assert!(!dog.admits_repair(), "the account's one resend is spent");
770 assert_eq!(
771 dog.on_provider_failure(&busy(), TimeMs::new(0)),
772 switched_to("b", &busy())
773 );
774 let mut single = Watchdog::new(Retries::AtMost(0), run(1));
775 single.open_round(None).unwrap();
776 single.repaired();
777 assert!(single.admits_repair());
778 }
779
780 #[test]
781 fn a_provider_that_names_its_wait_is_not_asked_sooner() {
782 let mut dog = Watchdog::new(Retries::UntilHalted, run(1));
783 let told = |wait_ms: u64| {
784 AxError::failure(AxCode::Provider, "call the model", "answered 429")
785 .retriable_after(wait_ms)
786 .with_recovery("wait, then ask again")
787 };
788 let waits: Vec<u64> = [30_000, 10]
789 .map(
790 |wait_ms| match dog.on_provider_failure(&told(wait_ms), TimeMs::new(1_000)) {
791 Disposal::BackOff { until, .. } => until.value().saturating_sub(1_000),
792 other => {
793 panic!("a provider that says when to come back is asked again: {other:?}")
794 }
795 },
796 )
797 .to_vec();
798 let own = waits_of_fresh_run();
799 assert_eq!(
800 waits,
801 [30_000, own[1]],
802 "the longer of the provider's word and the schedule's own wait"
803 );
804 }
805}