gateway_core/
error.rs

1use serde::{Deserialize, Serialize};
2
3/// The longest upstream diagnostic a [`ProviderError`] carries, before the
4/// truncation marker.
5///
6/// An upstream failure body is attacker-influenced and arrives over the
7/// network, so the diagnostic built from it is bounded *here* rather than only
8/// at the edge that read it: a provider error is logged, counted, and rendered
9/// into a response, and every one of those is a place an unbounded body would
10/// end up. The transport already truncates what it reads (`max_error_bytes`,
11/// 64 KiB), which makes this the second bound rather than the only one — and
12/// the one that holds for any caller, including a future non-HTTP one.
13pub const MAX_DIAGNOSTIC_BYTES: usize = 4 * 1024;
14
15/// Appended to a diagnostic that hit [`MAX_DIAGNOSTIC_BYTES`], so a reader can
16/// tell a truncated message from a short one.
17pub const DIAGNOSTIC_TRUNCATION_MARKER: &str = "… [truncated]";
18
19#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
20pub struct DependencyFailure {
21    pub provider: String,
22    pub status: Option<u16>,
23    pub message: String,
24}
25
26#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
27pub enum ProviderError {
28    #[error("invalid provider request: {0}")]
29    InvalidRequest(String),
30    #[error("context window exceeded: {0}")]
31    ContextWindowExceeded(String),
32    #[error("surface unsupported by provider: {0}")]
33    Unsupported(String),
34    #[error("upstream model unavailable")]
35    ModelUnavailable(Vec<DependencyFailure>),
36    #[error("provider dependency failed")]
37    Dependency(Vec<DependencyFailure>),
38    #[error("provider stream was invalid: {0}")]
39    InvalidStream(String),
40    #[error("provider stream was rate limited: {0}")]
41    RateLimitedStream(String),
42    #[error("all provider circuits are open")]
43    AllCircuitsOpen(Vec<String>),
44}
45
46impl ProviderError {
47    pub fn code(&self) -> &'static str {
48        match self {
49            Self::InvalidRequest(_) => "invalid_request",
50            Self::ContextWindowExceeded(_) => "context_window_exceeded",
51            Self::Unsupported(_) => "unsupported",
52            Self::ModelUnavailable(_) => "model_unavailable",
53            Self::Dependency(_) => "provider_dependency_failed",
54            Self::InvalidStream(_) => "invalid_stream",
55            Self::RateLimitedStream(_) => "provider_rate_limited",
56            Self::AllCircuitsOpen(_) => "all_provider_circuits_open",
57        }
58    }
59
60    pub fn from_upstream(provider: impl Into<String>, status: u16, body: &str) -> Self {
61        let provider = provider.into();
62        let message = bounded(extract_message(body));
63        if is_context_length_error(body) || is_context_length_error(&message) {
64            return Self::ContextWindowExceeded(message);
65        }
66        let failure = DependencyFailure {
67            provider,
68            status: Some(status),
69            message,
70        };
71        if status == 404 {
72            Self::ModelUnavailable(vec![failure])
73        } else if (400..500).contains(&status) && status != 429 {
74            Self::InvalidRequest(failure.message)
75        } else {
76            Self::Dependency(vec![failure])
77        }
78    }
79
80    /// A malformed provider stream, with its diagnostic bounded: the message
81    /// comes from the provider's own payload, so it is untrusted input the same
82    /// way a failure body is.
83    pub fn invalid_stream(message: impl Into<String>) -> Self {
84        Self::InvalidStream(bounded(message.into()))
85    }
86
87    /// A rate-limited provider stream, with its diagnostic bounded for the same
88    /// reason as [`Self::invalid_stream`].
89    pub fn rate_limited_stream(message: impl Into<String>) -> Self {
90        Self::RateLimitedStream(bounded(message.into()))
91    }
92
93    pub fn transport(provider: impl Into<String>, message: impl Into<String>) -> Self {
94        Self::Dependency(vec![DependencyFailure {
95            provider: provider.into(),
96            status: None,
97            message: bounded(message.into()),
98        }])
99    }
100
101    pub fn is_retryable(&self) -> bool {
102        match self {
103            Self::Dependency(failures) => {
104                !failures.is_empty()
105                    && failures.iter().all(|failure| {
106                        failure
107                            .status
108                            .is_none_or(|status| status == 429 || status >= 500)
109                    })
110            }
111            Self::ModelUnavailable(_) => true,
112            _ => false,
113        }
114    }
115
116    pub fn affects_provider_health(&self) -> bool {
117        matches!(self, Self::Dependency(_)) && self.is_retryable()
118    }
119
120    pub fn is_stream_rate_limited(&self) -> bool {
121        matches!(self, Self::RateLimitedStream(_))
122    }
123
124    pub fn is_credential_rate_limited(&self) -> bool {
125        match self {
126            Self::RateLimitedStream(_) => true,
127            Self::Dependency(failures) => {
128                failures.iter().any(|failure| failure.status == Some(429))
129            }
130            _ => false,
131        }
132    }
133}
134
135/// Cut a diagnostic down to [`MAX_DIAGNOSTIC_BYTES`] on a character boundary.
136///
137/// The cut is by bytes rather than characters because what is being bounded is
138/// the memory and the log line, not the glyph count.
139fn bounded(mut message: String) -> String {
140    if message.len() <= MAX_DIAGNOSTIC_BYTES {
141        return message;
142    }
143    let mut cut = MAX_DIAGNOSTIC_BYTES;
144    while !message.is_char_boundary(cut) {
145        cut -= 1;
146    }
147    message.truncate(cut);
148    message.push_str(DIAGNOSTIC_TRUNCATION_MARKER);
149    message
150}
151
152fn extract_message(body: &str) -> String {
153    serde_json::from_str::<serde_json::Value>(body)
154        .ok()
155        .and_then(|value| {
156            value
157                .pointer("/error/message")
158                .or_else(|| value.get("message"))
159                .and_then(serde_json::Value::as_str)
160                .map(str::to_owned)
161        })
162        .unwrap_or_else(|| body.to_owned())
163}
164
165fn is_context_length_error(text: &str) -> bool {
166    let text = text.to_ascii_lowercase();
167    [
168        "context_length_exceeded",
169        "context length",
170        "context window",
171        "prompt is too long",
172        "prompt too long",
173        "maximum number of tokens",
174        "too many tokens",
175        "maximum prompt length",
176    ]
177    .iter()
178    .any(|signal| text.contains(signal))
179}
180
181/// Recognize only explicit provider rate-limit markers in an SSE JSON payload.
182pub fn is_rate_limit_payload(value: &serde_json::Value) -> bool {
183    let error = value.get("error");
184    let error_shaped =
185        error.is_some() || value.get("type").and_then(serde_json::Value::as_str) == Some("error");
186    if !error_shaped {
187        return false;
188    }
189    let status_is_429 = [value.get("status"), value.pointer("/error/status")]
190        .into_iter()
191        .flatten()
192        .any(|status| {
193            status.as_u64() == Some(429) || status.as_str().is_some_and(|status| status == "429")
194        });
195    if status_is_429 {
196        return true;
197    }
198    [
199        value
200            .pointer("/error/type")
201            .and_then(serde_json::Value::as_str),
202        value.pointer("/error/code").and_then(|code| {
203            code.as_str()
204                .or_else(|| (code.as_u64() == Some(429)).then_some("429"))
205        }),
206        value.pointer("/type").and_then(serde_json::Value::as_str),
207        value.pointer("/code").and_then(serde_json::Value::as_str),
208    ]
209    .into_iter()
210    .flatten()
211    .any(|signal| signal.contains("rate_limit") || signal == "429")
212}
213
214#[cfg(test)]
215mod tests {
216    use super::*;
217
218    #[test]
219    fn normalizes_context_limit_signals_without_retrying_or_degrading_health() {
220        for body in [
221            r#"{"error":{"code":"context_length_exceeded","message":"too long"}}"#,
222            r#"{"error":{"message":"prompt is too long: 250000 tokens"}}"#,
223            r#"{"message":"input exceeds the maximum number of tokens"}"#,
224        ] {
225            let error = ProviderError::from_upstream("provider", 400, body);
226            assert!(matches!(error, ProviderError::ContextWindowExceeded(_)));
227            assert!(!error.is_retryable());
228            assert!(!error.affects_provider_health());
229        }
230    }
231
232    #[test]
233    fn rate_limits_and_server_failures_retry_and_affect_health() {
234        for status in [429, 500, 502, 503, 599] {
235            let error = ProviderError::from_upstream("provider", status, "upstream unavailable");
236            assert!(matches!(error, ProviderError::Dependency(_)));
237            assert!(error.is_retryable(), "status {status}");
238            assert!(error.affects_provider_health(), "status {status}");
239        }
240        let transport = ProviderError::transport("provider", "timeout");
241        assert!(transport.is_retryable());
242        assert!(transport.affects_provider_health());
243    }
244
245    #[test]
246    fn authentication_and_other_client_failures_are_permanent_but_not_unhealthy() {
247        for status in [400, 401, 403, 422] {
248            let error = ProviderError::from_upstream("provider", status, "invalid request");
249            assert!(matches!(error, ProviderError::InvalidRequest(_)));
250            assert!(!error.is_retryable(), "status {status}");
251            assert!(!error.affects_provider_health(), "status {status}");
252        }
253    }
254
255    #[test]
256    fn missing_model_fails_over_without_marking_provider_unhealthy() {
257        let error = ProviderError::from_upstream("foundry", 404, "missing deployment");
258        assert!(matches!(error, ProviderError::ModelUnavailable(_)));
259        assert!(error.is_retryable());
260        assert!(!error.affects_provider_health());
261    }
262
263    /// A provider that answers a failure with megabytes of HTML must not put
264    /// megabytes into a log line, a metric label, or a response body.
265    #[test]
266    fn upstream_diagnostics_are_bounded_however_large_the_body_is() {
267        let body = "x".repeat(4 * MAX_DIAGNOSTIC_BYTES);
268        for (status, message) in [
269            (
270                400,
271                diagnostic(&ProviderError::from_upstream("p", 400, &body)),
272            ),
273            (
274                404,
275                diagnostic(&ProviderError::from_upstream("p", 404, &body)),
276            ),
277            (
278                503,
279                diagnostic(&ProviderError::from_upstream("p", 503, &body)),
280            ),
281        ] {
282            assert!(
283                message.len() <= MAX_DIAGNOSTIC_BYTES + DIAGNOSTIC_TRUNCATION_MARKER.len(),
284                "status {status} carried a {}-byte diagnostic",
285                message.len()
286            );
287            assert!(
288                message.ends_with(DIAGNOSTIC_TRUNCATION_MARKER),
289                "status {status}"
290            );
291        }
292        // The JSON path is bounded too: the message a provider nests is as
293        // attacker-influenced as the body around it.
294        let nested = format!(r#"{{"error":{{"message":"{}"}}}}"#, "y".repeat(64 * 1024));
295        let error = ProviderError::from_upstream("p", 500, &nested);
296        assert!(
297            diagnostic(&error).len() <= MAX_DIAGNOSTIC_BYTES + DIAGNOSTIC_TRUNCATION_MARKER.len()
298        );
299        // So is a transport diagnostic, which is built from an error string
300        // rather than a body but reaches the same places.
301        let transport = ProviderError::transport("p", "z".repeat(1024 * 1024));
302        assert!(
303            diagnostic(&transport).len()
304                <= MAX_DIAGNOSTIC_BYTES + DIAGNOSTIC_TRUNCATION_MARKER.len()
305        );
306    }
307
308    /// Truncation cuts bytes, so it has to land on a character boundary: a
309    /// multi-byte character straddling the bound would panic the truncation.
310    #[test]
311    fn truncation_never_splits_a_character() {
312        // Three bytes each, so the 4096-byte bound falls mid-character.
313        let body = "€".repeat(MAX_DIAGNOSTIC_BYTES);
314        let message = diagnostic(&ProviderError::from_upstream("p", 500, &body));
315        assert!(message.len() <= MAX_DIAGNOSTIC_BYTES + DIAGNOSTIC_TRUNCATION_MARKER.len());
316        assert!(
317            message
318                .trim_end_matches(DIAGNOSTIC_TRUNCATION_MARKER)
319                .chars()
320                .all(|character| character == '€')
321        );
322    }
323
324    /// A body short enough to keep is kept whole: bounding must not silently
325    /// mangle the diagnostics operators actually read.
326    #[test]
327    fn short_diagnostics_are_untouched() {
328        let error = ProviderError::from_upstream("p", 500, "upstream unavailable");
329        assert_eq!(diagnostic(&error), "upstream unavailable");
330    }
331
332    fn diagnostic(error: &ProviderError) -> String {
333        match error {
334            ProviderError::InvalidRequest(message)
335            | ProviderError::ContextWindowExceeded(message)
336            | ProviderError::Unsupported(message)
337            | ProviderError::InvalidStream(message)
338            | ProviderError::RateLimitedStream(message) => message.clone(),
339            ProviderError::ModelUnavailable(failures) | ProviderError::Dependency(failures) => {
340                failures
341                    .iter()
342                    .map(|failure| failure.message.clone())
343                    .collect::<Vec<_>>()
344                    .join("\n")
345            }
346            ProviderError::AllCircuitsOpen(providers) => providers.join("\n"),
347        }
348    }
349
350    #[test]
351    fn recognizes_only_explicit_stream_rate_limit_shapes() {
352        for body in [
353            r#"{"type":"error","error":{"type":"rate_limit_error"}}"#,
354            r#"{"error":{"code":"rate_limit_exceeded"}}"#,
355            r#"{"error":{"code":429}}"#,
356            r#"{"error":{"status":429}}"#,
357        ] {
358            let value: serde_json::Value = serde_json::from_str(body).unwrap();
359            assert!(is_rate_limit_payload(&value), "{body}");
360        }
361        for body in [
362            r#"{"error":{"type":"overloaded_error"}}"#,
363            r#"{"error":{"message":"try again later"}}"#,
364            r#"{"status":500}"#,
365            r#"{"type":"rate_limits.updated","rate_limits":{"requests":10}}"#,
366        ] {
367            let value: serde_json::Value = serde_json::from_str(body).unwrap();
368            assert!(!is_rate_limit_payload(&value), "{body}");
369        }
370    }
371}
gateway_core/error.rs

gateway_core/
error.rs
