Read the request body after the source and key checks, with the gateway's 256 MiB as the only limit, v0.64.1

Cargo.toml

@@ -54,7 +54,7 @@ categories = ["network-programming", "web-programming::http-server"]
 
 # 二进制走 CalVer，crate 走 SemVer —— 两者是两回事，卖给不同的人。
 # 这里是 crate 的版本。
-version = "0.64.0"
+version = "0.64.1"
 
 [workspace.dependencies]
 # ── 内部 crate ───────────────────────────────────────────

crates/tw-api/msg-codes.txt

@@ -367,7 +367,9 @@ gw.probe.connect
 gw.probe.key_rejected
 gw.probe.request_failed
 gw.probe.timeout
-gw.request.body_too_large
+gw.request.body_declared_too_large
+gw.request.body_over_limit
+gw.request.body_unreadable
 gw.route.all_selected_disabled
 gw.route.denied
 gw.route.failed

crates/tw-gateway/src/error.rs

@@ -44,6 +44,9 @@ pub enum Source {
     /// 也没空出来。**和上游限流一样回 429，另外带 `Retry-After`**：请求本身没问题，过一会儿
     /// 再来就能发出去 —— 客户端该退避再试，不是放弃。对外的词表里算 `rate_limited`
     Busy,
+    /// 请求体超过网关的上限（见 `server::intake`）。**413，不是 400**：Anthropic 的格式里
+    /// 它是 `request_too_large`，和 Anthropic 自己嫌请求太大时一样。对外的词表里算 `request`
+    TooLarge,
 }
 
 /// 上游都满着时告诉客户端过几秒再来（`Retry-After`）。
@@ -62,7 +65,7 @@ impl Source {
             Source::Request => "request",
             Source::RateLimited | Source::Busy => "rate_limited",
             Source::Denied => "denied",
-            Source::NotSupported => "request",
+            Source::NotSupported | Source::TooLarge => "request",
         }
     }
     fn status(&self) -> StatusCode {
@@ -75,6 +78,7 @@ impl Source {
             Source::RateLimited | Source::Busy => StatusCode::TOO_MANY_REQUESTS,
             Source::Denied => StatusCode::FORBIDDEN,
             Source::NotSupported => StatusCode::NOT_IMPLEMENTED,
+            Source::TooLarge => StatusCode::PAYLOAD_TOO_LARGE,
         }
     }
 }
@@ -173,6 +177,9 @@ impl GatewayError {
     pub fn request(detail: Msg) -> Self {
         Self::new(Source::Request, detail)
     }
+    pub fn too_large(detail: Msg) -> Self {
+        Self::new(Source::TooLarge, detail)
+    }
 }
 
 impl GatewayError {

crates/tw-gateway/src/forward.rs

@@ -140,17 +140,6 @@ pub fn map_reqwest_error(e: reqwest::Error) -> GatewayError {
     }
 }
 
-/// 请求体原样转发，只在这里做一次大小检查。
-pub fn check_body_size(body: &Bytes, max: usize) -> Result<(), GatewayError> {
-    if body.len() > max {
-        return Err(GatewayError::request(msg!(
-            "gw.request.body_too_large", size = body.len(), max = max =>
-            "The request body is {size} bytes, over the {max}-byte limit."
-        )));
-    }
-    Ok(())
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -283,11 +272,4 @@ mod tests {
             "/v1beta/models/small:streamGenerateContent"
         );
     }
-
-    #[test]
-    fn body_size_limit_reports_both_numbers() {
-        let e = check_body_size(&Bytes::from(vec![0u8; 10]), 5).unwrap_err();
-        assert!(e.message().contains("10") && e.message().contains('5'));
-        assert!(check_body_size(&Bytes::from(vec![0u8; 5]), 5).is_ok());
-    }
 }

crates/tw-gateway/src/server.rs

@@ -7,7 +7,6 @@ use axum::extract::{OriginalUri, RawQuery, State};
 use axum::http::HeaderMap;
 use axum::response::{IntoResponse, Response};
 use axum::routing::{any, get};
-use bytes::Bytes;
 
 use crate::error::GatewayError;
 use crate::health::Health;
@@ -16,6 +15,7 @@ use listing::{get_model, list_models};
 pub(crate) use pipeline::stream_fault;
 use tw_types::msg;
 
+mod intake;
 mod listing;
 mod pipeline;
 mod upgrade;
@@ -98,11 +98,11 @@ async fn passthrough(
     method: axum::http::Method,
     OriginalUri(uri): OriginalUri,
     RawQuery(query): RawQuery,
-    // **必须排在 `body` 前面。**提取器按顺序跑，而 `Bytes` 会把体吃掉
-    // —— 一次升级要的是那条连接本身，体被读走之后就没得升了
+    // **必须排在 `body` 前面。**`body` 拿走的是整个请求
     crate::ws::MaybeUpgrade(upgrade): crate::ws::MaybeUpgrade,
     headers: HeaderMap,
-    body: Bytes,
+    // **这里还没读。**过完来源和密钥检查才读（见 `intake`）
+    body: axum::body::Body,
 ) -> Result<Response, GatewayError> {
     let started = std::time::Instant::now();
     // 从这一刻起它就算「在服务中」。**排队等并发名额的也算** —— 那时客户
@@ -135,9 +135,7 @@ async fn passthrough(
             .find(|c| c.name == client_name)
             .map(|c| tw_secret::mask_secret(&c.key)),
     };
-    // WebSocket 升级。**在鉴权之后、解体之前分叉** —— 鉴权
-    // 在前是因为一个不该连过来的地址不该有机会升级；解体之前是因为
-    // 升级要的是那条连接，而 `Bytes` 会把它读干净。
+    // WebSocket 升级。**在鉴权之后分叉** —— 一个不该连过来的地址不该有机会升级。
     if let Some(ws) = upgrade.filter(|_| crate::ws::is_upgrade(&headers)) {
         return upgrade::ws_upgrade(
             state,
@@ -189,6 +187,10 @@ async fn passthrough(
     // 交给管线里每一条 `return Err` 各自去报的话，漏掉一条的后果不是没报，
     // 而是被 Drop 报成「客户端取消」—— 一次策略拒绝会记到客户端头上。
     let mut ending: Option<crate::ending::Ending> = None;
+    // **到这里才读体**：来源、密钥、方法都过了。超了上限的还没有开始事件，不记一行
+    let body = intake::read(&headers, body, intake::MAX_BODY)
+        .await
+        .map_err(|e| e.in_dialect(dialect))?;
     let req = pipeline::Inbound {
         uri,
         query,