From d695877d55bd81e9a32736639e3f9de49458f321 Mon Sep 17 00:00:00 2001 From: Iain McGinniss <309153+iainmcgin@users.noreply.github.com> Date: Tue, 21 Jul 2026 10:21:39 -0700 Subject: [PATCH] client: classify post-deadline transport failures as deadline_exceeded MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A server enforcing the same deadline aborts the RPC independently, so its RST_STREAM can arrive before the client's own timer fires. The in-flight body read then failed first and the call reported internal, attributing to this client what was really a timeout the caller asked for. Which of the two won came down to timer coarseness and scheduler delay, so the same call could report either code run to run — the gRPC client conformance Timeouts cases failed between one and five times per run on a loaded host, on main as much as on any branch. The missing-grpc-status path already resolved this by deadline rather than by arrival order. This applies the same rule at the three body-read sites through one classifier, and the four places that ask "is the deadline past" now share a single predicate instead of three copies of it, so they cannot drift. Two Connect parse functions and collect_body_bounded had no deadline to consult, so it is threaded in. It is used only to classify; with_deadline still enforces. The transport cause is preserved in the upgraded error. A genuine transport fault that merely happened after the deadline should stay diagnosable, and the code is what callers branch on. Connect gets this too, not just gRPC: connect-timeout-ms produces the identical race, and collect_body_bounded is shared. Fixes #210. Signed-off-by: Iain McGinniss <309153+iainmcgin@users.noreply.github.com> --- .../unreleased/Fixed-20260721-100959.yaml | 20 ++ connectrpc/src/client/mod.rs | 192 +++++++++++++++--- 2 files changed, 180 insertions(+), 32 deletions(-) create mode 100644 .changes/unreleased/Fixed-20260721-100959.yaml diff --git a/.changes/unreleased/Fixed-20260721-100959.yaml b/.changes/unreleased/Fixed-20260721-100959.yaml new file mode 100644 index 00000000..d37369f8 --- /dev/null +++ b/.changes/unreleased/Fixed-20260721-100959.yaml @@ -0,0 +1,20 @@ +kind: Fixed +body: |- + **A Connect or gRPC call whose deadline expires now reports + `deadline_exceeded` when the transport fails first.** A server enforcing the + same deadline aborts the RPC independently, so its RST_STREAM can arrive + before the client's own timer fires. The in-flight body read then failed + first and the call reported `internal`, attributing to the client what was + really a timeout the caller asked for. Which code you got came down to timer + coarseness and scheduler delay, so the same call could report either one run + to run. + + Transport errors surfacing after the deadline has elapsed are now classified + as `deadline_exceeded`, the rule the missing-`grpc-status` path already + applied and which both now share. The transport cause is kept in the + message, so a + genuine transport fault that merely happened after the deadline is still + diagnosable, and a failure before the deadline is still `internal`. [#210] + + [#210]: https://github.com/connectrpc/connect-rust/issues/210 +time: 2026-07-21T10:09:59.192206315-07:00 diff --git a/connectrpc/src/client/mod.rs b/connectrpc/src/client/mod.rs index 416eff52..fe014102 100644 --- a/connectrpc/src/client/mod.rs +++ b/connectrpc/src/client/mod.rs @@ -1382,6 +1382,48 @@ where } } +/// Has the call's deadline passed? +/// +/// The single definition of "past the deadline" for classifying errors that +/// surface around a timeout. `with_deadline` decides when to *stop* waiting; +/// this decides how to *describe* a failure that arrived on its own, which +/// several paths need and which must agree between them. +/// Note for tests: this reads the real clock, while `with_deadline` runs on +/// tokio's. Under `#[tokio::test(start_paused = true)]` virtual time advances +/// and this does not, so a paused-time test that delivers a body *error* and +/// expects the timeout classification will not get it. Use real time for +/// those; paused time is fine when the timer is what you are exercising. +fn deadline_elapsed(deadline: Option) -> bool { + deadline.is_some_and(|d| std::time::Instant::now() >= d) +} + +/// Classify a transport error that surfaced while reading a response body. +/// +/// A server enforcing the same deadline aborts the RPC independently, so its +/// RST_STREAM can beat the local timer: the body read fails first and the +/// call reports a transport fault for what is really a timeout. Which of the +/// two wins is down to timer coarseness and scheduler delay, so the same call +/// can report either code run to run. +/// +/// The missing-`grpc-status` path a few hundred lines below already resolves +/// this by deadline rather than by arrival order; this applies the same rule +/// to the body-read sites, and both now share [`deadline_elapsed`] so they +/// cannot disagree. +/// +/// `internal` is the wrong answer here because it attributes the failure to +/// this client when the cause was a timeout the caller asked for. +fn classify_body_read_error( + context: &str, + error: &dyn std::fmt::Display, + deadline: Option, +) -> ConnectError { + if deadline_elapsed(deadline) { + ConnectError::deadline_exceeded(format!("{context} after the deadline elapsed: {error}")) + } else { + ConnectError::internal(format!("{context}: {error}")) + } +} + /// Response from a unary RPC call. /// /// Contains the decoded response message along with response headers and @@ -1629,7 +1671,9 @@ where .map_err(|e| map_transport_send_error(e, "request failed"))?; match config.protocol { - Protocol::Connect => parse_connect_unary_response(response, config, &options).await, + Protocol::Connect => { + parse_connect_unary_response(response, config, &options, deadline).await + } Protocol::Grpc | Protocol::GrpcWeb => { parse_grpc_unary_response(response, config, &options, deadline).await } @@ -1775,7 +1819,7 @@ where .map_err(|e| map_transport_send_error(e, "GET request failed"))?; // Response format is identical to POST unary Connect. - parse_connect_unary_response(response, config, &options).await + parse_connect_unary_response(response, config, &options, deadline).await }) .await } @@ -1865,6 +1909,7 @@ async fn parse_connect_unary_response( response: Response, config: &ClientConfig, options: &CallOptions, + deadline: Option, ) -> Result>, ConnectError> where B: Body + Send, @@ -1896,7 +1941,7 @@ where .max_message_size .unwrap_or(crate::service::DEFAULT_MAX_MESSAGE_SIZE); - let body = collect_body_bounded(response.into_body(), max_err_body_size) + let body = collect_body_bounded(response.into_body(), max_err_body_size, deadline) .await .map_err(|mut e| { e.set_response_headers(headers.clone()); @@ -2001,7 +2046,7 @@ where // every error out of it carries that response's metadata. let attach = with_response_metadata(&resp_headers, &resp_trailers); - let body = collect_body_bounded(response.into_body(), max_message_size) + let body = collect_body_bounded(response.into_body(), max_message_size, deadline) .await .map_err(&attach)?; @@ -2157,9 +2202,11 @@ where // resets the stream after its error headers) still reports the // status rather than the read failure. if status.is_success() { - return Err(ConnectError::internal(format!( - "failed to read response body: {e}" - ))); + return Err(classify_body_read_error( + "failed to read response body", + &e, + deadline, + )); } break; } @@ -2289,8 +2336,9 @@ where // spec: RST_STREAM CANCEL is upgraded to DeadlineExceeded when the deadline // has elapsed (matching grpc-go and connect-go behavior). if effective_trailers.get("grpc-status").is_none() { - let is_deadline_exceeded = deadline.is_some_and(|d| std::time::Instant::now() >= d); - let mut err = if is_deadline_exceeded { + let mut err = if deadline_elapsed(deadline) { + // Terser than the body-read path's message on purpose: there is + // no transport error here to carry, only a missing trailer. ConnectError::deadline_exceeded("request timeout") } else { ConnectError::internal("gRPC response missing grpc-status trailer") @@ -2746,10 +2794,7 @@ where || (trailers_only && has_status(&self.headers)) { Ok(()) - } else if self - .deadline - .is_some_and(|d| std::time::Instant::now() >= d) - { + } else if deadline_elapsed(self.deadline) { Err(ConnectError::deadline_exceeded("request timeout")) } else if trailers.is_some() { Err(ConnectError::new( @@ -2847,9 +2892,11 @@ where } } Some(Err(e)) => { - return Err(ConnectError::internal(format!( - "error reading response body: {e}" - ))); + return Err(classify_body_read_error( + "failed to read response body", + &e, + deadline, + )); } } } @@ -3053,7 +3100,8 @@ where let stream_max_err_size = max_message_size.unwrap_or(crate::service::DEFAULT_MAX_MESSAGE_SIZE); - let body = collect_body_bounded(response.into_body(), stream_max_err_size).await?; + let body = + collect_body_bounded(response.into_body(), stream_max_err_size, deadline).await?; // Decompress if the server set Content-Encoding. On failure, // fall through to the generic HTTP-status error below. @@ -3977,7 +4025,7 @@ where parse_grpc_unary_response(response, config, &options, deadline).await } Protocol::Connect => { - parse_connect_client_stream_response(response, config, &options).await + parse_connect_client_stream_response(response, config, &options, deadline).await } } }) @@ -4008,6 +4056,7 @@ async fn parse_connect_client_stream_response( response: Response, config: &ClientConfig, options: &CallOptions, + deadline: Option, ) -> Result>, ConnectError> where B: Body + Send, @@ -4029,7 +4078,7 @@ where .max_message_size .unwrap_or(crate::service::DEFAULT_MAX_MESSAGE_SIZE); - let body = collect_body_bounded(response.into_body(), max_err_size).await?; + let body = collect_body_bounded(response.into_body(), max_err_size, deadline).await?; // Decompress if the server set Content-Encoding. On failure, // fall through to the generic HTTP-status error below. @@ -4090,7 +4139,7 @@ where let body_limit = max_msg_size .saturating_add(2 * crate::envelope::HEADER_SIZE) .saturating_add(RESPONSE_BUFFER_TRAILER_SLACK); - let body = collect_body_bounded(response.into_body(), body_limit) + let body = collect_body_bounded(response.into_body(), body_limit, deadline) .await .map_err(with_response_metadata( &resp_headers, @@ -4553,8 +4602,19 @@ fn parse_grpc_error_from_trailers(trailers: &http::HeaderMap) -> Option(body: B, max_size: usize) -> Result +/// `deadline` is the call's absolute deadline, used only to classify a +/// transport failure — see [`classify_body_read_error`]. Pass `None` for a +/// call without one; this function does not enforce the deadline, which +/// [`with_deadline`] does around the whole read. +/// +/// Returns `ResourceExhausted` if the accumulated data exceeds `max_size`, +/// `DeadlineExceeded` if the body fails after the deadline has passed, and +/// `Internal` if it fails before. +async fn collect_body_bounded( + body: B, + max_size: usize, + deadline: Option, +) -> Result where B: Body, B::Error: std::fmt::Display, @@ -4578,9 +4638,11 @@ where } } Some(Err(e)) => { - return Err(ConnectError::internal(format!( - "failed to read response body: {e}", - ))); + return Err(classify_body_read_error( + "failed to read response body", + &e, + deadline, + )); } None => break, } @@ -5806,6 +5868,7 @@ mod tests { response, &config, &CallOptions::default(), + None, ) .await .expect_err("text/html is not a Connect response"); @@ -8087,28 +8150,28 @@ mod tests { #[tokio::test] async fn collect_body_bounded_within_limit() { let body = Full::new(Bytes::from_static(b"hello")); - let got = collect_body_bounded(body, 10).await.unwrap(); + let got = collect_body_bounded(body, 10, None).await.unwrap(); assert_eq!(&got[..], b"hello"); } #[tokio::test] async fn collect_body_bounded_at_exact_limit() { let body = Full::new(Bytes::from_static(b"hello")); - let got = collect_body_bounded(body, 5).await.unwrap(); + let got = collect_body_bounded(body, 5, None).await.unwrap(); assert_eq!(&got[..], b"hello"); } #[tokio::test] async fn collect_body_bounded_exceeds_limit() { let body = Full::new(Bytes::from_static(b"hello world")); - let err = collect_body_bounded(body, 5).await.unwrap_err(); + let err = collect_body_bounded(body, 5, None).await.unwrap_err(); assert_eq!(err.code, ErrorCode::ResourceExhausted); } #[tokio::test] async fn collect_body_bounded_empty() { let body = Full::new(Bytes::new()); - let got = collect_body_bounded(body, 0).await.unwrap(); + let got = collect_body_bounded(body, 0, None).await.unwrap(); assert!(got.is_empty()); } @@ -8121,7 +8184,7 @@ mod tests { tx.send(Ok(Bytes::from_static(b"ccc"))).await.unwrap(); drop(tx); // limit 7: first two frames (6 bytes) fit, third (3 more → 9) exceeds - let err = collect_body_bounded(body, 7).await.unwrap_err(); + let err = collect_body_bounded(body, 7, None).await.unwrap_err(); assert_eq!(err.code, ErrorCode::ResourceExhausted); } @@ -8132,7 +8195,7 @@ mod tests { tx.send(Ok(Bytes::from_static(b"foo"))).await.unwrap(); tx.send(Ok(Bytes::from_static(b"bar"))).await.unwrap(); drop(tx); - let got = collect_body_bounded(body, 10).await.unwrap(); + let got = collect_body_bounded(body, 10, None).await.unwrap(); assert_eq!(&got[..], b"foobar"); } @@ -8142,10 +8205,75 @@ mod tests { let body = ChannelBody { rx }; tx.send(Err(ConnectError::internal("io"))).await.unwrap(); drop(tx); - let err = collect_body_bounded(body, 1024).await.unwrap_err(); + let err = collect_body_bounded(body, 1024, None).await.unwrap_err(); assert_eq!(err.code, ErrorCode::Internal); } + /// The race this fixes: a server enforcing the same deadline aborts the + /// stream, and its RST_STREAM can arrive before the local timer fires. + /// The body read then fails first, and the caller sees a transport fault + /// for what is really a timeout. + #[tokio::test] + async fn a_body_error_after_the_deadline_is_deadline_exceeded() { + let (tx, rx) = tokio::sync::mpsc::channel(4); + let body = ChannelBody { rx }; + tx.send(Err(ConnectError::internal("io"))).await.unwrap(); + drop(tx); + + let elapsed = std::time::Instant::now() - Duration::from_millis(1); + let err = collect_body_bounded(body, 1024, Some(elapsed)) + .await + .unwrap_err(); + assert_eq!(err.code, ErrorCode::DeadlineExceeded); + // The transport cause survives the reclassification: a genuine + // transport fault that merely happened after the deadline is still + // diagnosable. + assert!( + err.message.as_deref().unwrap_or_default().contains("io"), + "got {:?}", + err.message + ); + } + + /// The other half, and the reason this is a deadline check rather than a + /// blanket remap: with the deadline still in the future the same failure + /// is a real transport error and must stay `internal`. + #[tokio::test] + async fn a_body_error_before_the_deadline_stays_internal() { + let (tx, rx) = tokio::sync::mpsc::channel(4); + let body = ChannelBody { rx }; + tx.send(Err(ConnectError::internal("io"))).await.unwrap(); + drop(tx); + + let future = std::time::Instant::now() + Duration::from_secs(60); + let err = collect_body_bounded(body, 1024, Some(future)) + .await + .unwrap_err(); + assert_eq!(err.code, ErrorCode::Internal); + } + + /// A call with no deadline can never be past one, and must not pick up + /// the timeout wording either — `collect_body_bounded_propagates_body_error` + /// already covers the code, so this covers the message. + #[tokio::test] + async fn a_body_error_without_a_deadline_is_not_described_as_a_timeout() { + let (tx, rx) = tokio::sync::mpsc::channel(4); + let body = ChannelBody { rx }; + tx.send(Err(ConnectError::internal("io"))).await.unwrap(); + drop(tx); + + let err = collect_body_bounded(body, 1024, None).await.unwrap_err(); + assert_eq!(err.code, ErrorCode::Internal); + assert!( + !err.message + .as_deref() + .unwrap_or_default() + .contains("deadline"), + "got {:?}", + err.message + ); + } + #[test] fn test_add_streaming_request_headers_grpc() { let config =