ruccl/global/
policy.rs

1//! Bounded control/data deadlines for the WebSocket collective API.
2use std::time::Duration;
3
4/// Startup retry is safe only before a request is submitted. Failed collectives
5/// are never replayed: discard the communicator and register a fresh group.
6#[derive(Debug, Clone, Copy)]
7pub struct GlobalFailurePolicy {
8    /// Maximum connection attempts per control channel, including the first.
9    pub connect_attempts: u32,
10    /// Per-attempt connection and handshake limit.
11    pub connect_timeout: Duration,
12    /// Initial retry backoff, doubled with a five-second cap.
13    pub retry_backoff: Duration,
14    /// Total bound for a queued control request and its reply.
15    pub request_timeout: Duration,
16    /// Total bound for a collective, including local serialization and sync.
17    pub collective_timeout: Duration,
18}
19impl Default for GlobalFailurePolicy {
20    fn default() -> Self {
21        Self { connect_attempts: 5, connect_timeout: Duration::from_secs(5),
22            retry_backoff: Duration::from_millis(100), request_timeout: Duration::from_secs(30),
23            collective_timeout: Duration::from_secs(300) }
24    }
25}
26impl GlobalFailurePolicy {
27    /// Environment overrides are validated instead of enabling infinite waits.
28    pub fn from_environment() -> Result<Self, String> {
29        fn number(name: &str, default: u64) -> Result<u64, String> {
30            match std::env::var(name) {
31                Ok(value) => value.parse::<u64>().map_err(|_| format!("{name} must be a positive integer")),
32                Err(std::env::VarError::NotPresent) => Ok(default),
33                Err(_) => Err(format!("{name} must be Unicode")),
34            }
35        }
36        let defaults = Self::default();
37        let attempts = number("RUCCL_CONNECT_ATTEMPTS", defaults.connect_attempts as u64)?;
38        if !(1..=64).contains(&attempts) { return Err("RUCCL_CONNECT_ATTEMPTS must be 1..64".into()); }
39        let duration = |name, default: Duration| -> Result<Duration, String> {
40            let value = number(name, default.as_millis() as u64)?;
41            if value == 0 || value > 86_400_000 { return Err(format!("{name} must be 1..86400000 ms")); }
42            Ok(Duration::from_millis(value))
43        };
44        Ok(Self {
45            connect_attempts: attempts as u32,
46            connect_timeout: duration("RUCCL_CONNECT_TIMEOUT_MS", defaults.connect_timeout)?,
47            retry_backoff: duration("RUCCL_RETRY_BACKOFF_MS", defaults.retry_backoff)?,
48            request_timeout: duration("RUCCL_REQUEST_TIMEOUT_MS", defaults.request_timeout)?,
49            collective_timeout: duration("RUCCL_COLLECTIVE_TIMEOUT_MS", defaults.collective_timeout)?,
50        })
51    }
52    pub(crate) fn backoff(self, attempt: u32) -> Duration {
53        self.retry_backoff.saturating_mul(1u32 << attempt.min(16)).min(Duration::from_secs(5))
54    }
55}
56#[cfg(test)]
57mod tests {
58    use super::*;
59    #[test] fn backoff_is_bounded() {
60        let p = GlobalFailurePolicy::default();
61        assert_eq!(p.backoff(0), Duration::from_millis(100));
62        assert_eq!(p.backoff(1), Duration::from_millis(200));
63        assert_eq!(p.backoff(u32::MAX), Duration::from_secs(5));
64    }
65}
ruccl/global/policy.rs

ruccl/global/
policy.rs
