From 06e94ef69b4c7f3166e7635677068208c0c90afb Mon Sep 17 00:00:00 2001 From: Zyy <39424122+zyycn@users.noreply.github.com> Date: Thu, 17 Sep 2026 15:29:14 +0800 Subject: [PATCH] fix(settings): make Responses decompression limit configurable (#141) Persist the output limit in runtime settings and use the authenticated snapshot for each compressed Responses request. Expose the setting in MiB and preserve bounded decoding and the independent zstd window limit. Co-authored-by: Zyy Co-authored-by: Codex --- .../gateway-admin/src/model/settings.rs | 2 + .../gateway-admin/src/use_case/settings.rs | 2 + .../gateway-admin/tests/use_case/accounts.rs | 1 + .../tests/use_case/freeze_recovery.rs | 1 + .../tests/use_case/observability.rs | 1 + .../gateway-admin/tests/use_case/settings.rs | 1 + .../crates/gateway-api/src/admin/settings.rs | 11 +++ .../gateway-api/src/openai/responses/error.rs | 11 ++- .../gateway-api/src/openai/responses/http.rs | 6 +- .../src/openai/responses/request.rs | 35 +++++--- backend/crates/gateway-api/tests/admin/mod.rs | 2 + .../gateway-api/tests/admin/settings.rs | 25 ++++++ .../crates/gateway-api/tests/openai/mod.rs | 12 ++- .../tests/openai/responses/http.rs | 49 ++++++++++- .../gateway-api/tests/openai/responses/mod.rs | 30 ++++--- .../tests/openai/responses/request.rs | 82 +++++++++++++++---- .../gateway-core/src/routing/snapshot.rs | 32 ++++++++ .../gateway-core/tests/routing/snapshot.rs | 46 +++++++++++ .../crates/gateway-store/src/admin_adapter.rs | 4 + .../src/postgres/runtime_settings.rs | 15 +++- .../gateway-store/src/postgres/snapshot.rs | 8 +- .../tests/postgres/runtime_settings.rs | 49 +++++++++++ backend/migrations/.frozen-sha256 | 1 + .../0013_responses_decompression_limit.sql | 4 + docs/api.md | 11 ++- frontend/src/api/modules/settings.ts | 1 + .../components/RuntimeSettingsCard.vue | 17 ++++ .../settings/composables/useSettingsForm.ts | 16 +++- frontend/src/views/settings/index.vue | 2 + 29 files changed, 425 insertions(+), 52 deletions(-) create mode 100644 backend/migrations/0013_responses_decompression_limit.sql 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" />