rocketmq_protocol/protocol/subscription/
exponential_retry_policy.rs

1// Copyright 2023 The RocketMQ Rust Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::time::Duration;
16
17use serde::Deserialize;
18use serde::Serialize;
19
20use crate::protocol::subscription::retry_policy::RetryPolicy;
21
22#[derive(Debug, Clone, Serialize, Deserialize)]
23pub struct ExponentialRetryPolicy {
24    initial: u64,
25    max: u64,
26    multiplier: u64,
27}
28
29impl Default for ExponentialRetryPolicy {
30    #[allow(clippy::incompatible_msrv)]
31    fn default() -> Self {
32        ExponentialRetryPolicy {
33            initial: Duration::from_secs(5).as_millis() as u64,
34            max: Duration::from_hours(2).as_millis() as u64,
35            multiplier: 2,
36        }
37    }
38}
39
40impl ExponentialRetryPolicy {
41    pub fn new(initial: u64, max: u64, multiplier: u64) -> Self {
42        ExponentialRetryPolicy {
43            initial,
44            max,
45            multiplier,
46        }
47    }
48
49    pub fn initial(&self) -> u64 {
50        self.initial
51    }
52
53    pub fn max(&self) -> u64 {
54        self.max
55    }
56
57    pub fn multiplier(&self) -> u64 {
58        self.multiplier
59    }
60
61    pub fn set_initial(&mut self, initial: u64) {
62        self.initial = initial;
63    }
64
65    pub fn set_max(&mut self, max: u64) {
66        self.max = max;
67    }
68
69    pub fn set_multiplier(&mut self, multiplier: u64) {
70        self.multiplier = multiplier;
71    }
72}
73
74impl RetryPolicy for ExponentialRetryPolicy {
75    fn next_delay_duration(&self, reconsume_times: i32) -> i64 {
76        let reconsume_times = reconsume_times.clamp(0, 32) as u32;
77        // Saturate instead of overflowing: a product only reaches `u64::MAX` when the exact value
78        // would exceed it, so the cap below still yields the right answer, and a zero `initial`
79        // keeps a zero delay because `0.saturating_mul(_)` is `0`.
80        let delay = self
81            .initial
82            .saturating_mul(self.multiplier.saturating_pow(reconsume_times));
83        delay.min(self.max).min(i64::MAX as u64) as i64
84    }
85}
86
87#[cfg(test)]
88mod exponential_retry_policy_tests {
89    use std::time::Duration;
90
91    use super::*;
92
93    #[test]
94    fn configuration_methods_and_serde_preserve_values() {
95        let default = ExponentialRetryPolicy::default();
96        assert_eq!(
97            (default.initial(), default.max(), default.multiplier()),
98            (
99                Duration::from_secs(5).as_millis() as u64,
100                Duration::from_hours(2).as_millis() as u64,
101                2,
102            )
103        );
104
105        let mut policy = ExponentialRetryPolicy::new(1_000, 3_600_000, 3);
106        policy.set_initial(2_000);
107        policy.set_max(7_200_000);
108        policy.set_multiplier(4);
109
110        let json = serde_json::to_string(&policy).unwrap();
111        assert_eq!(json, r#"{"initial":2000,"max":7200000,"multiplier":4}"#);
112
113        let decoded: ExponentialRetryPolicy = serde_json::from_str(&json).unwrap();
114        assert_eq!(
115            (decoded.initial(), decoded.max(), decoded.multiplier()),
116            (2_000, 7_200_000, 4)
117        );
118    }
119
120    #[test]
121    fn next_delay_duration_handles_boundaries() {
122        let policy = ExponentialRetryPolicy::new(1_000, 5_000, 2);
123        for (reconsume_times, expected_delay) in [(-1, 1_000), (0, 1_000), (1, 2_000), (2, 4_000), (3, 5_000)] {
124            assert_eq!(policy.next_delay_duration(reconsume_times), expected_delay);
125        }
126
127        let uncapped_max = ExponentialRetryPolicy::new(1, u64::MAX, 2);
128        assert_eq!(uncapped_max.next_delay_duration(100), 1_i64 << 32);
129    }
130
131    #[test]
132    fn next_delay_duration_saturates_instead_of_overflowing() {
133        // `multiplier.pow` overflows on its own well before the multiplication.
134        let exponentiation_overflow = ExponentialRetryPolicy::new(1_000, 5_000, 10);
135        assert_eq!(exponentiation_overflow.next_delay_duration(32), 5_000);
136
137        // The exponentiation fits, the multiplication by `initial` does not.
138        let multiplication_overflow = ExponentialRetryPolicy::new(u64::MAX, 7_200_000, 2);
139        assert_eq!(multiplication_overflow.next_delay_duration(1), 7_200_000);
140
141        // A zero `initial` keeps a zero delay even when the exponentiation saturates.
142        let zero_initial = ExponentialRetryPolicy::new(0, 5_000, 10);
143        assert_eq!(zero_initial.next_delay_duration(32), 0);
144
145        // A `max` above `i64::MAX` must not produce a negative delay.
146        let above_i64_max = ExponentialRetryPolicy::new(u64::MAX, u64::MAX, u64::MAX);
147        assert_eq!(above_i64_max.next_delay_duration(32), i64::MAX);
148    }
149}
rocketmq_protocol/protocol/subscription/exponential_retry_policy.rs

rocketmq_protocol/protocol/subscription/
exponential_retry_policy.rs
