knafeh 1.1.0

QUIC-based RPC library with Python bindings
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109

use std::time::Duration;

use crate::error::{KnafehError, RpcStatusCode};

/// Policy for retrying failed RPC calls.
#[derive(Debug, Clone)]
pub struct RetryPolicy {
    /// Maximum number of retry attempts (not counting the initial call).
    pub max_retries: u32,
    /// Initial backoff duration before the first retry.
    pub initial_backoff: Duration,
    /// Maximum backoff duration.
    pub max_backoff: Duration,
    /// Multiplier applied to the backoff on each retry.
    pub backoff_multiplier: f64,
    /// Whether to add random jitter (±25%) to the backoff.
    pub jitter: bool,
}

impl RetryPolicy {
    /// Create an exponential backoff policy with the given max retries.
    /// Starts at 100ms, doubles each retry, caps at 10s.
    pub fn exponential(max_retries: u32) -> Self {
        Self {
            max_retries,
            initial_backoff: Duration::from_millis(100),
            max_backoff: Duration::from_secs(10),
            backoff_multiplier: 2.0,
            jitter: true,
        }
    }

    /// No retries — every call is attempted exactly once.
    pub fn none() -> Self {
        Self {
            max_retries: 0,
            initial_backoff: Duration::ZERO,
            max_backoff: Duration::ZERO,
            backoff_multiplier: 1.0,
            jitter: false,
        }
    }

    pub fn with_initial_backoff(mut self, duration: Duration) -> Self {
        self.initial_backoff = duration;
        self
    }

    pub fn with_max_backoff(mut self, duration: Duration) -> Self {
        self.max_backoff = duration;
        self
    }

    pub fn with_jitter(mut self, jitter: bool) -> Self {
        self.jitter = jitter;
        self
    }

    /// Compute the backoff duration for the given retry attempt (0-indexed).
    pub fn backoff_for(&self, attempt: u32) -> Duration {
        let mut backoff =
            self.initial_backoff.as_secs_f64() * self.backoff_multiplier.powi(attempt as i32);

        if backoff > self.max_backoff.as_secs_f64() {
            backoff = self.max_backoff.as_secs_f64();
        }

        if self.jitter {
            // ±25% jitter using a simple deterministic-ish approach.
            // In production you'd use a proper RNG, but we keep dependencies minimal.
            let jitter_factor = 0.75 + (((attempt as f64 * 0.618).fract()) * 0.5);
            backoff *= jitter_factor;
        }

        Duration::from_secs_f64(backoff)
    }

    /// Whether the given RPC status code is retryable.
    pub fn is_retryable_status(code: RpcStatusCode) -> bool {
        matches!(
            code,
            RpcStatusCode::Unavailable | RpcStatusCode::ResourceExhausted
        )
    }

    /// Whether the given error is retryable.
    pub fn is_retryable(err: &KnafehError) -> bool {
        matches!(
            err,
            KnafehError::Transport(_)
                | KnafehError::ConnectionClosed
                | KnafehError::Timeout
                | KnafehError::Service {
                    code: RpcStatusCode::Unavailable,
                    ..
                }
                | KnafehError::Service {
                    code: RpcStatusCode::ResourceExhausted,
                    ..
                }
        )
    }
}

impl Default for RetryPolicy {
    fn default() -> Self {
        Self::none()
    }
}