ocpp_client/reconnect.rs
1//! Automatic-reconnect support for `Client`'s background read loop. `Reconnector` mirrors the
2//! `Executor`/`Timer` pattern in `src/runtime.rs` - a dyn-safe trait so `Client<E>` stays
3//! generic over one type parameter only. `connect_1_6`/`connect_2_0_1`/`connect_2_1` wire up a
4//! WebSocket-backed impl automatically; embedded users implement this trait for their own
5//! transport to get the same behavior.
6
7use crate::transport::{TransportError, TransportSink, TransportStream};
8use alloc::boxed::Box;
9use core::future::Future;
10use core::pin::Pin;
11use core::time::Duration;
12
13/// (Re-)establishes a transport connection from scratch. Called by `Client`'s background read
14/// loop after the current transport reports it closed (`TransportStream::recv` returning
15/// `Ok(None)` or `Err(_)`).
16pub trait Reconnector: Send + Sync + 'static {
17 #[allow(clippy::type_complexity)]
18 fn connect<'a>(
19 &'a self,
20 ) -> Pin<
21 Box<
22 dyn Future<
23 Output = Result<
24 (Box<dyn TransportSink>, Box<dyn TransportStream>),
25 TransportError,
26 >,
27 > + Send
28 + 'a,
29 >,
30 >;
31}
32
33/// Bounded exponential backoff between reconnect attempts. The delay doubles (by
34/// `multiplier`) after each attempt that didn't produce a working connection, capped at
35/// `max_delay` - but the number of attempts itself is unbounded: a charge point should keep
36/// trying to reach its CSMS indefinitely rather than giving up after N tries.
37///
38/// The backoff resets as soon as a connection carries any inbound traffic, so an ordinary
39/// transient drop costs one `initial_delay` rather than an escalating one. Escalation is
40/// reserved for connections that never work - notably a peer that accepts and then immediately
41/// closes, which is indistinguishable from a successful dial until nothing arrives on it.
42#[derive(Debug, Clone, Copy, PartialEq, Eq)]
43pub struct ReconnectPolicy {
44 pub initial_delay: Duration,
45 pub max_delay: Duration,
46 pub multiplier: u32,
47 /// Randomize each delay within `[delay / 2, delay]` ("equal jitter"). Defaults to `true`.
48 ///
49 /// Without it, every charge point that lost the same CSMS retries in lockstep, so the
50 /// endpoint coming back gets the whole fleet at once, repeatedly. Jitter spreads that out.
51 /// Half the delay is kept un-jittered so a randomly tiny value can't defeat the rate bound
52 /// the backoff exists to provide.
53 ///
54 /// Set `false` when exact retry timing matters more than fleet behavior - a deterministic
55 /// test, mostly.
56 pub jitter: bool,
57}
58
59impl Default for ReconnectPolicy {
60 fn default() -> Self {
61 Self {
62 initial_delay: Duration::from_secs(1),
63 max_delay: Duration::from_secs(60),
64 multiplier: 2,
65 jitter: true,
66 }
67 }
68}
69
70impl ReconnectPolicy {
71 /// The un-jittered delay before reconnect attempt number `attempt` (0-indexed: `0` is the
72 /// delay before the first retry, right after the disconnect).
73 pub(crate) fn delay_for(&self, attempt: u32) -> Duration {
74 let mut delay = self.initial_delay;
75 for _ in 0..attempt {
76 delay = match delay.checked_mul(self.multiplier) {
77 Some(d) if d < self.max_delay => d,
78 _ => return self.max_delay,
79 };
80 }
81 delay
82 }
83
84 /// [`ReconnectPolicy::delay_for`] with equal jitter applied: uniformly distributed over
85 /// `[delay / 2, delay]`, or exactly `delay` when `jitter` is off.
86 ///
87 /// Randomness comes from a throwaway v4 UUID because `uuid` is already a dependency (it
88 /// generates OCPP message ids) and its RNG already works on this crate's bare-metal target -
89 /// so this needs no additional RNG dependency and no new embedded plumbing. The arithmetic is
90 /// integer-only, avoiding a float dependency on no-FPU targets.
91 pub(crate) fn jittered_delay_for(&self, attempt: u32) -> Duration {
92 let delay = self.delay_for(attempt);
93 if !self.jitter {
94 return delay;
95 }
96 let half = delay / 2;
97 let fraction = uuid::Uuid::new_v4().as_u128() as u32;
98 let extra = (half.as_nanos() * fraction as u128) / u32::MAX as u128;
99 half + Duration::from_nanos(extra as u64)
100 }
101}
102
103/// Whether a `connect_*` call should reconnect automatically on disconnect. Defaults to
104/// `Enabled` with `ReconnectPolicy::default()` - production charge points are expected to keep
105/// retrying the CSMS connection, so that's the out-of-the-box behavior; set
106/// `ConnectOptions::reconnect` to `Disabled` to opt out.
107#[derive(Debug, Clone, Copy)]
108pub enum ReconnectBehavior {
109 Enabled(ReconnectPolicy),
110 Disabled,
111}
112
113impl Default for ReconnectBehavior {
114 fn default() -> Self {
115 Self::Enabled(ReconnectPolicy::default())
116 }
117}
118
119#[cfg(test)]
120mod tests {
121 use super::*;
122
123 #[test]
124 fn delay_doubles_and_caps() {
125 let policy = ReconnectPolicy {
126 initial_delay: Duration::from_secs(1),
127 max_delay: Duration::from_secs(10),
128 multiplier: 2,
129 jitter: false,
130 };
131 assert_eq!(policy.delay_for(0), Duration::from_secs(1));
132 assert_eq!(policy.delay_for(1), Duration::from_secs(2));
133 assert_eq!(policy.delay_for(2), Duration::from_secs(4));
134 assert_eq!(policy.delay_for(3), Duration::from_secs(8));
135 assert_eq!(policy.delay_for(4), Duration::from_secs(10));
136 assert_eq!(policy.delay_for(10), Duration::from_secs(10));
137 }
138
139 #[test]
140 fn jitter_off_is_exact() {
141 let policy = ReconnectPolicy {
142 initial_delay: Duration::from_secs(4),
143 jitter: false,
144 ..ReconnectPolicy::default()
145 };
146 for attempt in 0..6 {
147 assert_eq!(
148 policy.jittered_delay_for(attempt),
149 policy.delay_for(attempt)
150 );
151 }
152 }
153
154 #[test]
155 fn jitter_stays_within_half_the_delay_and_the_full_delay() {
156 let policy = ReconnectPolicy {
157 initial_delay: Duration::from_secs(8),
158 max_delay: Duration::from_secs(64),
159 multiplier: 2,
160 jitter: true,
161 };
162
163 for attempt in 0..6 {
164 let full = policy.delay_for(attempt);
165 for _ in 0..200 {
166 let jittered = policy.jittered_delay_for(attempt);
167 assert!(
168 jittered >= full / 2 && jittered <= full,
169 "attempt {attempt}: {jittered:?} outside [{:?}, {full:?}]",
170 full / 2
171 );
172 }
173 }
174 }
175
176 #[test]
177 fn jitter_actually_varies() {
178 // The floor matters more than the spread, but a constant "jitter" would defeat the
179 // point - a fleet would still retry in lockstep.
180 let policy = ReconnectPolicy::default();
181 let first = policy.jittered_delay_for(3);
182 let varies = (0..50).any(|_| policy.jittered_delay_for(3) != first);
183 assert!(varies, "jittered delays should not all be identical");
184 }
185
186 #[test]
187 fn a_zero_delay_survives_jittering() {
188 let policy = ReconnectPolicy {
189 initial_delay: Duration::ZERO,
190 max_delay: Duration::ZERO,
191 multiplier: 2,
192 jitter: true,
193 };
194 assert_eq!(policy.jittered_delay_for(0), Duration::ZERO);
195 }
196}