fix(proxy): preserve upstream error.code/type/param on 4xx + cross-wire translation (#322)

crates/aisix-gateway/src/bridge.rs

@@ -23,6 +23,63 @@ use std::time::Duration;
 
 use crate::chat::{ChatChunk, ChatFormat, ChatResponse, EmbeddingRequest, EmbeddingResponse};
 
+/// Maximum number of bytes read from an upstream error response body
+/// before attempting JSON envelope parse. Bounds memory and parser cost
+/// when an upstream returns something pathological (an HTML error page
+/// from a fronting WAF, or an unexpectedly large debug dump).
+pub const MAX_UPSTREAM_ERROR_BODY_BYTES: usize = 64 * 1024;
+
+/// Maximum length of the human-readable `message` string carried inside
+/// [`BridgeError::UpstreamStatus`]. The full body is parsed into
+/// [`UpstreamErrorView`] when JSON-shaped; the truncated string is the
+/// fallback shown to clients when parsing fails.
+pub const MAX_UPSTREAM_ERROR_MESSAGE_BYTES: usize = 1024;
+
+/// Which wire format the upstream that produced this error speaks. The
+/// envelope-rendering layer uses this together with [`UpstreamErrorView`]
+/// to decide whether the upstream `kind` / `code` can be forwarded
+/// verbatim or needs translation to the client's wire shape.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum UpstreamWire {
+    /// OpenAI-compatible envelope: `{error:{message,type,code,param}}`.
+    OpenAI,
+    /// Anthropic envelope: `{type:"error",error:{type,message}}`.
+    Anthropic,
+    /// Azure OpenAI envelope: OpenAI-like with `error.inner_error.code`
+    /// quirks for content policy violations.
+    AzureOpenAI,
+    /// AWS Bedrock structured error from the strongly-typed SDK; `kind`
+    /// carries the AWS exception code (e.g. `"ThrottlingException"`).
+    Bedrock,
+    /// Vertex AI envelope: `{error:{code:int,message,status}}` where
+    /// `status` is the canonical gRPC code string.
+    Vertex,
+    /// Wire format unknown / not applicable (tests, synthesised errors,
+    /// the legacy convenience constructors). Renders as the generic
+    /// `upstream_error` envelope with no translation attempt.
+    Unknown,
+}
+
+/// Structured view of an upstream error envelope, populated by each
+/// bridge after best-effort parsing of its provider's known shape.
+/// `None` everywhere means parsing failed (non-JSON body, malformed
+/// JSON, or unfamiliar envelope shape); callers fall back to the
+/// truncated raw message on [`BridgeError::UpstreamStatus::message`].
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct UpstreamErrorView {
+    /// Provider-native error-type token, unchanged from the upstream
+    /// envelope (e.g. Anthropic `"rate_limit_error"`, OpenAI
+    /// `"rate_limit_exceeded"`, Bedrock `"ThrottlingException"`).
+    pub kind: Option<String>,
+    /// Human-readable upstream message, post-parse.
+    pub message: Option<String>,
+    /// OpenAI envelope only. Other providers populate via the
+    /// translation table at render time.
+    pub code: Option<String>,
+    /// OpenAI envelope only.
+    pub param: Option<String>,
+}
+
 /// Context carried through the whole request lifecycle.
 ///
 /// The proxy layer fills this in after it has authenticated the request
@@ -77,10 +134,21 @@ pub enum BridgeError {
     /// backoff hints. Bridges that cannot parse the header (or where
     /// the header is absent) leave this `None`; the cooldown layer
     /// falls back to its configured default in that case.
+    /// `message` is a best-effort human-readable string for logs and
+    /// the fallback envelope when [`parsed`] is `None`. When [`parsed`]
+    /// is `Some`, the envelope-rendering layer (`error_translate`) uses
+    /// the structured fields and [`wire`] to produce a client-shape
+    /// envelope; `message` is kept around for logs and as a
+    /// last-resort fallback if a parsed field is missing.
     #[error("upstream returned HTTP {status}: {message}")]
     UpstreamStatus {
         status: u16,
         message: String,
+        /// Boxed to keep [`BridgeError`] small enough that
+        /// `Result<_, ProxyError>` doesn't trip `clippy::result_large_err`
+        /// once the four optional envelope fields are added.
+        parsed: Option<Box<UpstreamErrorView>>,
+        wire: UpstreamWire,
         retry_after: Option<Duration>,
     },
     #[error("upstream returned an unparseable body: {0}")]
@@ -94,18 +162,21 @@ pub enum BridgeError {
 }
 
 impl BridgeError {
-    /// Convenience constructor for upstream status errors when no
-    /// `Retry-After` is available. Keeps existing call sites readable.
+    /// Convenience constructor for synthesised upstream errors (tests,
+    /// cooldown fixtures) where no real upstream envelope is involved.
+    /// Sets [`UpstreamWire::Unknown`] and `parsed: None`.
     pub fn upstream_status(status: u16, message: impl Into<String>) -> Self {
         Self::UpstreamStatus {
             status,
             message: message.into(),
+            parsed: None,
+            wire: UpstreamWire::Unknown,
             retry_after: None,
         }
     }
 
-    /// Convenience constructor for upstream status errors that carry
-    /// a parsed `Retry-After` hint.
+    /// Convenience constructor for synthesised upstream errors that
+    /// carry a parsed `Retry-After` hint. See [`upstream_status`].
     pub fn upstream_status_with_retry_after(
         status: u16,
         message: impl Into<String>,
@@ -114,6 +185,8 @@ impl BridgeError {
         Self::UpstreamStatus {
             status,
             message: message.into(),
+            parsed: None,
+            wire: UpstreamWire::Unknown,
             retry_after,
         }
     }
@@ -139,6 +212,138 @@ pub fn parse_retry_after(headers: &http::HeaderMap) -> Option<Duration> {
     Some(Duration::from_secs(seconds))
 }
 
+/// Drain an upstream error response (capped at
+/// [`MAX_UPSTREAM_ERROR_BODY_BYTES`]) and produce a
+/// [`BridgeError::UpstreamStatus`] with a best-effort parsed view of
+/// the envelope.
+///
+/// The `parse` closure runs only when the response declares an
+/// `application/json` content-type — this guards against fronting WAFs
+/// or load balancers returning HTML error pages that would otherwise be
+/// fed to a JSON parser and either fail expensively or surface
+/// nonsensical fragments.
+///
+/// `parse` returning `None` is treated as "envelope shape unknown"; the
+/// fallback in that case is the truncated raw body string in
+/// [`BridgeError::UpstreamStatus::message`], same as for non-JSON
+/// bodies.
+pub async fn capture_upstream_error_http(
+    status: http::StatusCode,
+    resp: reqwest::Response,
+    wire: UpstreamWire,
+    parse: impl FnOnce(&[u8]) -> Option<UpstreamErrorView>,
+) -> BridgeError {
+    let retry_after = parse_retry_after(resp.headers());
+    let content_type = resp
+        .headers()
+        .get(http::header::CONTENT_TYPE)
+        .and_then(|v| v.to_str().ok())
+        .map(str::to_ascii_lowercase);
+    let body = read_body_capped(resp, MAX_UPSTREAM_ERROR_BODY_BYTES).await;
+    let parsed = content_type
+        .as_deref()
+        .map(content_type_is_json)
+        .unwrap_or(false)
+        .then(|| parse(&body))
+        .flatten()
+        // Truncate every parsed string at the same cap as the outer
+        // `message`. Otherwise a hostile or buggy upstream emitting a
+        // 60 KB `error.message` / `error.code` / `error.type` /
+        // `error.param` would reach the customer envelope verbatim —
+        // the cap exists exactly to prevent that. AWS exception codes
+        // / Anthropic types / OpenAI codes are bounded vocabulary in
+        // practice but the cap applies defensively.
+        .map(|mut v| {
+            let cap = MAX_UPSTREAM_ERROR_MESSAGE_BYTES;
+            v.message = v.message.map(|m| truncate_lossy(&m, cap));
+            v.kind = v.kind.map(|k| truncate_lossy(&k, cap));
+            v.code = v.code.map(|c| truncate_lossy(&c, cap));
+            v.param = v.param.map(|p| truncate_lossy(&p, cap));
+            v
+        });
+    let message = parsed
+        .as_ref()
+        .and_then(|v| v.message.clone())
+        .unwrap_or_else(|| String::from_utf8_lossy(&body).into_owned());
+    BridgeError::UpstreamStatus {
+        status: status.as_u16(),
+        message: truncate_lossy(&message, MAX_UPSTREAM_ERROR_MESSAGE_BYTES),
+        parsed: parsed.map(Box::new),
+        wire,
+        retry_after,
+    }
+}
+
+/// Read the response body, stopping after `limit` bytes. Used to bound
+/// upstream-error parsing cost regardless of `Content-Length`. Errors
+/// during read surface as an empty buffer — the caller falls through
+/// to a parse-failure path and emits the generic `upstream_error`
+/// envelope, which matches the pre-fix behaviour for that edge.
+///
+/// Public so non-OpenAI / non-Anthropic bridges (Vertex, Azure) can
+/// enforce the same cap when they need a custom parse path (e.g.
+/// extracting only `kind` from the upstream envelope while suppressing
+/// the `message` for operator-taxonomy redaction).
+pub async fn read_body_capped(resp: reqwest::Response, limit: usize) -> bytes::Bytes {
+    use futures::StreamExt;
+    let mut buf = bytes::BytesMut::with_capacity(limit.min(16 * 1024));
+    let mut stream = resp.bytes_stream();
+    // Continue draining the stream past `limit` so the underlying
+    // hyper connection can be returned to reqwest's keep-alive pool.
+    // Stopping the iteration mid-stream taints the connection and
+    // forces a new TCP handshake on the next upstream call — a real
+    // cost when an upstream is flapping and producing a burst of
+    // error responses. The extra reads only discard bytes; memory
+    // stays bounded by `limit`.
+    while let Some(chunk) = stream.next().await {
+        let Ok(chunk) = chunk else { break };
+        if buf.len() >= limit {
+            continue;
+        }
+        let remaining = limit - buf.len();
+        let take = chunk.len().min(remaining);
+        buf.extend_from_slice(&chunk[..take]);
+    }
+    buf.freeze()
+}
+
+/// Content-Type token starts with `application/json` (RFC 7231 §3.1.1.1
+/// allows a trailing `; charset=…` parameter, so a prefix match is the
+/// right shape here — exact equality misses `application/json; charset=utf-8`).
+///
+/// Public so non-OpenAI / non-Anthropic bridges (Vertex, Azure) can
+/// apply the same JSON-only guard when they need a custom parse path
+/// that doesn't route through [`capture_upstream_error_http`].
+pub fn content_type_is_json(ct: &str) -> bool {
+    let ct = ct.trim_start();
+    ct.starts_with("application/json")
+}
+
+/// Convenience: read the `Content-Type` header from a [`reqwest::Response`]
+/// and decide whether it's `application/json` per [`content_type_is_json`].
+/// Returns `false` when the header is missing or non-ASCII.
+pub fn response_is_json(resp: &reqwest::Response) -> bool {
+    resp.headers()
+        .get(http::header::CONTENT_TYPE)
+        .and_then(|v| v.to_str().ok())
+        .map(|ct| content_type_is_json(&ct.to_ascii_lowercase()))
+        .unwrap_or(false)
+}
+
+/// Truncate a string to at most `max` bytes, splitting only on a UTF-8
+/// boundary. Appends an ellipsis when truncation occurred so log
+/// readers can tell the message was cut.
+fn truncate_lossy(s: &str, max: usize) -> String {
+    if s.len() <= max {
+        return s.to_string();
+    }
+    let mut end = max;
+    while end > 0 && !s.is_char_boundary(end) {
+        end -= 1;
+    }
+    format!("{}…", &s[..end])
+}
+
 impl BridgeError {
     /// Stable HTTP status mapping. The proxy layer uses this to build
     /// its OpenAI-compatible `{error:{message,type,...}}` envelope.

crates/aisix-gateway/src/lib.rs

@@ -25,7 +25,11 @@ pub mod chat;
 pub mod hub;
 pub mod sse;
 
-pub use bridge::{parse_retry_after, Bridge, BridgeContext, BridgeError, ChatChunkStream};
+pub use bridge::{
+    capture_upstream_error_http, content_type_is_json, parse_retry_after, read_body_capped,
+    response_is_json, Bridge, BridgeContext, BridgeError, ChatChunkStream, UpstreamErrorView,
+    UpstreamWire, MAX_UPSTREAM_ERROR_BODY_BYTES, MAX_UPSTREAM_ERROR_MESSAGE_BYTES,
+};
 pub use chat::{
     ChatChunk, ChatDelta, ChatFormat, ChatMessage, ChatResponse, EmbeddingObject, EmbeddingRequest,
     EmbeddingResponse, EmbeddingUsage, FinishReason, Role, UsageStats,

crates/aisix-provider-anthropic/src/bridge.rs

@@ -131,21 +131,44 @@ fn upstream_model(ctx: &BridgeContext) -> Result<&str, BridgeError> {
 }
 
 async fn map_http_error(status: StatusCode, resp: reqwest::Response) -> BridgeError {
-    let retry_after = aisix_gateway::parse_retry_after(resp.headers());
-    let message = resp.text().await.unwrap_or_default();
-    BridgeError::upstream_status_with_retry_after(
-        status.as_u16(),
-        truncate(&message, 1024),
-        retry_after,
+    aisix_gateway::capture_upstream_error_http(
+        status,
+        resp,
+        aisix_gateway::UpstreamWire::Anthropic,
+        parse_anthropic_error_envelope,
     )
+    .await
 }
 
-fn truncate(s: &str, n: usize) -> String {
-    if s.len() <= n {
-        s.to_string()
-    } else {
-        format!("{}…", &s[..n])
+/// Parse the Anthropic error envelope:
+///
+/// ```json
+/// {"type": "error", "error": {"type": "...", "message": "..."}}
+/// ```
+///
+/// Anthropic does not carry `code` or `param` fields — those stay
+/// `None`. The translation table at render time derives an OpenAI
+/// `code` from `kind` when crossing wire formats.
+///
+/// Reference: <https://docs.anthropic.com/en/api/errors>
+fn parse_anthropic_error_envelope(body: &[u8]) -> Option<aisix_gateway::UpstreamErrorView> {
+    #[derive(serde::Deserialize)]
+    struct Outer {
+        error: Inner,
     }
+    #[derive(serde::Deserialize)]
+    struct Inner {
+        #[serde(rename = "type")]
+        kind: Option<String>,
+        message: Option<String>,
+    }
+    let outer: Outer = serde_json::from_slice(body).ok()?;
+    Some(aisix_gateway::UpstreamErrorView {
+        kind: outer.error.kind,
+        message: outer.error.message,
+        code: None,
+        param: None,
+    })
 }
 
 async fn with_deadline<T, F>(
@@ -371,10 +394,10 @@ mod tests {
         let server = MockServer::start().await;
         Mock::given(method("POST"))
             .and(path("/v1/messages"))
-            .respond_with(
-                ResponseTemplate::new(400)
-                    .set_body_string(r#"{"error":{"type":"invalid_request","message":"bad"}}"#),
-            )
+            .respond_with(ResponseTemplate::new(400).set_body_raw(
+                r#"{"error":{"type":"invalid_request","message":"bad"}}"#.as_bytes(),
+                "application/json",
+            ))
             .mount(&server)
             .await;
 
@@ -383,10 +406,19 @@ mod tests {
         let err = bridge.chat(&req(), &ctx).await.unwrap_err();
         match err {
             BridgeError::UpstreamStatus {
-                status, message, ..
+                status,
+                message,
+                parsed,
+                ..
             } => {
                 assert_eq!(status, 400);
-                assert!(message.contains("invalid_request"));
+                // After #322: bridge parses Anthropic envelope into a
+                // structured view; `message` is now the upstream's
+                // `error.message`, not the raw JSON body.
+                assert_eq!(message, "bad");
+                let parsed = parsed.expect("envelope parsed");
+                assert_eq!(parsed.kind.as_deref(), Some("invalid_request"));
+                assert_eq!(parsed.message.as_deref(), Some("bad"));
             }
             other => panic!("unexpected: {other:?}"),
         }