golem_rust/retry/local/
mod.rs

1//! Local retry policy interpretation and execution.
2
3use super::builder::Policy;
4use super::retry_api;
5use std::cmp::Ordering;
6use std::error::Error;
7use std::fmt;
8use std::ops::AsyncFnMut;
9use std::time::Duration;
10
11const MAX_COMPILED_NODES: usize = 4_096;
12const MAX_COMPILED_PAYLOAD_BYTES: usize = 1_048_576;
13const MAX_POLICY_DEPTH: usize = 256;
14
15/// A validated retry schedule for retrying fallible user code locally.
16///
17/// Unlike named retry policies installed through [`super::set_retry_policy`], this schedule is
18/// interpreted entirely by the component. It does not install a named policy or create a
19/// host-managed retry sequence. The loop is ordinary guest execution, so its host calls retain
20/// their normal durable replay and suspension behavior.
21#[derive(Clone, Debug)]
22pub struct RetrySchedule {
23    policy: CompiledPolicy,
24}
25
26/// An error found while validating a flattened raw retry policy.
27#[derive(Clone, Debug, PartialEq)]
28pub enum RetryPolicyError {
29    EmptyPolicy,
30    InvalidPolicyNodeIndex(i32),
31    CyclicPolicyNode(i32),
32    EmptyPredicate,
33    InvalidPredicateNodeIndex(i32),
34    CyclicPredicateNode(i32),
35    PolicyTooComplex,
36    InvalidExponentialFactor(f64),
37    InvalidJitterFactor(f64),
38    InvalidClampRange {
39        min_delay: Duration,
40        max_delay: Duration,
41    },
42}
43
44impl fmt::Display for RetryPolicyError {
45    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
46        match self {
47            Self::EmptyPolicy => write!(f, "retry policy has no root node"),
48            Self::InvalidPolicyNodeIndex(index) => {
49                write!(f, "retry policy references invalid node index {index}")
50            }
51            Self::CyclicPolicyNode(index) => {
52                write!(f, "retry policy contains a cycle at node index {index}")
53            }
54            Self::EmptyPredicate => write!(f, "retry predicate has no root node"),
55            Self::InvalidPredicateNodeIndex(index) => {
56                write!(f, "retry predicate references invalid node index {index}")
57            }
58            Self::CyclicPredicateNode(index) => {
59                write!(f, "retry predicate contains a cycle at node index {index}")
60            }
61            Self::PolicyTooComplex => write!(f, "retry policy is too deeply nested or expansive"),
62            Self::InvalidExponentialFactor(factor) => write!(
63                f,
64                "exponential factor must be finite and greater than 0, got {factor}"
65            ),
66            Self::InvalidJitterFactor(factor) => write!(
67                f,
68                "jitter factor must be finite and greater than or equal to 0, got {factor}"
69            ),
70            Self::InvalidClampRange {
71                min_delay,
72                max_delay,
73            } => write!(
74                f,
75                "clamp min delay {min_delay:?} must be less than or equal to max delay {max_delay:?}"
76            ),
77        }
78    }
79}
80
81impl Error for RetryPolicyError {}
82
83impl TryFrom<&retry_api::RetryPolicy> for RetrySchedule {
84    type Error = RetryPolicyError;
85
86    fn try_from(policy: &retry_api::RetryPolicy) -> Result<Self, Self::Error> {
87        if policy.nodes.is_empty() {
88            return Err(RetryPolicyError::EmptyPolicy);
89        }
90        if policy.nodes.len() > MAX_COMPILED_NODES {
91            return Err(RetryPolicyError::PolicyTooComplex);
92        }
93        let mut compiler = Compiler::new();
94        Ok(Self {
95            policy: compiler.compile_policy_node(&policy.nodes, 0, 0)?,
96        })
97    }
98}
99
100impl Policy {
101    /// Compiles this high-level policy into a reusable user-space retry schedule.
102    pub fn try_to_schedule(&self) -> Result<RetrySchedule, super::builder::RetryBuilderError> {
103        let raw = self.try_to_raw()?;
104        RetrySchedule::try_from(&raw)
105            .map_err(|_| super::builder::RetryBuilderError::PolicyTooComplex)
106    }
107}
108
109impl RetrySchedule {
110    /// Runs an async operation until it succeeds or this schedule gives up.
111    ///
112    /// Use [`Self::retry_with_properties`] when the schedule contains filtered predicates.
113    pub async fn retry<T, E, Operation>(&self, operation: Operation) -> Result<T, E>
114    where
115        Operation: AsyncFnMut() -> Result<T, E>,
116    {
117        self.retry_with_properties(operation, |_| std::iter::empty())
118            .await
119    }
120
121    /// Runs an async operation with retry properties projected from each failure.
122    ///
123    /// `properties` is evaluated after every failure, so predicates can react to the current
124    /// error. When the schedule gives up, the most recent operation error is returned unchanged.
125    pub async fn retry_with_properties<T, E, Operation, Properties, PropertyItems>(
126        &self,
127        operation: Operation,
128        properties: Properties,
129    ) -> Result<T, E>
130    where
131        Operation: AsyncFnMut() -> Result<T, E>,
132        Properties: FnMut(&E) -> PropertyItems,
133        PropertyItems: IntoIterator<Item = (String, retry_api::PredicateValue)>,
134    {
135        self.retry_with_runtime(
136            operation,
137            properties,
138            crate::wasip3::clocks::monotonic_clock::now,
139            || {
140                let value = crate::wasip3::random::random::get_random_u64() >> 11;
141                value as f64 * (1.0 / ((1_u64 << 53) as f64))
142            },
143            |delay| crate::wasip3::clocks::monotonic_clock::wait_for(duration_to_nanos(delay)),
144        )
145        .await
146    }
147
148    async fn retry_with_runtime<
149        T,
150        E,
151        Operation,
152        Properties,
153        PropertyItems,
154        Now,
155        Random,
156        Sleep,
157        SleepFuture,
158    >(
159        &self,
160        mut operation: Operation,
161        mut properties: Properties,
162        mut now: Now,
163        mut random: Random,
164        mut sleep: Sleep,
165    ) -> Result<T, E>
166    where
167        Operation: AsyncFnMut() -> Result<T, E>,
168        Properties: FnMut(&E) -> PropertyItems,
169        PropertyItems: IntoIterator<Item = (String, retry_api::PredicateValue)>,
170        Now: FnMut() -> u64,
171        Random: FnMut() -> f64,
172        Sleep: FnMut(Duration) -> SleepFuture,
173        SleepFuture: Future<Output = ()>,
174    {
175        let started = now();
176        let mut policy = self.policy.clone();
177        loop {
178            match operation().await {
179                Ok(value) => return Ok(value),
180                Err(error) => {
181                    let properties = properties(&error).into_iter().collect::<Vec<_>>();
182                    let elapsed = Duration::from_nanos(now().saturating_sub(started));
183                    match policy.step(elapsed, &properties, &mut random) {
184                        Step::Retry(delay) => sleep(delay).await,
185                        Step::GiveUp | Step::Error => return Err(error),
186                    }
187                }
188            }
189        }
190    }
191}
192
193#[derive(Clone, Debug)]
194enum CompiledPolicy {
195    Periodic(Duration),
196    Exponential {
197        base_delay: Duration,
198        factor: f64,
199        attempt: u32,
200    },
201    Fibonacci {
202        previous: Duration,
203        current: Duration,
204        attempt: u32,
205    },
206    Immediate,
207    Never,
208    CountBox {
209        max_retries: u32,
210        attempts: u32,
211        inner: Box<Self>,
212    },
213    TimeBox {
214        limit: Duration,
215        inner: Box<Self>,
216    },
217    Clamp {
218        min_delay: Duration,
219        max_delay: Duration,
220        inner: Box<Self>,
221    },
222    AddDelay {
223        delay: Duration,
224        inner: Box<Self>,
225    },
226    Jitter {
227        factor: f64,
228        inner: Box<Self>,
229    },
230    FilteredOn {
231        predicate: CompiledPredicate,
232        inner: Box<Self>,
233    },
234    AndThen {
235        left: Box<Self>,
236        right: Box<Self>,
237        on_right: bool,
238    },
239    Union(Box<Self>, Box<Self>),
240    Intersect(Box<Self>, Box<Self>),
241}
242
243#[derive(Clone, Copy, Debug, PartialEq, Eq)]
244enum Step {
245    Retry(Duration),
246    GiveUp,
247    Error,
248}
249
250impl CompiledPolicy {
251    fn step(
252        &mut self,
253        elapsed: Duration,
254        properties: &[(String, retry_api::PredicateValue)],
255        random: &mut impl FnMut() -> f64,
256    ) -> Step {
257        match self {
258            Self::Periodic(delay) => Step::Retry(*delay),
259            Self::Exponential {
260                base_delay,
261                factor,
262                attempt,
263            } => {
264                let delay = scale_duration(*base_delay, factor.powf(*attempt as f64));
265                *attempt = attempt.saturating_add(1);
266                Step::Retry(delay)
267            }
268            Self::Fibonacci {
269                previous,
270                current,
271                attempt,
272            } => {
273                let delay = match *attempt {
274                    0 => *previous,
275                    1 => *current,
276                    _ => {
277                        let next = saturating_add(*previous, *current);
278                        *previous = *current;
279                        *current = next;
280                        next
281                    }
282                };
283                *attempt = attempt.saturating_add(1);
284                Step::Retry(delay)
285            }
286            Self::Immediate => Step::Retry(Duration::ZERO),
287            Self::Never => Step::GiveUp,
288            Self::CountBox {
289                max_retries,
290                attempts,
291                inner,
292            } => {
293                if *attempts >= *max_retries {
294                    Step::GiveUp
295                } else {
296                    *attempts = attempts.saturating_add(1);
297                    inner.step(elapsed, properties, random)
298                }
299            }
300            Self::TimeBox { limit, inner } => {
301                if elapsed >= *limit {
302                    Step::GiveUp
303                } else {
304                    inner.step(elapsed, properties, random)
305                }
306            }
307            Self::Clamp {
308                min_delay,
309                max_delay,
310                inner,
311            } => match inner.step(elapsed, properties, random) {
312                Step::Retry(delay) => Step::Retry(delay.clamp(*min_delay, *max_delay)),
313                verdict => verdict,
314            },
315            Self::AddDelay { delay, inner } => match inner.step(elapsed, properties, random) {
316                Step::Retry(current) => Step::Retry(saturating_add(current, *delay)),
317                verdict => verdict,
318            },
319            Self::Jitter { factor, inner } => match inner.step(elapsed, properties, random) {
320                Step::Retry(delay) if *factor > 0.0 => Step::Retry(saturating_add(
321                    delay,
322                    scale_duration(delay, random() * *factor),
323                )),
324                verdict => verdict,
325            },
326            Self::FilteredOn { predicate, inner } => match predicate.matches(properties) {
327                Ok(true) => inner.step(elapsed, properties, random),
328                Ok(false) => Step::GiveUp,
329                Err(_) => Step::Error,
330            },
331            Self::AndThen {
332                left,
333                right,
334                on_right,
335            } => {
336                if *on_right {
337                    right.step(elapsed, properties, random)
338                } else {
339                    match left.step(elapsed, properties, random) {
340                        Step::Retry(delay) => Step::Retry(delay),
341                        Step::GiveUp => {
342                            *on_right = true;
343                            right.step(elapsed, properties, random)
344                        }
345                        Step::Error => Step::Error,
346                    }
347                }
348            }
349            Self::Union(left, right) => match (
350                left.step(elapsed, properties, random),
351                right.step(elapsed, properties, random),
352            ) {
353                (Step::Retry(left), Step::Retry(right)) => Step::Retry(left.min(right)),
354                (Step::Retry(delay), Step::GiveUp) | (Step::GiveUp, Step::Retry(delay)) => {
355                    Step::Retry(delay)
356                }
357                (Step::GiveUp, Step::GiveUp) => Step::GiveUp,
358                (Step::Error, _) | (_, Step::Error) => Step::Error,
359            },
360            Self::Intersect(left, right) => match (
361                left.step(elapsed, properties, random),
362                right.step(elapsed, properties, random),
363            ) {
364                (Step::Retry(left), Step::Retry(right)) => Step::Retry(left.max(right)),
365                (Step::Error, _) | (_, Step::Error) => Step::Error,
366                _ => Step::GiveUp,
367            },
368        }
369    }
370}
371
372#[derive(Clone, Debug)]
373enum CompiledPredicate {
374    Eq(String, retry_api::PredicateValue),
375    Neq(String, retry_api::PredicateValue),
376    Gt(String, retry_api::PredicateValue),
377    Gte(String, retry_api::PredicateValue),
378    Lt(String, retry_api::PredicateValue),
379    Lte(String, retry_api::PredicateValue),
380    Exists(String),
381    In(String, Vec<retry_api::PredicateValue>),
382    Matches(String, String),
383    StartsWith(String, String),
384    Contains(String, String),
385    And(Box<Self>, Box<Self>),
386    Or(Box<Self>, Box<Self>),
387    Not(Box<Self>),
388    True,
389    False,
390}
391
392#[derive(Clone, Copy, Debug)]
393struct PredicateEvaluationError;
394
395impl CompiledPredicate {
396    fn matches(
397        &self,
398        properties: &[(String, retry_api::PredicateValue)],
399    ) -> Result<bool, PredicateEvaluationError> {
400        use CompiledPredicate::*;
401        match self {
402            Eq(name, expected) => {
403                Ok(compare(required(properties, name)?, expected)? == Ordering::Equal)
404            }
405            Neq(name, expected) => {
406                Ok(compare(required(properties, name)?, expected)? != Ordering::Equal)
407            }
408            Gt(name, expected) => {
409                Ok(compare(required(properties, name)?, expected)? == Ordering::Greater)
410            }
411            Gte(name, expected) => {
412                Ok(compare(required(properties, name)?, expected)? != Ordering::Less)
413            }
414            Lt(name, expected) => {
415                Ok(compare(required(properties, name)?, expected)? == Ordering::Less)
416            }
417            Lte(name, expected) => {
418                Ok(compare(required(properties, name)?, expected)? != Ordering::Greater)
419            }
420            Exists(name) => Ok(get_property(properties, name).is_some()),
421            In(name, expected) => {
422                let actual = required(properties, name)?;
423                let mut had_error = false;
424                for candidate in expected {
425                    match compare(actual, candidate) {
426                        Ok(Ordering::Equal) => return Ok(true),
427                        Ok(_) => {}
428                        Err(_) => had_error = true,
429                    }
430                }
431                if had_error {
432                    Err(PredicateEvaluationError)
433                } else {
434                    Ok(false)
435                }
436            }
437            Matches(name, pattern) => Ok(glob_match::glob_match(
438                pattern,
439                &as_text(required(properties, name)?)?,
440            )),
441            StartsWith(name, prefix) => {
442                Ok(as_text(required(properties, name)?)?.starts_with(prefix))
443            }
444            Contains(name, substring) => {
445                Ok(as_text(required(properties, name)?)?.contains(substring))
446            }
447            And(left, right) => Ok(left.matches(properties)? && right.matches(properties)?),
448            Or(left, right) => Ok(left.matches(properties)? || right.matches(properties)?),
449            Not(inner) => Ok(!inner.matches(properties)?),
450            True => Ok(true),
451            False => Ok(false),
452        }
453    }
454}
455
456struct Compiler {
457    compiled_nodes: usize,
458    compiled_payload_bytes: usize,
459    policy_visiting: Vec<bool>,
460}
461
462impl Compiler {
463    fn new() -> Self {
464        Self {
465            compiled_nodes: 0,
466            compiled_payload_bytes: 0,
467            policy_visiting: Vec::new(),
468        }
469    }
470
471    fn enter(&mut self, depth: usize, payload_bytes: usize) -> Result<(), RetryPolicyError> {
472        self.compiled_nodes = self.compiled_nodes.saturating_add(1);
473        self.compiled_payload_bytes = self.compiled_payload_bytes.saturating_add(payload_bytes);
474        if depth > MAX_POLICY_DEPTH
475            || self.compiled_nodes > MAX_COMPILED_NODES
476            || self.compiled_payload_bytes > MAX_COMPILED_PAYLOAD_BYTES
477        {
478            Err(RetryPolicyError::PolicyTooComplex)
479        } else {
480            Ok(())
481        }
482    }
483
484    fn compile_policy_node(
485        &mut self,
486        nodes: &[retry_api::PolicyNode],
487        index: i32,
488        depth: usize,
489    ) -> Result<CompiledPolicy, RetryPolicyError> {
490        self.enter(depth, 0)?;
491        if self.policy_visiting.len() != nodes.len() {
492            self.policy_visiting = vec![false; nodes.len()];
493        }
494        let position = usize::try_from(index)
495            .ok()
496            .filter(|position| *position < nodes.len())
497            .ok_or(RetryPolicyError::InvalidPolicyNodeIndex(index))?;
498        if self.policy_visiting[position] {
499            return Err(RetryPolicyError::CyclicPolicyNode(index));
500        }
501        self.policy_visiting[position] = true;
502        let next_depth = depth + 1;
503        let result = match &nodes[position] {
504            retry_api::PolicyNode::Periodic(delay) => {
505                CompiledPolicy::Periodic(Duration::from_nanos(*delay))
506            }
507            retry_api::PolicyNode::Exponential(config) => {
508                if !config.factor.is_finite() || config.factor <= 0.0 {
509                    return Err(RetryPolicyError::InvalidExponentialFactor(config.factor));
510                }
511                CompiledPolicy::Exponential {
512                    base_delay: Duration::from_nanos(config.base_delay),
513                    factor: config.factor,
514                    attempt: 0,
515                }
516            }
517            retry_api::PolicyNode::Fibonacci(config) => CompiledPolicy::Fibonacci {
518                previous: Duration::from_nanos(config.first),
519                current: Duration::from_nanos(config.second),
520                attempt: 0,
521            },
522            retry_api::PolicyNode::Immediate => CompiledPolicy::Immediate,
523            retry_api::PolicyNode::Never => CompiledPolicy::Never,
524            retry_api::PolicyNode::CountBox(config) => CompiledPolicy::CountBox {
525                max_retries: config.max_retries,
526                attempts: 0,
527                inner: Box::new(self.compile_policy_node(nodes, config.inner, next_depth)?),
528            },
529            retry_api::PolicyNode::TimeBox(config) => CompiledPolicy::TimeBox {
530                limit: Duration::from_nanos(config.limit),
531                inner: Box::new(self.compile_policy_node(nodes, config.inner, next_depth)?),
532            },
533            retry_api::PolicyNode::ClampDelay(config) => {
534                let min_delay = Duration::from_nanos(config.min_delay);
535                let max_delay = Duration::from_nanos(config.max_delay);
536                if min_delay > max_delay {
537                    return Err(RetryPolicyError::InvalidClampRange {
538                        min_delay,
539                        max_delay,
540                    });
541                }
542                CompiledPolicy::Clamp {
543                    min_delay,
544                    max_delay,
545                    inner: Box::new(self.compile_policy_node(nodes, config.inner, next_depth)?),
546                }
547            }
548            retry_api::PolicyNode::AddDelay(config) => CompiledPolicy::AddDelay {
549                delay: Duration::from_nanos(config.delay),
550                inner: Box::new(self.compile_policy_node(nodes, config.inner, next_depth)?),
551            },
552            retry_api::PolicyNode::Jitter(config) => {
553                if !config.factor.is_finite() || config.factor < 0.0 {
554                    return Err(RetryPolicyError::InvalidJitterFactor(config.factor));
555                }
556                CompiledPolicy::Jitter {
557                    factor: config.factor,
558                    inner: Box::new(self.compile_policy_node(nodes, config.inner, next_depth)?),
559                }
560            }
561            retry_api::PolicyNode::FilteredOn(config) => CompiledPolicy::FilteredOn {
562                predicate: self.compile_predicate(&config.predicate, next_depth)?,
563                inner: Box::new(self.compile_policy_node(nodes, config.inner, next_depth)?),
564            },
565            retry_api::PolicyNode::AndThen((left, right)) => CompiledPolicy::AndThen {
566                left: Box::new(self.compile_policy_node(nodes, *left, next_depth)?),
567                right: Box::new(self.compile_policy_node(nodes, *right, next_depth)?),
568                on_right: false,
569            },
570            retry_api::PolicyNode::PolicyUnion((left, right)) => CompiledPolicy::Union(
571                Box::new(self.compile_policy_node(nodes, *left, next_depth)?),
572                Box::new(self.compile_policy_node(nodes, *right, next_depth)?),
573            ),
574            retry_api::PolicyNode::PolicyIntersect((left, right)) => CompiledPolicy::Intersect(
575                Box::new(self.compile_policy_node(nodes, *left, next_depth)?),
576                Box::new(self.compile_policy_node(nodes, *right, next_depth)?),
577            ),
578        };
579        self.policy_visiting[position] = false;
580        Ok(result)
581    }
582
583    fn compile_predicate(
584        &mut self,
585        predicate: &retry_api::RetryPredicate,
586        depth: usize,
587    ) -> Result<CompiledPredicate, RetryPolicyError> {
588        if predicate.nodes.is_empty() {
589            return Err(RetryPolicyError::EmptyPredicate);
590        }
591        if predicate.nodes.len() > MAX_COMPILED_NODES {
592            return Err(RetryPolicyError::PolicyTooComplex);
593        }
594        self.compile_predicate_node(
595            &predicate.nodes,
596            0,
597            depth,
598            &mut vec![false; predicate.nodes.len()],
599        )
600    }
601
602    fn compile_predicate_node(
603        &mut self,
604        nodes: &[retry_api::PredicateNode],
605        index: i32,
606        depth: usize,
607        visiting: &mut [bool],
608    ) -> Result<CompiledPredicate, RetryPolicyError> {
609        use retry_api::PredicateNode as Raw;
610        let position = usize::try_from(index)
611            .ok()
612            .filter(|position| *position < nodes.len())
613            .ok_or(RetryPolicyError::InvalidPredicateNodeIndex(index))?;
614        self.enter(depth, predicate_node_payload_bytes(&nodes[position]))?;
615        if visiting[position] {
616            return Err(RetryPolicyError::CyclicPredicateNode(index));
617        }
618        visiting[position] = true;
619        let next_depth = depth + 1;
620        let comparison = |comparison: &retry_api::PropertyComparison| {
621            (comparison.property_name.clone(), comparison.value.clone())
622        };
623        let result = match &nodes[position] {
624            Raw::PropEq(value) => {
625                let (name, value) = comparison(value);
626                CompiledPredicate::Eq(name, value)
627            }
628            Raw::PropNeq(value) => {
629                let (name, value) = comparison(value);
630                CompiledPredicate::Neq(name, value)
631            }
632            Raw::PropGt(value) => {
633                let (name, value) = comparison(value);
634                CompiledPredicate::Gt(name, value)
635            }
636            Raw::PropGte(value) => {
637                let (name, value) = comparison(value);
638                CompiledPredicate::Gte(name, value)
639            }
640            Raw::PropLt(value) => {
641                let (name, value) = comparison(value);
642                CompiledPredicate::Lt(name, value)
643            }
644            Raw::PropLte(value) => {
645                let (name, value) = comparison(value);
646                CompiledPredicate::Lte(name, value)
647            }
648            Raw::PropExists(name) => CompiledPredicate::Exists(name.clone()),
649            Raw::PropIn(check) => {
650                CompiledPredicate::In(check.property_name.clone(), check.values.clone())
651            }
652            Raw::PropMatches(pattern) => {
653                CompiledPredicate::Matches(pattern.property_name.clone(), pattern.pattern.clone())
654            }
655            Raw::PropStartsWith(pattern) => CompiledPredicate::StartsWith(
656                pattern.property_name.clone(),
657                pattern.pattern.clone(),
658            ),
659            Raw::PropContains(pattern) => {
660                CompiledPredicate::Contains(pattern.property_name.clone(), pattern.pattern.clone())
661            }
662            Raw::PredAnd((left, right)) => CompiledPredicate::And(
663                Box::new(self.compile_predicate_node(nodes, *left, next_depth, visiting)?),
664                Box::new(self.compile_predicate_node(nodes, *right, next_depth, visiting)?),
665            ),
666            Raw::PredOr((left, right)) => CompiledPredicate::Or(
667                Box::new(self.compile_predicate_node(nodes, *left, next_depth, visiting)?),
668                Box::new(self.compile_predicate_node(nodes, *right, next_depth, visiting)?),
669            ),
670            Raw::PredNot(inner) => CompiledPredicate::Not(Box::new(
671                self.compile_predicate_node(nodes, *inner, next_depth, visiting)?,
672            )),
673            Raw::PredTrue => CompiledPredicate::True,
674            Raw::PredFalse => CompiledPredicate::False,
675        };
676        visiting[position] = false;
677        Ok(result)
678    }
679}
680
681fn predicate_node_payload_bytes(node: &retry_api::PredicateNode) -> usize {
682    fn value_bytes(value: &retry_api::PredicateValue) -> usize {
683        match value {
684            retry_api::PredicateValue::Text(value) => value.len(),
685            retry_api::PredicateValue::Integer(_) | retry_api::PredicateValue::Boolean(_) => 0,
686        }
687    }
688
689    use retry_api::PredicateNode::*;
690    match node {
691        PropEq(value) | PropNeq(value) | PropGt(value) | PropGte(value) | PropLt(value)
692        | PropLte(value) => value
693            .property_name
694            .len()
695            .saturating_add(value_bytes(&value.value)),
696        PropExists(name) => name.len(),
697        PropIn(check) => {
698            let base = check.property_name.len().saturating_add(
699                check
700                    .values
701                    .len()
702                    .saturating_mul(std::mem::size_of::<retry_api::PredicateValue>()),
703            );
704            if base > MAX_COMPILED_PAYLOAD_BYTES {
705                usize::MAX
706            } else {
707                check.values.iter().fold(base, |total, value| {
708                    total.saturating_add(value_bytes(value))
709                })
710            }
711        }
712        PropMatches(pattern) | PropStartsWith(pattern) | PropContains(pattern) => pattern
713            .property_name
714            .len()
715            .saturating_add(pattern.pattern.len()),
716        PredAnd(_) | PredOr(_) | PredNot(_) | PredTrue | PredFalse => 0,
717    }
718}
719
720fn get_property<'a>(
721    properties: &'a [(String, retry_api::PredicateValue)],
722    name: &str,
723) -> Option<&'a retry_api::PredicateValue> {
724    properties
725        .iter()
726        .rev()
727        .find(|(key, _)| key == name)
728        .map(|(_, value)| value)
729}
730
731fn required<'a>(
732    properties: &'a [(String, retry_api::PredicateValue)],
733    name: &str,
734) -> Result<&'a retry_api::PredicateValue, PredicateEvaluationError> {
735    get_property(properties, name).ok_or(PredicateEvaluationError)
736}
737
738fn compare(
739    actual: &retry_api::PredicateValue,
740    expected: &retry_api::PredicateValue,
741) -> Result<Ordering, PredicateEvaluationError> {
742    use retry_api::PredicateValue::*;
743    match (actual, expected) {
744        (Integer(left), Integer(right)) => Ok(left.cmp(right)),
745        (Text(left), Text(right)) => Ok(left.cmp(right)),
746        (Boolean(left), Boolean(right)) => Ok(left.cmp(right)),
747        (Text(left), Integer(right)) => left
748            .parse::<i64>()
749            .map(|left| left.cmp(right))
750            .map_err(|_| PredicateEvaluationError),
751        (Integer(left), Text(right)) => Ok(left.to_string().cmp(right)),
752        _ => Err(PredicateEvaluationError),
753    }
754}
755
756fn as_text(value: &retry_api::PredicateValue) -> Result<String, PredicateEvaluationError> {
757    match value {
758        retry_api::PredicateValue::Text(value) => Ok(value.clone()),
759        retry_api::PredicateValue::Integer(value) => Ok(value.to_string()),
760        retry_api::PredicateValue::Boolean(_) => Err(PredicateEvaluationError),
761    }
762}
763
764fn scale_duration(duration: Duration, factor: f64) -> Duration {
765    if !factor.is_finite() || factor <= 0.0 {
766        return Duration::ZERO;
767    }
768    let value = duration.as_secs_f64() * factor;
769    if !value.is_finite() {
770        Duration::MAX
771    } else {
772        Duration::try_from_secs_f64(value).unwrap_or(Duration::MAX)
773    }
774}
775
776fn saturating_add(left: Duration, right: Duration) -> Duration {
777    left.checked_add(right).unwrap_or(Duration::MAX)
778}
779
780fn duration_to_nanos(duration: Duration) -> u64 {
781    duration.as_nanos().min(u64::MAX as u128) as u64
782}
783
784#[cfg(test)]
785mod tests;
golem_rust/retry/local/mod.rs

golem_rust/retry/local/
mod.rs
