diff --git a/litellm-rust/.agents/skills/rust-error/SKILL.md b/litellm-rust/.agents/skills/rust-error/SKILL.md new file mode 100644 index 00000000000..ea44704363c --- /dev/null +++ b/litellm-rust/.agents/skills/rust-error/SKILL.md @@ -0,0 +1,78 @@ +--- +name: rust-error +description: Define, wrap or refactor Rust error types in litellm-rust with thiserror, keeping messages in one place, causes typed, and classification done once +--- + +# Rust errors + +Use this skill when adding an error type or variant, mapping one crate's error into another, or replacing call sites that build errors by hand. `crates/http/src/error.rs` is the reference implementation of everything below + +## Where errors live + +A crate's errors live in `src/error.rs`, defined with `thiserror`, and are re-exported from `lib.rs`. Default to one top-level `Error` per crate with one variant per failure mode. A failure mode is something a caller handles differently, such as phase, status code, retry, or a message Python parity pins exactly. Failures no caller tells apart share one variant + +Keep shared error types minimal and provider-neutral. Provider names, credential types, configuration fields and setup guidance belong in caller-supplied data such as `MissingApiBase { provider: "Azure", guidance: "..." }`, not in dedicated variants or hardcoded shared messages. An exact parity message does not justify a provider-specific variant when caller-supplied context can preserve it + +Split into separate types only when different functions fail in disjoint ways that different callers see, so no caller has to match variants its function can never return. Name a split type after what went wrong, not after the function that returns it. A type whose fields are already typed data with the message in `#[error(...)]`, such as `litellm_http::request::HeaderError`, does not need folding into the crate error + +## Messages belong to the variant + +Every message template is written once, in the variant's `#[error(...)]`. A caller passes the typed facts the template needs and nothing else. None of these may appear at a call site + +```rust +Error::Network("WebSocket handshake returned a wrong Sec-WebSocket-Accept".into()) +Error::Request(format!("bad header {name}")) +Error::Signature(error.to_string()) +``` + +When a check fails for a distinct reason, that reason is a variant (`Repr::HandshakeAccept`), not a string inside a generic one. Format arguments can call functions and reach into fields (`#[error("could not read {}: {cause}", path.display())]`), so a hand-written `Display` is rarely needed + +## Keep causes typed + +Store the lower-level error itself, never its string. Flattening a cause loses `source()`, loses its type for downcasting, and for `reqwest::Error` leaks the request URL, which can carry credentials in the path or query. Strip the URL with `without_url()` where a `reqwest::Error` enters the type + +Render each cause once. Either the variant's message includes it or `source()` returns it, never both, or a reporter that walks the chain prints it twice. Our boundaries (the gateway error body, the Python bridge) stringify with `to_string()`, so a leaf variant shows its cause in its message and names the field `cause`. A field named `source` becomes the source automatically, and `#[from]` implies `#[source]`. A variant that only wraps another of our errors uses `#[error(transparent)]` so it adds no text + +## One conversion per foreign error + +Each foreign error type gets exactly one `From` impl in the crate that owns the wrapper, and that impl does all classification and sanitizing. `From for litellm_http::Error` decides connect versus network versus timeout and strips the URL, so every call site is `?` or `.map_err(litellm_http::Error::from)`. A call site never re-derives the classification with `is_connect()` checks or builds a variant from `error.to_string()` + +The same holds one layer up. When a lower error needs reclassifying for every caller, do it in the shared type's `From`, not per route. `From for litellm_inference::Error` turns a header the signer computes into `Unsupported`, so all four routes classify that input the same way + +## Nest, don't mirror + +Wrap a lower crate's error in one variant and delegate to it. Never re-declare its variants in the upper type with an upper-layer name and a 1:1 mapping function, as OCR once did for every media download failure. The upper type's predicates call through, as in `Self::Http(error) => error.is_request()` + +## Ask, don't match + +Downstream crates ask a lower error questions instead of matching its nested variants. `litellm_http::Error` answers `dispatched()`, `is_request()`, `is_response()`, `status()`, `upstream_response()` and `tls_source()`. Add a predicate when a real caller needs a new distinction, and name it after what the caller decides, not after the variant + +## Opaque errors + +Use the opaque newtype only when a crate wraps third-party causes that are not `Clone` or `PartialEq` and a consumer needs those traits, or when the variants must not become public API. `reqwest::Error`, `std::io::Error` and `serde_json::Error` are built this way + +```rust +#[derive(Clone, Debug, thiserror::Error)] +#[error(transparent)] +pub struct Error(Arc); + +#[derive(Debug, thiserror::Error)] +pub(crate) enum Repr { + #[error("upstream network error: {}", Chain(.0))] + Network(reqwest::Error), + #[error("upstream request failed with status {status}: {body}")] + Status { status: u16, body: String }, +} +``` + +`Repr` holds the real causes. `Arc` makes `Clone` a refcount bump, so `Repr` never has to be `Clone`. `#[error(transparent)]` forwards `Display` and `source()`. Other crates cannot name a variant, so they cannot build an error with a message of their own, and the compiler enforces the rules above. Public constructors exist only for errors another crate genuinely produces, take typed data (`Error::upstream(status, body)`, `Error::computed_header(name)`, `Error::signature(cause)`), and never take a message. `PartialEq` compares by `Arc::ptr_eq`, the same as `SecretError` and `ErrorSource`, so a wrapping enum can keep its derives + +A crate whose errors are its own data, which nobody outside matches on, keeps a plain `#[derive(Clone, Debug, thiserror::Error)]` enum + +## Testing errors + +Assert behavior through the predicates and accessors (`error.upstream_response() == Some((429, "slow down"))`, `!error.dispatched()`), not by constructing an expected variant with a guessed message. Produce errors the way production does, through a public constructor, a real failing call, or a real cause such as `tungstenite::Error::ConnectionClosed`. Inside the owning crate a test may match the private representation through a `#[cfg(test)]` accessor + +## thiserror reference + +The workspace used thiserror 2.0 when this was written. The [derive docs](https://docs.rs/thiserror/latest/thiserror/) cover format arguments, `#[from]`, `#[source]` and `#[error(transparent)]`. `#[from]` only works on a variant with no other fields besides the source and an optional backtrace, so a variant that needs context such as a path takes its cause through an explicit `map_err`. `#[backtrace]` needs a nightly compiler and is not used here diff --git a/litellm-rust/AGENTS.md b/litellm-rust/AGENTS.md index e62a9463cb5..9e7a55eb232 100644 --- a/litellm-rust/AGENTS.md +++ b/litellm-rust/AGENTS.md @@ -4,6 +4,8 @@ For diagnostic tracing changes, follow [.agents/skills/rust-tracing/SKILL.md](.a For string-valued enums and their Serde conversions, follow [.agents/skills/rust-string-enums/SKILL.md](.agents/skills/rust-string-enums/SKILL.md) +For error types, error variants and conversions between crate errors, follow [.agents/skills/rust-error/SKILL.md](.agents/skills/rust-error/SKILL.md) + For fieldless enums, derive `strum::VariantArray` and use `VARIANTS` instead of a hand-listed `ALL` array. Derive Strum string conversions instead of hand-written variant-to-string matches Use the derived conversions directly (`<&'static str>::from(x)` / `.into()`, `str::parse`) with no `as_str`/`parse` wrapper that only delegates, and spell each variant with explicit `#[strum(serialize = "...")]` instead of `serialize_all` @@ -24,13 +26,3 @@ Use [`#[rstest]`](https://docs.rs/rstest/latest/rstest/attr.rstest.html) for new - Never loop over inputs (`for`, `.iter().for_each`, `.all`) inside a test body. Give each input its own `#[case::name(...)]`, or use `#[values(...)]` for a cross product - Exception: a test pinning a Rust table against a repo-owned data file (for example `include_str!` of a JSON config) may iterate that file's entries - -## Error definitions - -- A crate's errors live in `src/error.rs`, defined with `thiserror`, and re-exported from `lib.rs` -- Put message templates in the variant's `#[error(...)]` declaration. Callers pass only the small typed arguments needed to fill them, never `Error::Variant(format!(...))` or a preformatted message. Keep the smallest set of neutral variants that callers need to distinguish. Different wording or providers do not justify new variants -- Default to one top-level `Error` enum per crate, with one variant per failure mode and a `#[error(...)]` message on each. A failure mode is something a caller handles differently (phase, status code, retry, a message Python parity pins exactly). Failures no caller tells apart share one variant and differ only in its message -- Keep shared error enums minimal and provider-neutral. Provider names, credential types, configuration fields, and setup guidance belong in caller-supplied data, not dedicated variants or hardcoded shared messages. Reuse a variant for the same failure mode across providers, such as `MissingApiBase { provider: "Azure", guidance: "..." }`. An exact parity message does not justify a provider-specific variant when caller-supplied context can preserve it -- Wrap a lower-level error as a variant with `#[from]` or `#[source]` instead of flattening it to a string -- Exception: split into separate types when different functions fail in disjoint ways, especially when different callers see them. A shared enum would force every caller to match variants its function can never return -- Name a split type after what went wrong (a unit struct is fine for a single failure mode), not after the function that returns it diff --git a/litellm-rust/Cargo.lock b/litellm-rust/Cargo.lock index 4f666854b56..d2b1354901c 100644 --- a/litellm-rust/Cargo.lock +++ b/litellm-rust/Cargo.lock @@ -3811,6 +3811,7 @@ dependencies = [ "serde_json", "thiserror 2.0.19", "tokio", + "tokio-tungstenite", "tower", "wiremock", ] @@ -3958,6 +3959,7 @@ dependencies = [ "tokio-rustls 0.26.4", "tokio-tungstenite", "tracing", + "url", "veil", "webpki-roots", ] @@ -4009,7 +4011,6 @@ dependencies = [ "litellm-llms-types", "litellm-secrets", "litellm-tracing", - "reqwest", "rstest", "serde_json", "tokio", @@ -4252,6 +4253,7 @@ dependencies = [ "serde_with", "sqlx", "strum", + "tempfile", "thiserror 2.0.19", "tokio", "tokio-tungstenite", diff --git a/litellm-rust/crates/auth-aws/src/signer.rs b/litellm-rust/crates/auth-aws/src/signer.rs index 6c2cb4fd162..13818ecf9b9 100644 --- a/litellm-rust/crates/auth-aws/src/signer.rs +++ b/litellm-rust/crates/auth-aws/src/signer.rs @@ -53,13 +53,13 @@ impl RequestSigner for SigV4Signer { .keys() .find(|name| is_sigv4_computed_header(name.as_str())) { - return Err(litellm_http::Error::ComputedHeader(name.to_string())); + return Err(litellm_http::Error::computed_header(name.as_str())); } let body = match request.body() { None => &[], - Some(body) => body.as_bytes().ok_or_else(|| { - litellm_http::Error::Signature("a streaming body cannot be signed".into()) - })?, + Some(body) => body + .as_bytes() + .ok_or_else(litellm_http::Error::unsignable_body)?, }; sign_headers( request.method().as_str(), @@ -73,7 +73,7 @@ impl RequestSigner for SigV4Signer { time: (self.clock)(), }, ) - .map_err(|error| litellm_http::Error::Signature(error.to_string())) + .map_err(litellm_http::Error::signature) } } @@ -183,10 +183,8 @@ mod tests { ) .unwrap_err(); - assert_eq!( - error, - litellm_http::Error::ComputedHeader("authorization".into()) - ); + assert!(error.is_computed_header()); + assert!(error.to_string().contains("authorization"), "{error}"); } #[test] @@ -198,7 +196,7 @@ mod tests { assert!(matches!( signed(request, &signer("textract")), - Err(litellm_http::Error::Signature(_)) + Err(error) if !error.dispatched() && !error.is_request() )); } } diff --git a/litellm-rust/crates/gateway-inference/Cargo.toml b/litellm-rust/crates/gateway-inference/Cargo.toml index 312d44b9a4b..5d0d364292f 100644 --- a/litellm-rust/crates/gateway-inference/Cargo.toml +++ b/litellm-rust/crates/gateway-inference/Cargo.toml @@ -34,5 +34,6 @@ litellm-cache-memory.workspace = true futures-util.workspace = true tokio = { workspace = true, features = ["io-util"] } rstest.workspace = true +tokio-tungstenite.workspace = true tower = { version = "0.5.3", features = ["util"] } wiremock.workspace = true diff --git a/litellm-rust/crates/gateway-inference/src/error.rs b/litellm-rust/crates/gateway-inference/src/error.rs index ec82a6ede12..a3d5e65b4f6 100644 --- a/litellm-rust/crates/gateway-inference/src/error.rs +++ b/litellm-rust/crates/gateway-inference/src/error.rs @@ -3,7 +3,6 @@ use axum::{ Json, response::{IntoResponse, Response}, }; -use litellm_http::transport::Error as TransportError; use litellm_inference::Error as InferenceError; use litellm_llms::base_llm::ocr::error::Error as OcrError; use serde_json::{Map, Value, json}; @@ -64,8 +63,8 @@ impl Error { .and_then(|status| StatusCode::from_u16(status).ok()) .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR), Self::InvalidBody(_) | Self::UnknownModel(_) => StatusCode::BAD_REQUEST, - Self::Route(InferenceError::Transport(TransportError::Http { status, .. })) => { - StatusCode::from_u16(*status).unwrap_or(StatusCode::BAD_GATEWAY) + Self::Route(error) if let Some((status, _)) = error.upstream_response() => { + StatusCode::from_u16(status).unwrap_or(StatusCode::BAD_GATEWAY) } Self::Route(InferenceError::Auth(litellm_auth::Error::MissingApiKey { .. })) => { StatusCode::UNAUTHORIZED @@ -97,8 +96,8 @@ impl Error { /// body already in that shape passes through, any other has its message extracted. pub fn body(&self, request_id: Option<&str>) -> Value { let raw = match self { - Self::Route(InferenceError::Transport(TransportError::Http { body, .. })) => { - body.clone() + Self::Route(error) if let Some((_, body)) = error.upstream_response() => { + body.to_owned() } other => other.to_string(), }; @@ -177,10 +176,7 @@ mod tests { use super::*; fn upstream(status: u16, body: &str) -> Error { - Error::Route(InferenceError::Transport(TransportError::Http { - status, - body: body.into(), - })) + Error::Route(litellm_http::Error::upstream(status, body.into()).into()) } #[rstest] @@ -236,7 +232,10 @@ mod tests { StatusCode::UNAUTHORIZED, )] #[case::lost_connection( - Error::Route(InferenceError::Transport(TransportError::Network("reset".into()))), + Error::Route( + litellm_http::Error::from(tokio_tungstenite::tungstenite::Error::ConnectionClosed) + .into() + ), StatusCode::INTERNAL_SERVER_ERROR, )] fn status_follows_who_is_at_fault(#[case] error: Error, #[case] status: StatusCode) { diff --git a/litellm-rust/crates/gateway/tests/mcp.rs b/litellm-rust/crates/gateway/tests/mcp.rs index 593fbfeb129..444f12c490b 100644 --- a/litellm-rust/crates/gateway/tests/mcp.rs +++ b/litellm-rust/crates/gateway/tests/mcp.rs @@ -313,9 +313,10 @@ fn applies_configured_http_environment_before_building_clients() { .unwrap(); let error = litellm_gateway::build_inference(&config).err().unwrap(); assert!(matches!( - error, - litellm_gateway::Error::Http(litellm_http::Error::Read { path, .. }) - if path == missing + &error, + litellm_gateway::Error::Http(http) + if http.tls_source() == Some(litellm_http::TlsSource::CaBundle) + && http.to_string().contains(&missing.display().to_string()) )); } diff --git a/litellm-rust/crates/http/Cargo.toml b/litellm-rust/crates/http/Cargo.toml index f787192f4c4..dbdb126ec94 100644 --- a/litellm-rust/crates/http/Cargo.toml +++ b/litellm-rust/crates/http/Cargo.toml @@ -21,6 +21,7 @@ strum.workspace = true thiserror.workspace = true tokio.workspace = true tracing.workspace = true +url.workspace = true veil.workspace = true webpki-roots.workspace = true diff --git a/litellm-rust/crates/http/src/config.rs b/litellm-rust/crates/http/src/config.rs index 6a2d80e3f85..b15c202ba42 100644 --- a/litellm-rust/crates/http/src/config.rs +++ b/litellm-rust/crates/http/src/config.rs @@ -135,7 +135,7 @@ mod tests { use rstest::rstest; use super::*; - use crate::TlsSource; + use crate::{TlsSource, error::Repr}; fn settings(ssl_verify: Option, ssl_cert_file: Option<&str>) -> HttpSettings { HttpSettings { @@ -305,11 +305,10 @@ mod tests { }; assert!(matches!( reqwest::ClientBuilder::try_from(&config), - Err(Error::Read { - path: reported, - tls_source: TlsSource::CaBundle, - .. - }) if reported == path + Err(error) if matches!( + error.repr(), + Repr::Read { path: reported, tls_source: TlsSource::CaBundle, .. } if *reported == path + ) )); } @@ -326,11 +325,10 @@ mod tests { std::fs::remove_file(&path).unwrap(); assert!(matches!( result, - Err(Error::InvalidPem { - path: reported, - tls_source: TlsSource::CaBundle, - .. - }) if reported == path + Err(error) if matches!( + error.repr(), + Repr::InvalidPem { path: reported, tls_source: TlsSource::CaBundle, .. } if *reported == path + ) )); } } diff --git a/litellm-rust/crates/http/src/error.rs b/litellm-rust/crates/http/src/error.rs index ae14756baa3..4e77eef7127 100644 --- a/litellm-rust/crates/http/src/error.rs +++ b/litellm-rust/crates/http/src/error.rs @@ -1,4 +1,6 @@ -use std::path::PathBuf; +use std::{fmt, io, path::PathBuf, sync::Arc}; + +use tokio_tungstenite::tungstenite; #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum TlsSource { @@ -6,34 +8,304 @@ pub enum TlsSource { ClientIdentity, } -#[derive(Clone, Debug, thiserror::Error, PartialEq, Eq)] -pub enum Error { - #[error("could not read {}: {message}", path.display())] +#[derive(Clone, Debug, thiserror::Error)] +#[error(transparent)] +pub struct Error(Arc); + +#[derive(Debug, thiserror::Error)] +pub(crate) enum Repr { + #[error("could not read {}: {cause}", path.display())] Read { path: PathBuf, - message: String, tls_source: TlsSource, + cause: io::Error, }, - #[error("{} is not a PEM file: {message}", path.display())] + #[error("{} is not a PEM file: {cause}", path.display())] InvalidPem { path: PathBuf, - message: String, tls_source: TlsSource, + cause: PemFault, }, + #[error("could not build the HTTP client: {}", Chain(.0))] + Client(reqwest::Error), #[error("could not build the HTTP client: {0}")] - Client(String), + TlsConfig(rustls::Error), + #[error("request body could not be serialized: {0}")] - RequestBody(String), + RequestBody(serde_json::Error), #[error("invalid request: {0}")] - Request(String), + Request(http::Error), + #[error("invalid request: {}", Chain(.0))] + RequestUrl(reqwest::Error), #[error("request forwards a header the signer computes: {0}")] ComputedHeader(String), + #[error("request signing failed: a streaming body cannot be signed")] + UnsignableBody, #[error("request signing failed: {0}")] - Signature(String), + Signature(Box), + + #[error("could not reach the provider: {}", Chain(.0))] + Connect(reqwest::Error), + #[error("upstream network error: {}", Chain(.0))] + Network(reqwest::Error), + #[error("upstream request timed out: {}", Chain(.0))] + Timeout(reqwest::Error), + #[error("upstream request failed with status {status}: {body}")] + Status { status: u16, body: String }, + + #[error("invalid WebSocket URL: {0}")] + WebSocketUrl(url::ParseError), + #[error("unsupported WebSocket URL scheme {0:?}")] + WebSocketScheme(String), + #[error("WebSocket handshake returned a wrong Sec-WebSocket-Accept")] + HandshakeAccept, + #[error("WebSocket handshake returned an invalid Upgrade header")] + HandshakeUpgrade, + #[error("WebSocket handshake returned an invalid Connection header")] + HandshakeConnection, + #[error("WebSocket handshake returned an invalid subprotocol")] + HandshakeSubprotocol, + #[error("WebSocket connection timed out")] + HandshakeTimeout, + #[error("upstream WebSocket error: {0}")] + WebSocket(tungstenite::Error), + + #[error("media URL rejected by network policy")] + BlockedUrl, + #[error("could not resolve the media host: {0}")] + Resolve(io::Error), + #[error("media download is disabled")] + DownloadDisabled, + #[error("media download exceeds the maximum size")] + DownloadTooLarge, + #[error("too many redirects while fetching media")] + TooManyRedirects, + #[error("media redirect is missing a Location header")] + MissingRedirectLocation, + #[error("invalid media redirect")] + InvalidRedirect, + #[error("media download failed with status {0}")] + DownloadStatus(u16), + #[error("media download timed out")] + DownloadTimeout, +} + +#[derive(Debug, thiserror::Error)] +pub(crate) enum PemFault { + #[error("no certificates found")] + NoCertificates, + #[error(transparent)] + Parse(#[from] rustls::pki_types::pem::Error), + #[error(transparent)] + Rejected(#[from] rustls::Error), +} + +struct Chain<'a>(&'a reqwest::Error); + +impl fmt::Display for Chain<'_> { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(formatter, "{}", self.0)?; + std::iter::successors(std::error::Error::source(self.0), |cause| cause.source()) + .try_for_each(|cause| write!(formatter, ": {cause}")) + } +} + +impl Error { + pub fn upstream(status: u16, body: String) -> Self { + Repr::Status { status, body }.into() + } + + pub fn computed_header(name: impl Into) -> Self { + Repr::ComputedHeader(name.into()).into() + } + + pub fn unsignable_body() -> Self { + Repr::UnsignableBody.into() + } + + pub fn signature(cause: impl std::error::Error + Send + Sync + 'static) -> Self { + Repr::Signature(Box::new(cause)).into() + } + + /// The request may have reached the provider, so retrying elsewhere could + /// repeat a billed call. + pub fn dispatched(&self) -> bool { + matches!( + *self.0, + Repr::Network(_) + | Repr::Timeout(_) + | Repr::Status { .. } + | Repr::HandshakeAccept + | Repr::HandshakeUpgrade + | Repr::HandshakeConnection + | Repr::HandshakeSubprotocol + | Repr::HandshakeTimeout + | Repr::WebSocket(_) + | Repr::TooManyRedirects + | Repr::MissingRedirectLocation + | Repr::InvalidRedirect + | Repr::DownloadTooLarge + | Repr::DownloadStatus(_) + | Repr::DownloadTimeout + ) + } + + pub fn is_request(&self) -> bool { + matches!( + *self.0, + Repr::Request(_) + | Repr::RequestUrl(_) + | Repr::ComputedHeader(_) + | Repr::WebSocketUrl(_) + | Repr::WebSocketScheme(_) + | Repr::BlockedUrl + | Repr::DownloadDisabled + | Repr::DownloadTooLarge + | Repr::TooManyRedirects + ) + } + + pub fn is_response(&self) -> bool { + matches!( + *self.0, + Repr::MissingRedirectLocation | Repr::InvalidRedirect + ) + } + + pub fn is_computed_header(&self) -> bool { + matches!(*self.0, Repr::ComputedHeader(_)) + } + + pub fn upstream_response(&self) -> Option<(u16, &str)> { + match &*self.0 { + Repr::Status { status, body } => Some((*status, body)), + _ => None, + } + } + + pub fn status(&self) -> Option { + match *self.0 { + Repr::Status { status, .. } | Repr::DownloadStatus(status) => Some(status), + Repr::Timeout(_) | Repr::DownloadTimeout => Some(408), + _ => None, + } + } + + pub fn tls_source(&self) -> Option { + match *self.0 { + Repr::Read { tls_source, .. } | Repr::InvalidPem { tls_source, .. } => Some(tls_source), + _ => None, + } + } + + #[cfg(test)] + pub(crate) fn repr(&self) -> &Repr { + &self.0 + } +} + +impl PartialEq for Error { + fn eq(&self, other: &Self) -> bool { + Arc::ptr_eq(&self.0, &other.0) + } +} + +impl Eq for Error {} + +impl From for Error { + fn from(repr: Repr) -> Self { + Self(Arc::new(repr)) + } } impl From for Error { fn from(error: reqwest::Error) -> Self { - Self::Client(error.without_url().to_string()) + let error = error.without_url(); + if error.is_timeout() { + Repr::Timeout(error).into() + } else if error.is_connect() || error.is_builder() { + Repr::Connect(error).into() + } else { + Repr::Network(error).into() + } + } +} + +impl From for Error { + fn from(error: tungstenite::Error) -> Self { + Repr::WebSocket(error).into() + } +} + +#[cfg(test)] +mod tests { + #[tokio::test] + async fn transport_errors_remove_urls_and_keep_dispatch_context() { + let error = reqwest::Client::builder() + .no_proxy() + .build() + .expect("client") + .get("http://localhost:invalid/private?api_key=secret") + .send() + .await + .expect_err("invalid port"); + let error = crate::Error::from(error); + assert!(!error.dispatched()); + assert!(!error.to_string().contains("secret")); + assert!(!error.to_string().contains("private")); + } + + fn root_cause(error: &dyn std::error::Error) -> Option { + match error.source() { + Some(cause) => root_cause(cause).or_else(|| Some(cause.to_string())), + None => None, + } + } + + #[tokio::test] + async fn network_error_message_names_the_underlying_cause() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind"); + let address = listener.local_addr().expect("address"); + drop(listener); + let error = reqwest::Client::builder() + .no_proxy() + .build() + .expect("client") + .get(format!("http://{address}/private?api_key=secret")) + .send() + .await + .expect_err("nothing listens on the port"); + let root_cause = root_cause(&error).expect("reqwest reports a cause"); + let message = crate::Error::from(error).to_string(); + assert!(message.contains(&root_cause), "{message}"); + assert!(!message.contains("secret")); + } + + #[tokio::test] + async fn request_timeout_is_not_safe_to_retry_as_a_connect_failure() { + use std::time::Duration; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let address = listener.local_addr().expect("address"); + let request = reqwest::Client::builder() + .no_proxy() + .build() + .expect("client") + .get(format!("http://{address}")) + .timeout(Duration::from_millis(200)) + .send(); + let (response, accepted) = tokio::join!( + request, + tokio::time::timeout(Duration::from_secs(2), listener.accept()) + ); + let _connection = accepted + .expect("accept deadline") + .expect("accepted connection"); + let error = response.expect_err("server does not respond"); + assert!(error.is_timeout()); + let error = crate::Error::from(error); + assert!(error.dispatched()); + assert_eq!(error.status(), Some(408)); } } diff --git a/litellm-rust/crates/http/src/lib.rs b/litellm-rust/crates/http/src/lib.rs index ef72420af9b..02129c62083 100644 --- a/litellm-rust/crates/http/src/lib.rs +++ b/litellm-rust/crates/http/src/lib.rs @@ -14,7 +14,6 @@ mod proxy; pub mod request; mod settings; mod tls; -pub mod transport; pub mod websocket; pub use client::Client; diff --git a/litellm-rust/crates/http/src/media.rs b/litellm-rust/crates/http/src/media.rs index 960a1d4044a..a69cb75793d 100644 --- a/litellm-rust/crates/http/src/media.rs +++ b/litellm-rust/crates/http/src/media.rs @@ -12,29 +12,7 @@ use reqwest::{ dns::{Addrs, Name, Resolve, Resolving}, }; -use crate::{Client, ClientVariant, HttpClientConfig, HttpClientPool}; - -#[derive(Debug, thiserror::Error)] -pub enum Error { - #[error("media URL rejected by network policy")] - BlockedUrl, - #[error("media download is disabled")] - DownloadDisabled, - #[error("media download exceeds the maximum size")] - DownloadTooLarge, - #[error("too many redirects while fetching media")] - TooManyRedirects, - #[error("media redirect is missing a Location header")] - MissingRedirectLocation, - #[error("invalid media redirect")] - InvalidRedirect, - #[error("media download failed with status {0}")] - Http(u16), - #[error("media download timed out")] - Timeout, - #[error("{0}")] - Transport(#[from] crate::transport::Error), -} +use crate::{Client, ClientVariant, Error, HttpClientConfig, HttpClientPool, error::Repr}; #[derive(Clone, Debug, PartialEq, Eq)] pub struct UrlPolicy { @@ -125,7 +103,7 @@ impl MediaFetcher { pool: &HttpClientPool, config: &HttpClientConfig, url_policy: UrlPolicy, - ) -> Result { + ) -> Result { let uses_proxy: ProxyMatch = Arc::new(config.proxies.matcher()); Self::with_resolution( pool, @@ -142,7 +120,7 @@ impl MediaFetcher { url_policy: UrlPolicy, address_resolver: Arc, uses_proxy: ProxyMatch, - ) -> Result { + ) -> Result { Ok(Self { pinned: pool.client(config, ClientVariant::Media)?, unpinned: pool.client(config, ClientVariant::UnpinnedMedia)?, @@ -167,11 +145,11 @@ impl MediaFetcher { pub async fn fetch(&self, url: Url, policy: DownloadPolicy) -> Result { if policy.max_bytes == 0 { - return Err(Error::DownloadDisabled); + return Err(Repr::DownloadDisabled.into()); } tokio::time::timeout(policy.timeout, self.fetch_before_deadline(url, policy)) .await - .map_err(|_| Error::Timeout)? + .map_err(|_| Repr::DownloadTimeout)? } async fn fetch_before_deadline( @@ -181,28 +159,22 @@ impl MediaFetcher { ) -> Result { let mut redirects_followed = 0; loop { - let mut response = self - .client_for(&url) - .await? - .get(url.clone()) - .send() - .await - .map_err(crate::transport::Error::from)?; + let mut response = self.client_for(&url).await?.get(url.clone()).send().await?; if response.status().is_redirection() { if redirects_followed == policy.max_redirects { - return Err(Error::TooManyRedirects); + return Err(Repr::TooManyRedirects.into()); } let location = response .headers() .get(reqwest::header::LOCATION) .and_then(|value| value.to_str().ok()) - .ok_or(Error::MissingRedirectLocation)?; - url = url.join(location).map_err(|_| Error::InvalidRedirect)?; + .ok_or(Repr::MissingRedirectLocation)?; + url = url.join(location).map_err(|_| Repr::InvalidRedirect)?; redirects_followed += 1; continue; } if !response.status().is_success() { - return Err(Error::Http(response.status().as_u16())); + return Err(Repr::DownloadStatus(response.status().as_u16()).into()); } enforce_download_size(response.content_length().unwrap_or(0), policy.max_bytes)?; let content_type = response @@ -215,11 +187,7 @@ impl MediaFetcher { .unwrap_or("application/octet-stream") .to_string(); let mut bytes = Vec::new(); - while let Some(chunk) = response - .chunk() - .await - .map_err(crate::transport::Error::from)? - { + while let Some(chunk) = response.chunk().await? { enforce_download_size(bytes.len() as u64 + chunk.len() as u64, policy.max_bytes)?; bytes.extend_from_slice(&chunk); } @@ -238,13 +206,13 @@ impl MediaFetcher { || !url.username().is_empty() || url.password().is_some() { - return Err(Error::BlockedUrl); + return Err(Repr::BlockedUrl.into()); } - let host = url.host_str().ok_or(Error::BlockedUrl)?; + let host = url.host_str().ok_or(Repr::BlockedUrl)?; if self.allow_private_network { return Ok(&self.pinned); } - let port = url.port_or_known_default().ok_or(Error::BlockedUrl)?; + let port = url.port_or_known_default().ok_or(Repr::BlockedUrl)?; if self.url_policy.allows(host, port) { return Ok(&self.unpinned); } @@ -262,27 +230,29 @@ impl MediaFetcher { .trim_end_matches(']') .parse::() { - return (!is_blocked_ip(ip)).then_some(()).ok_or(Error::BlockedUrl); + return (!is_blocked_ip(ip)) + .then_some(()) + .ok_or(Repr::BlockedUrl.into()); } let addresses = self .address_resolver .resolve(host, port) .await - .map_err(|error| crate::transport::Error::Network(error.to_string()))?; + .map_err(Repr::Resolve)?; validate_addresses(&addresses) } } fn enforce_download_size(length: u64, max_bytes: u64) -> Result<(), Error> { if length > max_bytes { - return Err(Error::DownloadTooLarge); + return Err(Repr::DownloadTooLarge.into()); } Ok(()) } fn validate_addresses(addresses: &[SocketAddr]) -> Result<(), Error> { if addresses.is_empty() || addresses.iter().any(|address| is_blocked_ip(address.ip())) { - return Err(Error::BlockedUrl); + return Err(Repr::BlockedUrl.into()); } Ok(()) } @@ -356,10 +326,8 @@ impl Resolve for PublicDnsResolver { .await .map_err(|error| Box::new(error) as Box)? .collect::>(); - validate_addresses(&addresses).map_err(|_| { - Box::new(io::Error::other("destination rejected by network policy")) - as Box - })?; + validate_addresses(&addresses) + .map_err(|error| Box::new(error) as Box)?; Ok(Box::new(addresses.into_iter()) as Addrs) }) } @@ -544,7 +512,7 @@ mod tests { .await .expect_err("oversize body is rejected"); server.await.expect("server completes"); - assert!(matches!(error, Error::DownloadTooLarge)); + assert!(matches!(error.repr(), Repr::DownloadTooLarge)); } #[tokio::test] @@ -559,7 +527,7 @@ mod tests { .await .expect_err("stream crossing limit is rejected"); server.await.expect("server completes"); - assert!(matches!(error, Error::DownloadTooLarge)); + assert!(matches!(error.repr(), Repr::DownloadTooLarge)); } #[tokio::test] @@ -595,7 +563,7 @@ mod tests { .expect_err("private redirect is rejected"); let requests = server.await.expect("server completes"); assert_eq!(requests.len(), 1); - assert!(matches!(error, Error::BlockedUrl)); + assert!(matches!(error.repr(), Repr::BlockedUrl)); } #[tokio::test] @@ -622,7 +590,7 @@ mod tests { .await .expect_err("fetch times out"); server.await.expect("server completes"); - assert!(matches!(error, Error::Timeout)); + assert!(matches!(error.repr(), Repr::DownloadTimeout)); } #[tokio::test] @@ -652,8 +620,13 @@ mod tests { let url = Url::parse("https://user:password@8.8.8.8/document").expect("credentialed URL parses"); assert!(matches!( - fetcher.fetch(url, policy(1, 0)).await, - Err(Error::BlockedUrl) + fetcher + .fetch(url, policy(1, 0)) + .await + .err() + .as_ref() + .map(Error::repr), + Some(Repr::BlockedUrl) )); } @@ -683,7 +656,10 @@ mod tests { let result = fetcher(UNROUTABLE, HashSet::from(["localhost"]), other_port, false) .fetch(url, policy(2, 0)) .await; - assert!(matches!(result, Err(Error::BlockedUrl))); + assert!(matches!( + result.as_ref().err().map(Error::repr), + Some(Repr::BlockedUrl) + )); } #[test] @@ -742,6 +718,9 @@ mod tests { ) .fetch(url, policy(2, 0)) .await; - assert!(matches!(blocked, Err(Error::BlockedUrl))); + assert!(matches!( + blocked.as_ref().err().map(Error::repr), + Some(Repr::BlockedUrl) + )); } } diff --git a/litellm-rust/crates/http/src/outbound.rs b/litellm-rust/crates/http/src/outbound.rs index 09b5d0179d0..0cf7212608e 100644 --- a/litellm-rust/crates/http/src/outbound.rs +++ b/litellm-rust/crates/http/src/outbound.rs @@ -14,7 +14,7 @@ use reqwest::{ }; use serde::Serialize; -use crate::Error; +use crate::{Error, error::Repr}; /// Returns the headers to add to the request; it never sees a mutable request. pub trait RequestSigner: Send + Sync { @@ -27,7 +27,7 @@ pub fn json( body: &impl Serialize, timeout: Option, ) -> Result { - let body = serde_json::to_vec(body).map_err(|error| Error::RequestBody(error.to_string()))?; + let body = serde_json::to_vec(body).map_err(Repr::RequestBody)?; let with_headers = headers .iter() .fold(http::Request::post(url), |builder, (name, value)| { @@ -41,13 +41,9 @@ pub fn json( } else { with_headers.header(CONTENT_TYPE, "application/json") }; - let mut request = builder - .body(body) - .map_err(|error| Error::Request(error.to_string())) - .and_then(|request| { - Request::try_from(request) - .map_err(|error| Error::Request(error.without_url().to_string())) - })?; + let request = builder.body(body).map_err(Repr::Request)?; + let mut request = + Request::try_from(request).map_err(|error| Repr::RequestUrl(error.without_url()))?; *request.timeout_mut() = timeout; Ok(request) } @@ -113,7 +109,7 @@ mod tests { .headers() .contains_key(CONTENT_TYPE) .then(HeaderMap::new) - .ok_or_else(|| Error::Signature("content-type was not signed".into())) + .ok_or_else(|| Error::signature(std::fmt::Error)) } } @@ -147,7 +143,7 @@ mod tests { struct Refuses; impl RequestSigner for Refuses { fn sign(&self, _request: &Request) -> Result { - Err(Error::ComputedHeader("authorization".into())) + Err(Error::computed_header("authorization")) } } @@ -156,14 +152,14 @@ mod tests { &Refuses, ) .unwrap_err(); - assert_eq!(error, Error::ComputedHeader("authorization".into())); + assert!(error.is_computed_header()); } #[test] fn a_request_that_cannot_be_sent_fails_when_it_is_built() { assert!(matches!( json("not a url", &[], &json!({}), None), - Err(Error::Request(_)) + Err(error) if error.is_request() )); assert!(matches!( json( @@ -172,7 +168,7 @@ mod tests { &json!({}), None ), - Err(Error::Request(_)) + Err(error) if error.is_request() )); } } diff --git a/litellm-rust/crates/http/src/pool.rs b/litellm-rust/crates/http/src/pool.rs index dc96e6713dd..a1ea8165d1f 100644 --- a/litellm-rust/crates/http/src/pool.rs +++ b/litellm-rust/crates/http/src/pool.rs @@ -6,7 +6,12 @@ use std::{ use reqwest::dns::Resolve; -use crate::{client::Client, config::HttpClientConfig, error::Error, proxy::EnvironmentProxies}; +use crate::{ + client::Client, + config::HttpClientConfig, + error::{Error, Repr}, + proxy::EnvironmentProxies, +}; #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] pub enum ClientVariant { @@ -74,7 +79,8 @@ impl HttpClientPool { } let client = self .apply(variant, reqwest::ClientBuilder::try_from(&key.0)?) - .build()?; + .build() + .map_err(Repr::Client)?; self.lock().insert( key, PooledClient { diff --git a/litellm-rust/crates/http/src/tls.rs b/litellm-rust/crates/http/src/tls.rs index 0b1c44bf6fe..01559b0d6ba 100644 --- a/litellm-rust/crates/http/src/tls.rs +++ b/litellm-rust/crates/http/src/tls.rs @@ -1,4 +1,4 @@ -use std::{fmt, path::Path, str::FromStr, sync::Arc}; +use std::{path::Path, str::FromStr, sync::Arc}; use rustls::{ CipherSuite, ClientConfig, DigitallySignedStruct, RootCertStore, SignatureScheme, @@ -9,7 +9,7 @@ use rustls::{ use crate::{ config::{ClientIdentity, HttpClientConfig, Verify}, - error::{Error, TlsSource}, + error::{Error, PemFault, Repr, TlsSource}, }; #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] @@ -187,7 +187,7 @@ impl TryFrom<&HttpClientConfig> for ClientConfig { }); let builder = ClientConfig::builder_with_provider(Arc::clone(&provider)) .with_safe_default_protocol_versions() - .map_err(|error| Error::Client(error.to_string()))?; + .map_err(Repr::TlsConfig)?; let verified = match &config.verify { Verify::Disabled => builder .dangerous() @@ -224,7 +224,7 @@ impl TryFrom<&HttpClientConfig> for ClientConfig { fn bundle_roots(path: &Path, source: TlsSource) -> Result { let certificates = certificates(path, source)?; if certificates.is_empty() { - return Err(invalid_pem(path, source, "no certificates found")); + return Err(invalid_pem(path, source, PemFault::NoCertificates)); } let mut store = RootCertStore::empty(); for certificate in certificates { @@ -242,7 +242,7 @@ fn client_identity( let source = TlsSource::ClientIdentity; let chain = certificates(certificate, source)?; if chain.is_empty() { - return Err(invalid_pem(certificate, source, "no certificates found")); + return Err(invalid_pem(certificate, source, PemFault::NoCertificates)); } let private_key = PrivateKeyDer::from_pem_slice(&read(key, source)?) .map_err(|error| invalid_pem(key, source, error))?; @@ -256,19 +256,23 @@ fn certificates(path: &Path, source: TlsSource) -> Result Result, Error> { - std::fs::read(path).map_err(|error| Error::Read { - path: path.to_path_buf(), - message: error.to_string(), - tls_source: source, + std::fs::read(path).map_err(|cause| { + Repr::Read { + path: path.to_path_buf(), + tls_source: source, + cause, + } + .into() }) } -fn invalid_pem(path: &Path, source: TlsSource, message: impl fmt::Display) -> Error { - Error::InvalidPem { +fn invalid_pem(path: &Path, source: TlsSource, cause: impl Into) -> Error { + Repr::InvalidPem { path: path.to_path_buf(), - message: message.to_string(), tls_source: source, + cause: cause.into(), } + .into() } #[derive(Debug)] @@ -448,11 +452,10 @@ mod tests { std::fs::remove_file(&path).unwrap(); assert!(matches!( result, - Err(Error::InvalidPem { - path: reported, - tls_source: TlsSource::ClientIdentity, - .. - }) if reported == path + Err(error) if matches!( + error.repr(), + Repr::InvalidPem { path: reported, tls_source: TlsSource::ClientIdentity, .. } if *reported == path + ) )); } diff --git a/litellm-rust/crates/http/src/transport.rs b/litellm-rust/crates/http/src/transport.rs deleted file mode 100644 index 8814925bbf2..00000000000 --- a/litellm-rust/crates/http/src/transport.rs +++ /dev/null @@ -1,109 +0,0 @@ -#[derive(Clone, Debug, thiserror::Error, PartialEq, Eq)] -pub enum Error { - #[error("upstream request failed with status {status}: {body}")] - Http { status: u16, body: String }, - #[error("upstream network error: {0}")] - Network(String), - #[error("could not reach the provider: {0}")] - Connect(String), -} - -impl Error { - pub fn from_reqwest_before_dispatch(error: reqwest::Error) -> Self { - let before_dispatch = !error.is_timeout() && (error.is_connect() || error.is_builder()); - let message = describe(error); - if before_dispatch { - Self::Connect(message) - } else { - Self::Network(message) - } - } -} - -impl From for Error { - fn from(error: reqwest::Error) -> Self { - Self::Network(describe(error)) - } -} - -fn describe(error: reqwest::Error) -> String { - let error = error.without_url(); - std::iter::successors(std::error::Error::source(&error), |cause| cause.source()) - .fold(error.to_string(), |message, cause| { - format!("{message}: {cause}") - }) -} - -#[cfg(test)] -mod tests { - #[tokio::test] - async fn transport_errors_remove_urls_and_keep_dispatch_context() { - let error = reqwest::Client::builder() - .no_proxy() - .build() - .expect("client") - .get("http://localhost:invalid/private?api_key=secret") - .send() - .await - .expect_err("invalid port"); - let error = crate::transport::Error::from_reqwest_before_dispatch(error); - assert!(matches!(error, crate::transport::Error::Connect(_))); - assert!(!error.to_string().contains("secret")); - assert!(!error.to_string().contains("private")); - } - - fn root_cause(error: &dyn std::error::Error) -> Option { - match error.source() { - Some(cause) => root_cause(cause).or_else(|| Some(cause.to_string())), - None => None, - } - } - - #[tokio::test] - async fn network_error_message_names_the_underlying_cause() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind"); - let address = listener.local_addr().expect("address"); - drop(listener); - let error = reqwest::Client::builder() - .no_proxy() - .build() - .expect("client") - .get(format!("http://{address}/private?api_key=secret")) - .send() - .await - .expect_err("nothing listens on the port"); - let root_cause = root_cause(&error).expect("reqwest reports a cause"); - let message = crate::transport::Error::from(error).to_string(); - assert!(message.contains(&root_cause), "{message}"); - assert!(!message.contains("secret")); - } - - #[tokio::test] - async fn request_timeout_is_not_safe_to_retry_as_a_connect_failure() { - use std::time::Duration; - let listener = tokio::net::TcpListener::bind("127.0.0.1:0") - .await - .expect("bind"); - let address = listener.local_addr().expect("address"); - let request = reqwest::Client::builder() - .no_proxy() - .build() - .expect("client") - .get(format!("http://{address}")) - .timeout(Duration::from_millis(200)) - .send(); - let (response, accepted) = tokio::join!( - request, - tokio::time::timeout(Duration::from_secs(2), listener.accept()) - ); - let _connection = accepted - .expect("accept deadline") - .expect("accepted connection"); - let error = response.expect_err("server does not respond"); - assert!(error.is_timeout()); - assert!(matches!( - crate::transport::Error::from_reqwest_before_dispatch(error), - crate::transport::Error::Network(_) - )); - } -} diff --git a/litellm-rust/crates/http/src/websocket.rs b/litellm-rust/crates/http/src/websocket.rs index d9ce183201d..e6d6a5e22a4 100644 --- a/litellm-rust/crates/http/src/websocket.rs +++ b/litellm-rust/crates/http/src/websocket.rs @@ -1,3 +1,5 @@ +use std::time::Duration; + use http::{ HeaderMap, HeaderValue, StatusCode, header::{ @@ -14,7 +16,10 @@ use tokio_tungstenite::{ }, }; -use crate::{client::Client, transport::Error}; +use crate::{ + client::Client, + error::{Error, Repr}, +}; pub type UpstreamWebSocket = WebSocketStream; @@ -23,6 +28,20 @@ pub async fn connect( client: &Client, url: &str, headers: HeaderMap, + timeout: Option, +) -> Result { + match timeout { + Some(timeout) => tokio::time::timeout(timeout, handshake(client, url, headers)) + .await + .map_err(|_| Repr::HandshakeTimeout)?, + None => handshake(client, url, headers).await, + } +} + +async fn handshake( + client: &Client, + url: &str, + headers: HeaderMap, ) -> Result { let offered_protocols = headers.get(SEC_WEBSOCKET_PROTOCOL).cloned(); let key = generate_key(); @@ -30,14 +49,13 @@ pub async fn connect( .get(http_url(url)?) .headers(handshake_headers(headers, &key)?) .send() - .await - .map_err(Error::from_reqwest_before_dispatch)?; + .await?; let status = response.status(); if status != StatusCode::SWITCHING_PROTOCOLS { - return Err(Error::Http { - status: status.as_u16(), - body: response.text().await.unwrap_or_default(), - }); + return Err(Error::upstream( + status.as_u16(), + response.text().await.unwrap_or_default(), + )); } let expected = derive_accept_key(key.as_bytes()); if response @@ -46,9 +64,7 @@ pub async fn connect( .map(HeaderValue::as_bytes) != Some(expected.as_bytes()) { - return Err(Error::Network( - "WebSocket handshake returned a wrong Sec-WebSocket-Accept".into(), - )); + return Err(Repr::HandshakeAccept.into()); } validate_response_headers(response.headers(), offered_protocols.as_ref())?; let upgraded = response.upgrade().await?; @@ -63,9 +79,7 @@ fn validate_response_headers( .get(UPGRADE) .is_some_and(|value| value.as_bytes().eq_ignore_ascii_case(b"websocket")) { - return Err(Error::Network( - "WebSocket handshake returned an invalid Upgrade header".into(), - )); + return Err(Repr::HandshakeUpgrade.into()); } if !headers.get_all(CONNECTION).iter().any(|value| { value.to_str().is_ok_and(|value| { @@ -74,9 +88,7 @@ fn validate_response_headers( .any(|token| token.trim().eq_ignore_ascii_case("upgrade")) }) }) { - return Err(Error::Network( - "WebSocket handshake returned an invalid Connection header".into(), - )); + return Err(Repr::HandshakeConnection.into()); } let protocol_matches = match (offered_protocols, headers.get(SEC_WEBSOCKET_PROTOCOL)) { (None, None) => true, @@ -92,31 +104,25 @@ fn validate_response_headers( _ => false, }; if !protocol_matches { - return Err(Error::Network( - "WebSocket handshake returned an invalid subprotocol".into(), - )); + return Err(Repr::HandshakeSubprotocol.into()); } Ok(()) } fn http_url(url: &str) -> Result { - let mut url = Url::parse(url).map_err(not_dispatched)?; + let mut url = Url::parse(url).map_err(Repr::WebSocketUrl)?; let scheme = match url.scheme() { "ws" | "http" => "http", "wss" | "https" => "https", - other => { - return Err(not_dispatched(format!( - "unsupported WebSocket URL scheme {other:?}" - ))); - } + other => return Err(Repr::WebSocketScheme(other.to_owned()).into()), }; url.set_scheme(scheme) - .map_err(|()| not_dispatched("WebSocket URL cannot be sent over HTTP"))?; + .map_err(|()| Repr::WebSocketScheme(url.scheme().to_owned()))?; Ok(url) } fn handshake_headers(headers: HeaderMap, key: &str) -> Result { - let key = HeaderValue::from_str(key).map_err(not_dispatched)?; + let key = HeaderValue::from_str(key).map_err(|error| Repr::Request(error.into()))?; Ok([ (CONNECTION, HeaderValue::from_static("Upgrade")), (UPGRADE, HeaderValue::from_static("websocket")), @@ -129,7 +135,3 @@ fn handshake_headers(headers: HeaderMap, key: &str) -> Result headers })) } - -fn not_dispatched(error: impl std::fmt::Display) -> Error { - Error::Connect(error.to_string()) -} diff --git a/litellm-rust/crates/http/tests/websocket.rs b/litellm-rust/crates/http/tests/websocket.rs index 03bf6c956a7..f3ee07bbb8b 100644 --- a/litellm-rust/crates/http/tests/websocket.rs +++ b/litellm-rust/crates/http/tests/websocket.rs @@ -7,9 +7,8 @@ use std::{ use futures_util::{SinkExt, StreamExt}; use http::{HeaderMap, HeaderValue}; use litellm_http::{ - ClientIdentity, ClientVariant, EnvironmentProxies, HttpClientConfig, HttpClientPool, - HttpSettings, Resolution, Verify, media::PublicDnsResolver, transport::Error, - websocket::connect, + ClientIdentity, ClientVariant, EnvironmentProxies, Error, HttpClientConfig, HttpClientPool, + HttpSettings, Resolution, Verify, media::PublicDnsResolver, websocket::connect, }; use rcgen::{BasicConstraints, CertificateParams, CertifiedIssuer, IsCa, KeyPair}; use rstest::rstest; @@ -122,7 +121,7 @@ async fn open( let client = HttpClientPool::new(Arc::new(PublicDnsResolver)) .client(config, ClientVariant::WebSocket) .unwrap(); - connect(&client, url, headers).await + connect(&client, url, headers, None).await } async fn echo_once( @@ -269,7 +268,7 @@ async fn wss_honors_ssl_verify(#[case] verify: fn(&Pki) -> Verify, #[case] trust match result { Ok(mut socket) if trusted => round_trip(&mut socket).await, - Err(Error::Connect(_)) if !trusted => {} + Err(error) if !trusted && !error.dispatched() => {} other => panic!("unexpected handshake outcome: {:?}", other.map(drop)), } } @@ -340,13 +339,8 @@ async fn refused_upgrade_reports_the_status_and_body() { ) .await; - assert_eq!( - result.map(drop), - Err(Error::Http { - status: 403, - body: "denied".into() - }) - ); + let error = result.map(drop).unwrap_err(); + assert_eq!(error.upstream_response(), Some((403, "denied"))); } #[rstest] @@ -364,7 +358,7 @@ async fn upgrade_with_a_wrong_accept_key_is_rejected() { ) .await; - assert!(matches!(result, Err(Error::Network(_)))); + assert!(matches!(result, Err(error) if error.dispatched() && !error.is_request())); } #[rstest] @@ -374,7 +368,7 @@ async fn upgrade_with_a_wrong_accept_key_is_rejected() { async fn invalid_url_is_rejected_before_dispatch(#[case] url: &str) { let result = open(&config(Verify::BuiltInRoots), url, HeaderMap::new()).await; - assert!(matches!(result, Err(Error::Connect(_)))); + assert!(matches!(result, Err(error) if !error.dispatched() && error.is_request())); } #[rstest] @@ -466,7 +460,34 @@ async fn upgrade_validates_headers_and_negotiated_protocol( match result { Ok(mut socket) if accepted => round_trip(&mut socket).await, - Err(Error::Network(_)) if !accepted => {} + Err(error) if !accepted && error.dispatched() => {} other => panic!("unexpected handshake outcome: {:?}", other.map(drop)), } } + +#[rstest] +#[tokio::test] +async fn a_handshake_that_never_answers_times_out_as_dispatched() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + read_head(&mut stream).await; + let _ = stream.read(&mut [0u8; 1]).await; + }); + let client = HttpClientPool::new(Arc::new(PublicDnsResolver)) + .client(&config(Verify::BuiltInRoots), ClientVariant::WebSocket) + .unwrap(); + + let error = connect( + &client, + &format!("ws://{address}/responses"), + HeaderMap::new(), + Some(std::time::Duration::from_millis(100)), + ) + .await + .map(drop) + .unwrap_err(); + + assert!(error.dispatched()); +} diff --git a/litellm-rust/crates/inference-chat/Cargo.toml b/litellm-rust/crates/inference-chat/Cargo.toml index fa3fad067a9..9815f3b5c64 100644 --- a/litellm-rust/crates/inference-chat/Cargo.toml +++ b/litellm-rust/crates/inference-chat/Cargo.toml @@ -18,7 +18,6 @@ litellm-secrets.workspace = true serde_json = { workspace = true, features = ["preserve_order"] } futures-util.workspace = true -reqwest.workspace = true tracing.workspace = true [dev-dependencies] diff --git a/litellm-rust/crates/inference-chat/src/handler.rs b/litellm-rust/crates/inference-chat/src/handler.rs index 15a45747997..689310e9765 100644 --- a/litellm-rust/crates/inference-chat/src/handler.rs +++ b/litellm-rust/crates/inference-chat/src/handler.rs @@ -90,39 +90,32 @@ pub(super) async fn execute( interceptors, observers, || async move { - let outbound = outbound_request( + let outbound = litellm_inference::outbound::outbound_request( Authenticated { headers: wire.headers, signer: authenticated.signer, }, &wire.url, &wire.body, - call.timeout, + Some( + call.timeout + .unwrap_or(Duration::from_secs(CHAT_COMPLETIONS_TIMEOUT_SECS)), + ), )?; let response = litellm_inference::outbound::send(outbound, http) .await - .map_err(|err| { - // Failing to establish the connection means the request never went out, - // so the host can still serve it. Everything else here, a timeout - // above all, may have reached the provider and been answered. - if err.is_connect() || err.is_builder() { - Error::Transport(litellm_http::transport::Error::Connect(err.to_string())) - } else { - Error::Transport(litellm_http::transport::Error::Network(err.to_string())) - } - })?; + .map_err(litellm_http::Error::from)?; let status = response.status(); - let text = response.text().await.map_err(|err| { - Error::Transport(litellm_http::transport::Error::Network(err.to_string())) - })?; + let text = response.text().await.map_err(litellm_http::Error::from)?; if !status.is_success() { - return Err(Error::Transport(litellm_http::transport::Error::Http { - status: status.as_u16(), - body: truncate_error_body(&text), - })); + return Err(litellm_http::Error::upstream( + status.as_u16(), + truncate_error_body(&text), + ) + .into()); } let raw = RawResponse { body: text.clone() }; if let Some(observers) = observers { @@ -161,34 +154,12 @@ pub(super) async fn execute( /// can only mean the provider was already called. pub(super) fn as_response_error(err: Error) -> Error { match err { - already @ (Error::InvalidResponse(_) - | Error::Transport(litellm_http::transport::Error::Http { .. })) => already, + already @ Error::InvalidResponse(_) => already, + upstream if upstream.upstream_response().is_some() => upstream, other => Error::InvalidResponse(other.to_string().into()), } } -pub(super) fn outbound_request( - authenticated: Authenticated, - url: &str, - body: &Value, - timeout: Option, -) -> Result { - litellm_inference::outbound::outbound_request( - authenticated, - url, - body, - Some(timeout.unwrap_or(Duration::from_secs(CHAT_COMPLETIONS_TIMEOUT_SECS))), - ) - .map_err(|error| match error { - // Python drops the caller's copy and prefers a forwarded Authorization - // over the signature, so leave the request to it. - litellm_http::Error::ComputedHeader(_) => { - Error::Unsupported("request forwards a header AWS SigV4 computes") - } - other => Error::Http(other), - }) -} - #[cfg(test)] mod tests { use std::sync::Mutex; @@ -341,10 +312,10 @@ mod tests { .await .expect_err("the upstream failure fails the call"); - assert!(matches!( - error, - Error::Transport(litellm_http::transport::Error::Http { status: 500, .. }) - )); + assert!( + matches!(&error, Error::Http(http) if http.upstream_response().is_some_and(|(status, _)| status == 500)), + "{error:?}" + ); assert!(interceptors.raw.into_inner().unwrap().is_empty()); } @@ -362,10 +333,7 @@ mod tests { "{label} must not stay retryable once the provider has answered" ); } - let upstream = Error::Transport(litellm_http::transport::Error::Http { - status: 500, - body: "boom".to_string(), - }); + let upstream = Error::from(litellm_http::Error::upstream(500, "boom".into())); assert_eq!(as_response_error(upstream.clone()), upstream); } } diff --git a/litellm-rust/crates/inference-chat/tests/chat_completions.rs b/litellm-rust/crates/inference-chat/tests/chat_completions.rs index 461d4146a63..ddcf2c0f5c9 100644 --- a/litellm-rust/crates/inference-chat/tests/chat_completions.rs +++ b/litellm-rust/crates/inference-chat/tests/chat_completions.rs @@ -5,7 +5,6 @@ use litellm_host::{ }; use std::time::Duration; -use litellm_http::transport::Error as TransportError; use litellm_inference_chat::{Error, types::ChatCompletionsRequest}; use litellm_llms_types::formats::chat_completions::ChatCompletionsResponse; use rstest::{fixture, rstest}; @@ -201,13 +200,7 @@ async fn an_upstream_error_status_keeps_its_code_and_body( .await .expect_err("upstream rejects"); - assert_eq!( - error, - Error::Transport(TransportError::Http { - status, - body: "slow down".into() - }) - ); + assert_eq!(error.upstream_response(), Some((status, "slow down"))); } #[rstest] @@ -223,7 +216,7 @@ async fn a_connection_that_is_never_established_returns_a_connect_error( .expect_err("nothing is listening"); assert!( - matches!(error, Error::Transport(TransportError::Connect(_))), + matches!(&error, Error::Http(http) if !http.dispatched()), "{error:?}" ); } @@ -244,7 +237,7 @@ async fn a_timeout_after_sending_returns_a_network_error(request: ChatCompletion .expect_err("the call times out"); assert!( - matches!(error, Error::Transport(TransportError::Network(_))), + matches!(&error, Error::Http(http) if http.dispatched()), "{error:?}" ); } diff --git a/litellm-rust/crates/inference-messages/src/handler.rs b/litellm-rust/crates/inference-messages/src/handler.rs index 292995d842c..979791f949e 100644 --- a/litellm-rust/crates/inference-messages/src/handler.rs +++ b/litellm-rust/crates/inference-messages/src/handler.rs @@ -10,7 +10,7 @@ use litellm_core_utils::{ use litellm_host::interceptors::{ Cost, ExecutionFacts, Interceptors, ProviderIdentity, RequestContext, ResultSource, WireRequest, }; -use litellm_http::{request::with_default_headers, transport::Error as TransportError}; +use litellm_http::request::with_default_headers; use litellm_llms::base_llm::{ auth::{Authenticated, ValidatedEnvironment, resolve_auth}, messages::{ @@ -276,7 +276,7 @@ fn serialize_failure(err: serde_json::Error) -> Error { } fn network(error: reqwest::Error) -> Error { - Error::Transport(TransportError::Network(error.to_string())) + litellm_http::Error::from(error).into() } async fn send( @@ -302,10 +302,7 @@ async fn provider_error(response: reqwest::Response) -> Error { match response.text().await { Ok(text) => { log_error_body(status, &text); - Error::Transport(TransportError::Http { - status, - body: truncate_error_body(&text), - }) + litellm_http::Error::upstream(status, truncate_error_body(&text)).into() } Err(error) => network(error), } diff --git a/litellm-rust/crates/inference-messages/tests/messages/response.rs b/litellm-rust/crates/inference-messages/tests/messages/response.rs index b5d623a5582..cd87b3c8c4c 100644 --- a/litellm-rust/crates/inference-messages/tests/messages/response.rs +++ b/litellm-rust/crates/inference-messages/tests/messages/response.rs @@ -2,7 +2,6 @@ use litellm_host::{ interceptors::{ExecutionFacts, ResultSource}, lifecycle::ExecutionEvent, }; -use litellm_http::transport::Error as TransportError; use litellm_inference_messages::{MessagesCallResponse, messages_body}; use litellm_inference_testing::{ RecordingSecrets, http_config, no_secrets, provider_http, resources, @@ -166,11 +165,11 @@ async fn a_json_error_envelope_is_kept_verbatim(call: MessagesCall) { .await .expect_err("upstream error propagates"); - let Error::Transport(TransportError::Http { status, body }) = error else { + let Some((status, body)) = error.upstream_response() else { panic!("{error:?}"); }; assert_eq!(status, 400); - assert_eq!(serde_json::from_str::(&body).unwrap(), envelope); + assert_eq!(serde_json::from_str::(body).unwrap(), envelope); } #[rstest] @@ -188,11 +187,8 @@ async fn a_long_error_body_is_truncated_at_the_documented_cap(call: MessagesCall .expect_err("upstream error propagates"); assert_eq!( - error, - Error::Transport(TransportError::Http { - status: 500, - body: format!("{}... (truncated)", &long[..256]) - }) + error.upstream_response(), + Some((500, format!("{}... (truncated)", &long[..256]).as_str())) ); } @@ -216,11 +212,8 @@ async fn an_upstream_error_keeps_its_status_and_body(call: MessagesCall, #[case] .expect_err("upstream error propagates"); assert_eq!( - error, - Error::Transport(TransportError::Http { - status, - body: "upstream said no".into() - }) + error.upstream_response(), + Some((status, "upstream said no")) ); } @@ -259,7 +252,7 @@ async fn a_provider_slower_than_the_timeout_fails_the_call(call: MessagesCall) { .await .expect_err("the call times out"); - assert!(matches!(error, Error::Transport(_)), "{error:?}"); + assert!(matches!(error, Error::Http(_)), "{error:?}"); } #[rstest] diff --git a/litellm-rust/crates/inference-messages/tests/messages/stream.rs b/litellm-rust/crates/inference-messages/tests/messages/stream.rs index cda688107b5..777f7e7c3ba 100644 --- a/litellm-rust/crates/inference-messages/tests/messages/stream.rs +++ b/litellm-rust/crates/inference-messages/tests/messages/stream.rs @@ -260,13 +260,7 @@ async fn an_upstream_error_fails_the_call_without_opening_the_stream( .await .expect_err("upstream error propagates"); - assert_eq!( - error, - Error::Transport(litellm_http::transport::Error::Http { - status: 429, - body: body.into() - }) - ); + assert_eq!(error.upstream_response(), Some((429, body))); assert!(host.seen.into_inner().unwrap().is_empty()); } @@ -332,7 +326,7 @@ async fn the_timeout_covers_a_stalled_stream_body(call: MessagesCall) { .expect("the stalled stream gives up within the timeout") .expect_err("a stalled body fails the call"); - assert!(matches!(error, Error::Transport(_)), "{error:?}"); + assert!(matches!(error, Error::Http(_)), "{error:?}"); let seen = host.seen.into_inner().unwrap(); assert!( matches!(seen.as_slice(), [Seen::Open(_), Seen::Deliver(chunk)] if chunk.as_ref() == b"event: message_start\ndata: {}\n\n"), @@ -390,13 +384,7 @@ async fn the_sdk_returns_http_errors_before_opening_a_stream(call: MessagesCall) .err() .expect("upstream failure is returned by messages()"); - assert_eq!( - error, - Error::Transport(litellm_http::transport::Error::Http { - status: 429, - body: "slow down".into(), - }) - ); + assert_eq!(error.upstream_response(), Some((429, "slow down"))); } #[rstest] @@ -470,7 +458,7 @@ async fn the_sdk_yields_a_body_error_once_after_delivered_chunks(call: MessagesC .expect("the stalled body times out") .unwrap() .unwrap_err(); - assert!(matches!(error, Error::Transport(_)), "{error:?}"); + assert!(matches!(error, Error::Http(_)), "{error:?}"); assert!(chunks.next().await.is_none()); tokio::time::timeout(Duration::from_secs(5), connection) .await diff --git a/litellm-rust/crates/inference-ocr/tests/ocr/documents.rs b/litellm-rust/crates/inference-ocr/tests/ocr/documents.rs index 338a47c5f57..b23c764d66c 100644 --- a/litellm-rust/crates/inference-ocr/tests/ocr/documents.rs +++ b/litellm-rust/crates/inference-ocr/tests/ocr/documents.rs @@ -216,7 +216,10 @@ async fn configured_client_preserves_document_url_policy(#[case] allowed: bool) .await; if !allowed { - assert!(matches!(result, Err(Error::BlockedDocumentUrl))); + assert!( + matches!(&result, Err(Error::Http(error)) if error.is_request()), + "{result:?}" + ); assert!(received(&documents).await.is_empty()); assert!(received(&upstream).await.is_empty()); return; diff --git a/litellm-rust/crates/inference-responses/src/handler.rs b/litellm-rust/crates/inference-responses/src/handler.rs index cbbc8be139d..b94638c4c04 100644 --- a/litellm-rust/crates/inference-responses/src/handler.rs +++ b/litellm-rust/crates/inference-responses/src/handler.rs @@ -89,10 +89,10 @@ pub(super) async fn execute( let status = response.status().as_u16(); if !response.status().is_success() { let body = response.text().await.map_err(network)?; - return Err(litellm_http::transport::Error::Http { + return Err(litellm_http::Error::upstream( status, - body: litellm_http::request::truncate_error_body(&body), - } + litellm_http::request::truncate_error_body(&body), + ) .into()); } if stream { @@ -135,5 +135,5 @@ pub(super) async fn execute( } fn network(error: reqwest::Error) -> Error { - litellm_http::transport::Error::Network(error.to_string()).into() + litellm_http::Error::from(error).into() } diff --git a/litellm-rust/crates/inference-responses/src/websocket.rs b/litellm-rust/crates/inference-responses/src/websocket.rs index c2ca89ecbaf..35f82af9ad7 100644 --- a/litellm-rust/crates/inference-responses/src/websocket.rs +++ b/litellm-rust/crates/inference-responses/src/websocket.rs @@ -7,7 +7,7 @@ use litellm_http::{ }; use tokio::sync::Mutex; use tokio_tungstenite::tungstenite::{ - Message, + self, Message, http::{HeaderMap, HeaderName, HeaderValue}, }; @@ -43,16 +43,7 @@ impl ResponsesWebSocketConnection { )) }) .collect::>()?; - let connect = websocket::connect(client, url, headers); - let socket = match timeout { - Some(timeout) => tokio::time::timeout(timeout, connect).await.map_err(|_| { - Error::Transport(litellm_http::transport::Error::Network( - "Responses WebSocket connection timed out".into(), - )) - })?, - None => connect.await, - } - .map_err(Error::Transport)?; + let socket = websocket::connect(client, url, headers, timeout).await?; Ok(Self { socket: Arc::new(Mutex::new(Some(socket))), }) @@ -70,13 +61,12 @@ impl ResponsesWebSocketConnection { litellm_inference::diagnostic::operation("litellm.websocket.send_text", async { let mut socket = self.socket.lock().await; let Some(socket) = socket.as_mut() else { - return Err(Error::Transport(litellm_http::transport::Error::Network( - "Responses WebSocket is closed".into(), - ))); + return Err(litellm_http::Error::from(tungstenite::Error::AlreadyClosed).into()); }; - socket.send(Message::Text(text)).await.map_err(|error| { - Error::Transport(litellm_http::transport::Error::Network(error.to_string())) - }) + socket + .send(Message::Text(text)) + .await + .map_err(|error| litellm_http::Error::from(error).into()) }) .await } @@ -100,9 +90,7 @@ impl ResponsesWebSocketConnection { .map_err(|error| Error::InvalidResponse(error.to_string().into())), Some(Ok(Message::Close(_))) | None => Ok(None), Some(Ok(_)) => Ok(None), - Some(Err(error)) => Err(Error::Transport(litellm_http::transport::Error::Network( - error.to_string(), - ))), + Some(Err(error)) => Err(litellm_http::Error::from(error).into()), } }) .await @@ -118,9 +106,10 @@ impl ResponsesWebSocketConnection { litellm_inference::diagnostic::operation("litellm.websocket.close", async { let mut socket = self.socket.lock().await; if let Some(socket) = socket.as_mut() { - socket.close(None).await.map_err(|error| { - Error::Transport(litellm_http::transport::Error::Network(error.to_string())) - })?; + socket + .close(None) + .await + .map_err(litellm_http::Error::from)?; } *socket = None; Ok(()) diff --git a/litellm-rust/crates/inference-transcription/src/handler.rs b/litellm-rust/crates/inference-transcription/src/handler.rs index 6541ae8263e..7ad1f63b4fb 100644 --- a/litellm-rust/crates/inference-transcription/src/handler.rs +++ b/litellm-rust/crates/inference-transcription/src/handler.rs @@ -57,18 +57,13 @@ pub(super) async fn execute( )?; let response = litellm_inference::outbound::send(outbound, http) .await - .map_err(|error| { - Error::Transport(litellm_http::transport::Error::Network(error.to_string())) - })?; + .map_err(litellm_http::Error::from)?; let status = response.status(); - let text = response.text().await.map_err(|error| { - Error::Transport(litellm_http::transport::Error::Network(error.to_string())) - })?; + let text = response.text().await.map_err(litellm_http::Error::from)?; if !status.is_success() { - return Err(Error::Transport(litellm_http::transport::Error::Http { - status: status.as_u16(), - body: truncate_error_body(&text), - })); + return Err( + litellm_http::Error::upstream(status.as_u16(), truncate_error_body(&text)).into(), + ); } let response_json = serde_json::from_str(&text).map_err(|error| { Error::InvalidResponse(litellm_llms::ErrorDetail::invalid( diff --git a/litellm-rust/crates/inference-transcription/tests/audio_transcription.rs b/litellm-rust/crates/inference-transcription/tests/audio_transcription.rs index eb252bdff7e..dd40ac84056 100644 --- a/litellm-rust/crates/inference-transcription/tests/audio_transcription.rs +++ b/litellm-rust/crates/inference-transcription/tests/audio_transcription.rs @@ -222,11 +222,8 @@ async fn an_upstream_error_keeps_its_status_and_body( .expect_err("upstream error propagates"); assert_eq!( - error, - Error::Transport(litellm_http::transport::Error::Http { - status, - body: "upstream said no".into() - }) + error.upstream_response(), + Some((status, "upstream said no")) ); } diff --git a/litellm-rust/crates/inference/src/error.rs b/litellm-rust/crates/inference/src/error.rs index 895c8c55e01..99511881639 100644 --- a/litellm-rust/crates/inference/src/error.rs +++ b/litellm-rust/crates/inference/src/error.rs @@ -7,7 +7,6 @@ use std::sync::Arc; -use litellm_http::transport::Error as TransportError; use litellm_llms::Error as LlmError; #[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)] @@ -30,11 +29,9 @@ pub enum Error { #[error(transparent)] Auth(#[from] litellm_auth::Error), #[error(transparent)] - Transport(#[from] TransportError), - #[error(transparent)] Headers(#[from] litellm_http::request::HeaderError), #[error(transparent)] - Http(#[from] litellm_http::Error), + Http(litellm_http::Error), #[error(transparent)] Secret(#[from] SecretError), #[error("post-call hook failed: {0}")] @@ -67,15 +64,32 @@ impl Error { | Self::Unsupported(_) | Self::Headers(_) => true, Self::Auth(error) => !matches!(error, litellm_auth::Error::MissingApiKey { .. }), - Self::InvalidResponse(_) - | Self::Transport(_) - | Self::Http(_) - | Self::Secret(_) - | Self::PostCallHook(_) => false, + Self::Http(error) => error.is_request(), + Self::InvalidResponse(_) | Self::Secret(_) | Self::PostCallHook(_) => false, } } } +impl Error { + pub fn upstream_response(&self) -> Option<(u16, &str)> { + match self { + Self::Http(error) => error.upstream_response(), + _ => None, + } + } +} + +impl From for Error { + fn from(error: litellm_http::Error) -> Self { + // Python drops the caller's copy and prefers a forwarded Authorization + // over the signature, so leave the request to it. + if error.is_computed_header() { + return Self::Unsupported("request forwards a header AWS SigV4 computes"); + } + Self::Http(error) + } +} + impl From for Error { fn from(error: LlmError) -> Self { match error { @@ -132,6 +146,13 @@ mod tests { assert!(Error::InvalidRequest("top_k".into()).is_request()); assert!(!Error::InvalidResponse("bad json".into()).is_request()); } + #[test] + fn a_header_the_signer_computes_is_left_to_the_host_on_every_route() { + let error = Error::from(litellm_http::Error::computed_header("authorization")); + assert!(matches!(error, Error::Unsupported(_)), "{error:?}"); + assert!(error.is_request()); + } + #[rstest] #[case::request(true)] #[case::response(false)] diff --git a/litellm-rust/crates/llms/src/azure_ai/ocr/document_intelligence/transformation.rs b/litellm-rust/crates/llms/src/azure_ai/ocr/document_intelligence/transformation.rs index 2a50989f4e4..48de38c44c5 100644 --- a/litellm-rust/crates/llms/src/azure_ai/ocr/document_intelligence/transformation.rs +++ b/litellm-rust/crates/llms/src/azure_ai/ocr/document_intelligence/transformation.rs @@ -505,7 +505,7 @@ async fn poll_operation( tokio::time::timeout_at(deadline, litellm_http::request::http_request(builder)) .await .map_err(|_| Error::PollTimeout)? - .map_err(litellm_http::transport::Error::from)?; + .map_err(litellm_http::Error::from)?; let retry = response .headers() .get(reqwest::header::RETRY_AFTER) diff --git a/litellm-rust/crates/llms/src/base_llm/ocr/document.rs b/litellm-rust/crates/llms/src/base_llm/ocr/document.rs index 57976240e1b..bed760b3781 100644 --- a/litellm-rust/crates/llms/src/base_llm/ocr/document.rs +++ b/litellm-rust/crates/llms/src/base_llm/ocr/document.rs @@ -1,9 +1,6 @@ use base64::{Engine, engine::general_purpose::STANDARD}; use data_url::{DataUrl, DataUrlError, forgiving_base64::DecodeError, mime::Mime}; -use litellm_http::{ - media::{DownloadPolicy, Error as MediaError, MediaFetcher}, - transport::Error as TransportError, -}; +use litellm_http::media::{DownloadPolicy, MediaFetcher}; use reqwest::Url; use crate::base_llm::ocr::{ @@ -73,8 +70,7 @@ pub async fn inline_remote_document( max_redirects: OCR_MAX_FETCH_REDIRECTS, }, ) - .await - .map_err(map_media_error)?; + .await?; let result = document.with_source(format!( "data:{};base64,{}", downloaded.content_type, @@ -84,28 +80,6 @@ pub async fn inline_remote_document( Ok(result) } -fn map_media_error(error: MediaError) -> Error { - match error { - MediaError::BlockedUrl => Error::BlockedDocumentUrl, - MediaError::DownloadDisabled => Error::DownloadDisabled, - MediaError::DownloadTooLarge => Error::DownloadTooLarge, - MediaError::TooManyRedirects => Error::TooManyRedirects, - MediaError::MissingRedirectLocation => Error::MissingRedirectLocation, - MediaError::InvalidRedirect => Error::InvalidRedirect, - MediaError::Http(status) => TransportError::Http { - status, - body: "OCR document download failed".into(), - } - .into(), - MediaError::Timeout => TransportError::Http { - status: 408, - body: "OCR document download timed out".into(), - } - .into(), - MediaError::Transport(error) => error.into(), - } -} - #[cfg(test)] mod tests { use std::collections::BTreeMap as Map; diff --git a/litellm-rust/crates/llms/src/base_llm/ocr/error.rs b/litellm-rust/crates/llms/src/base_llm/ocr/error.rs index 392dd56412c..3c5c692e76c 100644 --- a/litellm-rust/crates/llms/src/base_llm/ocr/error.rs +++ b/litellm-rust/crates/llms/src/base_llm/ocr/error.rs @@ -38,14 +38,6 @@ pub enum Error { ReductoSource, #[error("inline OCR document exceeds the size limit")] InlineDocumentTooLarge, - #[error("OCR document URL is blocked by network policy")] - BlockedDocumentUrl, - #[error("OCR document downloads are disabled")] - DownloadDisabled, - #[error("OCR document download exceeds the size limit")] - DownloadTooLarge, - #[error("OCR document download exceeded the redirect limit")] - TooManyRedirects, #[error("invalid OCR pages: {0}")] Pages(String), #[error("invalid OCR features")] @@ -58,10 +50,6 @@ pub enum Error { ResponseField { path: String }, #[error("OCR response is missing non-empty content")] EmptyContent, - #[error("OCR document redirect is missing a location")] - MissingRedirectLocation, - #[error("OCR document redirect location is invalid")] - InvalidRedirect, #[error("OCR operation ended with status {0}")] OperationStatus(String), #[error("OCR response numeric value is out of range: {0}")] @@ -103,8 +91,6 @@ pub enum Error { #[error(transparent)] Auth(#[from] litellm_auth::Error), #[error(transparent)] - Transport(#[from] litellm_http::transport::Error), - #[error(transparent)] Params(#[from] litellm_core_utils::params::Error), #[error(transparent)] Headers(#[from] litellm_http::request::HeaderError), @@ -133,57 +119,52 @@ impl From for Error { impl Error { pub fn http_status_code(&self) -> Option { match self { - Self::Provider { status, .. } - | Self::Transport(litellm_http::transport::Error::Http { status, .. }) => Some(*status), + Self::Provider { status, .. } => Some(*status), + Self::Http(error) if error.status().is_some() => error.status(), error if error.is_request() => Some(400), _ => None, } } pub fn is_request(&self) -> bool { - matches!( - self, + match self { Self::EmptyFile - | Self::InvalidMimeType(_) - | Self::CohereImageOnly - | Self::RequestFormat - | Self::RequestField { .. } - | Self::MissingField(_) - | Self::MissingDocumentUrl - | Self::InvalidDataUri - | Self::ReductoSource - | Self::InlineDocumentTooLarge - | Self::BlockedDocumentUrl - | Self::DownloadDisabled - | Self::DownloadTooLarge - | Self::TooManyRedirects - | Self::Pages(_) - | Self::Features - | Self::DotModel - | Self::InvalidRequest(_) - | Self::InvalidProvider(_) - | Self::InvalidModel { .. } - | Self::Params(_) - | Self::Headers(_) - | Self::Http(_) - | Self::Auth(litellm_auth::Error::MissingParam { .. }) - ) + | Self::InvalidMimeType(_) + | Self::CohereImageOnly + | Self::RequestFormat + | Self::RequestField { .. } + | Self::MissingField(_) + | Self::MissingDocumentUrl + | Self::InvalidDataUri + | Self::ReductoSource + | Self::InlineDocumentTooLarge + | Self::Pages(_) + | Self::Features + | Self::DotModel + | Self::InvalidRequest(_) + | Self::InvalidProvider(_) + | Self::InvalidModel { .. } + | Self::Params(_) + | Self::Headers(_) + | Self::Auth(litellm_auth::Error::MissingParam { .. }) => true, + Self::Http(error) => error.is_request(), + _ => false, + } } pub fn is_response(&self) -> bool { - matches!( - self, + match self { Self::TooLarge { .. } - | Self::ResponseField { .. } - | Self::EmptyContent - | Self::MissingRedirectLocation - | Self::InvalidRedirect - | Self::OperationStatus(_) - | Self::NumericRange(_) - | Self::PollLocation - | Self::PollOrigin - | Self::PollTimeout - | Self::InvalidResponse(_) - ) + | Self::ResponseField { .. } + | Self::EmptyContent + | Self::OperationStatus(_) + | Self::NumericRange(_) + | Self::PollLocation + | Self::PollOrigin + | Self::PollTimeout + | Self::InvalidResponse(_) => true, + Self::Http(error) => error.is_response(), + _ => false, + } } } diff --git a/litellm-rust/crates/llms/src/base_llm/ocr/handler.rs b/litellm-rust/crates/llms/src/base_llm/ocr/handler.rs index a9ddcb87086..6ff886f7f84 100644 --- a/litellm-rust/crates/llms/src/base_llm/ocr/handler.rs +++ b/litellm-rust/crates/llms/src/base_llm/ocr/handler.rs @@ -9,7 +9,6 @@ use litellm_http::{ media::{MediaFetcher, UrlPolicy}, outbound::{RequestSigner, json, signed}, request::header_pairs, - transport, }; use serde::{Serialize, de::DeserializeOwned}; use serde_json::Value; @@ -134,13 +133,14 @@ pub async fn ocr( .provider_http() .execute(http) .await - .map_err(transport_error)?; + .map_err(litellm_http::Error::from)?; if !response.status().is_success() { let headers = header_pairs(response.headers()); return match read_response_bytes(response, request.connection.max_response_bytes).await { - Err(Error::Transport(transport::Error::Http { status, body })) => { - Err(config.get_error_class(body, status, headers)) - } + Err(Error::Http(error)) => Err(match error.upstream_response() { + Some((status, body)) => config.get_error_class(body.to_owned(), status, headers), + None => Error::Http(error), + }), Err(error) => Err(error), Ok(_) => unreachable!("non-success response produces an HTTP error"), }; @@ -180,7 +180,7 @@ pub async fn read_response_bytes( return Err(Error::TooLarge { limit }); } let mut bytes = BytesMut::new(); - while let Some(chunk) = response.chunk().await.map_err(transport_error)? { + while let Some(chunk) = response.chunk().await.map_err(litellm_http::Error::from)? { let remaining = limit.saturating_sub(bytes.len()); if status.is_success() && chunk.len() > remaining { return Err(Error::TooLarge { limit }); @@ -191,25 +191,15 @@ pub async fn read_response_bytes( } } if !status.is_success() { - return Err(transport::Error::Http { - status: status.as_u16(), - body: String::from_utf8_lossy(&bytes).into_owned(), - } + return Err(litellm_http::Error::upstream( + status.as_u16(), + String::from_utf8_lossy(&bytes).into_owned(), + ) .into()); } Ok(bytes.freeze()) } -pub fn transport_error(error: reqwest::Error) -> Error { - if error.is_timeout() { - return Error::Transport(transport::Error::Http { - status: 408, - body: "OCR request timed out".into(), - }); - } - transport::Error::from(error).into() -} - pub async fn transform_request_body( config: &C, request: &PreparedOcrRequest, @@ -309,8 +299,6 @@ mod tests { net::TcpListener, }; - use super::*; - #[rstest::rstest] #[tokio::test] async fn request_timeout_has_an_http_408_status() { @@ -326,10 +314,10 @@ mod tests { .send() .await .unwrap_err(); - assert!(matches!( - transport_error(error), - Error::Transport(transport::Error::Http { status: 408, .. }) - )); + assert_eq!( + Error::from(litellm_http::Error::from(error)).http_status_code(), + Some(408) + ); server.abort(); } @@ -400,10 +388,9 @@ mod tests { .await .unwrap_err(); - let Error::Transport(litellm_http::transport::Error::Http { status, body }) = error else { + let Error::Http(error) = error else { panic!("unexpected error: {error}"); }; - assert_eq!(status, 429); - assert_eq!(body, prefix); + assert_eq!(error.upstream_response(), Some((429, prefix.as_str()))); } } diff --git a/litellm-rust/crates/llms/src/reducto/ocr/transformation.rs b/litellm-rust/crates/llms/src/reducto/ocr/transformation.rs index fc73ce7c014..8580e9e9fe0 100644 --- a/litellm-rust/crates/llms/src/reducto/ocr/transformation.rs +++ b/litellm-rust/crates/llms/src/reducto/ocr/transformation.rs @@ -530,7 +530,7 @@ async fn upload_bytes_async( ); let response = litellm_http::request::http_request(builder) .await - .map_err(litellm_http::transport::Error::from)?; + .map_err(litellm_http::Error::from)?; let uploaded = crate::base_llm::ocr::handler::read_json_response::( response, false, diff --git a/litellm-rust/crates/python-bridge/Cargo.toml b/litellm-rust/crates/python-bridge/Cargo.toml index 6fc05c43146..ad86d1f5e79 100644 --- a/litellm-rust/crates/python-bridge/Cargo.toml +++ b/litellm-rust/crates/python-bridge/Cargo.toml @@ -70,6 +70,7 @@ criterion.workspace = true futures-util.workspace = true rstest.workspace = true sqlx = { workspace = true, features = ["migrate"] } +tempfile.workspace = true tokio-tungstenite.workspace = true wiremock.workspace = true aws-sdk-secretsmanager = "1.117.0" diff --git a/litellm-rust/crates/python-bridge/src/errors.rs b/litellm-rust/crates/python-bridge/src/errors.rs index 4dab0297e73..0335742c699 100644 --- a/litellm-rust/crates/python-bridge/src/errors.rs +++ b/litellm-rust/crates/python-bridge/src/errors.rs @@ -1,4 +1,3 @@ -use litellm_http::transport::Error as TransportError; use litellm_inference::Error; use pyo3::{ exceptions::{PyRuntimeError, PyValueError}, @@ -20,11 +19,9 @@ pyo3::create_exception!( ); pub(crate) fn route_error_to_pyerr(error: Error) -> PyErr { - match error { - Error::Transport(TransportError::Http { status, body }) => { - RustUpstreamError::new_err((status, body)) - } - other => by_fault(other.is_request(), other.to_string()), + match error.upstream_response() { + Some((status, body)) => RustUpstreamError::new_err((status, body.to_owned())), + None => by_fault(error.is_request(), error.to_string()), } } @@ -45,8 +42,10 @@ mod tests { #[case::unsupported(Error::Unsupported("test capability"), true)] #[case::invalid_provider(Error::InvalidProvider("unknown".into()), true)] #[case::invalid_request(Error::InvalidRequest("empty messages".into()), true)] - #[case::connection(TransportError::Connect("unreachable".into()).into(), false)] - #[case::network(TransportError::Network("timed out".into()).into(), false)] + #[case::lost_connection( + litellm_http::Error::from(tokio_tungstenite::tungstenite::Error::ConnectionClosed).into(), + false + )] #[case::invalid_response(Error::InvalidResponse("missing usage".into()), false)] fn route_failures_are_terminal(#[case] error: Error, #[case] is_request: bool) { Python::initialize(); @@ -62,13 +61,8 @@ mod tests { fn transport_status_survives_python_mapping() { Python::initialize(); Python::attach(|py| { - let upstream = route_error_to_pyerr( - TransportError::Http { - status: 429, - body: "slow down".into(), - } - .into(), - ); + let upstream = + route_error_to_pyerr(litellm_http::Error::upstream(429, "slow down".into()).into()); assert!(upstream.is_instance_of::(py)); assert_eq!( upstream diff --git a/litellm-rust/crates/python-bridge/src/http.rs b/litellm-rust/crates/python-bridge/src/http.rs index 57bf8aafc4e..d1e6221ef0c 100644 --- a/litellm-rust/crates/python-bridge/src/http.rs +++ b/litellm-rust/crates/python-bridge/src/http.rs @@ -132,26 +132,14 @@ pub(crate) fn host_client(py: Python<'_>, variant: ClientVariant) -> PyResult PyErr { - match error { - litellm_http::Error::Read { - tls_source: TlsSource::ClientIdentity, - .. - } - | litellm_http::Error::InvalidPem { - tls_source: TlsSource::ClientIdentity, - .. - } => PyValueError::new_err( + match error.tls_source() { + Some(TlsSource::ClientIdentity) => PyValueError::new_err( "http_settings.ssl_certificate: expected a readable PEM certificate and private key", ), - litellm_http::Error::Read { - tls_source: TlsSource::CaBundle, - .. + Some(TlsSource::CaBundle) => { + PyValueError::new_err("http_settings.ssl_verify: expected a readable PEM CA bundle") } - | litellm_http::Error::InvalidPem { - tls_source: TlsSource::CaBundle, - .. - } => PyValueError::new_err("http_settings.ssl_verify: expected a readable PEM CA bundle"), - _ => PyValueError::new_err("http_settings: native HTTP client configuration is invalid"), + None => PyValueError::new_err("http_settings: native HTTP client configuration is invalid"), } } @@ -274,21 +262,28 @@ settings = types.SimpleNamespace(**defaults) fn client_error_uses_tls_source_when_paths_match() { Python::initialize(); Python::attach(|py| { - let path = PathBuf::from("/shared.pem"); - let ca_error = client_error(litellm_http::Error::InvalidPem { - path: path.clone(), - message: "invalid".into(), - tls_source: TlsSource::CaBundle, - }); + let file = tempfile::NamedTempFile::new().unwrap(); + std::fs::write(file.path(), b"not a certificate").unwrap(); + let path = file.path().to_path_buf(); + let defaults = Resolution::from(&HttpSettings::default()).config; + let build = |config: HttpClientConfig| { + HttpClientPool::new(Arc::new(PublicDnsResolver)) + .client(&config, ClientVariant::Provider) + .map(drop) + .unwrap_err() + }; + let ca_error = client_error(build(HttpClientConfig { + verify: Verify::CaBundle(path.clone()), + ..defaults.clone() + })); assert_eq!( ca_error.to_string(), "ValueError: http_settings.ssl_verify: expected a readable PEM CA bundle" ); - let client_error = client_error(litellm_http::Error::InvalidPem { - path, - message: "invalid".into(), - tls_source: TlsSource::ClientIdentity, - }); + let client_error = client_error(build(HttpClientConfig { + client_certificate: Some(litellm_http::ClientIdentity::Pem(path)), + ..defaults + })); assert!(client_error.is_instance_of::(py)); assert_eq!( client_error.to_string(), diff --git a/litellm-rust/crates/python-bridge/src/routes/inference.rs b/litellm-rust/crates/python-bridge/src/routes/inference.rs index c9eb60e47ed..8e02995b920 100644 --- a/litellm-rust/crates/python-bridge/src/routes/inference.rs +++ b/litellm-rust/crates/python-bridge/src/routes/inference.rs @@ -6,7 +6,6 @@ use litellm_callbacks_legacy_python::LoggingOperation; use litellm_core_utils::get_llm_provider_logic::get_custom_llm_provider; use litellm_host::{call::HostedMachine, protocol::Protocol}; use litellm_host_python::{PythonBinding, PythonHostCalls, from_py}; -use litellm_http::transport::Error as TransportError; use litellm_inference::Error; use litellm_secrets::source::SecretSource; use pyo3::{ @@ -125,15 +124,15 @@ impl InferenceHost { { return Ok(original); } - let native = match error { - Error::Transport(TransportError::Http { status, body }) => { - let error = RustUpstreamError::new_err((status, body)); + let native = match error.upstream_response() { + Some((status, body)) => { + let error = RustUpstreamError::new_err((status, body.to_owned())); error .value(py) .setattr("headers", Vec::<(String, String)>::new())?; error } - other => route_error_to_pyerr(other), + None => route_error_to_pyerr(error), }; let mapped = py .import(self.module)? diff --git a/litellm-rust/crates/python-bridge/src/routes/messages/host.rs b/litellm-rust/crates/python-bridge/src/routes/messages/host.rs index 5e445fcaa7d..8d5505514ad 100644 --- a/litellm-rust/crates/python-bridge/src/routes/messages/host.rs +++ b/litellm-rust/crates/python-bridge/src/routes/messages/host.rs @@ -3,7 +3,6 @@ use litellm_host_python::{PythonHostCalls, PythonOwned}; use bytes::Bytes; use litellm_host_python::{InvokeError, PythonBinding, from_py, to_py}; -use litellm_http::transport::Error as TransportError; use litellm_inference_messages::{ Error, MessagesCall, MessagesSettings, MessagesShaping, litellm_params, messages_body, route::{Messages, MessagesStreamHead}, @@ -69,14 +68,14 @@ fn project_litellm_params<'py>( } fn native_error(py: Python<'_>, error: Error) -> PyResult { + if let Some((status, body)) = error.upstream_response() { + let error = RustUpstreamError::new_err((status, body.to_owned())); + error + .value(py) + .setattr("headers", Vec::<(String, String)>::new())?; + return Ok(error); + } match error { - Error::Transport(TransportError::Http { status, body }) => { - let error = RustUpstreamError::new_err((status, body)); - error - .value(py) - .setattr("headers", Vec::<(String, String)>::new())?; - Ok(error) - } Error::InvalidRequest(message) => { let error = PyValueError::new_err(message.to_string()); error.value(py).setattr(REQUEST_ERROR_MARKER, true)?; @@ -470,7 +469,7 @@ mod tests { #[case::missing_field(Error::MissingField("max_tokens"), true)] #[case::unresolvable_provider(Error::InvalidProvider("openai".into()), false)] #[case::upstream_failure( - Error::Transport(TransportError::Http { status: 400, body: "bad".into() }), + litellm_http::Error::upstream(400, "bad".into()).into(), false, )] fn only_request_rejections_carry_the_request_error_marker( diff --git a/litellm-rust/crates/python-bridge/src/routes/ocr/errors.rs b/litellm-rust/crates/python-bridge/src/routes/ocr/errors.rs index 2068b6a6e4b..59662ba646e 100644 --- a/litellm-rust/crates/python-bridge/src/routes/ocr/errors.rs +++ b/litellm-rust/crates/python-bridge/src/routes/ocr/errors.rs @@ -15,7 +15,10 @@ pub(super) fn to_pyerr(error: Error) -> PyErr { body, headers, } => upstream_error(py, status, body, headers)?, - Error::Transport(litellm_http::transport::Error::Http { status, body }) => { + Error::Http(http) if let Some(status) = http.status() => { + let body = http + .upstream_response() + .map_or_else(|| http.to_string(), |(_, body)| body.to_owned()); upstream_error(py, status, body, Vec::new())? } Error::RequestFormat => { diff --git a/litellm-rust/crates/secrets-cyberark/src/error.rs b/litellm-rust/crates/secrets-cyberark/src/error.rs index 4f21647225b..72ca3416e28 100644 --- a/litellm-rust/crates/secrets-cyberark/src/error.rs +++ b/litellm-rust/crates/secrets-cyberark/src/error.rs @@ -19,7 +19,7 @@ pub enum Error { #[error("CyberArk client certificate could not be loaded")] ClientCertificate, #[error("CyberArk Conjur HTTP client could not be built")] - Client(#[redact] Box), + Client(#[redact] litellm_http::Error), #[error("invalid refresh interval")] RefreshInterval, #[error("invalid CyberArk Conjur endpoint")] diff --git a/litellm-rust/crates/secrets-cyberark/src/secret_manager/client.rs b/litellm-rust/crates/secrets-cyberark/src/secret_manager/client.rs index ea2262392d5..6c31ff73498 100644 --- a/litellm-rust/crates/secrets-cyberark/src/secret_manager/client.rs +++ b/litellm-rust/crates/secrets-cyberark/src/secret_manager/client.rs @@ -64,19 +64,12 @@ impl CyberArkSecretManager { }), ..config.clone() }; - let client = - pool.client(&config, ClientVariant::Provider) - .map_err(|error| match error { - litellm_http::Error::Read { - tls_source: TlsSource::ClientIdentity, - .. - } - | litellm_http::Error::InvalidPem { - tls_source: TlsSource::ClientIdentity, - .. - } => Error::ClientCertificate, - other => Error::Client(Box::new(other)), - })?; + let client = pool + .client(&config, ClientVariant::Provider) + .map_err(|error| match error.tls_source() { + Some(TlsSource::ClientIdentity) => Error::ClientCertificate, + _ => Error::Client(error), + })?; let endpoint = reqwest::Url::parse( &environment .get(CYBERARK_API_BASE)