diff --git a/crates/aisix-proxy/AGENTS.md b/crates/aisix-proxy/AGENTS.md index e0615f96..412081d2 100644 --- a/crates/aisix-proxy/AGENTS.md +++ b/crates/aisix-proxy/AGENTS.md @@ -28,6 +28,29 @@ interval) so a model that is slow to its first token doesn't look like an abandoned connection to a proxy in front. Only for SSE: the same wrapper on an opaque binary passthrough (audio, images) corrupts it. +## Every terminal path emits the access log — including the ones that give up early + +The access log and `record_request` are emitted **by the handler**, at the end of +dispatch, because that is the only place that knows the provider, model and token +counts. A path that returns before reaching that tail therefore logs nothing, and +nothing errors: the caller gets a correct status while the gateway keeps no record +of the request, which is indistinguishable from the request never arriving. + +Two shapes give up early, and both must answer through +`reject::reject_before_dispatch` (it renders the envelope *and* emits the +telemetry, so the two can't drift apart): + +- **Middleware short-circuits** — anything that returns instead of calling + `next.run(request)` (see `enforce_request_body_limit`). These run ahead of + authentication, so they pass `api_key_id: None`. +- **Extractor rejections a handler unwraps at its top** — the + `Result, JsonRejection>` / `Result` parameters. + Auth already ran here, so pass the key id. + +A handler that instead wraps its whole dispatch and logs the wrapper's status +(`/mcp`, `/a2a`, `/passthrough`, `/v1/videos`, `/v1/files`) is already covered — +don't add a second emit to those, or the request logs twice. + ## A per-model gate must say whether it binds the requested entry or each target `resolve_attempt_models` expands a routing model into targets, so `model_entry` / diff --git a/crates/aisix-proxy/src/audio.rs b/crates/aisix-proxy/src/audio.rs index 3c4e3592..452aa1e1 100644 --- a/crates/aisix-proxy/src/audio.rs +++ b/crates/aisix-proxy/src/audio.rs @@ -286,17 +286,23 @@ pub async fn speech( // envelope — see completions.rs. body: Result, axum::extract::rejection::JsonRejection>, ) -> Response { + let started = Instant::now(); let Json(body) = match body { Ok(json) => json, + // Answer through `reject` — see completions.rs. Err(rej) => { - return crate::error::proxy_error_from_json_rejection( - rej, - state.request_body_limit_bytes, - ) - .into_response(); + return crate::reject::reject_before_dispatch( + &state, + "POST", + "/v1/audio/speech", + &client.request_id, + Some(&auth.entry.id), + started, + crate::reject::Envelope::OpenAi, + crate::error::proxy_error_from_json_rejection(rej, state.request_body_limit_bytes), + ); } }; - let started = Instant::now(); let request_id = client.request_id.clone(); let api_key_id = auth.entry.id.clone(); let model_name = body diff --git a/crates/aisix-proxy/src/chat.rs b/crates/aisix-proxy/src/chat.rs index 8e740db2..97a4dd4b 100644 --- a/crates/aisix-proxy/src/chat.rs +++ b/crates/aisix-proxy/src/chat.rs @@ -94,30 +94,21 @@ pub async fn chat_completions( let path = "/v1/chat/completions"; let mut req = match body { Ok(Json(r)) => r, + // Classify the body-extractor failure (malformed JSON vs 413 cap + // vs transport read error) via the shared helper, then answer + // through `reject` so the rejection lands in the access log and + // the request metrics like every other terminal path here. Err(rej) => { - use axum::extract::rejection::JsonRejection; - use axum::http::StatusCode; - // BytesRejection → distinguish 413 (PAYLOAD_TOO_LARGE, - // real per-extractor cap exceeded) from 400 (transport- - // side read failure). `JsonRejection` is `#[non_exhaustive]` - // so the fallback `_` arm catches today's JsonDataError - // (the #324 case) / JsonSyntaxError / MissingJsonContentType - // AND any future variant axum adds, defaulting to 400 - // until each new variant gets an explicit policy decision. - return match rej { - JsonRejection::BytesRejection(inner) - if inner.status() == StatusCode::PAYLOAD_TOO_LARGE => - { - ProxyError::RequestTooLarge { - limit_bytes: state.request_body_limit_bytes, - } - } - JsonRejection::BytesRejection(_) => { - ProxyError::InvalidRequest("failed to read request body".into()) - } - _ => ProxyError::InvalidRequest("invalid JSON request body".into()), - } - .into_response(); + return crate::reject::reject_before_dispatch( + &state, + method, + path, + &client.request_id, + Some(&auth.entry.id), + started, + crate::reject::Envelope::OpenAi, + crate::error::proxy_error_from_json_rejection(rej, state.request_body_limit_bytes), + ); } }; let request_id = client.request_id.clone(); diff --git a/crates/aisix-proxy/src/completions.rs b/crates/aisix-proxy/src/completions.rs index e81b5d1e..5ba0fa18 100644 --- a/crates/aisix-proxy/src/completions.rs +++ b/crates/aisix-proxy/src/completions.rs @@ -92,17 +92,25 @@ pub async fn completions( // chat.rs / messages.rs. body: Result, axum::extract::rejection::JsonRejection>, ) -> Response { + let started = Instant::now(); let Json(body) = match body { Ok(json) => json, + // Answer through `reject` so the refusal still produces the access + // log line + request metrics the handler tail emits for a served + // request — the tail it never reaches. Err(rej) => { - return crate::error::proxy_error_from_json_rejection( - rej, - state.request_body_limit_bytes, - ) - .into_response(); + return crate::reject::reject_before_dispatch( + &state, + "POST", + "/v1/completions", + &client.request_id, + Some(&auth.entry.id), + started, + crate::reject::Envelope::OpenAi, + crate::error::proxy_error_from_json_rejection(rej, state.request_body_limit_bytes), + ); } }; - let started = Instant::now(); let request_id = client.request_id.clone(); let api_key_id = auth.entry.id.clone(); let model_name = body diff --git a/crates/aisix-proxy/src/count_tokens.rs b/crates/aisix-proxy/src/count_tokens.rs index 86da2eab..ea0f8bf8 100644 --- a/crates/aisix-proxy/src/count_tokens.rs +++ b/crates/aisix-proxy/src/count_tokens.rs @@ -69,18 +69,24 @@ pub async fn count_tokens( Ok(a) => a, Err(e) => return e.into_anthropic_response(), }; + let started = Instant::now(); let Json(body) = match body { Ok(j) => j, + // Answer through `reject` — see messages.rs. Err(rej) => { - return crate::error::proxy_error_from_json_rejection( - rej, - state.request_body_limit_bytes, - ) - .into_anthropic_response(); + return crate::reject::reject_before_dispatch( + &state, + "POST", + "/v1/messages/count_tokens", + &client.request_id, + Some(&auth.entry.id), + started, + crate::reject::Envelope::Anthropic, + crate::error::proxy_error_from_json_rejection(rej, state.request_body_limit_bytes), + ); } }; - let started = Instant::now(); let request_id = client.request_id.clone(); let api_key_id = auth.entry.id.clone(); diff --git a/crates/aisix-proxy/src/embeddings.rs b/crates/aisix-proxy/src/embeddings.rs index d8827a2d..738e4c82 100644 --- a/crates/aisix-proxy/src/embeddings.rs +++ b/crates/aisix-proxy/src/embeddings.rs @@ -98,29 +98,20 @@ pub async fn embeddings( let api_key_id = auth.entry.id.clone(); let body = match body { Ok(Json(b)) => b, + // Classification stays in the shared helper so this route can't + // drift from its siblings on the 413-vs-400 rules; `reject` gives + // the refusal the same access log + metrics a served request gets. Err(rej) => { - use axum::extract::rejection::JsonRejection; - // BytesRejection → distinguish 413 (PAYLOAD_TOO_LARGE, - // real per-extractor cap exceeded) from 400 (transport- - // side read failure). `JsonRejection` is `#[non_exhaustive]` - // so the fallback `_` arm catches today's JsonDataError - // (the #401 case) / JsonSyntaxError / MissingJsonContentType - // AND any future variant axum adds, defaulting to 400 - // until each new variant gets an explicit policy decision. - return match rej { - JsonRejection::BytesRejection(inner) - if inner.status() == StatusCode::PAYLOAD_TOO_LARGE => - { - ProxyError::RequestTooLarge { - limit_bytes: state.request_body_limit_bytes, - } - } - JsonRejection::BytesRejection(_) => { - ProxyError::InvalidRequest("failed to read request body".into()) - } - _ => ProxyError::InvalidRequest("invalid JSON request body".into()), - } - .into_response(); + return crate::reject::reject_before_dispatch( + &state, + "POST", + "/v1/embeddings", + &request_id, + Some(&api_key_id), + started, + crate::reject::Envelope::OpenAi, + crate::error::proxy_error_from_json_rejection(rej, state.request_body_limit_bytes), + ); } }; let model_name = body.model.clone(); diff --git a/crates/aisix-proxy/src/images.rs b/crates/aisix-proxy/src/images.rs index 3708247a..5019831e 100644 --- a/crates/aisix-proxy/src/images.rs +++ b/crates/aisix-proxy/src/images.rs @@ -70,17 +70,23 @@ pub async fn image_generations( // envelope — see completions.rs. body: Result, axum::extract::rejection::JsonRejection>, ) -> Response { + let started = Instant::now(); let Json(body) = match body { Ok(json) => json, + // Answer through `reject` — see completions.rs. Err(rej) => { - return crate::error::proxy_error_from_json_rejection( - rej, - state.request_body_limit_bytes, - ) - .into_response(); + return crate::reject::reject_before_dispatch( + &state, + "POST", + "/v1/images/generations", + &client.request_id, + Some(&auth.entry.id), + started, + crate::reject::Envelope::OpenAi, + crate::error::proxy_error_from_json_rejection(rej, state.request_body_limit_bytes), + ); } }; - let started = Instant::now(); let request_id = client.request_id.clone(); let api_key_id = auth.entry.id.clone(); let model_name = body diff --git a/crates/aisix-proxy/src/jobs.rs b/crates/aisix-proxy/src/jobs.rs index 7626b413..7519b399 100644 --- a/crates/aisix-proxy/src/jobs.rs +++ b/crates/aisix-proxy/src/jobs.rs @@ -1027,17 +1027,23 @@ pub(crate) async fn create_batch( // text/plain rejection — see completions.rs. body: Result, ) -> Response { + let started = Instant::now(); let body = match body { Ok(bytes) => bytes, + // Answer through `reject` — see completions.rs. Err(rej) => { - return crate::error::proxy_error_from_bytes_rejection( - rej, - state.request_body_limit_bytes, - ) - .into_response(); + return crate::reject::reject_before_dispatch( + &state, + "POST", + "/v1/batches", + &client.request_id, + Some(&auth.entry.id), + started, + crate::reject::Envelope::OpenAi, + crate::error::proxy_error_from_bytes_rejection(rej, state.request_body_limit_bytes), + ); } }; - let started = Instant::now(); let request_id = client.request_id.clone(); let mut monitor_hits: Vec = Vec::new(); @@ -1267,17 +1273,23 @@ pub(crate) async fn create_ft_job( // text/plain rejection — see completions.rs. body: Result, ) -> Response { + let started = Instant::now(); let body = match body { Ok(bytes) => bytes, + // Answer through `reject` — see completions.rs. Err(rej) => { - return crate::error::proxy_error_from_bytes_rejection( - rej, - state.request_body_limit_bytes, - ) - .into_response(); + return crate::reject::reject_before_dispatch( + &state, + "POST", + "/v1/fine_tuning/jobs", + &client.request_id, + Some(&auth.entry.id), + started, + crate::reject::Envelope::OpenAi, + crate::error::proxy_error_from_bytes_rejection(rej, state.request_body_limit_bytes), + ); } }; - let started = Instant::now(); let request_id = client.request_id.clone(); let mut monitor_hits: Vec = Vec::new(); diff --git a/crates/aisix-proxy/src/lib.rs b/crates/aisix-proxy/src/lib.rs index 750cbdcd..f0a97726 100644 --- a/crates/aisix-proxy/src/lib.rs +++ b/crates/aisix-proxy/src/lib.rs @@ -53,6 +53,7 @@ mod passthrough; mod quota; mod realtime; mod redact; +mod reject; mod render; mod request_id; mod rerank; @@ -78,7 +79,7 @@ pub use state::{CacheBackends, ProxyState}; use axum::extract::State; use axum::http::{header, HeaderValue, Request}; use axum::middleware::{self, Next}; -use axum::response::{IntoResponse, Response}; +use axum::response::Response; use axum::routing::{any, get, post}; use axum::Router; use tower_http::set_header::SetResponseHeaderLayer; @@ -326,11 +327,18 @@ impl Drop for InFlightGuard { /// `fetch` both set Content-Length for non-streamed POSTs, and /// without this middleware they see ECONNRESET (indistinguishable /// from a network failure or a gateway crash) instead of 413. +/// +/// Both short-circuits answer through [`crate::reject`], so a request +/// refused here still produces the access-log line and request metrics +/// every other terminal path emits — the handler it never reached can't +/// do it. The logged latency spans the body drain below, which is what +/// an oversize request actually costs the gateway. async fn enforce_request_body_limit( State(state): State, request: Request, next: Next, ) -> Response { + let started = std::time::Instant::now(); // /v1/messages must emit the Anthropic-shape error envelope // (closes #336). The middleware runs BEFORE the handler so the // handler's `into_anthropic_response()` would never see the @@ -349,12 +357,10 @@ async fn enforce_request_body_limit( let path = request.uri().path(); let is_anthropic_path = path == "/v1/messages" || path == "/v1/messages/" || path == "/v1/messages/count_tokens"; - let render = |e: ProxyError| -> Response { - if is_anthropic_path { - e.into_anthropic_response() - } else { - e.into_response() - } + let envelope = if is_anthropic_path { + reject::Envelope::Anthropic + } else { + reject::Envelope::OpenAi }; // RFC 9110 §8.6 — a server SHOULD reject a request that carries // duplicate or conflicting `Content-Length` values rather than @@ -365,9 +371,16 @@ async fn enforce_request_body_limit( .iter(); let first = content_lengths.next(); if content_lengths.next().is_some() { - return render(ProxyError::InvalidRequest( - "conflicting Content-Length headers".into(), - )); + return reject::reject_before_dispatch( + &state, + request.method().as_str(), + request.uri().path(), + &request_id_of(&request), + None, + started, + envelope, + ProxyError::InvalidRequest("conflicting Content-Length headers".into()), + ); } // `0` = the cap is disabled; the duplicate-Content-Length rejection // above still applies — that one is request-smuggling hygiene, not a @@ -377,19 +390,45 @@ async fn enforce_request_body_limit( .and_then(|s| s.parse::().ok()) { if state.request_body_limit_bytes > 0 && declared > state.request_body_limit_bytes { + // Capture what the access log needs before the body move + // below consumes the request. + let method = request.method().clone(); + let path = request.uri().path().to_string(); + let request_id = request_id_of(&request); // Drain the inbound body so hyper can flush the 413 response // on the same HTTP/1.1 connection. Without this, hyper closes // the socket while the client is still writing, and the client // sees EPIPE/ECONNRESET instead of the 413. drain_body(request.into_body()).await; - return render(ProxyError::RequestTooLarge { - limit_bytes: state.request_body_limit_bytes, - }); + return reject::reject_before_dispatch( + &state, + method.as_str(), + &path, + &request_id, + None, + started, + envelope, + ProxyError::RequestTooLarge { + limit_bytes: state.request_body_limit_bytes, + }, + ); } } next.run(request).await } +/// The id `ensure_request_id` (the outermost layer) minted for this +/// request, so a rejection logged here joins the `x-aisix-request-id` the +/// caller was handed. The fallback only covers a router assembled without +/// that layer — every shipped path has it. +fn request_id_of(request: &Request) -> String { + request + .extensions() + .get::() + .map(|r| r.0.clone()) + .unwrap_or_else(request_id::new_request_id) +} + /// Read and discard the inbound body, bounded by both bytes and time. /// /// Byte cap (32 MiB) prevents a huge `Content-Length` from consuming @@ -1522,6 +1561,75 @@ mod tests { assert!(resp.status().is_server_error()); } + /// A body-cap rejection short-circuits BEFORE any handler runs, and + /// both the access log and the request metrics are emitted BY the + /// handlers — so pre-fix a caller got a 413 the gateway kept no + /// record of: nothing in the log, no `aisix_requests_total` sample. + /// "Client reports 413, gateway shows nothing" was indistinguishable + /// from the request never arriving. + /// + /// The metric is what this asserts, because it is per-`ProxyState` + /// and so unaffected by whatever else the suite is doing; the log + /// line — process-global tracing state, not safely assertable from a + /// parallel unit test — is pinned end-to-end in the `body-edges` E2E. + #[tokio::test] + async fn body_cap_short_circuit_is_counted_like_any_other_terminal_path() { + let hub = Arc::new(Hub::new()); + let snap = seed_snapshot("my-gpt4", &["my-gpt4"], "http://unused"); + let state = build_state(snap, hub); // 1 MiB cap from cfg() + let app = build_router(state.clone()); + + let oversized = 2 * 1024 * 1024; + let req = Request::builder() + .method("POST") + .uri("/v1/chat/completions") + .header("authorization", "Bearer sk-caller") + .header("content-type", "application/json") + .header("content-length", oversized.to_string()) + .body(Body::from( + r#"{"model":"my-gpt4","messages":[{"role":"user","content":"hi"}]}"#, + )) + .unwrap(); + let resp = run(app, req).await; + assert_eq!(resp.status(), StatusCode::PAYLOAD_TOO_LARGE); + + let scrape = state.metrics.render(); + assert!( + scrape.contains("aisix_requests_total") && scrape.contains(r#"status="413""#), + "the 413 must be counted, got: {scrape}" + ); + } + + /// The chunked path reaches the handler, which rejects at its body + /// extractor and returns before the dispatch tail — silent for the + /// same reason, one layer further in. Locks the handler-side half. + #[tokio::test] + async fn chunked_oversize_rejection_is_counted_like_any_other_terminal_path() { + let hub = Arc::new(Hub::new()); + let snap = seed_snapshot("my-gpt4", &["my-gpt4"], "http://unused"); + let state = build_state(snap, hub); + let app = build_router(state.clone()); + + let chunk = vec![b'x'; 200 * 1024]; + let stream = + futures::stream::iter((0..10).map(move |_| Ok::<_, std::io::Error>(chunk.clone()))); + let req = Request::builder() + .method("POST") + .uri("/v1/completions") + .header("authorization", "Bearer sk-caller") + .header("content-type", "application/json") + .body(Body::from_stream(stream)) + .unwrap(); + let resp = run(app, req).await; + assert_eq!(resp.status(), StatusCode::PAYLOAD_TOO_LARGE); + + let scrape = state.metrics.render(); + assert!( + scrape.contains(r#"status="413""#), + "the 413 must be counted, got: {scrape}" + ); + } + /// The duplicate-Content-Length rejection is smuggling hygiene, not /// a size limit — it must keep firing when the cap is disabled. #[tokio::test] diff --git a/crates/aisix-proxy/src/messages.rs b/crates/aisix-proxy/src/messages.rs index 94685a07..91bd75a4 100644 --- a/crates/aisix-proxy/src/messages.rs +++ b/crates/aisix-proxy/src/messages.rs @@ -84,22 +84,28 @@ pub async fn messages( Ok(a) => a, Err(e) => return e.into_anthropic_response(), }; + let started = Instant::now(); let Json(mut body) = match body { Ok(j) => j, Err(rej) => { // Classify the body-extractor failure (malformed JSON vs // 413 cap vs transport read error) via the shared helper so // /v1/messages and /v1/messages/count_tokens stay in lockstep - // on the discrimination rules, then render the Anthropic- - // shape envelope the Claude SDK can parse (#336). - return crate::error::proxy_error_from_json_rejection( - rej, - state.request_body_limit_bytes, - ) - .into_anthropic_response(); + // on the discrimination rules, then answer through `reject`, + // which renders the Anthropic-shape envelope the Claude SDK + // can parse (#336) and emits the access log + metrics. + return crate::reject::reject_before_dispatch( + &state, + "POST", + "/v1/messages", + &client.request_id, + Some(&auth.entry.id), + started, + crate::reject::Envelope::Anthropic, + crate::error::proxy_error_from_json_rejection(rej, state.request_body_limit_bytes), + ); } }; - let started = Instant::now(); let request_id = client.request_id.clone(); let api_key_id = auth.entry.id.clone(); diff --git a/crates/aisix-proxy/src/reject.rs b/crates/aisix-proxy/src/reject.rs new file mode 100644 index 00000000..36fcde3d --- /dev/null +++ b/crates/aisix-proxy/src/reject.rs @@ -0,0 +1,93 @@ +//! Terminal handling for requests rejected BEFORE dispatch. +//! +//! Every dispatching handler ends by emitting one access-log line plus the +//! request metrics for whatever it did. The paths that give up *before* +//! dispatch — the body-cap middleware's `Content-Length` short-circuit and +//! the body-extractor rejection each handler unwraps at its top — used to +//! `return` a bare response instead, so an oversize request was invisible: +//! a caller saw `413`, the operator saw nothing in the access log and no +//! `aisix_proxy_requests_total` sample. "Client reports 413, gateway has no +//! record of the request" was indistinguishable from the request never +//! arriving. +//! +//! Route every pre-dispatch rejection through [`reject_before_dispatch`] so +//! the family can't drift again: the rendered envelope and the telemetry are +//! produced by the same call. + +use std::time::Instant; + +use aisix_obs::{AccessLog, RequestOutcome}; +use axum::response::{IntoResponse, Response}; + +use crate::error::ProxyError; +use crate::state::ProxyState; +use crate::usage_attr::UNRESOLVED_MODEL_LABEL; + +/// Metric `provider` label for a rejection that never reached routing. +/// Matches what the handlers' own pre-dispatch error paths (auth, 404) +/// already record, so the series doesn't fork. +const UNRESOLVED_PROVIDER_LABEL: &str = "unknown"; + +/// Which wire envelope the caller expects. The Anthropic-protocol routes +/// (`/v1/messages`, `/v1/messages/count_tokens`) must answer in Anthropic +/// shape or the Claude SDK can't parse the error (#336) — the rejection +/// path is no exception. +#[derive(Clone, Copy)] +pub(crate) enum Envelope { + OpenAi, + Anthropic, +} + +/// Emit the access log + request metrics for a request refused before +/// dispatch, and render `err` into the caller's envelope. +/// +/// `api_key_id` is `None` for the middleware short-circuit, which runs +/// ahead of authentication — the request is refused on its declared size +/// alone, before any credential is read. +#[allow(clippy::too_many_arguments)] +pub(crate) fn reject_before_dispatch( + state: &ProxyState, + method: &str, + path: &str, + request_id: &str, + api_key_id: Option<&str>, + started: Instant, + envelope: Envelope, + err: ProxyError, +) -> Response { + let status = err.status().as_u16(); + let elapsed = started.elapsed(); + let (error_kind, error) = crate::attempt::access_log_error(&err); + AccessLog { + method, + path, + status, + latency: elapsed, + // Nothing is resolved this early: no upstream was picked, and the + // body naming the model is exactly what we refused to read. + provider: None, + model: None, + api_key_id, + prompt_tokens: None, + completion_tokens: None, + total_tokens: None, + request_id, + served_by_model: None, + routing_attempt_count: None, + routing_fallback_count: None, + error_kind: Some(error_kind), + error: Some(&error), + } + .emit(); + state.metrics.record_request( + UNRESOLVED_PROVIDER_LABEL, + UNRESOLVED_MODEL_LABEL, + status, + RequestOutcome::from_status(status), + elapsed, + ); + match envelope { + Envelope::OpenAi => err.into_response(), + Envelope::Anthropic => err.into_anthropic_response(), + } +} diff --git a/crates/aisix-proxy/src/rerank.rs b/crates/aisix-proxy/src/rerank.rs index 730dad90..4be1aef9 100644 --- a/crates/aisix-proxy/src/rerank.rs +++ b/crates/aisix-proxy/src/rerank.rs @@ -76,17 +76,23 @@ pub async fn rerank( // envelope — see completions.rs. body: Result, axum::extract::rejection::JsonRejection>, ) -> Response { + let started = Instant::now(); let Json(mut body) = match body { Ok(json) => json, + // Answer through `reject` — see completions.rs. Err(rej) => { - return crate::error::proxy_error_from_json_rejection( - rej, - state.request_body_limit_bytes, - ) - .into_response(); + return crate::reject::reject_before_dispatch( + &state, + "POST", + "/v1/rerank", + &client.request_id, + Some(&auth.entry.id), + started, + crate::reject::Envelope::OpenAi, + crate::error::proxy_error_from_json_rejection(rej, state.request_body_limit_bytes), + ); } }; - let started = Instant::now(); let request_id = client.request_id.clone(); let api_key_id = auth.entry.id.clone(); diff --git a/crates/aisix-proxy/src/responses.rs b/crates/aisix-proxy/src/responses.rs index d941bcf4..dd521ddf 100644 --- a/crates/aisix-proxy/src/responses.rs +++ b/crates/aisix-proxy/src/responses.rs @@ -154,17 +154,23 @@ pub async fn responses( // envelope — see completions.rs. body: Result, axum::extract::rejection::JsonRejection>, ) -> Response { + let started = Instant::now(); let Json(mut body) = match body { Ok(json) => json, + // Answer through `reject` — see completions.rs. Err(rej) => { - return crate::error::proxy_error_from_json_rejection( - rej, - state.request_body_limit_bytes, - ) - .into_response(); + return crate::reject::reject_before_dispatch( + &state, + "POST", + "/v1/responses", + &client.request_id, + Some(&auth.entry.id), + started, + crate::reject::Envelope::OpenAi, + crate::error::proxy_error_from_json_rejection(rej, state.request_body_limit_bytes), + ); } }; - let started = Instant::now(); let request_id = client.request_id.clone(); let api_key_id = auth.entry.id.clone(); diff --git a/tests/e2e/src/cases/body-edges-e2e.test.ts b/tests/e2e/src/cases/body-edges-e2e.test.ts index a8a8c6c4..da864dcc 100644 --- a/tests/e2e/src/cases/body-edges-e2e.test.ts +++ b/tests/e2e/src/cases/body-edges-e2e.test.ts @@ -46,6 +46,15 @@ const CALLER_KEY_HASH = createHash("sha256") .update(CALLER_PLAINTEXT) .digest("hex"); +/** An access-log line for a `/v1/completions` request the gateway refused. */ +const isRefusal = (line: string): boolean => + line.includes("proxy request completed") && + line.includes('path="/v1/completions"') && + line.includes("status=413"); + +const countRefusals = (output: string): number => + output.split("\n").filter(isRefusal).length; + describe("body edges e2e: multi-turn, oversize body, empty messages", () => { let app: SpawnedApp | undefined; let upstream: OpenAiUpstream | undefined; @@ -58,7 +67,9 @@ describe("body edges e2e: multi-turn, oversize body, empty messages", () => { if (!etcdReachable) return; upstream = await startOpenAiUpstream(); - app = await spawnApp(); + // `info` so the oversize case can assert on the access-log line the + // gateway emits for the request it refuses. + app = await spawnApp({ logLevel: "info" }); seed = new SeedClient(etcd, app.etcdPrefix); const pk = await seed.createProviderKey({ @@ -220,6 +231,113 @@ describe("body edges e2e: multi-turn, oversize body, empty messages", () => { expect(upstream.receivedRequests.length).toBe(upstreamHitsBefore); }); + test("oversize body: the gateway records the request it refused", async (ctx) => { + if (!etcdReachable || !app) { + ctx.skip(); + return; + } + + // The 413 above is answered by the body-cap middleware, which + // short-circuits BEFORE any handler runs — and the access log is + // emitted BY the handlers. That left an operator with nothing to + // look at: a caller reporting a 413 the gateway had no record of is + // indistinguishable from the request never arriving. + // + // (`observability.access_log` in the harness config is the reserved + // field nothing reads today — the access log is gated by the log + // level alone, which is why this suite raises it to `info`.) + const filler = "x".repeat(10 * 1024 * 1024 + 512 * 1024); + const res = await fetch(`${app.proxyUrl}/v1/chat/completions`, { + method: "POST", + headers: { + authorization: `Bearer ${CALLER_PLAINTEXT}`, + "content-type": "application/json", + }, + body: JSON.stringify({ + model: "body-edges", + messages: [{ role: "user", content: filler }], + }), + }); + expect(res.status).toBe(413); + + // Join the log line to THIS request instead of grepping for a bare + // `413` any other case in the suite could also have produced. + const requestId = res.headers.get("x-aisix-request-id"); + expect(requestId).toBeTruthy(); + + let line: string | undefined; + await waitConfigPropagation(async () => { + line = app! + .output() + .split("\n") + .find( + (l) => + l.includes(requestId!) && l.includes("proxy request completed"), + ); + return line !== undefined; + }); + expect(line).toMatch(/status=413/); + expect(line).toContain("/v1/chat/completions"); + // The reason, not just the status: `error_kind` carries the OpenAI + // envelope's coarse `invalid_request_error`, so a cap hit is only + // nameable through the message. + expect(line).toMatch(/request body exceeds/); + + // Same blindness on the metrics plane — no handler ran, so nothing + // counted the refusal either. + const scrape = await fetch(`${app.metricsUrl}/metrics`).then((r) => + r.text(), + ); + expect(scrape).toMatch(/aisix_requests_total\{[^}]*status="413"/); + }); + + test("chunked oversize body: the handler that rejected it records the refusal", async (ctx) => { + if (!etcdReachable || !app) { + ctx.skip(); + return; + } + + // No Content-Length, so the middleware can't judge the request up + // front: it reaches the handler, whose body extractor rejects once + // the cap is crossed. Same blindness one layer further in, so it + // needs its own coverage. `/v1/completions` isolates the count — + // nothing else in this suite calls it. + const before = countRefusals(app.output()); + + const chunk = "x".repeat(512 * 1024); + const body = new ReadableStream({ + start(controller) { + for (let i = 0; i < 22; i++) controller.enqueue(new TextEncoder().encode(chunk)); + controller.close(); + }, + }); + // A client streaming into a cap can legitimately lose the connection + // mid-write instead of reading the 413 — the failure mode the + // Content-Length path exists to avoid and this one cannot. The + // gateway's record is the subject here, so tolerate either outcome. + await fetch(`${app.proxyUrl}/v1/completions`, { + method: "POST", + headers: { + authorization: `Bearer ${CALLER_PLAINTEXT}`, + "content-type": "application/json", + }, + body, + duplex: "half", + } as RequestInit & { duplex: "half" }).catch(() => undefined); + + await waitConfigPropagation(async () => countRefusals(app!.output()) > before); + const line = app + .output() + .split("\n") + .filter((l) => isRefusal(l)) + .at(-1)!; + expect(line).toMatch(/status=413/); + expect(line).toMatch(/request body exceeds/); + // Auth ran before the body extractor here, so unlike the + // middleware's short-circuit the refusal is attributable to a caller. + expect(line).toMatch(/api_key_id/); + }); + test("empty messages array: 4xx with OpenAI-shape error envelope, upstream untouched", async (ctx) => { if (!etcdReachable || !app || !upstream) { ctx.skip(); diff --git a/tests/e2e/src/harness/app.ts b/tests/e2e/src/harness/app.ts index 7ef89c0a..d5bc9ea8 100644 --- a/tests/e2e/src/harness/app.ts +++ b/tests/e2e/src/harness/app.ts @@ -58,6 +58,18 @@ export interface AppOverrides { * AccessKey deliberately never travels on the config path. */ extraEnv?: Record; + /** + * Log level for the spawned binary. Defaults to `warn` — quiet enough + * that the suite's output stays readable. Tests that assert on a line + * the gateway emits at `info` (the access log) raise it here. + * + * Applied to BOTH `observability.log_level` and `RUST_LOG`: the DP's + * `init_tracing` tries `EnvFilter::try_from_default_env()` first, so + * the env var wins and setting the config key alone would silently do + * nothing. It also outranks an ambient `RUST_LOG`, so a developer + * debugging with `RUST_LOG=error` can't turn a test's subject off. + */ + logLevel?: string; /** * `observability.metrics.client_type_rules` (AISIX-Cloud#1045): operator * UA→client_type regex rules, tried before the built-in allowlist. @@ -205,7 +217,7 @@ async function spawnAppOnce(overrides: AppOverrides = {}): Promise { : { addr: `127.0.0.1:${adminPort}`, enabled: false }, observability: { service_name: "aisix-e2e", - log_level: "warn", + log_level: overrides.logLevel ?? "warn", access_log: false, metrics: { prometheus: { @@ -233,7 +245,7 @@ async function spawnAppOnce(overrides: AppOverrides = {}): Promise { for (const [k, v] of Object.entries(process.env)) { if (v !== undefined && !k.startsWith("AISIX_")) childEnv[k] = v; } - childEnv.RUST_LOG = process.env.RUST_LOG ?? "warn"; + childEnv.RUST_LOG = overrides.logLevel ?? process.env.RUST_LOG ?? "warn"; childEnv.HTTP_PROXY = ""; childEnv.HTTPS_PROXY = ""; childEnv.ALL_PROXY = "";