1use core::fmt;
4
5use super::{
6 FingerprintRef, IdempotencyBinding, MonotonicDuration, MonotonicInstant, RetryPermit,
7 RetrySubject,
8};
9use crate::operation::{
10 BodyReplayability, OperationImpact, OperationMetadata, PreparedRequest, RequestSemantics,
11 RetryEligibility,
12};
13use crate::transport::{DeliveryPhase, StatusCode};
14
15#[derive(Clone, Copy, Debug, Eq, PartialEq)]
17pub enum MaxAttemptsError {
18 Zero,
20}
21
22impl_static_error!(MaxAttemptsError, Self::Zero => "retry maximum attempts must be nonzero");
23
24#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
26pub struct MaxAttempts(u16);
27
28impl MaxAttempts {
29 pub const fn new(value: u16) -> Result<Self, MaxAttemptsError> {
31 if value == 0 {
32 return Err(MaxAttemptsError::Zero);
33 }
34 Ok(Self(value))
35 }
36
37 #[must_use]
39 pub const fn get(self) -> u16 {
40 self.0
41 }
42}
43
44#[derive(Clone, Copy, Debug, Eq, PartialEq)]
46pub struct RetryPolicy {
47 max_attempts: MaxAttempts,
48 max_cumulative_delay: MonotonicDuration,
49 max_elapsed: MonotonicDuration,
50}
51
52impl RetryPolicy {
53 #[must_use]
55 pub const fn new(
56 max_attempts: MaxAttempts,
57 max_cumulative_delay: MonotonicDuration,
58 max_elapsed: MonotonicDuration,
59 ) -> Self {
60 Self {
61 max_attempts,
62 max_cumulative_delay,
63 max_elapsed,
64 }
65 }
66
67 #[must_use]
69 pub const fn max_attempts(self) -> MaxAttempts {
70 self.max_attempts
71 }
72
73 #[must_use]
75 pub const fn max_cumulative_delay(self) -> MonotonicDuration {
76 self.max_cumulative_delay
77 }
78
79 #[must_use]
81 pub const fn max_elapsed(self) -> MonotonicDuration {
82 self.max_elapsed
83 }
84}
85
86#[derive(Clone, Copy, Debug, Eq, PartialEq)]
88pub enum RetryEvent {
89 Transport(DeliveryPhase),
91 Response(StatusCode),
93}
94
95#[derive(Clone, Copy, Debug, Eq, PartialEq)]
97pub enum RetryStopReason {
98 IneligibleOperation,
100 NonReplayableBody,
102 MutationRequiresIntent,
104 NonTransientResponse,
106 AttemptsExhausted,
108 CumulativeDelayExhausted,
110 ElapsedBudgetExhausted,
112}
113
114#[derive(Debug)]
116pub enum RetryDecision<'controller, 'request, 'subject> {
117 Retry(RetryPermit<'controller, 'request, 'subject>),
119 Stop(RetryStopReason),
121}
122
123#[derive(Clone, Copy, Debug, Eq, PartialEq)]
125pub enum RetryPolicyError {
126 MissingMutationIntent,
128 FingerprintMismatch,
130 ReplayPolicyMismatch,
132 IdempotencyFingerprintMismatch,
134 MonotonicRollback,
136 CumulativeDelayOverflow,
138}
139
140impl_static_error!(RetryPolicyError,
141 Self::MissingMutationIntent => "retrying a mutation requires an idempotency intent",
142 Self::FingerprintMismatch => "retry request fingerprint does not match the initial request",
143 Self::ReplayPolicyMismatch => "retry request policy does not match the initial request",
144 Self::IdempotencyFingerprintMismatch => "idempotency binding does not match the initial request",
145 Self::MonotonicRollback => "retry monotonic observation moved backward",
146 Self::CumulativeDelayOverflow => "retry cumulative delay overflowed",
147);
148
149pub struct RetryController<'request, 'binding> {
159 prepared: PreparedRequest<'request>,
160 metadata: OperationMetadata,
161 body: BodyReplayability,
162 fingerprint: FingerprintRef<'binding>,
163 idempotency: Option<IdempotencyBinding<'binding>>,
164 policy: RetryPolicy,
165 started: MonotonicInstant,
166 last_observed: MonotonicInstant,
167 attempts: u16,
168 cumulative_delay: u64,
169}
170
171impl<'request, 'binding> RetryController<'request, 'binding> {
172 pub fn new(
177 subject: RetrySubject<'request, 'binding>,
178 idempotency: Option<IdempotencyBinding<'binding>>,
179 policy: RetryPolicy,
180 started: MonotonicInstant,
181 ) -> Result<Self, RetryPolicyError> {
182 Self::from_parts(
183 subject.prepared(),
184 subject.fingerprint(),
185 idempotency,
186 policy,
187 started,
188 )
189 }
190
191 fn from_parts(
192 prepared: &PreparedRequest<'request>,
193 fingerprint: FingerprintRef<'binding>,
194 idempotency: Option<IdempotencyBinding<'binding>>,
195 policy: RetryPolicy,
196 started: MonotonicInstant,
197 ) -> Result<Self, RetryPolicyError> {
198 let metadata = prepared.metadata();
199 let body = prepared.body_replayability();
200 if policy.max_attempts().get() > 1
201 && metadata.impact() != OperationImpact::ReadOnly
202 && idempotency.is_none()
203 {
204 return Err(RetryPolicyError::MissingMutationIntent);
205 }
206 if idempotency
207 .as_ref()
208 .is_some_and(|binding| !binding.matches(fingerprint))
209 {
210 return Err(RetryPolicyError::IdempotencyFingerprintMismatch);
211 }
212 Ok(Self {
213 prepared: *prepared,
214 metadata,
215 body,
216 fingerprint,
217 idempotency,
218 policy,
219 started,
220 last_observed: started,
221 attempts: 1,
222 cumulative_delay: 0,
223 })
224 }
225
226 #[must_use]
228 pub const fn attempts(&self) -> u16 {
229 self.attempts
230 }
231
232 #[must_use]
234 pub const fn cumulative_delay(&self) -> MonotonicDuration {
235 MonotonicDuration::new(self.cumulative_delay)
236 }
237
238 pub fn decide_retry<'controller, 'replay, 'subject>(
243 &'controller mut self,
244 event: RetryEvent,
245 replay: RetrySubject<'replay, 'subject>,
246 delay: MonotonicDuration,
247 now: MonotonicInstant,
248 ) -> Result<RetryDecision<'controller, 'replay, 'subject>, RetryPolicyError> {
249 if !self.fingerprint.matches(replay.fingerprint()) {
250 return Err(RetryPolicyError::FingerprintMismatch);
251 }
252 if !self.prepared.has_same_retry_policy(replay.prepared()) {
253 return Err(RetryPolicyError::ReplayPolicyMismatch);
254 }
255 let elapsed = observe_monotonic(&mut self.last_observed, self.started, now)?;
256
257 if self.metadata.retry_eligibility() != RetryEligibility::ExplicitPolicy {
258 return Ok(RetryDecision::Stop(RetryStopReason::IneligibleOperation));
259 }
260 if self.body != BodyReplayability::Replayable {
261 return Ok(RetryDecision::Stop(RetryStopReason::NonReplayableBody));
262 }
263 if self.metadata.impact() != OperationImpact::ReadOnly && self.idempotency.is_none() {
264 return Ok(RetryDecision::Stop(RetryStopReason::MutationRequiresIntent));
265 }
266 if let RetryEvent::Response(status) = event
267 && status != StatusCode::TOO_MANY_REQUESTS
268 && !(500..=599).contains(&status.get())
269 {
270 return Ok(RetryDecision::Stop(RetryStopReason::NonTransientResponse));
271 }
272 if matches!(
273 event,
274 RetryEvent::Transport(DeliveryPhase::PossiblySent | DeliveryPhase::ResponseStarted)
275 ) && self.metadata.impact() != OperationImpact::ReadOnly
276 && self.metadata.semantics() != RequestSemantics::Idempotent
277 {
278 return Ok(RetryDecision::Stop(RetryStopReason::IneligibleOperation));
279 }
280 if self.attempts >= self.policy.max_attempts().get() {
281 return Ok(RetryDecision::Stop(RetryStopReason::AttemptsExhausted));
282 }
283 let projected_elapsed = elapsed.get().checked_add(delay.get());
284 if projected_elapsed.is_none_or(|value| value > self.policy.max_elapsed().get()) {
285 return Ok(RetryDecision::Stop(RetryStopReason::ElapsedBudgetExhausted));
286 }
287 let Some(not_before) = now.checked_add(delay) else {
288 return Ok(RetryDecision::Stop(RetryStopReason::ElapsedBudgetExhausted));
289 };
290 let cumulative = self
291 .cumulative_delay
292 .checked_add(delay.get())
293 .ok_or(RetryPolicyError::CumulativeDelayOverflow)?;
294 if cumulative > self.policy.max_cumulative_delay().get() {
295 return Ok(RetryDecision::Stop(
296 RetryStopReason::CumulativeDelayExhausted,
297 ));
298 }
299 self.cumulative_delay = cumulative;
300 self.attempts = self
301 .attempts
302 .checked_add(1)
303 .ok_or(RetryPolicyError::CumulativeDelayOverflow)?;
304 Ok(RetryDecision::Retry(RetryPermit::new(
305 &mut self.last_observed,
306 replay.prepared(),
307 self.attempts,
308 delay,
309 not_before,
310 self.started,
311 self.policy.max_elapsed(),
312 )))
313 }
314}
315
316pub(crate) fn observe_monotonic(
317 last_observed: &mut MonotonicInstant,
318 started: MonotonicInstant,
319 now: MonotonicInstant,
320) -> Result<MonotonicDuration, RetryPolicyError> {
321 if now < *last_observed {
322 return Err(RetryPolicyError::MonotonicRollback);
323 }
324 let elapsed = now
325 .checked_duration_since(started)
326 .ok_or(RetryPolicyError::MonotonicRollback)?;
327 *last_observed = now;
328 Ok(elapsed)
329}
330
331impl fmt::Debug for RetryController<'_, '_> {
332 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
333 formatter
334 .debug_struct("RetryController")
335 .field("metadata", &self.metadata)
336 .field("body", &self.body)
337 .field("fingerprint", &"[redacted]")
338 .field(
339 "intent_len",
340 &self
341 .idempotency
342 .as_ref()
343 .map(IdempotencyBinding::intent_len),
344 )
345 .field("policy", &self.policy)
346 .field("attempts", &self.attempts)
347 .field("cumulative_delay", &self.cumulative_delay)
348 .finish()
349 }
350}