sui_rpc/client/
unsent_request_retry.rs

1//! Transparent retry of RPCs the server never processed, following gRPC's
2//! transparent retry design (gRFC A6, "Transparent Retries").
3//!
4//! A server or load balancer that drains an HTTP/2 connection sends GOAWAY
5//! carrying the id of the last stream it will process. Streams the client
6//! opened above that id, and streams it tries to open once the GOAWAY has
7//! arrived, fail without the server having processed them (RFC 9113 §6.8);
8//! the same holds for a stream the server resets with REFUSED_STREAM (RFC
9//! 9113 §8.7). Load balancers drain connections routinely (on a maximum
10//! connection age or request count, or during deploys), so a long-lived
11//! client hits this regularly. tonic reports these failures as `Internal` or
12//! `Unavailable` and does not resend the request (hyperium/tonic#2185); the
13//! next request reconnects, but the failed one reaches the caller.
14//!
15//! [`UnsentRequestRetry`] resends such requests the way grpc-go and grpc-java
16//! do. A6 distinguishes two cases, which are retried differently:
17//!
18//! - The request never left the client: hyper cancelled it before writing it
19//!   because its connection closed. This costs the server nothing, so it is
20//!   retried a few times.
21//! - The request reached the server, which rejected it unprocessed: a
22//!   GOAWAY whose last stream id is below the request's stream, or a
23//!   REFUSED_STREAM reset. This puts load on the wire, so it is retried once
24//!   per call.
25//!
26//! As in grpc-java, only a GOAWAY with the NO_ERROR code counts: a GOAWAY
27//! carrying an error code may report a wrong last stream id (Netty servers do
28//! so for HPACK failures, netty/netty#10670), so a request it fails may have
29//! been processed. A request the server may have processed is never resent,
30//! so this retry is safe for every RPC, including transaction execution.
31//!
32//! Also as in A6, the call's deadline (the `grpc-timeout` header set by
33//! [`tonic::Request::set_timeout`]) covers all attempts: it runs from the
34//! first attempt, and each resend carries the remaining time in its
35//! `grpc-timeout` header. The client's response-headers timeout is enforced
36//! by tonic inside the channel and so applies to each attempt.
37
38use std::error::Error as StdError;
39use std::task::Context;
40use std::task::Poll;
41use std::time::Duration;
42
43use futures::future::BoxFuture;
44use http_body_util::BodyExt;
45use http_body_util::Full;
46use tokio::time::Instant;
47use tonic::body::Body;
48use tower::Service;
49use tower::ServiceExt;
50
51use super::BoxError;
52use super::watchdog::parse_grpc_timeout;
53
54/// Retries per call of a request that never left the client. One is enough
55/// for a drained connection: the retry after it reconnects. grpc-java allows
56/// up to 1000, but its reconnects back off exponentially; tonic reconnects
57/// immediately, so a connection that keeps closing before the request is
58/// written would turn a large bound into a burst of back-to-back connections.
59const MAX_NOT_SENT_RETRIES: usize = 3;
60
61/// Sends each request on the channel, resending it when it fails before the
62/// server processed it.
63///
64/// Request bodies are buffered so they can be resent. Every `sui.rpc` RPC
65/// sends a single request message, so no request is streamed.
66#[derive(Clone)]
67pub(super) struct UnsentRequestRetry {
68    channel: tonic::transport::Channel,
69}
70
71impl UnsentRequestRetry {
72    pub(super) fn new(channel: tonic::transport::Channel) -> Self {
73        Self { channel }
74    }
75}
76
77impl Service<http::Request<Body>> for UnsentRequestRetry {
78    type Response = http::Response<Body>;
79    type Error = BoxError;
80    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
81
82    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
83        // The channel's readiness is awaited for every attempt in `call`.
84        Poll::Ready(Ok(()))
85    }
86
87    fn call(&mut self, request: http::Request<Body>) -> Self::Future {
88        let call_deadline =
89            parse_grpc_timeout(request.headers()).map(|timeout| Instant::now() + timeout);
90        let attempts = send_until_processed(self.channel.clone(), request, call_deadline);
91        Box::pin(async move {
92            let Some(call_deadline) = call_deadline else {
93                return attempts.await;
94            };
95            tokio::time::timeout_at(call_deadline, attempts)
96                .await
97                .unwrap_or_else(|_| {
98                    Err(tonic::Status::deadline_exceeded(
99                        "timeout expired before response headers were received",
100                    )
101                    .into())
102                })
103        })
104    }
105}
106
107async fn send_until_processed(
108    channel: tonic::transport::Channel,
109    request: http::Request<Body>,
110    call_deadline: Option<Instant>,
111) -> Result<http::Response<Body>, BoxError> {
112    let (mut parts, body) = request.into_parts();
113    let body = body.collect().await?.to_bytes();
114
115    let mut refused_retried = false;
116    let mut not_sent_retries = 0;
117    loop {
118        let request = http::Request::from_parts(parts.clone(), Body::new(Full::new(body.clone())));
119        let error = match channel.clone().oneshot(request).await {
120            Ok(response) => return Ok(response),
121            Err(error) => error,
122        };
123        match unprocessed_failure(&error) {
124            Some(UnprocessedFailure::NotSent) if not_sent_retries < MAX_NOT_SENT_RETRIES => {
125                not_sent_retries += 1;
126            }
127            Some(UnprocessedFailure::Refused) if !refused_retried => {
128                refused_retried = true;
129            }
130            _ => return Err(error.into()),
131        }
132
133        if let Some(call_deadline) = call_deadline {
134            let remaining = call_deadline.saturating_duration_since(Instant::now());
135            parts.headers.insert(
136                "grpc-timeout",
137                http::HeaderValue::from_str(&format_grpc_timeout(remaining))
138                    .expect("grpc-timeout values are ASCII"),
139            );
140        }
141    }
142}
143
144/// How a request failed without the server processing it.
145#[derive(Debug, PartialEq, Eq)]
146enum UnprocessedFailure {
147    /// hyper cancelled the request before writing it to the connection.
148    NotSent,
149    /// The server rejected the request unprocessed: a NO_ERROR GOAWAY that
150    /// excluded its stream, or a REFUSED_STREAM reset.
151    Refused,
152}
153
154/// Classify `error` by the first h2 or hyper error in its source chain.
155///
156/// hyper does not expose the h2 error kind through its own API, so this
157/// inspects the `h2::Error` in hyper's source chain, which needs the same h2
158/// version as hyper. h2 reports a remote GOAWAY on a stream only for streams
159/// above the GOAWAY's last stream id and for streams opened after it.
160fn unprocessed_failure(error: &(dyn StdError + 'static)) -> Option<UnprocessedFailure> {
161    let mut source = Some(error);
162    while let Some(error) = source {
163        if let Some(h2_error) = error.downcast_ref::<h2::Error>() {
164            let refused = h2_error.is_remote()
165                && match h2_error.reason() {
166                    Some(h2::Reason::NO_ERROR) => h2_error.is_go_away(),
167                    Some(h2::Reason::REFUSED_STREAM) => h2_error.is_reset(),
168                    _ => false,
169                };
170            return refused.then_some(UnprocessedFailure::Refused);
171        }
172        if let Some(hyper_error) = error.downcast_ref::<hyper::Error>()
173            && hyper_error.is_canceled()
174        {
175            return Some(UnprocessedFailure::NotSent);
176        }
177        source = error.source();
178    }
179    None
180}
181
182/// Format `duration` as a `grpc-timeout` header value: at most 8 digits and
183/// a unit, using the most precise unit that fits, as tonic does.
184fn format_grpc_timeout(duration: Duration) -> String {
185    const MAX_VALUE: u128 = 99_999_999;
186    [
187        (duration.as_micros(), 'u'),
188        (duration.as_millis(), 'm'),
189        (u128::from(duration.as_secs()), 'S'),
190        (u128::from(duration.as_secs() / 60), 'M'),
191    ]
192    .into_iter()
193    .find(|(value, _)| *value <= MAX_VALUE)
194    .map(|(value, unit)| format!("{value}{unit}"))
195    .unwrap_or_else(|| format!("{}H", (duration.as_secs() / 3600).min(99_999_999)))
196}
197
198#[cfg(test)]
199mod tests {
200    use super::*;
201
202    #[test]
203    fn grpc_timeout_formatting_round_trips() {
204        for duration in [
205            Duration::ZERO,
206            Duration::from_micros(1),
207            Duration::from_millis(1500),
208            Duration::from_secs(99),
209            Duration::from_secs(100),
210            Duration::from_secs(60 * 60 * 24 * 30),
211        ] {
212            let mut headers = http::HeaderMap::new();
213            headers.insert(
214                "grpc-timeout",
215                format_grpc_timeout(duration).parse().unwrap(),
216            );
217            let parsed = parse_grpc_timeout(&headers).unwrap();
218            assert!(parsed <= duration, "{duration:?} formatted as {parsed:?}");
219            assert!(
220                duration - parsed < Duration::from_secs(60),
221                "{duration:?} lost precision: {parsed:?}"
222            );
223        }
224    }
225}
sui_rpc/client/unsent_request_retry.rs

sui_rpc/client/
unsent_request_retry.rs
