cloud_sdk/action_polling/
driver.rs

1use core::fmt;
2
3use crate::rate_limit::RateLimit;
4use crate::retry::{MonotonicDuration, MonotonicInstant};
5
6use super::progress::{ProgressError, ProgressTracker};
7use super::{
8    PollBackoff, PollContext, PollControl, PollRequestStep, ProgressObservation, ProgressPolicy,
9    ProviderTimeObservation,
10};
11
12/// One provider action observation.
13#[derive(Eq, PartialEq)]
14pub enum ActionUpdate<E> {
15    /// The action remains in progress.
16    Running,
17    /// The action completed successfully.
18    Success,
19    /// The action completed with a provider failure.
20    Failed(E),
21}
22
23impl<E> fmt::Debug for ActionUpdate<E> {
24    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
25        match self {
26            Self::Running => formatter.write_str("Running"),
27            Self::Success => formatter.write_str("Success"),
28            Self::Failed(_) => formatter.write_str("Failed([redacted])"),
29        }
30    }
31}
32
33/// Invalid hard action-poll limits.
34#[derive(Clone, Copy, Debug, Eq, PartialEq)]
35pub enum ActionPollLimitsError {
36    /// Every hard limit must be nonzero.
37    Zero,
38    /// One delay may not exceed the cumulative delay budget.
39    DelayExceedsCumulative,
40    /// One delay may not meet or exceed the elapsed budget.
41    DelayExceedsElapsed,
42}
43
44impl_static_error!(ActionPollLimitsError,
45    Self::Zero => "action polling limits must be nonzero",
46    Self::DelayExceedsCumulative => "maximum poll delay exceeds the cumulative budget",
47    Self::DelayExceedsElapsed => "maximum poll delay reaches the elapsed budget",
48);
49
50/// Unconditional request, delay, and elapsed limits.
51#[derive(Clone, Copy, Debug, Eq, PartialEq)]
52pub struct ActionPollLimits {
53    max_observations: u32,
54    max_delay: MonotonicDuration,
55    max_cumulative_delay: MonotonicDuration,
56    max_elapsed: MonotonicDuration,
57}
58
59impl ActionPollLimits {
60    /// Creates complete hard limits selected before the first request.
61    pub const fn new(
62        max_observations: u32,
63        max_delay: MonotonicDuration,
64        max_cumulative_delay: MonotonicDuration,
65        max_elapsed: MonotonicDuration,
66    ) -> Result<Self, ActionPollLimitsError> {
67        if max_observations == 0
68            || max_delay.get() == 0
69            || max_cumulative_delay.get() == 0
70            || max_elapsed.get() == 0
71        {
72            return Err(ActionPollLimitsError::Zero);
73        }
74        if max_delay.get() > max_cumulative_delay.get() {
75            return Err(ActionPollLimitsError::DelayExceedsCumulative);
76        }
77        if max_delay.get() >= max_elapsed.get() {
78            return Err(ActionPollLimitsError::DelayExceedsElapsed);
79        }
80        Ok(Self {
81            max_observations,
82            max_delay,
83            max_cumulative_delay,
84            max_elapsed,
85        })
86    }
87
88    /// Returns the maximum accepted provider observations.
89    #[must_use]
90    pub const fn max_observations(self) -> u32 {
91        self.max_observations
92    }
93
94    /// Returns the maximum delay admitted from any backoff policy.
95    #[must_use]
96    pub const fn max_delay(self) -> MonotonicDuration {
97        self.max_delay
98    }
99
100    /// Returns the cumulative requested-delay budget.
101    #[must_use]
102    pub const fn max_cumulative_delay(self) -> MonotonicDuration {
103        self.max_cumulative_delay
104    }
105
106    /// Returns the monotonic elapsed-time budget.
107    #[must_use]
108    pub const fn max_elapsed(self) -> MonotonicDuration {
109        self.max_elapsed
110    }
111}
112
113/// Structural action workflow failure.
114#[derive(Clone, Copy, Debug, Eq, PartialEq)]
115pub enum ActionPollError {
116    /// Another request cannot start before its response is observed.
117    ResponsePending,
118    /// A response was supplied without one admitted request.
119    UnexpectedObservation,
120    /// The workflow already reached a terminal state.
121    Terminal,
122    /// Caller monotonic time moved backwards.
123    MonotonicRollback,
124    /// Monotonic timestamp arithmetic overflowed.
125    TimeOverflow,
126    /// The unconditional observation limit was reached while still running.
127    ObservationLimitExceeded,
128    /// Provider progress exceeded 100.
129    InvalidProgress,
130    /// Provider progress regressed without an explicit reset.
131    ProgressRegressed,
132    /// Provider progress reset was not admitted.
133    ProgressResetForbidden,
134    /// Provider progress exhausted its reset budget.
135    ProgressResetLimitExceeded,
136    /// Backoff requested a zero delay.
137    ZeroDelay,
138    /// Backoff requested more than the per-delay bound.
139    DelayLimitExceeded,
140    /// Backoff exhausted the cumulative delay budget.
141    CumulativeDelayExceeded,
142    /// Backoff would reach or exceed the elapsed budget.
143    ElapsedBudgetExceeded,
144}
145
146impl_static_error!(ActionPollError,
147    Self::ResponsePending => "an action response is still pending",
148    Self::UnexpectedObservation => "action observation has no admitted request",
149    Self::Terminal => "action polling already reached a terminal state",
150    Self::MonotonicRollback => "action polling monotonic time moved backwards",
151    Self::TimeOverflow => "action polling monotonic time overflowed",
152    Self::ObservationLimitExceeded => "action polling observation limit was exceeded",
153    Self::InvalidProgress => "action progress exceeds 100",
154    Self::ProgressRegressed => "action progress moved backwards",
155    Self::ProgressResetForbidden => "action progress reset is forbidden",
156    Self::ProgressResetLimitExceeded => "action progress reset limit was exceeded",
157    Self::ZeroDelay => "action poll backoff requested a zero delay",
158    Self::DelayLimitExceeded => "action poll backoff exceeded the delay limit",
159    Self::CumulativeDelayExceeded => "action polling cumulative delay was exceeded",
160    Self::ElapsedBudgetExceeded => "action polling elapsed budget would be exceeded",
161);
162
163/// Observation failure with payload-redacted backoff diagnostics.
164#[derive(Eq, PartialEq)]
165pub enum ActionObserveError<E> {
166    /// Driver validation failed.
167    Driver(ActionPollError),
168    /// Caller-owned backoff policy failed.
169    Backoff(E),
170}
171
172impl<E> fmt::Debug for ActionObserveError<E> {
173    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
174        match self {
175            Self::Driver(error) => formatter.debug_tuple("Driver").field(error).finish(),
176            Self::Backoff(_) => formatter.write_str("Backoff([redacted])"),
177        }
178    }
179}
180
181impl<E> fmt::Display for ActionObserveError<E> {
182    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
183        match self {
184            Self::Driver(error) => fmt::Display::fmt(error, formatter),
185            Self::Backoff(_) => formatter.write_str("action poll backoff policy failed"),
186        }
187    }
188}
189
190impl<E> core::error::Error for ActionObserveError<E> {}
191
192/// Step returned after one accepted provider observation.
193#[derive(Eq, PartialEq)]
194pub enum ActionPollStep<E> {
195    /// Wait before asking the driver to admit another request.
196    Delay(MonotonicDuration),
197    /// The provider action completed successfully.
198    Complete,
199    /// The provider action completed with an error.
200    Failed(E),
201}
202
203impl<E> fmt::Debug for ActionPollStep<E> {
204    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
205        match self {
206            Self::Delay(delay) => formatter.debug_tuple("Delay").field(delay).finish(),
207            Self::Complete => formatter.write_str("Complete"),
208            Self::Failed(_) => formatter.write_str("Failed([redacted])"),
209        }
210    }
211}
212
213#[derive(Clone, Copy, Debug, Eq, PartialEq)]
214enum PollPhase {
215    Ready,
216    AwaitingResponse,
217    Delayed(MonotonicInstant),
218    Terminal,
219}
220
221/// Single-owner bounded action workflow driver.
222///
223/// ```compile_fail
224/// use cloud_sdk::action_polling::ActionPoller;
225/// fn duplicate(driver: ActionPoller) {
226///     let _copy = driver.clone();
227/// }
228/// ```
229pub struct ActionPoller {
230    limits: ActionPollLimits,
231    progress: ProgressTracker,
232    started: MonotonicInstant,
233    last_observed: MonotonicInstant,
234    observations: u32,
235    cumulative_delay: u64,
236    phase: PollPhase,
237}
238
239impl ActionPoller {
240    /// Creates a workflow without reading a clock or issuing a request.
241    #[must_use]
242    pub const fn new(
243        limits: ActionPollLimits,
244        progress_policy: ProgressPolicy,
245        started: MonotonicInstant,
246    ) -> Self {
247        Self {
248            limits,
249            progress: ProgressTracker::new(progress_policy),
250            started,
251            last_observed: started,
252            observations: 0,
253            cumulative_delay: 0,
254            phase: PollPhase::Ready,
255        }
256    }
257
258    /// Returns accepted provider observations.
259    #[must_use]
260    pub const fn observations(&self) -> u32 {
261        self.observations
262    }
263
264    /// Returns cumulative delay selected by backoff.
265    #[must_use]
266    pub const fn cumulative_delay(&self) -> MonotonicDuration {
267        MonotonicDuration::new(self.cumulative_delay)
268    }
269
270    /// Reports whether no further request can be admitted.
271    #[must_use]
272    pub const fn is_terminal(&self) -> bool {
273        matches!(self.phase, PollPhase::Terminal)
274    }
275
276    /// Admits, delays, cancels, or times out the next request.
277    pub fn next_request(
278        &mut self,
279        control: PollControl,
280        now: MonotonicInstant,
281    ) -> Result<PollRequestStep, ActionPollError> {
282        if self.is_terminal() {
283            return Err(ActionPollError::Terminal);
284        }
285        if now < self.last_observed {
286            self.phase = PollPhase::Terminal;
287            return Err(ActionPollError::MonotonicRollback);
288        }
289        self.last_observed = now;
290        if self.elapsed_exhausted(now)? {
291            self.phase = PollPhase::Terminal;
292            return Ok(PollRequestStep::TimedOut);
293        }
294        if control == PollControl::Cancel {
295            self.phase = PollPhase::Terminal;
296            return Ok(PollRequestStep::Cancelled);
297        }
298        match self.phase {
299            PollPhase::AwaitingResponse => Err(ActionPollError::ResponsePending),
300            PollPhase::Delayed(not_before) if now < not_before => {
301                let remaining = not_before
302                    .checked_duration_since(now)
303                    .ok_or(ActionPollError::MonotonicRollback)?;
304                Ok(PollRequestStep::Delay(remaining))
305            }
306            PollPhase::Ready | PollPhase::Delayed(_) => {
307                self.phase = PollPhase::AwaitingResponse;
308                Ok(PollRequestStep::Request)
309            }
310            PollPhase::Terminal => Err(ActionPollError::Terminal),
311        }
312    }
313
314    /// Accepts exactly one response for the last admitted request.
315    pub fn observe<E, B>(
316        &mut self,
317        update: ActionUpdate<E>,
318        progress: ProgressObservation,
319        rate_limit: Option<RateLimit>,
320        provider_time: ProviderTimeObservation,
321        now: MonotonicInstant,
322        backoff: &mut B,
323    ) -> Result<ActionPollStep<E>, ActionObserveError<B::Error>>
324    where
325        B: PollBackoff,
326    {
327        if self.phase != PollPhase::AwaitingResponse {
328            return Err(ActionObserveError::Driver(if self.is_terminal() {
329                ActionPollError::Terminal
330            } else {
331                ActionPollError::UnexpectedObservation
332            }));
333        }
334        self.validate_observation_time(now)
335            .map_err(ActionObserveError::Driver)?;
336        let observations = self.observations.checked_add(1).ok_or_else(|| {
337            self.phase = PollPhase::Terminal;
338            ActionObserveError::Driver(ActionPollError::ObservationLimitExceeded)
339        })?;
340        self.observations = observations;
341        match update {
342            ActionUpdate::Success => {
343                self.phase = PollPhase::Terminal;
344                Ok(ActionPollStep::Complete)
345            }
346            ActionUpdate::Failed(error) => {
347                self.phase = PollPhase::Terminal;
348                Ok(ActionPollStep::Failed(error))
349            }
350            ActionUpdate::Running => self.observe_running(
351                observations,
352                progress,
353                rate_limit,
354                provider_time,
355                now,
356                backoff,
357            ),
358        }
359    }
360
361    fn observe_running<E, B>(
362        &mut self,
363        observations: u32,
364        progress: ProgressObservation,
365        rate_limit: Option<RateLimit>,
366        provider_time: ProviderTimeObservation,
367        now: MonotonicInstant,
368        backoff: &mut B,
369    ) -> Result<ActionPollStep<E>, ActionObserveError<B::Error>>
370    where
371        B: PollBackoff,
372    {
373        if observations >= self.limits.max_observations {
374            self.phase = PollPhase::Terminal;
375            return Err(ActionObserveError::Driver(
376                ActionPollError::ObservationLimitExceeded,
377            ));
378        }
379        let progress_change = self.progress.observe(progress).map_err(|error| {
380            self.phase = PollPhase::Terminal;
381            ActionObserveError::Driver(map_progress_error(error))
382        })?;
383        let context = PollContext {
384            observation: observations,
385            progress,
386            progress_change,
387            rate_limit,
388            provider_time,
389        };
390        let delay = backoff.delay(context).map_err(|error| {
391            self.phase = PollPhase::Terminal;
392            ActionObserveError::Backoff(error)
393        })?;
394        self.schedule(delay, now)
395            .map_err(ActionObserveError::Driver)?;
396        Ok(ActionPollStep::Delay(delay))
397    }
398
399    fn schedule(
400        &mut self,
401        delay: MonotonicDuration,
402        now: MonotonicInstant,
403    ) -> Result<(), ActionPollError> {
404        if delay.get() == 0 {
405            self.phase = PollPhase::Terminal;
406            return Err(ActionPollError::ZeroDelay);
407        }
408        if delay > self.limits.max_delay {
409            self.phase = PollPhase::Terminal;
410            return Err(ActionPollError::DelayLimitExceeded);
411        }
412        let Some(cumulative) = self.cumulative_delay.checked_add(delay.get()) else {
413            self.phase = PollPhase::Terminal;
414            return Err(ActionPollError::CumulativeDelayExceeded);
415        };
416        if cumulative > self.limits.max_cumulative_delay.get() {
417            self.phase = PollPhase::Terminal;
418            return Err(ActionPollError::CumulativeDelayExceeded);
419        }
420        let Some(not_before) = now.checked_add(delay) else {
421            self.phase = PollPhase::Terminal;
422            return Err(ActionPollError::TimeOverflow);
423        };
424        let Some(elapsed) = not_before.checked_duration_since(self.started) else {
425            self.phase = PollPhase::Terminal;
426            return Err(ActionPollError::MonotonicRollback);
427        };
428        if elapsed >= self.limits.max_elapsed {
429            self.phase = PollPhase::Terminal;
430            return Err(ActionPollError::ElapsedBudgetExceeded);
431        }
432        self.cumulative_delay = cumulative;
433        self.phase = PollPhase::Delayed(not_before);
434        Ok(())
435    }
436
437    fn validate_observation_time(&mut self, now: MonotonicInstant) -> Result<(), ActionPollError> {
438        if now < self.last_observed {
439            self.phase = PollPhase::Terminal;
440            return Err(ActionPollError::MonotonicRollback);
441        }
442        self.last_observed = now;
443        if self.elapsed_exhausted(now)? {
444            self.phase = PollPhase::Terminal;
445            return Err(ActionPollError::ElapsedBudgetExceeded);
446        }
447        Ok(())
448    }
449
450    fn elapsed_exhausted(&self, now: MonotonicInstant) -> Result<bool, ActionPollError> {
451        let elapsed = now
452            .checked_duration_since(self.started)
453            .ok_or(ActionPollError::MonotonicRollback)?;
454        Ok(elapsed >= self.limits.max_elapsed)
455    }
456}
457
458impl fmt::Debug for ActionPoller {
459    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
460        formatter
461            .debug_struct("ActionPoller")
462            .field("limits", &self.limits)
463            .field("observations", &self.observations)
464            .field("cumulative_delay", &self.cumulative_delay())
465            .field("phase", &self.phase)
466            .finish_non_exhaustive()
467    }
468}
469
470fn map_progress_error(error: ProgressError) -> ActionPollError {
471    match error {
472        ProgressError::Invalid => ActionPollError::InvalidProgress,
473        ProgressError::Regressed => ActionPollError::ProgressRegressed,
474        ProgressError::ResetForbidden => ActionPollError::ProgressResetForbidden,
475        ProgressError::ResetLimit => ActionPollError::ProgressResetLimitExceeded,
476    }
477}
cloud_sdk/action_polling/driver.rs

cloud_sdk/action_polling/
driver.rs
