diff --git a/backend/crates/gateway-admin/src/model/settings.rs b/backend/crates/gateway-admin/src/model/settings.rs index 52ecc1c26..0bb8de20b 100644 --- a/backend/crates/gateway-admin/src/model/settings.rs +++ b/backend/crates/gateway-admin/src/model/settings.rs @@ -28,6 +28,7 @@ pub struct RuntimeSettings { pub max_waiting_per_key: u32, pub max_waiting_per_account: u32, pub concurrency_wait_timeout_seconds: u32, + pub responses_max_decompressed_body_bytes: u64, pub rotation_strategy: RotationStrategy, pub min_codex_desktop_version: Option, pub min_codex_cli_version: Option, @@ -57,6 +58,7 @@ pub struct ReplaceRuntimeSettings { pub max_waiting_per_key: u32, pub max_waiting_per_account: u32, pub concurrency_wait_timeout_seconds: u32, + pub responses_max_decompressed_body_bytes: u64, pub rotation_strategy: RotationStrategy, pub min_codex_desktop_version: Option, pub min_codex_cli_version: Option, diff --git a/backend/crates/gateway-admin/src/use_case/settings.rs b/backend/crates/gateway-admin/src/use_case/settings.rs index af9dc5623..61000ceb4 100644 --- a/backend/crates/gateway-admin/src/use_case/settings.rs +++ b/backend/crates/gateway-admin/src/use_case/settings.rs @@ -115,6 +115,8 @@ impl SettingsService for DefaultSettingsService { fn validate_settings(command: &ReplaceRuntimeSettings) -> Result<(), AdminError> { let valid = command.request_location.validate().is_ok() + && command.responses_max_decompressed_body_bytes > 0 + && isize::try_from(command.responses_max_decompressed_body_bytes).is_ok() && command.refresh_margin_seconds > 0 && command.refresh_concurrency > 0 && command.max_concurrent_per_account > 0 diff --git a/backend/crates/gateway-admin/tests/use_case/accounts.rs b/backend/crates/gateway-admin/tests/use_case/accounts.rs index a8fc4e810..990e5bf34 100644 --- a/backend/crates/gateway-admin/tests/use_case/accounts.rs +++ b/backend/crates/gateway-admin/tests/use_case/accounts.rs @@ -984,6 +984,7 @@ impl SettingsStore for StaticSettingsStore { max_waiting_per_key: 0, max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, + responses_max_decompressed_body_bytes: 64 * 1024 * 1024, rotation_strategy: RotationStrategy::Smart, min_codex_desktop_version: None, min_codex_cli_version: None, diff --git a/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs b/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs index 5734fae0e..78757ec0b 100644 --- a/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs +++ b/backend/crates/gateway-admin/tests/use_case/freeze_recovery.rs @@ -36,6 +36,7 @@ fn runtime_settings(enabled: bool, probe_enabled: bool, adaptive: bool) -> Runti max_waiting_per_key: 0, max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, + responses_max_decompressed_body_bytes: 64 * 1024 * 1024, rotation_strategy: gateway_admin::model::settings::RotationStrategy::Smart, min_codex_desktop_version: None, min_codex_cli_version: None, diff --git a/backend/crates/gateway-admin/tests/use_case/observability.rs b/backend/crates/gateway-admin/tests/use_case/observability.rs index 85b9e0a37..a81029269 100644 --- a/backend/crates/gateway-admin/tests/use_case/observability.rs +++ b/backend/crates/gateway-admin/tests/use_case/observability.rs @@ -764,6 +764,7 @@ impl SettingsStore for FixtureSettingsStore { max_waiting_per_key: 0, max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, + responses_max_decompressed_body_bytes: 64 * 1024 * 1024, rotation_strategy: RotationStrategy::Smart, min_codex_desktop_version: None, min_codex_cli_version: None, diff --git a/backend/crates/gateway-admin/tests/use_case/settings.rs b/backend/crates/gateway-admin/tests/use_case/settings.rs index 90819b979..b3bd6134a 100644 --- a/backend/crates/gateway-admin/tests/use_case/settings.rs +++ b/backend/crates/gateway-admin/tests/use_case/settings.rs @@ -71,6 +71,7 @@ async fn settings_should_reject_zero_refresh_margin_before_store_call() { max_waiting_per_key: 0, max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, + responses_max_decompressed_body_bytes: 64 * 1024 * 1024, rotation_strategy: RotationStrategy::Smart, min_codex_desktop_version: None, min_codex_cli_version: None, diff --git a/backend/crates/gateway-api/src/admin/settings.rs b/backend/crates/gateway-api/src/admin/settings.rs index 9953a3638..ee60c49b4 100644 --- a/backend/crates/gateway-api/src/admin/settings.rs +++ b/backend/crates/gateway-api/src/admin/settings.rs @@ -44,6 +44,7 @@ pub struct RuntimeSettingsView { pub max_waiting_per_key: u32, pub max_waiting_per_account: u32, pub concurrency_wait_timeout_seconds: u32, + pub responses_max_decompressed_body_bytes: u64, pub rotation_strategy: String, pub min_codex_desktop_version: Option, pub min_codex_cli_version: Option, @@ -74,6 +75,7 @@ pub struct UpdateRuntimeSettingsRequest { pub max_waiting_per_key: u32, pub max_waiting_per_account: u32, pub concurrency_wait_timeout_seconds: u32, + pub responses_max_decompressed_body_bytes: u64, pub rotation_strategy: String, pub min_codex_desktop_version: Option, pub min_codex_cli_version: Option, @@ -104,6 +106,13 @@ impl UpdateRuntimeSettingsRequest { return Err(WireValidationError::new(field)); } } + if self.responses_max_decompressed_body_bytes == 0 + || isize::try_from(self.responses_max_decompressed_body_bytes).is_err() + { + return Err(WireValidationError::new( + "responsesMaxDecompressedBodyBytes", + )); + } if !(1..=120).contains(&self.concurrency_wait_timeout_seconds) { return Err(WireValidationError::new("concurrencyWaitTimeoutSeconds")); } @@ -181,6 +190,7 @@ impl UpdateRuntimeSettingsRequest { max_waiting_per_key: self.max_waiting_per_key, max_waiting_per_account: self.max_waiting_per_account, concurrency_wait_timeout_seconds: self.concurrency_wait_timeout_seconds, + responses_max_decompressed_body_bytes: self.responses_max_decompressed_body_bytes, rotation_strategy: RotationStrategy::parse(&self.rotation_strategy) .ok_or_else(|| WireValidationError::new("rotationStrategy"))?, min_codex_desktop_version: self.min_codex_desktop_version, @@ -216,6 +226,7 @@ impl From for RuntimeSettingsView { max_waiting_per_key: settings.max_waiting_per_key, max_waiting_per_account: settings.max_waiting_per_account, concurrency_wait_timeout_seconds: settings.concurrency_wait_timeout_seconds, + responses_max_decompressed_body_bytes: settings.responses_max_decompressed_body_bytes, rotation_strategy: settings.rotation_strategy.as_str().to_owned(), min_codex_desktop_version: settings.min_codex_desktop_version, min_codex_cli_version: settings.min_codex_cli_version, diff --git a/backend/crates/gateway-api/src/openai/responses/error.rs b/backend/crates/gateway-api/src/openai/responses/error.rs index a6998519b..6a0d899d8 100644 --- a/backend/crates/gateway-api/src/openai/responses/error.rs +++ b/backend/crates/gateway-api/src/openai/responses/error.rs @@ -42,7 +42,10 @@ pub enum RequestDecodeError { MalformedJson, /// 请求体解压后超过允许上限。 #[error("decompressed request body is too large")] - DecompressedBodyTooLarge, + DecompressedBodyTooLarge { + /// 当前请求快照中的解压输出上限。 + limit_bytes: usize, + }, /// 请求使用了网关不支持的 `Content-Encoding`。 #[error("request content-encoding `{encoding}` is not supported")] UnsupportedContentEncoding { @@ -108,9 +111,11 @@ impl RequestDecodeError { "Request body must be valid JSON.".to_owned(), None, ), - Self::DecompressedBodyTooLarge => ( + Self::DecompressedBodyTooLarge { limit_bytes } => ( "request_too_large", - "Decompressed request body exceeds the allowed size.".to_owned(), + format!( + "Decompressed request body exceeds the allowed size ({limit_bytes} bytes)." + ), None, ), Self::UnsupportedContentEncoding { encoding } => ( diff --git a/backend/crates/gateway-api/src/openai/responses/http.rs b/backend/crates/gateway-api/src/openai/responses/http.rs index 5cd23029e..20eb111a9 100644 --- a/backend/crates/gateway-api/src/openai/responses/http.rs +++ b/backend/crates/gateway-api/src/openai/responses/http.rs @@ -52,7 +52,11 @@ pub(crate) async fn responses( Ok(client) => client, Err(error) => return client_access_error_response(error), }; - let decoded = match decode_request_with_headers(&body, &headers) { + let decoded = match decode_request_with_headers( + &body, + &headers, + client.snapshot().responses_max_decompressed_body_bytes(), + ) { Ok(decoded) => decoded, Err(error) => { return protocol_error_response(StatusCode::BAD_REQUEST, error.protocol_body()); diff --git a/backend/crates/gateway-api/src/openai/responses/request.rs b/backend/crates/gateway-api/src/openai/responses/request.rs index 5710fa2b1..55f18aafa 100644 --- a/backend/crates/gateway-api/src/openai/responses/request.rs +++ b/backend/crates/gateway-api/src/openai/responses/request.rs @@ -327,13 +327,14 @@ impl fmt::Debug for DecodedResponsesRequest { pub fn decode_request_with_headers( body: &[u8], headers: &HeaderMap, + max_decompressed_bytes: usize, ) -> Result { - let body = decompress_request_body(body, headers)?; + let body = decompress_request_body(body, headers, max_decompressed_bytes)?; decode_request_inner(&body, &OpenAiRequestHeaders::from_headers(headers)) } -/// 下游请求体解压后的最大字节数；防御解压炸弹。 -const MAX_DECOMPRESSED_REQUEST_BYTES: usize = 64 * 1024 * 1024; +/// zstd 回溯窗口独立于输出上限，调整设置不能放大解码器内部窗口分配。 +const MAX_ZSTD_WINDOW_LOG: u32 = 26; /// 按 `Content-Encoding` 解压下游请求体。 /// @@ -342,6 +343,7 @@ const MAX_DECOMPRESSED_REQUEST_BYTES: usize = 64 * 1024 * 1024; fn decompress_request_body<'a>( body: &'a [u8], headers: &HeaderMap, + max_decompressed_bytes: usize, ) -> Result, RequestDecodeError> { let encoding = headers .get_all(axum::http::header::CONTENT_ENCODING) @@ -357,20 +359,20 @@ fn decompress_request_body<'a>( "" | "identity" => Ok(Cow::Borrowed(body)), "gzip" => { let decoder = flate2::read::MultiGzDecoder::new(body); - read_bounded(decoder).map(Cow::Owned) + read_bounded(decoder, max_decompressed_bytes).map(Cow::Owned) } "deflate" => { let decoder = flate2::read::ZlibDecoder::new(body); - read_bounded(decoder).map(Cow::Owned) + read_bounded(decoder, max_decompressed_bytes).map(Cow::Owned) } "zstd" => { let mut decoder = zstd::stream::read::Decoder::with_buffer(body) .map_err(|_| RequestDecodeError::MalformedJson)?; // 输出有界之外，也限制压缩帧声明的回溯窗口，防止解码器内部过量分配。 decoder - .window_log_max(MAX_DECOMPRESSED_REQUEST_BYTES.ilog2()) + .window_log_max(MAX_ZSTD_WINDOW_LOG) .map_err(|_| RequestDecodeError::MalformedJson)?; - read_bounded(decoder).map(Cow::Owned) + read_bounded(decoder, max_decompressed_bytes).map(Cow::Owned) } other => Err(RequestDecodeError::UnsupportedContentEncoding { encoding: other.to_owned(), @@ -378,12 +380,15 @@ fn decompress_request_body<'a>( } } -fn read_bounded(mut reader: R) -> Result, RequestDecodeError> { +fn read_bounded( + mut reader: R, + max_decompressed_bytes: usize, +) -> Result, RequestDecodeError> { let mut decoded = Vec::new(); let mut chunk = [0; 8192]; loop { - let remaining = MAX_DECOMPRESSED_REQUEST_BYTES - decoded.len(); - let read_limit = chunk.len().min(remaining + 1); + let remaining = max_decompressed_bytes - decoded.len(); + let read_limit = chunk.len().min(remaining.saturating_add(1)); let read = reader .read(&mut chunk[..read_limit]) .map_err(|_| RequestDecodeError::MalformedJson)?; @@ -391,15 +396,19 @@ fn read_bounded(mut reader: R) -> Result, RequestDecod return Ok(decoded); } if read > remaining { - return Err(RequestDecodeError::DecompressedBodyTooLarge); + return Err(RequestDecodeError::DecompressedBodyTooLarge { + limit_bytes: max_decompressed_bytes, + }); } // 默认 Vec 扩容和 read_to_end 的 EOF 探测可能突破输出上限； // 容量增长也受相同边界约束，越界探测只使用栈上分块缓冲区。 let required = decoded.len() + read; if required > decoded.capacity() { - let capacity = (decoded.capacity() * 2) + let capacity = decoded + .capacity() + .saturating_mul(2) .max(required) - .min(MAX_DECOMPRESSED_REQUEST_BYTES); + .min(max_decompressed_bytes); decoded.reserve_exact(capacity - decoded.len()); } decoded.extend_from_slice(&chunk[..read]); diff --git a/backend/crates/gateway-api/tests/admin/mod.rs b/backend/crates/gateway-api/tests/admin/mod.rs index 58a96f08f..017adcdc3 100644 --- a/backend/crates/gateway-api/tests/admin/mod.rs +++ b/backend/crates/gateway-api/tests/admin/mod.rs @@ -413,6 +413,7 @@ impl SettingsStore for MemorySettingsStore { max_waiting_per_key: command.max_waiting_per_key, max_waiting_per_account: command.max_waiting_per_account, concurrency_wait_timeout_seconds: command.concurrency_wait_timeout_seconds, + responses_max_decompressed_body_bytes: command.responses_max_decompressed_body_bytes, rotation_strategy: command.rotation_strategy, min_codex_desktop_version: command.min_codex_desktop_version, min_codex_cli_version: command.min_codex_cli_version, @@ -1281,6 +1282,7 @@ fn test_runtime_settings() -> RuntimeSettings { max_waiting_per_key: 0, max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, + responses_max_decompressed_body_bytes: 64 * 1024 * 1024, rotation_strategy: RotationStrategy::Smart, min_codex_desktop_version: None, min_codex_cli_version: None, diff --git a/backend/crates/gateway-api/tests/admin/settings.rs b/backend/crates/gateway-api/tests/admin/settings.rs index 7337bd095..2e284f634 100644 --- a/backend/crates/gateway-api/tests/admin/settings.rs +++ b/backend/crates/gateway-api/tests/admin/settings.rs @@ -50,6 +50,7 @@ fn update_body() -> Value { "maxWaitingPerKey": 0, "maxWaitingPerAccount": 0, "concurrencyWaitTimeoutSeconds": 30, + "responsesMaxDecompressedBodyBytes": 67108864, "rotationStrategy": "round_robin", "minCodexDesktopVersion": "26.825.6671", "minCodexCliVersion": "0.40.0", @@ -121,6 +122,7 @@ fn settings_response_should_cover_the_full_runtime_settings_contract() { max_waiting_per_key: 0, max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, + responses_max_decompressed_body_bytes: 64 * 1024 * 1024, rotation_strategy: RotationStrategy::RoundRobin, min_codex_desktop_version: Some("26.825.6671".to_owned()), min_codex_cli_version: Some("0.40.0".to_owned()), @@ -157,6 +159,7 @@ fn settings_response_should_cover_the_full_runtime_settings_contract() { "maxWaitingPerKey": 0, "maxWaitingPerAccount": 0, "concurrencyWaitTimeoutSeconds": 30, + "responsesMaxDecompressedBodyBytes": 67108864, "rotationStrategy": "round_robin", "minCodexDesktopVersion": "26.825.6671", "minCodexCliVersion": "0.40.0", @@ -217,6 +220,7 @@ fn settings_request_and_response_fields_should_stay_in_lockstep() { max_waiting_per_key: 0, max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, + responses_max_decompressed_body_bytes: 64 * 1024 * 1024, rotation_strategy: RotationStrategy::parse(&request.rotation_strategy) .expect("fixture rotation strategy"), min_codex_desktop_version: request.min_codex_desktop_version, @@ -575,3 +579,24 @@ async fn request_location_should_reject_invalid_or_missing_fields_without_replac original ); } + +#[test] +fn decompression_setting_should_reject_invalid_values() { + let field = "responsesMaxDecompressedBodyBytes"; + for invalid in [ + json!(0), + json!(-1), + json!(1.5), + json!(u64::MAX), + Value::Null, + ] { + let mut body = update_body(); + body[field] = invalid; + if let Ok(request) = serde_json::from_value::(body) { + assert_eq!(request.validate().unwrap_err().field(), field); + } + } + let mut body = update_body(); + body.as_object_mut().unwrap().remove(field); + assert!(serde_json::from_value::(body).is_err()); +} diff --git a/backend/crates/gateway-api/tests/openai/mod.rs b/backend/crates/gateway-api/tests/openai/mod.rs index d514d6b65..7347a6aa9 100644 --- a/backend/crates/gateway-api/tests/openai/mod.rs +++ b/backend/crates/gateway-api/tests/openai/mod.rs @@ -135,8 +135,18 @@ pub(super) fn authenticated_client_for_provider( plaintext: &str, provider_name: &str, ) -> AuthenticatedClient { + authenticated_client_for_provider_with_limit(plaintext, provider_name, 64 * 1024 * 1024) +} + +pub(super) fn authenticated_client_for_provider_with_limit( + plaintext: &str, + provider_name: &str, + bytes: usize, +) -> AuthenticatedClient { + let snapshot = snapshot(plaintext, provider_name) + .with_responses_max_decompressed_body_bytes(std::num::NonZeroUsize::new(bytes).unwrap()); let source = DefaultExecutionService::new( - RuntimeSnapshotHandle::new(snapshot(plaintext, provider_name)), + RuntimeSnapshotHandle::new(snapshot), Arc::new(UnusedExecutionStore), ProviderRegistry::default(), Arc::new(UnusedAdmissions), diff --git a/backend/crates/gateway-api/tests/openai/responses/http.rs b/backend/crates/gateway-api/tests/openai/responses/http.rs index 9395d7f65..06ec9eb04 100644 --- a/backend/crates/gateway-api/tests/openai/responses/http.rs +++ b/backend/crates/gateway-api/tests/openai/responses/http.rs @@ -548,7 +548,7 @@ async fn invalid_compressed_http_requests_should_fail_before_execution() { super::request::encode_body("zstd", std::io::repeat(0).take(64 * 1024 * 1024 + 1)) .into(), "request_too_large", - "Decompressed request body exceeds the allowed size.", + "Decompressed request body exceeds the allowed size (67108864 bytes).", ), ] { let headers = HeaderMap::from_iter([ @@ -2288,3 +2288,50 @@ async fn local_execution_failure_uses_the_existing_model_id() { } } } + +#[tokio::test] +async fn compressed_http_request_above_default_limit_should_reach_execution_after_increase() { + let body = json!({"model": "model-a", "input": "x".repeat(64 * 1024 * 1024)}).to_string(); + let compressed = super::request::encode_body("zstd", body.as_bytes()); + for limit in [64 * 1024 * 1024, 128 * 1024 * 1024] { + let observed = Arc::new(Mutex::new(None)); + let execution = Arc::new(ContextCaptureExecution { + observed: Arc::clone(&observed), + client: crate::openai::authenticated_client_for_provider_with_limit( + "sk_context_test", + "openai", + limit, + ), + }); + let response = api_router(execution) + .await + .oneshot( + Request::post("/v1/responses") + .header(AUTHORIZATION, "Bearer sk_context_test") + .header(CONTENT_ENCODING, "zstd") + .body(Body::from(compressed.clone())) + .unwrap(), + ) + .await + .unwrap(); + if limit == 64 * 1024 * 1024 { + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + assert!(observed.lock().unwrap().is_none()); + let error: Value = + serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap()) + .unwrap(); + assert_eq!(error["error"]["code"], "request_too_large"); + } else { + // 捕获执行服务刻意返回 500；断言完整输入到达执行层，而不是只验证路由状态码。 + let captured = observed + .lock() + .unwrap() + .take() + .expect("request reached execution"); + assert_eq!( + captured.input.unwrap().as_str().unwrap().len(), + 64 * 1024 * 1024 + ); + } + } +} diff --git a/backend/crates/gateway-api/tests/openai/responses/mod.rs b/backend/crates/gateway-api/tests/openai/responses/mod.rs index 114f7e97b..9fc6e7586 100644 --- a/backend/crates/gateway-api/tests/openai/responses/mod.rs +++ b/backend/crates/gateway-api/tests/openai/responses/mod.rs @@ -19,7 +19,7 @@ use gateway_protocol::openai::sse::parse_sse_events; use serde_json::{Value, json}; fn decode_request(body: &[u8]) -> Result { - decode_request_with_headers(body, &HeaderMap::new()) + decode_request_with_headers(body, &HeaderMap::new(), 64 * 1024 * 1024) } fn decode_response_create( @@ -205,8 +205,9 @@ fn decoder_should_preserve_connection_metadata_outside_the_openai_wire_body() { } }); - let decoded = decode_request_with_headers(body.to_string().as_bytes(), &headers) - .expect("request should decode"); + let decoded = + decode_request_with_headers(body.to_string().as_bytes(), &headers, 64 * 1024 * 1024) + .expect("request should decode"); assert_eq!(Value::Object(openai_wire_body(&decoded).clone()), body); assert_eq!( @@ -317,9 +318,12 @@ fn decoder_should_preserve_ordinary_request_headers_as_opaque_multivalues() { HeaderValue::from_static("client-is-update"), ); - let decoded = - decode_request_with_headers(br#"{"model":"smart-code","input":"hello"}"#, &headers) - .expect("opaque request headers should not affect body decoding"); + let decoded = decode_request_with_headers( + br#"{"model":"smart-code","input":"hello"}"#, + &headers, + 64 * 1024 * 1024, + ) + .expect("opaque request headers should not affect body decoding"); let entries = openai_protocol_context(&decoded) .get("opaque_request_headers") .and_then(Value::as_array) @@ -420,9 +424,12 @@ fn decoder_should_strip_http_transport_but_leave_source_headers_for_provider() { headers.insert("content-encoding", HeaderValue::from_static("identity")); headers.insert("x-openai-future-mode", HeaderValue::from_static("keep")); - let decoded = - decode_request_with_headers(br#"{"model":"smart-code","input":"hello"}"#, &headers) - .expect("decode request behind a reverse proxy"); + let decoded = decode_request_with_headers( + br#"{"model":"smart-code","input":"hello"}"#, + &headers, + 64 * 1024 * 1024, + ) + .expect("decode request behind a reverse proxy"); assert_eq!( openai_protocol_context(&decoded).get("opaque_request_headers"), @@ -482,7 +489,8 @@ fn downstream_client_headers_should_remain_opaque_without_losing_session_semanti frame["type"] = json!("response.create"); let opening = OpenAiRequestHeaders::from_headers(&headers); for decoded in [ - decode_request_with_headers(body.to_string().as_bytes(), &headers).unwrap(), + decode_request_with_headers(body.to_string().as_bytes(), &headers, 64 * 1024 * 1024) + .unwrap(), decode_response_create_with_context(&frame.to_string(), &opening).unwrap(), decode_response_create_with_context(&frame.to_string(), &opening).unwrap(), ] { @@ -565,7 +573,7 @@ fn decoder_should_preserve_unknown_arbitrary_precision_numbers() { let number = "12345678901234567890123456789012345678901234567890"; let body = format!(r#"{{"model":"smart-code","input":"hello","future_number":{number}}}"#); - let decoded = decode_request_with_headers(body.as_bytes(), &HeaderMap::new()) + let decoded = decode_request_with_headers(body.as_bytes(), &HeaderMap::new(), 64 * 1024 * 1024) .expect("opaque numeric field should decode"); let encoded = serde_json::to_string(openai_wire_body(&decoded)).expect("encode request body"); diff --git a/backend/crates/gateway-api/tests/openai/responses/request.rs b/backend/crates/gateway-api/tests/openai/responses/request.rs index f0ba6480d..cf499e7a9 100644 --- a/backend/crates/gateway-api/tests/openai/responses/request.rs +++ b/backend/crates/gateway-api/tests/openai/responses/request.rs @@ -41,7 +41,9 @@ fn http_decode_should_preserve_compressed_request_fields() { let expected = serde_json::from_slice::(REQUEST).expect("fixture JSON"); for encoding in ENCODINGS { let compressed = encode_body(encoding, REQUEST); - let decoded = decode_request_with_headers(&compressed, &headers(encoding)).expect(encoding); + let decoded = + decode_request_with_headers(&compressed, &headers(encoding), 64 * 1024 * 1024) + .expect(encoding); assert_eq!( openai_wire_body(&decoded), expected.as_object().expect("object") @@ -52,10 +54,10 @@ fn http_decode_should_preserve_compressed_request_fields() { #[test] fn http_decode_should_accept_plain_identity_and_case_insensitive_encodings() { for request_headers in [HeaderMap::new(), headers(""), headers(" identity ")] { - assert!(decode_request_with_headers(REQUEST, &request_headers).is_ok()); + assert!(decode_request_with_headers(REQUEST, &request_headers, 64 * 1024 * 1024).is_ok()); } let compressed = encode_body("gzip", REQUEST); - assert!(decode_request_with_headers(&compressed, &headers(" GZip ")).is_ok()); + assert!(decode_request_with_headers(&compressed, &headers(" GZip "), 64 * 1024 * 1024).is_ok()); } #[test] @@ -63,7 +65,7 @@ fn http_decode_should_reject_unknown_and_stacked_encodings() { let mut repeated = headers("identity"); repeated.append(CONTENT_ENCODING, HeaderValue::from_static("gzip")); for request_headers in [headers("br"), headers("gzip, zstd"), repeated] { - let error = decode_request_with_headers(REQUEST, &request_headers) + let error = decode_request_with_headers(REQUEST, &request_headers, 64 * 1024 * 1024) .expect_err("unsupported encoding"); assert!(matches!( error, @@ -82,7 +84,8 @@ fn http_decode_should_reject_corrupted_and_truncated_compressed_bodies() { let mut truncated = encode_body(encoding, REQUEST); truncated.truncate(truncated.len() - 4); for body in [b"not-a-compressed-stream".as_slice(), truncated.as_slice()] { - let error = decode_request_with_headers(body, &headers(encoding)).expect_err(encoding); + let error = decode_request_with_headers(body, &headers(encoding), 64 * 1024 * 1024) + .expect_err(encoding); assert_eq!( error.protocol_body().into_value(), json!({"error": { @@ -99,7 +102,9 @@ fn http_decode_should_read_all_gzip_members_and_zstd_frames() { for encoding in ["gzip", "zstd"] { let mut compressed = encode_body(encoding, &REQUEST[..12]); compressed.extend(encode_body(encoding, &REQUEST[12..])); - let decoded = decode_request_with_headers(&compressed, &headers(encoding)).expect(encoding); + let decoded = + decode_request_with_headers(&compressed, &headers(encoding), 64 * 1024 * 1024) + .expect(encoding); assert_eq!(openai_wire_body(&decoded)["input"], "hello"); } } @@ -110,7 +115,8 @@ fn http_decode_should_not_ignore_invalid_trailing_members() { let mut compressed = encode_body(encoding, REQUEST); compressed.extend(encode_body(encoding, b"invalid trailing JSON".as_slice())); assert_eq!( - decode_request_with_headers(&compressed, &headers(encoding)).expect_err(encoding), + decode_request_with_headers(&compressed, &headers(encoding), 64 * 1024 * 1024) + .expect_err(encoding), RequestDecodeError::MalformedJson ); } @@ -129,7 +135,8 @@ fn http_decode_should_reject_zstd_frames_with_an_excessive_window() { REQUEST ); assert_eq!( - decode_request_with_headers(&compressed, &headers("zstd")).expect_err("oversized window"), + decode_request_with_headers(&compressed, &headers("zstd"), 128 * 1024 * 1024) + .expect_err("oversized window"), RequestDecodeError::MalformedJson ); } @@ -138,7 +145,8 @@ fn http_decode_should_reject_zstd_frames_with_an_excessive_window() { fn http_decode_should_accept_the_exact_decompressed_limit() { let padding = repeat(b' ').take(DECOMPRESSED_LIMIT - REQUEST.len() as u64); let compressed = encode_body("zstd", REQUEST.chain(padding)); - let decoded = decode_request_with_headers(&compressed, &headers("zstd")).expect("exact limit"); + let decoded = decode_request_with_headers(&compressed, &headers("zstd"), 64 * 1024 * 1024) + .expect("exact limit"); assert_eq!(decoded.metadata().requested_model(), "gpt-test"); } @@ -146,11 +154,13 @@ fn http_decode_should_accept_the_exact_decompressed_limit() { fn http_decode_should_reject_bodies_above_the_decompressed_limit() { for encoding in ENCODINGS { let compressed = encode_body(encoding, repeat(0).take(DECOMPRESSED_LIMIT + 1)); - let error = - decode_request_with_headers(&compressed, &headers(encoding)).expect_err(encoding); + let error = decode_request_with_headers(&compressed, &headers(encoding), 64 * 1024 * 1024) + .expect_err(encoding); assert_eq!( error, - RequestDecodeError::DecompressedBodyTooLarge, + RequestDecodeError::DecompressedBodyTooLarge { + limit_bytes: 64 * 1024 * 1024 + }, "{encoding}" ); } @@ -163,12 +173,54 @@ fn http_decode_should_stop_at_limit_before_reading_invalid_trailing_frames() { // 这个断言验证越界时已经停止读取，而不只是最终返回了某个超限错误。 let mut compressed = encode_body(encoding, repeat(0).take(DECOMPRESSED_LIMIT + 1)); compressed.extend_from_slice(b"invalid trailing frame"); - let error = - decode_request_with_headers(&compressed, &headers(encoding)).expect_err(encoding); + let error = decode_request_with_headers(&compressed, &headers(encoding), 64 * 1024 * 1024) + .expect_err(encoding); assert_eq!( error, - RequestDecodeError::DecompressedBodyTooLarge, + RequestDecodeError::DecompressedBodyTooLarge { + limit_bytes: 64 * 1024 * 1024 + }, "{encoding}" ); } } + +#[test] +fn http_decode_should_apply_configured_limits_to_every_encoding_and_frame() { + for encoding in ENCODINGS { + let compressed = encode_body(encoding, REQUEST); + assert!( + decode_request_with_headers(&compressed, &headers(encoding), REQUEST.len()).is_ok() + ); + let error = decode_request_with_headers(&compressed, &headers(encoding), REQUEST.len() - 1) + .unwrap_err(); + assert_eq!( + error, + RequestDecodeError::DecompressedBodyTooLarge { + limit_bytes: REQUEST.len() - 1 + } + ); + assert!( + error + .protocol_body() + .error + .message + .contains(&(REQUEST.len() - 1).to_string()) + ); + } + for encoding in ["gzip", "zstd"] { + let split = REQUEST.len() / 2; + let mut compressed = encode_body(encoding, &REQUEST[..split]); + compressed.extend(encode_body(encoding, &REQUEST[split..])); + assert!( + decode_request_with_headers(&compressed, &headers(encoding), REQUEST.len()).is_ok() + ); + assert!(matches!( + decode_request_with_headers(&compressed, &headers(encoding), REQUEST.len() - 1), + Err(RequestDecodeError::DecompressedBodyTooLarge { .. }) + )); + } + for request_headers in [HeaderMap::new(), headers("identity")] { + assert!(decode_request_with_headers(REQUEST, &request_headers, 1).is_ok()); + } +} diff --git a/backend/crates/gateway-core/src/routing/snapshot.rs b/backend/crates/gateway-core/src/routing/snapshot.rs index 60875a02b..3ea7a9878 100644 --- a/backend/crates/gateway-core/src/routing/snapshot.rs +++ b/backend/crates/gateway-core/src/routing/snapshot.rs @@ -34,6 +34,7 @@ pub struct SnapshotSettingsFacts { max_waiting_per_key: u32, max_waiting_per_account: u32, concurrency_wait_timeout_seconds: u32, + responses_max_decompressed_body_bytes: u64, request_interval_ms: u64, rotation_strategy: String, model_mappings: BTreeMap, @@ -42,6 +43,12 @@ pub struct SnapshotSettingsFacts { } impl SnapshotSettingsFacts { + #[must_use] + pub const fn with_responses_max_decompressed_body_bytes(mut self, bytes: u64) -> Self { + self.responses_max_decompressed_body_bytes = bytes; + self + } + #[must_use] pub fn with_request_location( mut self, @@ -82,6 +89,7 @@ impl SnapshotSettingsFacts { max_waiting_per_key: 0, max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, + responses_max_decompressed_body_bytes: 64 * 1024 * 1024, request_interval_ms, rotation_strategy: rotation_strategy.into(), model_mappings, @@ -391,6 +399,12 @@ async fn compile_runtime_snapshot( { return Err(RuntimeSnapshotCompileError::InvalidData); } + let decompressed_body_limit = + isize::try_from(facts.settings.responses_max_decompressed_body_bytes) + .ok() + .and_then(|bytes| usize::try_from(bytes).ok()) + .and_then(std::num::NonZeroUsize::new) + .ok_or(RuntimeSnapshotCompileError::InvalidData)?; let queue_timeout = Duration::from_secs(u64::from(facts.settings.concurrency_wait_timeout_seconds)); let client_queue_policy = ConcurrencyQueuePolicy { @@ -492,6 +506,7 @@ async fn compile_runtime_snapshot( .map(|snapshot| { snapshot .with_request_location(request_location) + .with_responses_max_decompressed_body_bytes(decompressed_body_limit) .with_client_queue_policy(client_queue_policy) .with_model_mappings(model_mappings) .with_account_directory(account_directory) @@ -503,6 +518,7 @@ async fn compile_runtime_snapshot( /// 数据面使用的不可变配置快照。 #[derive(Debug, Clone)] pub struct RuntimeSnapshot { + responses_max_decompressed_body_bytes: std::num::NonZeroUsize, request_location: Option, revision: ConfigRevision, client_queue_policy: ConcurrencyQueuePolicy, @@ -520,6 +536,20 @@ pub struct RuntimeSnapshot { } impl RuntimeSnapshot { + #[must_use] + pub const fn responses_max_decompressed_body_bytes(&self) -> usize { + self.responses_max_decompressed_body_bytes.get() + } + + #[must_use] + pub const fn with_responses_max_decompressed_body_bytes( + mut self, + bytes: std::num::NonZeroUsize, + ) -> Self { + self.responses_max_decompressed_body_bytes = bytes; + self + } + #[must_use] pub fn with_request_location( mut self, @@ -608,6 +638,8 @@ impl RuntimeSnapshot { client_policy_map.retain(|_, policy| policy.enabled()); Ok(Self { + responses_max_decompressed_body_bytes: std::num::NonZeroUsize::new(64 * 1024 * 1024) + .expect("positive default limit"), request_location: None, revision, account_selection_policy, diff --git a/backend/crates/gateway-core/tests/routing/snapshot.rs b/backend/crates/gateway-core/tests/routing/snapshot.rs index 5732f2967..b3b638d0f 100644 --- a/backend/crates/gateway-core/tests/routing/snapshot.rs +++ b/backend/crates/gateway-core/tests/routing/snapshot.rs @@ -407,3 +407,49 @@ fn global_request_location_should_be_frozen_when_snapshot_is_published() { "Asia/Tokyo" ); } + +#[test] +fn decompression_setting_should_validate_and_remain_frozen_across_publication() { + use gateway_core::runtime::RuntimeSnapshotHandle; + let compile = |version, bytes| { + let facts = SnapshotFacts::new( + revision(version), + revision(version), + SnapshotSettingsFacts::new(3, 0, "smart", BTreeMap::new(), None, None) + .with_responses_max_decompressed_body_bytes(bytes), + Vec::new(), + Vec::new(), + Vec::new(), + Vec::new(), + ); + block_on(compiler(Arc::new(TestSnapshotStore::new(Ok(facts)))).compile()) + }; + for invalid in [0, u64::MAX] { + assert!(matches!( + compile(1, invalid), + Err(RuntimeSnapshotCompileError::InvalidData) + )); + } + let handle = RuntimeSnapshotHandle::new(compile(1, 64 * 1024 * 1024).unwrap()); + let frozen = handle.acquire().unwrap(); + handle.publish(compile(2, 128 * 1024 * 1024).unwrap()); + assert_eq!( + handle + .acquire() + .unwrap() + .responses_max_decompressed_body_bytes(), + 128 * 1024 * 1024 + ); + assert_eq!( + frozen.responses_max_decompressed_body_bytes(), + 64 * 1024 * 1024 + ); + handle.publish(compile(3, 1024 * 1024).unwrap()); + assert_eq!( + handle + .acquire() + .unwrap() + .responses_max_decompressed_body_bytes(), + 1024 * 1024 + ); +} diff --git a/backend/crates/gateway-store/src/admin_adapter.rs b/backend/crates/gateway-store/src/admin_adapter.rs index 7e37d7222..ee195af34 100644 --- a/backend/crates/gateway-store/src/admin_adapter.rs +++ b/backend/crates/gateway-store/src/admin_adapter.rs @@ -49,6 +49,8 @@ impl SettingsStore for AdminSettingsStoreAdapter { max_waiting_per_key: command.max_waiting_per_key, max_waiting_per_account: command.max_waiting_per_account, concurrency_wait_timeout_seconds: command.concurrency_wait_timeout_seconds, + responses_max_decompressed_body_bytes: command + .responses_max_decompressed_body_bytes, rotation_strategy: command.rotation_strategy.as_str().to_owned(), model_mappings: store_model_mappings(command.model_mappings), min_codex_desktop_version: command.min_codex_desktop_version, @@ -81,6 +83,7 @@ impl SettingsStore for AdminSettingsStoreAdapter { "max_waiting_per_key".to_owned(), "max_waiting_per_account".to_owned(), "concurrency_wait_timeout_seconds".to_owned(), + "responses_max_decompressed_body_bytes".to_owned(), "rotation_strategy".to_owned(), "min_codex_desktop_version".to_owned(), "min_codex_cli_version".to_owned(), @@ -190,6 +193,7 @@ pub(crate) fn admin_runtime_settings( max_waiting_per_key: settings.max_waiting_per_key, max_waiting_per_account: settings.max_waiting_per_account, concurrency_wait_timeout_seconds: settings.concurrency_wait_timeout_seconds, + responses_max_decompressed_body_bytes: settings.responses_max_decompressed_body_bytes, rotation_strategy, min_codex_desktop_version: settings.min_codex_desktop_version, min_codex_cli_version: settings.min_codex_cli_version, diff --git a/backend/crates/gateway-store/src/postgres/runtime_settings.rs b/backend/crates/gateway-store/src/postgres/runtime_settings.rs index 34adf6138..e4ddff743 100644 --- a/backend/crates/gateway-store/src/postgres/runtime_settings.rs +++ b/backend/crates/gateway-store/src/postgres/runtime_settings.rs @@ -28,6 +28,7 @@ pub struct RuntimeSettings { pub max_waiting_per_key: u32, pub max_waiting_per_account: u32, pub concurrency_wait_timeout_seconds: u32, + pub responses_max_decompressed_body_bytes: u64, pub rotation_strategy: String, pub request_location_enabled: bool, pub request_location: gateway_core::account::RequestLocation, @@ -115,6 +116,7 @@ pub struct RuntimeSettingsUpdate { pub max_waiting_per_key: u32, pub max_waiting_per_account: u32, pub concurrency_wait_timeout_seconds: u32, + pub responses_max_decompressed_body_bytes: u64, pub rotation_strategy: String, pub request_location_enabled: bool, pub request_location: gateway_core::account::RequestLocation, @@ -152,6 +154,8 @@ impl fmt::Debug for RuntimeSettingsUpdate { impl RuntimeSettingsUpdate { pub fn validate(&self) -> StoreResult<()> { if self.request_location.validate().is_err() + || self.responses_max_decompressed_body_bytes == 0 + || isize::try_from(self.responses_max_decompressed_body_bytes).is_err() || self.refresh_margin_seconds == 0 || self.refresh_concurrency == 0 || self.max_concurrent_per_account == 0 @@ -229,7 +233,7 @@ pub(crate) async fn load_runtime_settings_from_pool(pool: &PgPool) -> StoreResul refresh_concurrency, max_concurrent_per_account, request_interval_ms, rotation_strategy, model_mappings_json, usage_retention_days, ops_event_retention_days, audit_retention_days, min_codex_desktop_version, - min_codex_cli_version, updated_at, max_waiting_per_key, max_waiting_per_account, concurrency_wait_timeout_seconds, + min_codex_cli_version, updated_at, responses_max_decompressed_body_bytes, max_waiting_per_key, max_waiting_per_account, concurrency_wait_timeout_seconds, account_auto_freeze_enabled, account_auto_freeze_threshold, account_auto_freeze_window_seconds, account_auto_freeze_duration_seconds, account_auto_freeze_probe_enabled, account_auto_freeze_probe_model, @@ -291,7 +295,7 @@ pub(crate) async fn load_runtime_settings_in_transaction( refresh_concurrency, max_concurrent_per_account, request_interval_ms, rotation_strategy, model_mappings_json, usage_retention_days, ops_event_retention_days, audit_retention_days, min_codex_desktop_version, - min_codex_cli_version, updated_at, max_waiting_per_key, max_waiting_per_account, concurrency_wait_timeout_seconds, + min_codex_cli_version, updated_at, responses_max_decompressed_body_bytes, max_waiting_per_key, max_waiting_per_account, concurrency_wait_timeout_seconds, account_auto_freeze_enabled, account_auto_freeze_threshold, account_auto_freeze_window_seconds, account_auto_freeze_duration_seconds, account_auto_freeze_probe_enabled, account_auto_freeze_probe_model, @@ -342,6 +346,7 @@ pub(crate) async fn update_runtime_settings_in_transaction( account_auto_freeze_adaptive_concurrency = $22, request_location_json = $23, request_location_enabled = $24, + responses_max_decompressed_body_bytes = $25, updated_at = now() where id = 1 returning config_revision", @@ -379,6 +384,10 @@ pub(crate) async fn update_runtime_settings_in_transaction( .map_err(|_| invalid_location())?, )) .bind(update.request_location_enabled) + .bind( + i64::try_from(update.responses_max_decompressed_body_bytes) + .map_err(|_| invalid_numeric())?, + ) .fetch_optional(&mut **transaction) .await .map_err(|_| postgres_unavailable("update runtime settings in transaction"))? @@ -447,6 +456,7 @@ struct RuntimeSettingsRow { max_waiting_per_key: i64, max_waiting_per_account: i64, concurrency_wait_timeout_seconds: i64, + responses_max_decompressed_body_bytes: i64, account_auto_freeze_enabled: bool, account_auto_freeze_threshold: i64, account_auto_freeze_window_seconds: i64, @@ -481,6 +491,7 @@ fn runtime_settings_from_row(row: RuntimeSettingsRow) -> StoreResult, pub min_codex_desktop_version: Option, @@ -154,6 +155,9 @@ impl SnapshotStorePort for PgRuntimeSnapshotRepository { data.settings.min_codex_desktop_version, data.settings.min_codex_cli_version, ) + .with_responses_max_decompressed_body_bytes( + data.settings.responses_max_decompressed_body_bytes, + ) .with_request_location( data.settings.request_location, data.settings.request_location_enabled, @@ -249,12 +253,13 @@ async fn load_settings( i64, i64, i64, sqlx::types::Json, bool, + i64, ), >( "select config_revision, refresh_margin_seconds, refresh_concurrency, max_concurrent_per_account, request_interval_ms, rotation_strategy, model_mappings_json, min_codex_desktop_version, - min_codex_cli_version, max_waiting_per_key, max_waiting_per_account, concurrency_wait_timeout_seconds, request_location_json, request_location_enabled + min_codex_cli_version, max_waiting_per_key, max_waiting_per_account, concurrency_wait_timeout_seconds, request_location_json, request_location_enabled, responses_max_decompressed_body_bytes from runtime_settings where id = 1", ) .fetch_optional(&mut **transaction) @@ -267,6 +272,7 @@ async fn load_settings( Ok(( revision_from_i64(row.0)?, SnapshotRuntimeSettings { + responses_max_decompressed_body_bytes: to_u64(row.14)?, request_location_enabled: row.13, request_location: row.12.0, refresh_margin_seconds: to_u64(row.1)?, diff --git a/backend/crates/gateway-store/tests/postgres/runtime_settings.rs b/backend/crates/gateway-store/tests/postgres/runtime_settings.rs index 47518906f..bb6256075 100644 --- a/backend/crates/gateway-store/tests/postgres/runtime_settings.rs +++ b/backend/crates/gateway-store/tests/postgres/runtime_settings.rs @@ -19,6 +19,7 @@ fn settings_with_margin(refresh_margin_seconds: u64) -> RuntimeSettingsUpdate { max_waiting_per_key: 0, max_waiting_per_account: 0, concurrency_wait_timeout_seconds: 30, + responses_max_decompressed_body_bytes: 64 * 1024 * 1024, rotation_strategy: "smart".to_owned(), model_mappings: BTreeMap::from([ ("gpt-5.4".to_owned(), "gpt-5.5".to_owned()), @@ -380,3 +381,51 @@ async fn auto_freeze_defaults_off_and_explicit_opt_in_round_trips() { ); database.close().await; } + +#[tokio::test] +async fn decompression_setting_should_persist_and_reach_snapshot_facts() { + use gateway_store::postgres::{PgRuntimeSnapshotRepository, RuntimeSnapshotRepository}; + let Some(database) = TestDatabase::create("decompression_settings").await else { + return; + }; + let repository = PgRuntimeSettingsRepository::new(database.pool.clone()); + let before = repository.load_runtime_settings().await.unwrap(); + assert_eq!( + before.responses_max_decompressed_body_bytes, + 64 * 1024 * 1024 + ); + let mut update = settings_with_margin(3600); + update.responses_max_decompressed_body_bytes = 128 * 1024 * 1024; + repository.update_runtime_settings(update).await.unwrap(); + let reloaded = PgRuntimeSettingsRepository::new(database.pool.clone()) + .load_runtime_settings() + .await + .unwrap(); + assert_eq!( + reloaded.responses_max_decompressed_body_bytes, + 128 * 1024 * 1024 + ); + let snapshot = PgRuntimeSnapshotRepository::new(database.pool.clone()) + .load_runtime_snapshot() + .await + .unwrap(); + assert_eq!( + snapshot.settings.responses_max_decompressed_body_bytes, + reloaded.responses_max_decompressed_body_bytes + ); + assert!(snapshot.config_revision > before.config_revision); + for invalid in [0, u64::MAX] { + let mut update = settings_with_margin(3600); + update.responses_max_decompressed_body_bytes = invalid; + assert!(repository.update_runtime_settings(update).await.is_err()); + assert_eq!( + repository + .load_runtime_settings() + .await + .unwrap() + .config_revision, + reloaded.config_revision + ); + } + database.close().await; +} diff --git a/backend/migrations/.frozen-sha256 b/backend/migrations/.frozen-sha256 index 3e5c9bfd3..da7aa4bf6 100644 --- a/backend/migrations/.frozen-sha256 +++ b/backend/migrations/.frozen-sha256 @@ -10,3 +10,4 @@ b7d3df7b48d835a00a3b74cbefc958a039d38fea5529cdad04bfbc703dca72aa 0001_initial.s 3751a802e80b5d538abc5619a2397aa4bf13f52076068a489892f664907adf49 0010_account_auto_freeze.sql ae7f7acb3014f0a60d979dd9c04b3065322b206e42eb6256505baf78caf9dd22 0011_proxy_request_location.sql b20e425c0b062b72d35fdebe294e023eb17ceb44ac93083ff3d48400715bf430 0012_upstream_response_model.sql +6e23fceef1096b50c1af3584cb1687eded2ad37e31e65391397ec984bf78c7e3 0013_responses_decompression_limit.sql diff --git a/backend/migrations/0013_responses_decompression_limit.sql b/backend/migrations/0013_responses_decompression_limit.sql new file mode 100644 index 000000000..de818412e --- /dev/null +++ b/backend/migrations/0013_responses_decompression_limit.sql @@ -0,0 +1,4 @@ +-- 解压输出上限属于运行设置，升级后沿用原有 64 MiB 默认值。 +alter table runtime_settings + add column responses_max_decompressed_body_bytes bigint not null default 67108864 + check (responses_max_decompressed_body_bytes > 0); diff --git a/docs/api.md b/docs/api.md index b72bb3948..88f5b4241 100644 --- a/docs/api.md +++ b/docs/api.md @@ -162,8 +162,9 @@ Codex 的 review 等子代理请求仍使用 `/v1/responses`，并通过 `x-open `POST /v1/responses` 在鉴权后按 `Content-Encoding` 解压，再解析 JSON；支持单一 `gzip`、 `deflate`（zlib 封装）和 `zstd`，缺省、空值或 `identity` 直接使用原始正文。gzip 多成员与 zstd -多帧连续解码，整体展开结果最多 64 MiB，超限在继续展开前返回 `400 request_too_large`；zstd -回溯窗口同样最多 64 MiB，不能满足该限制的帧按解码失败处理。这个限制保护入站解压资源，不是 +多帧连续解码，整体展开结果受运行设置 `responsesMaxDecompressedBodyBytes` 约束（默认 64 MiB）， +超限在继续展开前返回 `400 request_too_large`，错误信息包含当前请求的上限字节数。zstd +回溯窗口独立固定为最多 64 MiB，不能满足该限制的帧按解码失败处理。这个限制保护入站解压资源，不是 模型上下文或 Token 上限；未压缩正文不受此长度限制。 不支持的编码、逗号分隔的叠加编码和重复 `Content-Encoding` 头返回 `400 unsupported_content_encoding`；压缩正文损坏、截断或解压后不是合法 JSON 返回 @@ -943,6 +944,7 @@ maxConcurrentPerAccount maxWaitingPerKey maxWaitingPerAccount concurrencyWaitTimeoutSeconds +responsesMaxDecompressedBodyBytes requestIntervalMs rotationStrategy minCodexDesktopVersion @@ -974,6 +976,11 @@ accountAutoFreezeAdaptiveConcurrency 切换账号或内部重试不重新计时，等待同时计入请求总超时。该时限不用于中断已开始的上游生成。 设置更新请求须包含这三个字段，新请求使用更新后的快照。 +`responsesMaxDecompressedBodyBytes` 是压缩 Responses HTTP 请求的解压输出上限，单位字节，默认 +67108864（64 MiB）。必须为正整数，且可表示为进程平台的 `isize`；管理端以整数 MiB 编辑。 +保存并发布成功后，新请求使用新值；已鉴权请求沿用原快照，无需重启。调高上限会增加大请求的内存占用， +它不代表整个进程的内存预算。 + `rotationStrategy` 可取 `smart`、`quota_reset_priority`、`round_robin`、`sticky`。 两个 `minCodex*Version` 字段为 `string | null`，只设置最低版本，不存在最大版本字段。 diff --git a/frontend/src/api/modules/settings.ts b/frontend/src/api/modules/settings.ts index 913688e19..a41f8ebe5 100644 --- a/frontend/src/api/modules/settings.ts +++ b/frontend/src/api/modules/settings.ts @@ -15,6 +15,7 @@ export interface RuntimeSettings { maxWaitingPerKey: number maxWaitingPerAccount: number concurrencyWaitTimeoutSeconds: number + responsesMaxDecompressedBodyBytes: number rotationStrategy: RotationStrategy minCodexDesktopVersion: string | null minCodexCliVersion: string | null diff --git a/frontend/src/views/settings/components/RuntimeSettingsCard.vue b/frontend/src/views/settings/components/RuntimeSettingsCard.vue index c9dee4f1b..3cc42a144 100644 --- a/frontend/src/views/settings/components/RuntimeSettingsCard.vue +++ b/frontend/src/views/settings/components/RuntimeSettingsCard.vue @@ -13,6 +13,7 @@ const requestIntervalMs = defineModel('requestIntervalMs', { required: t const maxWaitingPerKey = defineModel('maxWaitingPerKey', { required: true }) const maxWaitingPerAccount = defineModel('maxWaitingPerAccount', { required: true }) const concurrencyWaitTimeoutSeconds = defineModel('concurrencyWaitTimeoutSeconds', { required: true }) +const responsesMaxDecompressedBodyMiB = defineModel('responsesMaxDecompressedBodyMiB', { required: true }) + + + + +
diff --git a/frontend/src/views/settings/composables/useSettingsForm.ts b/frontend/src/views/settings/composables/useSettingsForm.ts index 99b111604..2a8e4fb89 100644 --- a/frontend/src/views/settings/composables/useSettingsForm.ts +++ b/frontend/src/views/settings/composables/useSettingsForm.ts @@ -11,6 +11,8 @@ import { normalizeRequestLocation, requestLocationError } from '@/utils/request- type RotationStrategy = (typeof rotationOptions)[number]['value'] +const MIB = 1024 * 1024 + export function useSettingsForm() { const loading = shallowRef(true) const saveAction = useAsyncAction() @@ -28,6 +30,7 @@ export function useSettingsForm() { maxWaitingPerKey: null as number | null, maxWaitingPerAccount: null as number | null, concurrencyWaitTimeoutSeconds: null as number | null, + responsesMaxDecompressedBodyMiB: null as number | null, rotationStrategy: '' as RotationStrategy | '', minCodexDesktopVersion: '', @@ -45,7 +48,7 @@ export function useSettingsForm() { accountAutoFreezeAdaptiveConcurrency: true, }) - function numericModel(key: 'refreshMarginSeconds' | 'refreshConcurrency' | 'maxConcurrentPerAccount' | 'requestIntervalMs' | 'maxWaitingPerKey' | 'maxWaitingPerAccount' | 'concurrencyWaitTimeoutSeconds' | 'accountAutoFreezeThreshold' | 'accountAutoFreezeWindowSeconds' | 'accountAutoFreezeDurationSeconds') { + function numericModel(key: 'refreshMarginSeconds' | 'refreshConcurrency' | 'maxConcurrentPerAccount' | 'requestIntervalMs' | 'maxWaitingPerKey' | 'maxWaitingPerAccount' | 'concurrencyWaitTimeoutSeconds' | 'responsesMaxDecompressedBodyMiB' | 'accountAutoFreezeThreshold' | 'accountAutoFreezeWindowSeconds' | 'accountAutoFreezeDurationSeconds') { return computed({ get: () => (form[key] === null ? '' : String(form[key])), set: (value: string) => { @@ -65,6 +68,7 @@ export function useSettingsForm() { const requestIntervalMsValue = numericModel('requestIntervalMs') const maxWaitingPerKeyValue = numericModel('maxWaitingPerKey') const maxWaitingPerAccountValue = numericModel('maxWaitingPerAccount') + const responsesMaxDecompressedBodyMiBValue = numericModel('responsesMaxDecompressedBodyMiB') const concurrencyWaitTimeoutSecondsValue = numericModel('concurrencyWaitTimeoutSeconds') const accountAutoFreezeThresholdValue = numericModel('accountAutoFreezeThreshold') const accountAutoFreezeWindowSecondsValue = numericModel('accountAutoFreezeWindowSeconds') @@ -89,6 +93,7 @@ export function useSettingsForm() { form.maxWaitingPerKey = data.maxWaitingPerKey form.maxWaitingPerAccount = data.maxWaitingPerAccount form.concurrencyWaitTimeoutSeconds = data.concurrencyWaitTimeoutSeconds + form.responsesMaxDecompressedBodyMiB = data.responsesMaxDecompressedBodyBytes / MIB form.rotationStrategy = data.rotationStrategy form.minCodexDesktopVersion = data.minCodexDesktopVersion ?? '' @@ -158,11 +163,16 @@ export function useSettingsForm() { async function saveSettings() { if (saving.value || loading.value || !savedRequestLocation.value) return - const { refreshMarginSeconds, refreshConcurrency, maxConcurrentPerAccount, requestIntervalMs, rotationStrategy, maxWaitingPerKey, maxWaitingPerAccount, concurrencyWaitTimeoutSeconds, accountAutoFreezeThreshold, accountAutoFreezeWindowSeconds, accountAutoFreezeDurationSeconds } = form + const { refreshMarginSeconds, refreshConcurrency, maxConcurrentPerAccount, requestIntervalMs, rotationStrategy, maxWaitingPerKey, maxWaitingPerAccount, concurrencyWaitTimeoutSeconds, responsesMaxDecompressedBodyMiB, accountAutoFreezeThreshold, accountAutoFreezeWindowSeconds, accountAutoFreezeDurationSeconds } = form if (refreshMarginSeconds === null || refreshConcurrency === null || maxConcurrentPerAccount === null || requestIntervalMs === null || !rotationStrategy || maxWaitingPerKey === null || maxWaitingPerAccount === null || concurrencyWaitTimeoutSeconds === null) { toast.warning('请完整填写运行参数和调度策略') return } + if (responsesMaxDecompressedBodyMiB === null || !Number.isInteger(responsesMaxDecompressedBodyMiB) || responsesMaxDecompressedBodyMiB < 1 + || !Number.isSafeInteger(responsesMaxDecompressedBodyMiB * MIB)) { + toast.warning('Responses 解压上限应为有效的正整数（MiB）') + return + } if (![maxWaitingPerKey, maxWaitingPerAccount].every(value => Number.isInteger(value) && value >= 0 && value <= 1000) || !Number.isInteger(concurrencyWaitTimeoutSeconds) || concurrencyWaitTimeoutSeconds < 1 || concurrencyWaitTimeoutSeconds > 120) { toast.warning('最大排队数应为 0～1000 的整数，最长排队时间应为 1～120 秒的整数') @@ -208,6 +218,7 @@ export function useSettingsForm() { maxWaitingPerKey, maxWaitingPerAccount, concurrencyWaitTimeoutSeconds, + responsesMaxDecompressedBodyBytes: responsesMaxDecompressedBodyMiB * MIB, rotationStrategy, minCodexDesktopVersion: form.minCodexDesktopVersion.trim() || null, minCodexCliVersion: form.minCodexCliVersion.trim() || null, @@ -248,6 +259,7 @@ export function useSettingsForm() { maxWaitingPerKeyValue, maxWaitingPerAccountValue, concurrencyWaitTimeoutSecondsValue, + responsesMaxDecompressedBodyMiBValue, accountAutoFreezeThresholdValue, accountAutoFreezeWindowSecondsValue, accountAutoFreezeDurationSecondsValue, diff --git a/frontend/src/views/settings/index.vue b/frontend/src/views/settings/index.vue index 517018a3f..b63a25447 100644 --- a/frontend/src/views/settings/index.vue +++ b/frontend/src/views/settings/index.vue @@ -54,6 +54,7 @@ const { maxWaitingPerKeyValue, maxWaitingPerAccountValue, concurrencyWaitTimeoutSecondsValue, + responsesMaxDecompressedBodyMiBValue, accountAutoFreezeThresholdValue, accountAutoFreezeWindowSecondsValue, accountAutoFreezeDurationSecondsValue, @@ -139,6 +140,7 @@ watch( v-model:max-waiting-per-key="maxWaitingPerKeyValue" v-model:max-waiting-per-account="maxWaitingPerAccountValue" v-model:concurrency-wait-timeout-seconds="concurrencyWaitTimeoutSecondsValue" + v-model:responses-max-decompressed-body-mi-b="responsesMaxDecompressedBodyMiBValue" />