imsg_session/
retry.rs

1//! Connection-retry primitives: transient/permanent classification and the backoff schedule.
2//!
3//! These are the policy pieces the broker actor *drives*; the reconnect loop that consumes them
4//! lives in `imsg-broker`, not here. Keeping classification in session is mandatory — it is the
5//! only layer that knows what `MapError`/`ObexError`/`io::ErrorKind` mean.
6
7use std::io;
8use std::time::Duration;
9
10use map_core::{MapError, ObexError};
11use tokio_retry::strategy::ExponentialBackoff;
12
13use crate::{SessionError, TransportError};
14
15/// Whether a failed connection attempt should be retried.
16#[derive(Debug, Clone, Copy, PartialEq, Eq)]
17pub enum Disposition {
18    /// Recoverable (link timeout/reset, transient protocol hiccup) — retry with backoff.
19    Transient,
20    /// Non-recoverable (device refused, auth/pairing, bad input) — fail fast.
21    Permanent,
22}
23
24/// Classifies a session-establishment failure as transient (retry) or permanent (fail fast).
25///
26/// Inspects the inner `io::ErrorKind` and OBEX/server opcodes rather than the top-level variant,
27/// so "device asleep / out of range" (transient) is distinguished from "wrong channel / pairing
28/// rejected" (permanent). Anything not explicitly recognised defaults to [`Disposition::Transient`]:
29/// the retry budget bounds the cost of a wrong guess, whereas a wrong "permanent" verdict abandons
30/// a recoverable link.
31#[must_use]
32pub fn classify(e: &SessionError) -> Disposition {
33    match e {
34        SessionError::Transport(t) => classify_transport(t),
35        SessionError::Map(m) => classify_map(m),
36        SessionError::Pbap(_) => Disposition::Transient,
37    }
38}
39
40fn classify_map(e: &MapError) -> Disposition {
41    match e {
42        MapError::Transport(t) => classify_transport(t),
43        // OBEX CONNECT refused, server refusal, or rejected input: retrying will not help.
44        MapError::Obex(ObexError::ConnectRejected(_))
45        | MapError::ServerError(_)
46        | MapError::InvalidInput(_) => Disposition::Permanent,
47        _ => Disposition::Transient,
48    }
49}
50
51fn classify_transport(e: &TransportError) -> Disposition {
52    match e {
53        TransportError::Io(io) => classify_io(io.kind()),
54        _ => Disposition::Transient,
55    }
56}
57
58const fn classify_io(kind: io::ErrorKind) -> Disposition {
59    match kind {
60        // No service on the channel, auth denied, or a malformed address — all permanent.
61        io::ErrorKind::ConnectionRefused
62        | io::ErrorKind::PermissionDenied
63        | io::ErrorKind::InvalidInput => Disposition::Permanent,
64        _ => Disposition::Transient,
65    }
66}
67
68/// Builds the inter-attempt backoff schedule: delays doubling from `initial`, capped at `max`,
69/// yielding exactly `max_attempts - 1` values (the gaps between `max_attempts` attempts).
70///
71/// Empty when `max_attempts <= 1`. No jitter — the kernel-atomic bind election guarantees one
72/// broker per device, so there is no thundering herd to spread. The returned iterator is itself
73/// `#[must_use]`.
74pub fn backoff(
75    initial: Duration,
76    max: Duration,
77    max_attempts: u32,
78) -> impl Iterator<Item = Duration> + Clone {
79    let initial_ms = u64::try_from(initial.as_millis()).unwrap_or(u64::MAX);
80    // base=2 gives a doubling ratio; factor scales the first delay to `initial` (2 * factor).
81    let factor = (initial_ms / 2).max(1);
82    let gaps = usize::try_from(max_attempts.saturating_sub(1)).unwrap_or(usize::MAX);
83    ExponentialBackoff::from_millis(2).factor(factor).max_delay(max).take(gaps)
84}
85
86#[cfg(test)]
87mod tests {
88    use super::*;
89
90    fn io(kind: io::ErrorKind) -> SessionError {
91        SessionError::Transport(TransportError::Io(io::Error::new(kind, "x")))
92    }
93
94    #[test]
95    fn timeouts_and_resets_are_transient() {
96        for k in
97            [io::ErrorKind::TimedOut, io::ErrorKind::ConnectionReset, io::ErrorKind::BrokenPipe]
98        {
99            assert_eq!(classify(&io(k)), Disposition::Transient);
100        }
101    }
102
103    #[test]
104    fn refusal_and_auth_are_permanent() {
105        for k in [io::ErrorKind::ConnectionRefused, io::ErrorKind::PermissionDenied] {
106            assert_eq!(classify(&io(k)), Disposition::Permanent);
107        }
108    }
109
110    #[test]
111    fn obex_connect_rejected_is_permanent() {
112        let e = SessionError::Map(MapError::Obex(ObexError::ConnectRejected(0xC3)));
113        assert_eq!(classify(&e), Disposition::Permanent);
114    }
115
116    #[test]
117    fn server_error_is_permanent_but_eof_is_transient() {
118        assert_eq!(
119            classify(&SessionError::Map(MapError::ServerError(0xC0))),
120            Disposition::Permanent
121        );
122        assert_eq!(classify(&SessionError::Map(MapError::UnexpectedEof)), Disposition::Transient);
123    }
124
125    #[test]
126    fn backoff_doubles_from_initial_capped_and_bounded() {
127        let delays: Vec<_> =
128            backoff(Duration::from_millis(500), Duration::from_secs(2), 5).collect();
129        // 500ms → 1s → 2s (cap) → 2s (cap); exactly max_attempts - 1 = 4 gaps.
130        assert_eq!(
131            delays.as_slice(),
132            [
133                Duration::from_millis(500),
134                Duration::from_secs(1),
135                Duration::from_secs(2),
136                Duration::from_secs(2),
137            ]
138        );
139    }
140
141    #[test]
142    fn backoff_is_empty_for_single_attempt() {
143        assert_eq!(backoff(Duration::from_millis(10), Duration::from_secs(1), 1).count(), 0);
144    }
145}
imsg_session/retry.rs

imsg_session/
retry.rs
