harmos_domain/model/operations/
policy.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::time::Duration;
18// Third Party Imports
19use serde::{Deserialize, Serialize};
20// Local Imports
21
22/* === Constants === */
23
24/* === Macros === */
25
26/* === Type Aliases === */
27
28/* === Structs === */
29
30#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
31#[serde(deny_unknown_fields)]
32pub struct OperationPolicy {
33    pub timeout: Duration,
34    pub attempt_timeout: Duration,
35    pub retries: RetryPolicy,
36}
37
38/* === Enums === */
39
40#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
41#[serde(rename_all = "snake_case", deny_unknown_fields)]
42pub enum RetryPolicy {
43    Never,
44    OnTransient { max_retries: usize, backoff: Backoff },
45}
46
47#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
48#[serde(rename_all = "snake_case", deny_unknown_fields)]
49pub enum Backoff {
50    Fixed { delay: Duration, jitter: Jitter },
51    Exponential { initial: Duration, maximum: Duration, jitter: Jitter },
52}
53
54#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
55#[serde(rename_all = "snake_case", deny_unknown_fields)]
56pub enum Jitter {
57    None,
58    Deterministic,
59}
60
61#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
62#[serde(rename_all = "snake_case")]
63pub enum FailureKind {
64    Permanent,
65    Transient,
66    Ambiguous,
67    Unauthorized,
68    Invalid,
69    Cancelled,
70}
71
72#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
73#[serde(rename_all = "snake_case", deny_unknown_fields)]
74pub enum ReplaySafety {
75    Unknown,
76    Idempotent,
77    Deduplicated,
78}
79
80#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
81pub enum PolicyError {
82    #[error("operation deadlines must be positive (overall {timeout:?}, attempt {attempt_timeout:?})")]
83    Deadline { timeout: Duration, attempt_timeout: Duration },
84
85    #[error("a retry policy needs a positive retry count; use Never for zero")]
86    EmptyRetries,
87
88    #[error("a restart policy needs positive credit and sustained health; use Never for zero")]
89    EmptyRestarts,
90
91    #[error("retry delays must be positive and ordered (initial {initial:?}, maximum {maximum:?})")]
92    Backoff { initial: Duration, maximum: Duration },
93}
94
95/* === Traits === */
96
97/* === Implementations === */
98
99impl OperationPolicy {
100    pub fn validate(self) -> Result<(), PolicyError> {
101        if self.timeout.is_zero() || self.attempt_timeout.is_zero() {
102            return Err(PolicyError::Deadline {
103                timeout: self.timeout,
104                attempt_timeout: self.attempt_timeout,
105            });
106        }
107
108        self.retries.validate()
109    }
110}
111
112impl RetryPolicy {
113    pub fn validate(self) -> Result<(), PolicyError> {
114        match self {
115            Self::Never => Ok(()),
116            Self::OnTransient { max_retries: 0, .. } => Err(PolicyError::EmptyRetries),
117            Self::OnTransient { backoff, .. } => backoff.validate(),
118        }
119    }
120
121    pub fn retry_after(self, failed_attempt: usize, failure: FailureKind, replay: ReplaySafety, entropy: u64) -> Option<Duration> {
122        let Self::OnTransient { max_retries, backoff } = self else {
123            return None;
124        };
125        if self.validate().is_err() || failed_attempt >= max_retries {
126            return None;
127        }
128        if !matches!(failure, FailureKind::Transient) && !(failure == FailureKind::Ambiguous && replay != ReplaySafety::Unknown) {
129            return None;
130        }
131
132        backoff.delay(failed_attempt, entropy).ok()
133    }
134}
135
136impl Backoff {
137    pub fn validate(self) -> Result<(), PolicyError> {
138        let (initial, maximum) = match self {
139            Self::Fixed { delay, .. } => (delay, delay),
140            Self::Exponential { initial, maximum, .. } => (initial, maximum),
141        };
142        if initial.is_zero() || maximum < initial {
143            return Err(PolicyError::Backoff { initial, maximum });
144        }
145
146        Ok(())
147    }
148
149    pub fn delay(self, failed_attempt: usize, entropy: u64) -> Result<Duration, PolicyError> {
150        self.validate()?;
151
152        let (nominal, maximum, jitter) = match self {
153            Self::Fixed { delay, jitter } => (delay, delay.saturating_add(from_nanos(delay.as_nanos() / 4)), jitter),
154            Self::Exponential { initial, maximum, jitter } => {
155                let shift = u32::try_from(failed_attempt).unwrap_or(u32::MAX);
156                let nanos = if shift >= 128 || initial.as_nanos() > (u128::MAX >> shift) {
157                    maximum.as_nanos()
158                } else {
159                    (initial.as_nanos() << shift).min(maximum.as_nanos())
160                };
161
162                (from_nanos(nanos), maximum, jitter)
163            }
164        };
165        if jitter == Jitter::None {
166            return Ok(nominal.min(maximum));
167        }
168        let mixed = u128::from(entropy)
169            .wrapping_mul(0x9e37_79b9_7f4a_7c15)
170            .wrapping_add((failed_attempt as u128).wrapping_mul(17).wrapping_add(11));
171        let nanos = mixed % (nominal.as_nanos() / 4 + 1);
172
173        Ok(nominal.saturating_add(from_nanos(nanos)).min(maximum))
174    }
175}
176
177/* === Public Functions === */
178
179/* === Private Functions === */
180
181// Every caller bounds nanos by a representable Duration before this conversion.
182fn from_nanos(nanos: u128) -> Duration {
183    Duration::new((nanos / 1_000_000_000) as u64, (nanos % 1_000_000_000) as u32)
184}
185
186/* === Tests === */
harmos_domain/model/operations/policy.rs

harmos_domain/model/operations/
policy.rs
