diff --git a/crates/tw-api/msg-codes.txt b/crates/tw-api/msg-codes.txt index 0d752798..a1c07e9c 100644 --- a/crates/tw-api/msg-codes.txt +++ b/crates/tw-api/msg-codes.txt @@ -409,6 +409,7 @@ gw.upstream.aws_profile_expired gw.upstream.aws_token_expired gw.upstream.bedrock_refused gw.upstream.bedrock_refused_unnamed +gw.upstream.body_broken gw.upstream.eventstream_broken gw.upstream.forward_failed gw.upstream.idle_timeout diff --git a/crates/tw-api/src/lib.rs b/crates/tw-api/src/lib.rs index e7383382..9396f4dc 100644 --- a/crates/tw-api/src/lib.rs +++ b/crates/tw-api/src/lib.rs @@ -1749,7 +1749,7 @@ pub struct AttemptView { } /// 发出去之后才失败的一跳报的码(见 [`AttemptView::sent`]):上游没在时限内回话、请求在路上 -/// 断了、流在第一段内容之前报了错。**上游可能已经收下了它、在算了**。 +/// 断了、回答读到一半连接断了、流在第一段内容之前报了错。**上游可能已经收下了它、在算了**。 /// /// 别的 `error` 都没发出去:这家满着(`skipped`)、规则拒绝、格式对不上、发给它的名字对不上、 /// 插件拒绝、转换不了、凭据取不到、签不了名,还有连不上(地址不通、握手失败)—— 那时上游 @@ -1757,6 +1757,7 @@ pub struct AttemptView { pub const FAILED_AFTER_SENDING: &[&str] = &[ "gw.upstream.timeout", "gw.upstream.forward_failed", + "gw.upstream.body_broken", "gw.upstream.stream_opening_error", ]; @@ -6626,6 +6627,7 @@ mod tests { attempt(Aborted, None, Some("gw.request.aborted")), attempt(Estimated, Some(404), None), attempt(Error, None, Some("gw.upstream.timeout")), + attempt(Error, None, Some("gw.upstream.body_broken")), attempt(Error, None, Some("gw.upstream.stream_opening_error")), ] { assert!(a.sent(), "{a:?}"); diff --git a/crates/tw-gateway/src/forward.rs b/crates/tw-gateway/src/forward.rs index 99cfe6e1..fb491ac0 100644 --- a/crates/tw-gateway/src/forward.rs +++ b/crates/tw-gateway/src/forward.rs @@ -152,13 +152,58 @@ pub fn map_reqwest_error(e: reqwest::Error) -> GatewayError { "The upstream could not be reached at {url}. Check the endpoint address, the network \ and the proxy settings." )) + } else if broke_reading(&e) { + // reqwest 把读正文时的一切失败都叫「error decoding response body」,哪怕根本没有 + // 在解码:上游关了连接、连接被重置、HTTP/2 的流被重置,都是这一句。用户照字面去 + // 查「解码」只会走错路,所以单给一个码,说清是连接在回答途中断了,再接上断法 + GatewayError::upstream(msg!( + "gw.upstream.body_broken", detail = match causes(&e) { + c if c.is_empty() => e.to_string(), + c => c, + } => + "The connection to the upstream broke while the answer was being received: {detail}" + )) } else { GatewayError::upstream(msg!( - "gw.upstream.forward_failed", detail = e => "Forwarding failed: {detail}" + "gw.upstream.forward_failed", detail = with_causes(&e) => "Forwarding failed: {detail}" )) } } +/// 读正文的半路断了。reqwest 0.13 里读正文的失败一律是 decode;`json()` 解析失败也是 +/// decode,那一种的原因是 serde 的错误,不是连接断了 +fn broke_reading(e: &reqwest::Error) -> bool { + e.is_decode() && !sources(e).any(|c| c.is::()) +} + +fn sources(e: &reqwest::Error) -> impl Iterator { + std::iter::successors(std::error::Error::source(e), |c| c.source()) +} + +/// 错误本身连同每一层原因。**reqwest 自己只说最外面一层**,真正的原因(「end of file +/// before message length reached」「connection reset by peer」「stream error received: …」) +/// 在 `source()` 里,只报外层的话这一句什么都没说。带着的地址照 `unreachable` 那样打码 +fn with_causes(e: &reqwest::Error) -> String { + let mut top = e.to_string(); + if let Some(u) = e.url() { + top = top.replace(u.as_str(), &tw_secret::redact_url(u.as_str())); + } + match causes(e) { + c if c.is_empty() => top, + c => format!("{top}: {c}"), + } +} + +/// 只要原因那几层,用 `: ` 连起来。**中间套着的 reqwest 错误跳过**:接回开头字节的那条流 +/// (`opening`)把上游的错误又包了一层,它只会把「error decoding response body」再说一遍 +fn causes(e: &reqwest::Error) -> String { + sources(e) + .filter(|c| !c.is::()) + .map(|c| c.to_string()) + .collect::>() + .join(": ") +} + #[cfg(test)] mod tests { use super::*; @@ -488,4 +533,84 @@ mod tests { } assert!(changed > 1_000 && unusual > 50, "{changed} {unusual}"); } + + /// 回了响应头和一段正文,说好的长度没发完就关连接:上游在回答途中走了的样子 + async fn cut_short() -> reqwest::Response { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let at = l.local_addr().unwrap(); + tokio::spawn(async move { + let (mut s, _) = l.accept().await.unwrap(); + let mut buf = [0u8; 4096]; + let _ = s.read(&mut buf).await; + let _ = s + .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 900\r\n\r\ndata: {}\n\n") + .await; + }); + reqwest::Client::new() + .post(format!("http://{at}/v1/responses?key=secret")) + .send() + .await + .unwrap() + } + + async fn first_error(r: reqwest::Response) -> reqwest::Error { + use futures::StreamExt; + let mut s = r.bytes_stream(); + loop { + match s.next().await { + Some(Ok(_)) => continue, + Some(Err(e)) => return e, + None => panic!("the body ended without an error"), + } + } + } + + #[tokio::test] + async fn a_body_cut_short_names_the_break_not_a_decoding_error() { + let err = map_reqwest_error(first_error(cut_short().await).await); + assert_eq!(err.detail.code, "gw.upstream.body_broken"); + let text = &err.detail.text; + assert!( + text.ends_with( + "error reading a body from connection: end of file before message length reached" + ), + "{text}" + ); + assert!(!text.contains("decoding"), "{text}"); + } + + #[tokio::test] + async fn a_break_behind_the_replayed_opening_is_told_once() { + // `opening` 把读过的字节接回流的前面,上游的错误因此又被 reqwest 包了一层 + let r = cut_short().await; + let replayed = http::Response::new(reqwest::Body::wrap_stream(r.bytes_stream())); + let err = map_reqwest_error(first_error(reqwest::Response::from(replayed)).await); + assert_eq!(err.detail.code, "gw.upstream.body_broken"); + assert!(!err.detail.text.contains("decoding"), "{}", err.detail.text); + assert!( + err.detail.text.contains("end of file"), + "{}", + err.detail.text + ); + } + + #[tokio::test] + async fn every_cause_is_spelled_out_and_the_address_redacted() { + // 整包读(`bytes()`)的错误带着地址:query 里可能就是密钥 + let e = cut_short().await.bytes().await.unwrap_err(); + let detail = with_causes(&e); + assert!( + detail.starts_with("error decoding response body for url (http://127.0.0.1:"), + "{detail}" + ); + assert!( + !detail.contains("secret") && !detail.contains("/v1/"), + "{detail}" + ); + assert!( + detail.ends_with("end of file before message length reached"), + "{detail}" + ); + } } diff --git a/crates/tw-gateway/tests/passthrough.rs b/crates/tw-gateway/tests/passthrough.rs index 657fd977..58e7193f 100644 --- a/crates/tw-gateway/tests/passthrough.rs +++ b/crates/tw-gateway/tests/passthrough.rs @@ -1825,6 +1825,14 @@ async fn a_stream_that_dies_midway_says_so_instead_of_just_stopping() { match tokio::time::timeout(Duration::from_secs(2), rx.recv()).await { Ok(Ok(tw_api::Event::RequestFailed { message, .. })) => { assert!(message.text.contains("stream broke"), "{message}"); + // 不是 reqwest 那句「error decoding response body」:说清是连接断了、怎么断的 + assert_eq!(message.code, "gw.upstream.body_broken", "{message}"); + assert!( + message + .text + .contains("end of file before message length reached"), + "{message}" + ); saw_failed = true; break; }