sui_rpc/client/ledger_streams/
retry.rs

1use std::collections::hash_map::RandomState;
2use std::error::Error as _;
3use std::hash::BuildHasher;
4use std::time::Duration;
5
6use tonic::Code;
7use tonic::Status;
8
9use super::observability::LedgerStreamEvent;
10use super::observability::LedgerStreamFamily;
11use super::observability::LedgerStreamObservability;
12use super::observability::LedgerStreamOperation;
13use super::observability::LedgerStreamStage;
14use super::types::LedgerStreamConfig;
15
16/// Where an error was encountered during an RPC interaction.
17///
18/// Tonic surfaces errors differently depending on whether the RPC failed while initiating the call
19/// or while reading frames from an already-open response stream.
20#[derive(Clone, Copy, Debug, PartialEq, Eq)]
21pub(super) enum FailurePhase {
22    /// Failed while sending the initial RPC request.
23    Dispatch,
24    /// Failed while reading chunks from the response stream.
25    Body,
26}
27
28/// Classifies failures eligible for automatic retry.
29///
30/// `Unavailable`, `DeadlineExceeded`, `ResourceExhausted`, and `Aborted` are always retried.
31///
32/// `Cancelled` and `Unknown` use two heuristics based on where the error was observed:
33/// - During `Dispatch` (initiating the RPC), server-sent `Cancelled` and `Unknown` are
34///   conservatively treated as terminal. We only retry if Tonic attached an underlying error source
35///   (`status.source().is_some()`), treating that as evidence of a lower-layer transport or connection
36///   failure.
37/// - During `Body` (streaming responses), stream interruptions (such as Envoy, ALB, or NAT gateway
38///   timeouts resetting an HTTP/2 stream) surface through Tonic as `Cancelled` or `Unknown`. Because
39///   `stream.next()` surfaces transport drops and server trailers identically, we treat body-phase
40///   `Cancelled` and `Unknown` as retryable based on the observation phase alone, allowing the stream
41///   to reconnect and resume from the latest cursor.
42///
43/// All other statuses are terminal.
44fn transient(status: &Status, phase: FailurePhase) -> bool {
45    let dispatch_transient = matches!(
46        status.code(),
47        Code::Unavailable | Code::DeadlineExceeded | Code::ResourceExhausted | Code::Aborted
48    );
49    let reset_code = matches!(status.code(), Code::Cancelled | Code::Unknown);
50    let transport_reset = reset_code && (phase == FailurePhase::Body || status.source().is_some());
51    dispatch_transient || transport_reset
52}
53
54pub(super) struct RetryState {
55    family: LedgerStreamFamily,
56    operation: LedgerStreamOperation,
57    consecutive_errors: u32,
58    first_failure: Option<tokio::time::Instant>,
59    started_in: Option<LedgerStreamStage>,
60}
61
62impl RetryState {
63    pub(super) fn new(family: LedgerStreamFamily, operation: LedgerStreamOperation) -> Self {
64        Self {
65            family,
66            operation,
67            consecutive_errors: 0,
68            first_failure: None,
69            started_in: None,
70        }
71    }
72
73    pub(super) fn reset(&mut self, observability: &LedgerStreamObservability) {
74        if self.consecutive_errors == 0 {
75            return;
76        }
77        let consecutive_failures = self.consecutive_errors;
78        self.consecutive_errors = 0;
79        let first_failure = self.first_failure.take();
80        let started_in = self.started_in.take();
81        if let (Some(first_failure), Some(started_in)) = (first_failure, started_in) {
82            observability.emit(|| LedgerStreamEvent::RetryRecovered {
83                family: self.family,
84                operation: self.operation,
85                started_in,
86                consecutive_failures,
87                elapsed: first_failure.elapsed(),
88            });
89        }
90    }
91
92    pub(super) fn retry_delay(
93        &mut self,
94        status: &Status,
95        phase: FailurePhase,
96        config: &LedgerStreamConfig,
97        observability: &LedgerStreamObservability,
98        stage: LedgerStreamStage,
99    ) -> Option<Duration> {
100        if !transient(status, phase) {
101            return None;
102        }
103
104        let attempt = self.consecutive_errors;
105        self.consecutive_errors = self.consecutive_errors.saturating_add(1);
106        if attempt == 0 {
107            let first_failure = observability.start_timer();
108            self.started_in = first_failure.as_ref().map(|_| stage);
109            self.first_failure = first_failure;
110        }
111        let delay = backoff_delay(config, attempt);
112        let consecutive_failures = self.consecutive_errors;
113        observability.emit(|| LedgerStreamEvent::RetryScheduled {
114            family: self.family,
115            operation: self.operation,
116            stage,
117            status: status.clone(),
118            consecutive_failures,
119            delay,
120        });
121        Some(delay)
122    }
123}
124
125pub(super) fn backoff_delay(config: &LedgerStreamConfig, attempt: u32) -> Duration {
126    let multiplier = 2_u32.saturating_pow(attempt);
127    let exponential = config.base_retry_delay.saturating_mul(multiplier);
128    let capped = exponential.min(config.max_retry_delay);
129    capped.saturating_add(random_jitter(config.retry_jitter))
130}
131
132fn random_jitter(maximum: Duration) -> Duration {
133    if maximum.is_zero() {
134        return Duration::ZERO;
135    }
136    let maximum_nanos = u64::try_from(maximum.as_nanos()).unwrap_or(u64::MAX);
137    // Hashing with a fresh `RandomState`, randomly seeded per instance, yields an unpredictable
138    // sample without a `rand` dependency.
139    let sample = RandomState::new().hash_one(());
140    let jitter_nanos = if maximum_nanos == u64::MAX {
141        sample
142    } else {
143        sample % (maximum_nanos + 1)
144    };
145    Duration::from_nanos(jitter_nanos)
146}
147
148#[cfg(test)]
149mod tests {
150    use super::*;
151
152    #[test]
153    fn internal_is_terminal_in_both_phases() {
154        let status = Status::internal("client idle-body watchdog task panicked");
155        assert!(!transient(&status, FailurePhase::Body));
156        assert!(!transient(&status, FailurePhase::Dispatch));
157    }
158
159    #[test]
160    fn server_reported_stream_reset_codes_are_transient_only_mid_body() {
161        for status in [
162            Status::cancelled("server cancelled request"),
163            Status::unknown("server failed request"),
164        ] {
165            assert!(transient(&status, FailurePhase::Body));
166            assert!(!transient(&status, FailurePhase::Dispatch));
167        }
168    }
169
170    #[test]
171    fn transport_sourced_unknown_is_transient_during_dispatch() {
172        let status = Status::from_error(Box::new(std::io::Error::from(
173            std::io::ErrorKind::ConnectionReset,
174        )));
175        assert_eq!(status.code(), Code::Unknown);
176        assert!(status.source().is_some());
177        assert!(transient(&status, FailurePhase::Dispatch));
178    }
179
180    #[test]
181    fn dispatch_set_is_transient_in_both_phases() {
182        let status = Status::unavailable("connection refused");
183        assert!(transient(&status, FailurePhase::Dispatch));
184        assert!(transient(&status, FailurePhase::Body));
185    }
186}
sui_rpc/client/ledger_streams/retry.rs

sui_rpc/client/ledger_streams/
retry.rs
