graphrefly/
resilience.rs

1//! Passive resilience policies for environment adapters (D130).
2//!
3//! These types do not schedule work and do not own graph state. Adapter-native
4//! retry/reconnect code can use them while surfacing attempts/status/errors as
5//! graph-visible data.
6
7use crate::ctx::DepTerminal;
8use crate::graph::{Graph, GraphNodeOpts};
9use crate::node::Node;
10use crate::time::timeout;
11
12/// Delay strategy for retry/reconnect attempts.
13#[derive(Debug, Clone, Default, PartialEq, Eq)]
14pub enum BackoffPolicy {
15    #[default]
16    /// `None` variant.
17    None,
18    /// `Constant` variant.
19    Constant {
20        /// `delay_ms` field for delay ms.
21        delay_ms: u64,
22    },
23    /// `Linear` variant.
24    Linear {
25        /// `initial_ms` field for initial ms.
26        initial_ms: u64,
27        /// `step_ms` field for step ms.
28        step_ms: u64,
29        /// `max_ms` field for max ms.
30        max_ms: Option<u64>,
31    },
32    /// `Exponential` variant.
33    Exponential {
34        /// `initial_ms` field for initial ms.
35        initial_ms: u64,
36        /// `factor` field for factor.
37        factor: u32,
38        /// `max_ms` field for max ms.
39        max_ms: Option<u64>,
40    },
41    /// `Fibonacci` variant.
42    Fibonacci {
43        /// `unit_ms` field for unit ms.
44        unit_ms: u64,
45        /// `max_ms` field for max ms.
46        max_ms: Option<u64>,
47    },
48}
49
50impl BackoffPolicy {
51    /// Updates or reads `delay_ms`.
52    pub fn delay_ms(&self, attempt: u32) -> u64 {
53        match self {
54            BackoffPolicy::None => 0,
55            BackoffPolicy::Constant { delay_ms } => *delay_ms,
56            BackoffPolicy::Linear {
57                initial_ms,
58                step_ms,
59                max_ms,
60            } => cap(
61                initial_ms.saturating_add(step_ms.saturating_mul(attempt.saturating_sub(1) as u64)),
62                *max_ms,
63            ),
64            BackoffPolicy::Exponential {
65                initial_ms,
66                factor,
67                max_ms,
68            } => {
69                let multiplier = factor.saturating_pow(attempt.saturating_sub(1));
70                cap(initial_ms.saturating_mul(multiplier as u64), *max_ms)
71            }
72            BackoffPolicy::Fibonacci { unit_ms, max_ms } => {
73                cap(unit_ms.saturating_mul(fibonacci(attempt) as u64), *max_ms)
74            }
75        }
76    }
77}
78
79/// Passive retry policy shared by environment adapters.
80#[derive(Debug, Clone, PartialEq, Eq)]
81pub struct RetryPolicy {
82    /// `max_attempts` field for max attempts.
83    pub max_attempts: u32,
84    /// `backoff` field for backoff.
85    pub backoff: BackoffPolicy,
86}
87
88impl RetryPolicy {
89    /// Creates or computes `new`.
90    pub fn new(max_attempts: u32, backoff: BackoffPolicy) -> Self {
91        assert!(max_attempts > 0, "RetryPolicy: max_attempts must be > 0");
92        Self {
93            max_attempts,
94            backoff,
95        }
96    }
97
98    /// Updates or reads `should_retry`.
99    pub fn should_retry(&self, failed_attempt: u32) -> bool {
100        failed_attempt < self.max_attempts
101    }
102
103    /// Updates or reads `next_delay_ms`.
104    pub fn next_delay_ms(&self, next_attempt: u32) -> Option<u64> {
105        if next_attempt == 0 || next_attempt > self.max_attempts {
106            return None;
107        }
108        Some(self.backoff.delay_ms(next_attempt))
109    }
110}
111
112impl Default for RetryPolicy {
113    fn default() -> Self {
114        Self {
115            max_attempts: 1,
116            backoff: BackoffPolicy::None,
117        }
118    }
119}
120
121/// Graph-visible retry/reconnect status payload shape.
122#[derive(Debug, Clone, PartialEq, Eq)]
123pub struct RetryStatus {
124    /// `attempt` field for attempt.
125    pub attempt: u32,
126    /// `max_attempts` field for max attempts.
127    pub max_attempts: u32,
128    /// `delay_ms` field for delay ms.
129    pub delay_ms: Option<u64>,
130    /// `state` field for state.
131    pub state: RetryState,
132}
133
134#[derive(Debug, Clone, PartialEq, Eq)]
135/// `RetryState` variants.
136pub enum RetryState {
137    /// `Idle` variant.
138    Idle,
139    /// `Running` variant.
140    Running,
141    /// `Waiting` variant.
142    Waiting,
143    /// `Succeeded` variant.
144    Succeeded,
145    /// `Failed` variant.
146    Failed,
147    /// `Exhausted` variant.
148    Exhausted,
149}
150
151#[derive(Debug, Clone, PartialEq, Eq)]
152/// `RetryEvent` variants.
153pub enum RetryEvent {
154    /// `Attempt` variant.
155    Attempt {
156        /// `attempt` field for attempt.
157        attempt: u32,
158    },
159    /// `Retry` variant.
160    Retry {
161        /// `attempt` field for attempt.
162        attempt: u32,
163        /// `delay_ms` field for delay ms.
164        delay_ms: u64,
165        /// `error` field for error.
166        error: String,
167    },
168    /// `Success` variant.
169    Success {
170        /// `attempt` field for attempt.
171        attempt: u32,
172    },
173    /// `Failure` variant.
174    Failure {
175        /// `attempt` field for attempt.
176        attempt: u32,
177        /// `error` field for error.
178        error: String,
179    },
180    /// `Exhausted` variant.
181    Exhausted {
182        /// `attempt` field for attempt.
183        attempt: u32,
184        /// `error` field for error.
185        error: String,
186    },
187}
188
189#[derive(Debug, Clone, PartialEq, Eq)]
190/// `BreakerStatus` data container.
191pub struct BreakerStatus {
192    /// `state` field for state.
193    pub state: BreakerState,
194    /// `failures` field for failures.
195    pub failures: u32,
196    /// `opened_at_ms` field for opened at ms.
197    pub opened_at_ms: Option<u64>,
198}
199
200#[derive(Debug, Clone, PartialEq, Eq)]
201/// `BreakerState` variants.
202pub enum BreakerState {
203    /// `Closed` variant.
204    Closed,
205    /// `Open` variant.
206    Open,
207    /// `HalfOpen` variant.
208    HalfOpen,
209}
210
211#[derive(Debug, Clone, PartialEq, Eq)]
212/// `RateLimitStatus` data container.
213pub struct RateLimitStatus {
214    /// `allowed` field for allowed.
215    pub allowed: u64,
216    /// `dropped` field for dropped.
217    pub dropped: u64,
218    /// `remaining` field for remaining.
219    pub remaining: u32,
220    /// `reset_at_ms` field for reset at ms.
221    pub reset_at_ms: u64,
222}
223
224/// `RateLimitBundle` data container.
225pub struct RateLimitBundle<T: 'static> {
226    /// `allowed` field for allowed.
227    pub allowed: Node<T>,
228    /// `dropped` field for dropped.
229    pub dropped: Node<T>,
230    /// `status` field for status.
231    pub status: Node<RateLimitStatus>,
232}
233
234#[derive(Debug, Clone, PartialEq, Eq)]
235/// `TimeoutStatus` variants.
236pub enum TimeoutStatus {
237    /// `Running` variant.
238    Running,
239    /// `Completed` variant.
240    Completed,
241    /// `Errored` variant.
242    Errored,
243}
244
245/// `TimeoutBundle` data container.
246pub struct TimeoutBundle<T: 'static> {
247    /// `node` field for node.
248    pub node: Node<T>,
249    /// `status` field for status.
250    pub status: Node<TimeoutStatus>,
251    /// `errors` field for errors.
252    pub errors: Node<String>,
253}
254
255#[derive(Clone)]
256struct RateLimitEvent<T> {
257    kind: RateLimitEventKind,
258    value: T,
259    status: RateLimitStatus,
260}
261
262#[derive(Clone, Copy)]
263enum RateLimitEventKind {
264    Allowed,
265    Dropped,
266}
267
268#[derive(Clone)]
269struct RateLimitState {
270    count: u32,
271    reset_at_ms: u64,
272    allowed: u64,
273    dropped: u64,
274}
275
276/// Creates or computes `retry_status_node`.
277pub fn retry_status_node(
278    graph: &Graph,
279    events: &Node<RetryEvent>,
280    policy: RetryPolicy,
281    name: impl Into<String>,
282) -> Node<RetryStatus> {
283    let name = name.into();
284    graph.node_opts::<RetryStatus, _>(
285        vec![events.erased()],
286        move |ctx| {
287            let mut next = RetryStatus {
288                attempt: 0,
289                max_attempts: policy.max_attempts,
290                delay_ms: None,
291                state: RetryState::Idle,
292            };
293            for event in ctx.batch::<RetryEvent>(0) {
294                next = match event.as_ref() {
295                    RetryEvent::Attempt { attempt } => RetryStatus {
296                        attempt: *attempt,
297                        max_attempts: policy.max_attempts,
298                        delay_ms: None,
299                        state: RetryState::Running,
300                    },
301                    RetryEvent::Retry {
302                        attempt, delay_ms, ..
303                    } => RetryStatus {
304                        attempt: *attempt,
305                        max_attempts: policy.max_attempts,
306                        delay_ms: Some(*delay_ms),
307                        state: RetryState::Waiting,
308                    },
309                    RetryEvent::Success { attempt } => RetryStatus {
310                        attempt: *attempt,
311                        max_attempts: policy.max_attempts,
312                        delay_ms: None,
313                        state: RetryState::Succeeded,
314                    },
315                    RetryEvent::Failure { attempt, .. } => RetryStatus {
316                        attempt: *attempt,
317                        max_attempts: policy.max_attempts,
318                        delay_ms: policy.next_delay_ms(attempt.saturating_add(1)),
319                        state: if policy.should_retry(*attempt) {
320                            RetryState::Failed
321                        } else {
322                            RetryState::Exhausted
323                        },
324                    },
325                    RetryEvent::Exhausted { attempt, .. } => RetryStatus {
326                        attempt: *attempt,
327                        max_attempts: policy.max_attempts,
328                        delay_ms: None,
329                        state: RetryState::Exhausted,
330                    },
331                };
332            }
333            ctx.emit(next);
334        },
335        GraphNodeOpts::named(format!("{name}/status")),
336    )
337}
338
339/// Creates or computes `breaker_status_node`.
340pub fn breaker_status_node(
341    graph: &Graph,
342    events: &Node<RetryEvent>,
343    failure_threshold: u32,
344    now_ms: impl Fn() -> u64 + 'static,
345    name: impl Into<String>,
346) -> Node<BreakerStatus> {
347    assert!(
348        failure_threshold > 0,
349        "breaker_status_node: failure_threshold must be > 0"
350    );
351    let name = name.into();
352    graph.node_opts::<BreakerStatus, _>(
353        vec![events.erased()],
354        move |ctx| {
355            let mut status = ctx.state_get::<BreakerStatus>().map_or(
356                BreakerStatus {
357                    state: BreakerState::Closed,
358                    failures: 0,
359                    opened_at_ms: None,
360                },
361                |value| (*value).clone(),
362            );
363            for event in ctx.batch::<RetryEvent>(0) {
364                match event.as_ref() {
365                    RetryEvent::Success { .. } => {
366                        status = BreakerStatus {
367                            state: BreakerState::Closed,
368                            failures: 0,
369                            opened_at_ms: None,
370                        };
371                    }
372                    RetryEvent::Failure { .. } | RetryEvent::Exhausted { .. } => {
373                        status.failures = status.failures.saturating_add(1);
374                        if status.failures >= failure_threshold {
375                            status.state = BreakerState::Open;
376                            status.opened_at_ms = Some(now_ms());
377                        }
378                    }
379                    RetryEvent::Attempt { .. } | RetryEvent::Retry { .. } => {}
380                }
381            }
382            ctx.state_set(status.clone());
383            ctx.emit(status);
384        },
385        GraphNodeOpts::named(format!("{name}/status")),
386    )
387}
388
389/// Creates or computes `rate_limit_bundle`.
390pub fn rate_limit_bundle<T>(
391    graph: &Graph,
392    source: &Node<T>,
393    max: u32,
394    window_ms: u64,
395    now_ms: impl Fn() -> u64 + 'static,
396    name: impl Into<String>,
397) -> RateLimitBundle<T>
398where
399    T: Clone + 'static,
400{
401    assert!(max > 0, "rate_limit_bundle: max must be > 0");
402    assert!(window_ms > 0, "rate_limit_bundle: window_ms must be > 0");
403    let name = name.into();
404    let events = graph.node_opts::<RateLimitEvent<T>, _>(
405        vec![source.erased()],
406        move |ctx| {
407            let current = now_ms();
408            let mut state = ctx.state_get::<RateLimitState>().map_or(
409                RateLimitState {
410                    count: 0,
411                    reset_at_ms: current.saturating_add(window_ms),
412                    allowed: 0,
413                    dropped: 0,
414                },
415                |value| (*value).clone(),
416            );
417            if current >= state.reset_at_ms {
418                state.count = 0;
419                state.reset_at_ms = current.saturating_add(window_ms);
420            }
421            for value in ctx.batch::<T>(0) {
422                let kind = if state.count < max {
423                    state.count = state.count.saturating_add(1);
424                    state.allowed = state.allowed.saturating_add(1);
425                    RateLimitEventKind::Allowed
426                } else {
427                    state.dropped = state.dropped.saturating_add(1);
428                    RateLimitEventKind::Dropped
429                };
430                let status = RateLimitStatus {
431                    allowed: state.allowed,
432                    dropped: state.dropped,
433                    remaining: max.saturating_sub(state.count),
434                    reset_at_ms: state.reset_at_ms,
435                };
436                ctx.emit(RateLimitEvent {
437                    kind,
438                    value: (*value).clone(),
439                    status,
440                });
441            }
442            ctx.state_set(state);
443        },
444        GraphNodeOpts::named(format!("{name}/events")),
445    );
446    let allowed = graph.node_opts::<T, _>(
447        vec![events.erased()],
448        move |ctx| {
449            for event in ctx.batch::<RateLimitEvent<T>>(0) {
450                if matches!(event.kind, RateLimitEventKind::Allowed) {
451                    ctx.emit(event.value.clone());
452                }
453            }
454        },
455        GraphNodeOpts::named(format!("{name}/allowed")),
456    );
457    let dropped = graph.node_opts::<T, _>(
458        vec![events.erased()],
459        move |ctx| {
460            for event in ctx.batch::<RateLimitEvent<T>>(0) {
461                if matches!(event.kind, RateLimitEventKind::Dropped) {
462                    ctx.emit(event.value.clone());
463                }
464            }
465        },
466        GraphNodeOpts::named(format!("{name}/dropped")),
467    );
468    let status = graph.node_opts::<RateLimitStatus, _>(
469        vec![events.erased()],
470        move |ctx| {
471            for event in ctx.batch::<RateLimitEvent<T>>(0) {
472                ctx.emit(event.status.clone());
473            }
474        },
475        GraphNodeOpts::named(format!("{name}/status")),
476    );
477    RateLimitBundle {
478        allowed,
479        dropped,
480        status,
481    }
482}
483
484/// Creates or computes `timeout_bundle`.
485pub fn timeout_bundle<T>(
486    graph: &Graph,
487    source: &Node<T>,
488    ms: u64,
489    name: impl Into<String>,
490) -> TimeoutBundle<T>
491where
492    T: Clone + 'static,
493{
494    let name = name.into();
495    let node = timeout(source, ms);
496    let status = graph.node_opts::<TimeoutStatus, _>(
497        vec![node.erased()],
498        move |ctx| match ctx.terminal(0) {
499            Some(DepTerminal::Complete) => ctx.emit(TimeoutStatus::Completed),
500            Some(DepTerminal::Error(_)) => ctx.emit(TimeoutStatus::Errored),
501            None => {
502                if !ctx.batch::<T>(0).is_empty() {
503                    ctx.emit(TimeoutStatus::Running);
504                }
505            }
506        },
507        timeout_projection_opts(format!("{name}/status")),
508    );
509    let errors = graph.node_opts::<String, _>(
510        vec![node.erased()],
511        move |ctx| {
512            if let Some(DepTerminal::Error(error)) = ctx.terminal(0) {
513                ctx.emit(error.to_string());
514            }
515        },
516        timeout_projection_opts(format!("{name}/errors")),
517    );
518    TimeoutBundle {
519        node,
520        status,
521        errors,
522    }
523}
524
525fn timeout_projection_opts(name: String) -> GraphNodeOpts {
526    GraphNodeOpts {
527        name: Some(name),
528        node: crate::node::NodeOpts {
529            complete_when_deps_complete: false,
530            error_when_deps_error: false,
531            terminal_as_real_input: true,
532            ..crate::node::NodeOpts::default()
533        },
534        ..GraphNodeOpts::default()
535    }
536}
537
538fn cap(value: u64, max: Option<u64>) -> u64 {
539    max.map_or(value, |max| value.min(max))
540}
541
542fn fibonacci(n: u32) -> u32 {
543    match n {
544        0 => 0,
545        1 => 1,
546        _ => {
547            let mut prev = 0u32;
548            let mut curr = 1u32;
549            for _ in 1..n {
550                let next = prev.saturating_add(curr);
551                prev = curr;
552                curr = next;
553            }
554            curr
555        }
556    }
557}
558
559#[cfg(test)]
560mod tests {
561    use super::*;
562
563    #[test]
564    fn backoff_policy_calculates_bounded_delays() {
565        assert_eq!(BackoffPolicy::None.delay_ms(1), 0);
566        assert_eq!(BackoffPolicy::Constant { delay_ms: 25 }.delay_ms(3), 25);
567        assert_eq!(
568            BackoffPolicy::Linear {
569                initial_ms: 10,
570                step_ms: 5,
571                max_ms: Some(18),
572            }
573            .delay_ms(4),
574            18
575        );
576        assert_eq!(
577            BackoffPolicy::Exponential {
578                initial_ms: 10,
579                factor: 2,
580                max_ms: Some(50),
581            }
582            .delay_ms(4),
583            50
584        );
585        assert_eq!(
586            BackoffPolicy::Fibonacci {
587                unit_ms: 10,
588                max_ms: None,
589            }
590            .delay_ms(6),
591            80
592        );
593    }
594
595    #[test]
596    fn retry_policy_bounds_attempts() {
597        let policy = RetryPolicy::new(
598            3,
599            BackoffPolicy::Linear {
600                initial_ms: 10,
601                step_ms: 10,
602                max_ms: None,
603            },
604        );
605
606        assert!(policy.should_retry(1));
607        assert!(policy.should_retry(2));
608        assert!(!policy.should_retry(3));
609        assert_eq!(policy.next_delay_ms(1), Some(10));
610        assert_eq!(policy.next_delay_ms(3), Some(30));
611        assert_eq!(policy.next_delay_ms(4), None);
612    }
613
614    #[test]
615    fn retry_and_breaker_status_nodes_project_event_facts() {
616        let g = crate::graph::graph();
617        let events = g.state_empty::<RetryEvent>();
618        let policy = RetryPolicy::new(2, BackoffPolicy::Constant { delay_ms: 25 });
619        let retry = retry_status_node(&g, &events, policy, "retry");
620        let breaker = breaker_status_node(&g, &events, 1, || 100, "breaker");
621        let _retry_sub = retry.subscribe(|_| {});
622        let _breaker_sub = breaker.subscribe(|_| {});
623
624        events.set(RetryEvent::Attempt { attempt: 1 });
625        events.set(RetryEvent::Failure {
626            attempt: 1,
627            error: "nope".to_owned(),
628        });
629
630        assert_eq!(
631            retry.cache(),
632            Some(RetryStatus {
633                attempt: 1,
634                max_attempts: 2,
635                delay_ms: Some(25),
636                state: RetryState::Failed,
637            })
638        );
639        assert_eq!(
640            breaker.cache(),
641            Some(BreakerStatus {
642                state: BreakerState::Open,
643                failures: 1,
644                opened_at_ms: Some(100),
645            })
646        );
647    }
648
649    #[test]
650    fn rate_limit_bundle_projects_allowed_dropped_and_status() {
651        let g = crate::graph::graph();
652        let source = g.state_empty::<i32>();
653        let now = std::rc::Rc::new(std::cell::Cell::new(0_u64));
654        let now_for_bundle = now.clone();
655        let bundle = rate_limit_bundle(&g, &source, 2, 100, move || now_for_bundle.get(), "limit");
656        let _allowed = bundle.allowed.subscribe(|_| {});
657        let _dropped = bundle.dropped.subscribe(|_| {});
658        let _status = bundle.status.subscribe(|_| {});
659
660        source.set(1);
661        source.set(2);
662        source.set(3);
663        now.set(101);
664        source.set(4);
665
666        assert_eq!(bundle.allowed.cache(), Some(4));
667        assert_eq!(bundle.dropped.cache(), Some(3));
668        assert_eq!(
669            bundle.status.cache(),
670            Some(RateLimitStatus {
671                allowed: 3,
672                dropped: 1,
673                remaining: 1,
674                reset_at_ms: 201,
675            })
676        );
677    }
678}
graphrefly/resilience.rs

graphrefly/
resilience.rs
