From 2480f07f2d6ba8cfa08b31a911ba91a1d0f2ccb9 Mon Sep 17 00:00:00 2001 From: ZQiannnn <604922962@qq.com> Date: Wed, 21 Oct 2026 00:51:13 +0800 Subject: [PATCH] fix(plugins): honor configured model and HTTP request body limits (#496) * fix(plugins): honor configured request body limits * test(plugin-sdk): mirror upstream adapter module layout Co-authored-by: Codex * fix(plugins): preserve frozen request body budgets across callbacks Carry the execution body budget through policy and attempt callbacks, and validate explicit-key model requests against their bound snapshot. Add regression coverage for configuration publication and document budget ownership and lifetime. Co-authored-by: Codex --------- Co-authored-by: Codex Co-authored-by: Zyy --- backend/apps/gateway/src/bootstrap/command.rs | 4 + backend/apps/gateway/src/bootstrap/server.rs | 4 + .../gateway-core/src/engine/coordinator.rs | 1 + .../src/engine/execution/scope.rs | 9 + .../src/engine/execution/service.rs | 6 + .../src/engine/middleware/contract.rs | 14 ++ backend/crates/gateway-core/src/engine/mod.rs | 16 +- .../crates/gateway-core/src/engine/policy.rs | 30 +++ .../gateway-core/src/routing/extensions.rs | 9 + .../crates/gateway-core/src/routing/mod.rs | 6 + .../gateway-core/src/routing/snapshot.rs | 10 +- .../crates/gateway-core/src/runtime/mod.rs | 4 +- .../gateway-core/src/runtime/publications.rs | 36 ++- .../gateway-core/tests/routing/snapshot.rs | 14 +- .../crates/gateway-host/src/outbound/http.rs | 4 +- .../src/plugin_distribution/http.rs | 1 + .../gateway-host/tests/outbound/connector.rs | 1 + .../gateway-host/tests/outbound/http.rs | 38 +++ .../gateway-host/tests/outbound/network.rs | 1 + .../gateway-host/tests/outbound/websocket.rs | 1 + .../runtime/src/adapter/middleware/request.rs | 1 + .../runtime/src/adapter/policy/retry.rs | 1 + .../src/adapter/policy/route_schedule.rs | 2 + .../src/adapter/upstream_adapter/execution.rs | 31 ++- .../runtime/src/callback/http.rs | 14 ++ .../runtime/src/callback/mod.rs | 30 ++- .../runtime/src/callback/model.rs | 20 +- .../runtime/src/callback/request_limits.rs | 49 ++++ .../runtime/src/callback/scope.rs | 2 + .../runtime/src/callback/upstream/mod.rs | 7 +- .../src/generation/prepare/instance.rs | 8 +- .../src/generation/prepare/management.rs | 2 +- .../runtime/src/generation/prepare/runtime.rs | 10 + .../src/generation/prepare/snapshot.rs | 34 ++- .../tests/adapter/command_line/execution.rs | 238 ++++++++++++++++++ .../runtime/tests/generation/policy/mounts.rs | 161 ++++++++++++ .../tests/generation/prepare/snapshot.rs | 38 +++ .../tests/generation/upstream_adapter.rs | 130 +++++++++- .../runtime/tests/support/environment.rs | 2 + .../runtime/tests/support/worker/mod.rs | 104 +++++++- .../gateway-plugin/sdk/docs/capabilities.md | 18 ++ .../sdk/src/call/model/codec.rs | 52 ++-- .../gateway-plugin/sdk/src/call/model/mod.rs | 9 + .../sdk/src/call/upstream_adapter/codec.rs | 7 +- .../gateway-plugin/sdk/tests/call/mod.rs | 1 + .../sdk/tests/call/upstream_adapter/codec.rs | 46 ++++ .../sdk/tests/call/upstream_adapter/mod.rs | 3 + docs/api.md | 16 +- docs/architecture.md | 4 + 49 files changed, 1187 insertions(+), 62 deletions(-) create mode 100644 backend/crates/gateway-plugin/runtime/src/callback/request_limits.rs create mode 100644 backend/crates/gateway-plugin/sdk/tests/call/upstream_adapter/codec.rs create mode 100644 backend/crates/gateway-plugin/sdk/tests/call/upstream_adapter/mod.rs diff --git a/backend/apps/gateway/src/bootstrap/command.rs b/backend/apps/gateway/src/bootstrap/command.rs index ba1745cc8..27a3807df 100644 --- a/backend/apps/gateway/src/bootstrap/command.rs +++ b/backend/apps/gateway/src/bootstrap/command.rs @@ -96,6 +96,10 @@ async fn execute( Some(plugin_runtime.policy_registry()), Some(plugin_runtime.execution_registry()), ); + or_shutdown!( + plugin_runtime, + plugin_runtime.bind_snapshots(&core.snapshots()) + ); let management = ManagementPorts::new(&store, providers.admin.clone(), core.snapshot_control()); or_shutdown!(plugin_runtime, management.bind(&plugin_runtime)); let core = or_shutdown!(plugin_runtime, core.activate().await); diff --git a/backend/apps/gateway/src/bootstrap/server.rs b/backend/apps/gateway/src/bootstrap/server.rs index f3a92971b..97d353075 100644 --- a/backend/apps/gateway/src/bootstrap/server.rs +++ b/backend/apps/gateway/src/bootstrap/server.rs @@ -31,6 +31,10 @@ pub async fn run() -> Result<(), BootstrapError> { Some(plugin_runtime.execution_registry()), Some(plugin_runtime.frontend_authentication_registry()), ); + or_shutdown!( + plugin_runtime, + plugin_runtime.bind_snapshots(&core.snapshots()) + ); let management = ManagementPorts::new(&store, providers.admin.clone(), core.snapshot_control()); or_shutdown!(plugin_runtime, management.bind(&plugin_runtime)); let mut core = or_shutdown!(plugin_runtime, core.activate().await); diff --git a/backend/crates/gateway-core/src/engine/coordinator.rs b/backend/crates/gateway-core/src/engine/coordinator.rs index 0af86b7be..b5747c284 100644 --- a/backend/crates/gateway-core/src/engine/coordinator.rs +++ b/backend/crates/gateway-core/src/engine/coordinator.rs @@ -1132,6 +1132,7 @@ where .with_pricing(self.plan.pricing()) .with_request_location(self.plan.request_location().cloned()) .with_privacy(self.plan.privacy()) + .with_request_body_limit(self.plan.request_body_limit()) .with_concurrency_wait_budget(self.concurrency_wait_budget.clone()) .with_connection_budget(if connection_rotation_reserved { self.connection_budget.clone().reserve_next_opening() diff --git a/backend/crates/gateway-core/src/engine/execution/scope.rs b/backend/crates/gateway-core/src/engine/execution/scope.rs index ecf096d75..f9bcb3c7c 100644 --- a/backend/crates/gateway-core/src/engine/execution/scope.rs +++ b/backend/crates/gateway-core/src/engine/execution/scope.rs @@ -197,6 +197,15 @@ pub struct BoundModelExecutionContext { } impl BoundModelExecutionContext { + /// 与实际执行共用绑定时的快照，不能回退到后续发布的配置 + #[must_use] + pub fn request_body_limit(&self) -> usize { + self.authority + .client + .snapshot + .responses_max_decompressed_body_bytes() + } + #[must_use] pub fn request_settings(&self) -> Option<&RequestSettings> { self.authority.client.request_settings() diff --git a/backend/crates/gateway-core/src/engine/execution/service.rs b/backend/crates/gateway-core/src/engine/execution/service.rs index 0fa94f878..8075fbf57 100644 --- a/backend/crates/gateway-core/src/engine/execution/service.rs +++ b/backend/crates/gateway-core/src/engine/execution/service.rs @@ -448,6 +448,12 @@ impl DefaultExecutionService { account_group_ids.clone(), ) .with_extension_scope(authorization.extension_scope.clone()) + .with_request_body_limit( + request + .client + .snapshot + .responses_max_decompressed_body_bytes(), + ) .with_execution_effects(Arc::clone(&authorization.graph.as_ref()?.effects)), ) }); diff --git a/backend/crates/gateway-core/src/engine/middleware/contract.rs b/backend/crates/gateway-core/src/engine/middleware/contract.rs index 386c338af..6146083fe 100644 --- a/backend/crates/gateway-core/src/engine/middleware/contract.rs +++ b/backend/crates/gateway-core/src/engine/middleware/contract.rs @@ -559,9 +559,22 @@ pub struct MiddlewareContext { deadline: crate::lifecycle::Deadline, extension_scope: ExtensionCallScope, execution_effects: Option>, + request_body_limit: Option, } impl MiddlewareContext { + /// attempt 只继承执行预算，不开放 request 设置改写 + #[must_use] + pub const fn with_request_body_limit(mut self, bytes: Option) -> Self { + self.request_body_limit = bytes; + self + } + + #[must_use] + pub const fn request_body_limit(&self) -> Option { + self.request_body_limit + } + #[must_use] pub fn plan(&self) -> Option { self.plan.clone() @@ -590,6 +603,7 @@ impl MiddlewareContext { } = authority; Self { plan: None, + request_body_limit: None, request_id, mount, attempt_index, diff --git a/backend/crates/gateway-core/src/engine/mod.rs b/backend/crates/gateway-core/src/engine/mod.rs index 0a8b6e7aa..0709840e5 100644 --- a/backend/crates/gateway-core/src/engine/mod.rs +++ b/backend/crates/gateway-core/src/engine/mod.rs @@ -286,6 +286,7 @@ impl ContinuationAttempt { /// Provider 每次执行可见的 request-local context #[derive(Debug, Clone)] pub struct RequestAttemptContext { + request_body_limit: Option, lease_metadata: Option>, response_control: Option, pricing: Arc, @@ -311,6 +312,12 @@ pub struct RequestAttemptContext { } impl RequestAttemptContext { + #[must_use] + pub const fn with_request_body_limit(mut self, bytes: usize) -> Self { + self.request_body_limit = Some(bytes); + self + } + #[must_use] pub fn with_lease_metadata( mut self, @@ -387,6 +394,7 @@ impl RequestAttemptContext { pub fn new(request_id: ModelRequestId, client_api_key_ref: ClientApiKeyId) -> Self { Self { response_control: None, + request_body_limit: None, lease_metadata: None, request_id, client_api_key_ref, @@ -542,6 +550,11 @@ pub struct AttemptContext { } impl AttemptContext { + #[must_use] + pub const fn request_body_limit(&self) -> Option { + self.request.request_body_limit + } + #[must_use] pub fn lease_metadata(&self) -> Option> { self.request.lease_metadata.clone() @@ -747,7 +760,8 @@ impl AttemptContext { extension_scope: self.request.extension_scope.clone(), execution_effects: self.request.execution_effects.as_ref().map(Arc::clone), }, - ); + ) + .with_request_body_limit(self.request.request_body_limit); provider::execute_attempt_middleware( self.request.middleware.as_ref(), context, diff --git a/backend/crates/gateway-core/src/engine/policy.rs b/backend/crates/gateway-core/src/engine/policy.rs index 70170e2d9..1d3b70167 100644 --- a/backend/crates/gateway-core/src/engine/policy.rs +++ b/backend/crates/gateway-core/src/engine/policy.rs @@ -34,9 +34,15 @@ pub struct ModelRouteInput { requested_model: PublicModelId, available_providers: BTreeSet, execution_effects: Arc, + request_body_limit: Option, } impl ModelRouteInput { + #[must_use] + pub const fn request_body_limit(&self) -> Option { + self.request_body_limit + } + #[must_use] pub const fn request_id(&self) -> &super::ModelRequestId { &self.request_id @@ -188,9 +194,15 @@ pub struct AccountScheduleInput { model: Option, candidates: Vec, execution_effects: Arc, + request_body_limit: Option, } impl AccountScheduleInput { + #[must_use] + pub const fn request_body_limit(&self) -> Option { + self.request_body_limit + } + #[must_use] pub const fn request_id(&self) -> &super::ModelRequestId { &self.request_id @@ -282,9 +294,15 @@ pub struct RetryInput { pub extension_scope: ExtensionCallScope, pub facts: RetryFacts, execution_effects: Arc, + request_body_limit: Option, } impl RetryInput { + #[must_use] + pub const fn request_body_limit(&self) -> Option { + self.request_body_limit + } + #[must_use] pub fn execution_effects(&self) -> Arc { Arc::clone(&self.execution_effects) @@ -339,6 +357,7 @@ pub struct RequestPolicyContext { account_group_ids: Arc<[AccountGroupId]>, extension_scope: ExtensionCallScope, execution_effects: Arc, + request_body_limit: Option, } impl fmt::Debug for RequestPolicyContext { @@ -353,6 +372,13 @@ impl fmt::Debug for RequestPolicyContext { } impl RequestPolicyContext { + /// 路由、调度和重试沿用同一次模型执行的冻结预算 + #[must_use] + pub const fn with_request_body_limit(mut self, bytes: usize) -> Self { + self.request_body_limit = Some(bytes); + self + } + pub async fn retry_decision(&self, facts: RetryFacts) -> RetryDecision { self.plan .retry_decision(RetryInput { @@ -362,6 +388,7 @@ impl RequestPolicyContext { extension_scope: self.extension_scope.clone(), facts, execution_effects: Arc::clone(&self.execution_effects), + request_body_limit: self.request_body_limit, }) .await .unwrap_or_else(|_| { @@ -391,6 +418,7 @@ impl RequestPolicyContext { account_group_ids: account_group_ids.into(), extension_scope: ExtensionCallScope::default(), execution_effects: Arc::new(ExecutionEffects::default()), + request_body_limit: None, } } @@ -428,6 +456,7 @@ impl RequestPolicyContext { requested_model, available_providers, execution_effects: Arc::clone(&self.execution_effects), + request_body_limit: self.request_body_limit, }) .await } @@ -474,6 +503,7 @@ impl RequestPolicyContext { model: model.map(str::to_owned), candidates: projected, execution_effects: Arc::clone(&self.execution_effects), + request_body_limit: self.request_body_limit, }) .await .map_err(|_| AccountPolicyError::Fault)?; diff --git a/backend/crates/gateway-core/src/routing/extensions.rs b/backend/crates/gateway-core/src/routing/extensions.rs index e44169104..15ea33ed7 100644 --- a/backend/crates/gateway-core/src/routing/extensions.rs +++ b/backend/crates/gateway-core/src/routing/extensions.rs @@ -88,6 +88,15 @@ pub struct ExtensionPreparationError; /// 准备阶段读取与请求快照相同的持久 revision；不得自行发布当前代次 pub trait ExtensionPreparationPort: Send + Sync { + /// 初始发布尚无可读快照，注册阶段沿用本次编译的正文预算 + fn prepare_with_request_body_limit( + &self, + revision: ConfigRevision, + _request_body_limit: usize, + ) -> BoxFuture<'_, Result> { + self.prepare(revision) + } + fn prepare( &self, revision: ConfigRevision, diff --git a/backend/crates/gateway-core/src/routing/mod.rs b/backend/crates/gateway-core/src/routing/mod.rs index d81dbcdb5..375a9dade 100644 --- a/backend/crates/gateway-core/src/routing/mod.rs +++ b/backend/crates/gateway-core/src/routing/mod.rs @@ -561,6 +561,7 @@ impl ProviderCandidate { /// 一次请求冻结的 Provider 尝试顺序 #[derive(Debug, Clone)] pub struct RoutingPlan { + request_body_limit: usize, pricing: Arc, request_location: Option, privacy: Option>, @@ -573,6 +574,11 @@ pub struct RoutingPlan { } impl RoutingPlan { + #[must_use] + pub const fn request_body_limit(&self) -> usize { + self.request_body_limit + } + #[must_use] pub fn pricing(&self) -> Arc { Arc::clone(&self.pricing) diff --git a/backend/crates/gateway-core/src/routing/snapshot.rs b/backend/crates/gateway-core/src/routing/snapshot.rs index 0e080981f..00e9873c1 100644 --- a/backend/crates/gateway-core/src/routing/snapshot.rs +++ b/backend/crates/gateway-core/src/routing/snapshot.rs @@ -331,7 +331,13 @@ impl RuntimeSnapshotCompiler { }); if must_prepare { let prepared = preparer - .prepare(revision) + .prepare_with_request_body_limit( + revision, + usize::try_from( + facts.settings.responses_max_decompressed_body_bytes, + ) + .unwrap_or(usize::MAX), + ) .await .map_err(|_| RuntimeSnapshotCompileError::ExtensionsUnavailable)?; prepared_extensions = Some((revision, prepared)); @@ -1201,6 +1207,7 @@ impl RuntimeSnapshot { } Ok(RoutingPlan { + request_body_limit: self.responses_max_decompressed_body_bytes(), config_revision: self.revision, pricing: Arc::clone(&self.settings.values.pricing), request_location: self.settings.request_location.clone(), @@ -1282,6 +1289,7 @@ impl RuntimeSnapshot { account_scope: Arc::clone(&account_scope), }; Ok(RoutingPlan { + request_body_limit: self.responses_max_decompressed_body_bytes(), config_revision: self.revision, pricing: Arc::clone(&self.settings.values.pricing), request_location: self.settings.request_location.clone(), diff --git a/backend/crates/gateway-core/src/runtime/mod.rs b/backend/crates/gateway-core/src/runtime/mod.rs index d43baa5ce..7ffaf46db 100644 --- a/backend/crates/gateway-core/src/runtime/mod.rs +++ b/backend/crates/gateway-core/src/runtime/mod.rs @@ -3,7 +3,9 @@ mod publications; mod publisher; -pub use publications::{RuntimeSnapshotHandle, RuntimeSnapshotUnavailable}; +pub use publications::{ + RuntimeSnapshotHandle, RuntimeSnapshotUnavailable, WeakRuntimeSnapshotHandle, +}; pub use publisher::{ RuntimeSnapshotPublisher, SnapshotControl, SnapshotRevisionStream, SnapshotSubscriptionError, SnapshotSubscriptionPort, runtime_revision_needs_refresh, diff --git a/backend/crates/gateway-core/src/runtime/publications.rs b/backend/crates/gateway-core/src/runtime/publications.rs index 693c68776..7a43688d1 100644 --- a/backend/crates/gateway-core/src/runtime/publications.rs +++ b/backend/crates/gateway-core/src/runtime/publications.rs @@ -8,7 +8,7 @@ use crate::{ use futures::future::BoxFuture; use std::{ collections::BTreeMap, - sync::{Arc, Mutex as SyncMutex, RwLock}, + sync::{Arc, Mutex as SyncMutex, RwLock, Weak}, }; /// 请求级冻结失败；此状态必须 fail closed @@ -23,7 +23,27 @@ pub struct RuntimeSnapshotHandle { publications: Arc>>>, } +/// 不延长发布集合生命周期的只读句柄，供集合持有的扩展读取后续配置 +#[derive(Clone, Default)] +pub struct WeakRuntimeSnapshotHandle { + current: Weak>>>, +} + +impl WeakRuntimeSnapshotHandle { + pub fn acquire(&self) -> Result, RuntimeSnapshotUnavailable> { + let current = self.current.upgrade().ok_or(RuntimeSnapshotUnavailable)?; + acquire_current(¤t) + } +} + impl RuntimeSnapshotHandle { + #[must_use] + pub fn downgrade(&self) -> WeakRuntimeSnapshotHandle { + WeakRuntimeSnapshotHandle { + current: Arc::downgrade(&self.current), + } + } + #[must_use] pub fn new(initial: RuntimeSnapshot) -> Self { Self { @@ -86,10 +106,7 @@ impl RuntimeSnapshotHandle { /// 冻结当前 Arc；后续发布不改变已经开始的请求 pub fn acquire(&self) -> Result, RuntimeSnapshotUnavailable> { - read_unpoisoned(&self.current) - .clone() - .filter(|snapshot| snapshot.extensions().is_none_or(|set| set.can_serve())) - .ok_or(RuntimeSnapshotUnavailable) + acquire_current(&self.current) } } @@ -109,6 +126,15 @@ impl HealthProbe for RuntimeSnapshotHandle { } } +fn acquire_current( + current: &RwLock>>, +) -> Result, RuntimeSnapshotUnavailable> { + read_unpoisoned(current) + .clone() + .filter(|snapshot| snapshot.extensions().is_none_or(|set| set.can_serve())) + .ok_or(RuntimeSnapshotUnavailable) +} + fn read_unpoisoned(lock: &RwLock) -> std::sync::RwLockReadGuard<'_, T> { lock.read() .unwrap_or_else(std::sync::PoisonError::into_inner) diff --git a/backend/crates/gateway-core/tests/routing/snapshot.rs b/backend/crates/gateway-core/tests/routing/snapshot.rs index f19ca164f..7e2386de1 100644 --- a/backend/crates/gateway-core/tests/routing/snapshot.rs +++ b/backend/crates/gateway-core/tests/routing/snapshot.rs @@ -1253,7 +1253,8 @@ fn decompression_setting_should_validate_and_remain_frozen_across_publication() )); } let handle = RuntimeSnapshotHandle::new(compile(1, 64 * 1024 * 1024).unwrap()); - let frozen = handle.acquire().unwrap(); + let reader = handle.downgrade(); + let frozen = reader.acquire().unwrap(); handle.publish(compile(2, 128 * 1024 * 1024).unwrap()); assert_eq!( handle @@ -1266,6 +1267,13 @@ fn decompression_setting_should_validate_and_remain_frozen_across_publication() frozen.responses_max_decompressed_body_bytes(), 64 * 1024 * 1024 ); + assert_eq!( + reader + .acquire() + .unwrap() + .responses_max_decompressed_body_bytes(), + 128 * 1024 * 1024 + ); handle.publish(compile(3, 1024 * 1024).unwrap()); assert_eq!( handle @@ -1274,6 +1282,10 @@ fn decompression_setting_should_validate_and_remain_frozen_across_publication() .responses_max_decompressed_body_bytes(), 1024 * 1024 ); + handle.suspend(); + assert!(reader.acquire().is_err()); + drop(handle); + assert!(reader.acquire().is_err()); } #[test] diff --git a/backend/crates/gateway-host/src/outbound/http.rs b/backend/crates/gateway-host/src/outbound/http.rs index 9fc5f6762..910e98c3e 100644 --- a/backend/crates/gateway-host/src/outbound/http.rs +++ b/backend/crates/gateway-host/src/outbound/http.rs @@ -40,6 +40,8 @@ pub struct HttpRequest { pub url: String, pub headers: Vec<(String, Vec)>, pub body: Vec, + /// 调用方从有效配置派生的请求体上限；零只允许空正文 + pub maximum_body_bytes: usize, } pub struct HttpResponse { @@ -191,7 +193,7 @@ impl HttpClient { scope: Option<&str>, websocket_key: Option<&str>, ) -> Result { - if request.body.len() > 16 * 1024 * 1024 + if request.body.len() > request.maximum_body_bytes || request.headers.len() > 128 || timeout.is_zero() || timeout > Duration::from_secs(120) diff --git a/backend/crates/gateway-host/src/plugin_distribution/http.rs b/backend/crates/gateway-host/src/plugin_distribution/http.rs index ef0cd0901..fceedd538 100644 --- a/backend/crates/gateway-host/src/plugin_distribution/http.rs +++ b/backend/crates/gateway-host/src/plugin_distribution/http.rs @@ -96,6 +96,7 @@ impl Downloads { url: url.to_string(), headers, body: vec![], + maximum_body_bytes: 0, }, egress.map(|egress| &egress.proxy), &self.network, diff --git a/backend/crates/gateway-host/tests/outbound/connector.rs b/backend/crates/gateway-host/tests/outbound/connector.rs index aa04d606a..ccb75073c 100644 --- a/backend/crates/gateway-host/tests/outbound/connector.rs +++ b/backend/crates/gateway-host/tests/outbound/connector.rs @@ -16,6 +16,7 @@ fn request(port: u16, secure: bool) -> HttpRequest { ), headers: vec![], body: vec![], + maximum_body_bytes: 0, } } diff --git a/backend/crates/gateway-host/tests/outbound/http.rs b/backend/crates/gateway-host/tests/outbound/http.rs index e0ed35cc3..2c1a73f10 100644 --- a/backend/crates/gateway-host/tests/outbound/http.rs +++ b/backend/crates/gateway-host/tests/outbound/http.rs @@ -12,6 +12,7 @@ fn request(url: String) -> HttpRequest { url, headers: vec![], body: vec![], + maximum_body_bytes: 0, } } @@ -316,3 +317,40 @@ async fn cancelling_dns_releases_the_request_slot_without_starting_http() { ); } } + +#[tokio::test] +async fn managed_http_enforces_the_callers_body_limit_before_sending() { + let server = MockServer::start().await; + Mock::given(any()) + .respond_with(ResponseTemplate::new(200)) + .expect(2) + .mount(&server) + .await; + let client = HttpClient::new().unwrap(); + // 覆盖旧固定上限之外的正文，并验证同一个客户端逐次使用调用方的预算 + let maximum = 17 * 1024 * 1024; + for (size, limit, accepted) in [ + (maximum, maximum, true), + (maximum + 1, maximum, false), + (maximum + 1, maximum + 1, true), + ] { + let mut request = request(server.uri()); + request.method = "POST".into(); + request.body = vec![b'x'; size]; + request.maximum_body_bytes = limit; + let result = client + .open(request, None, &local_network(), Duration::from_secs(10)) + .await; + if accepted { + assert_eq!(result.unwrap().status, 200); + } else { + let error = result.err().unwrap(); + assert_eq!(error.kind(), HttpErrorKind::InvalidRequest); + assert_eq!(error.send_state, UpstreamSendState::NotSent); + } + } + let requests = server.received_requests().await.unwrap(); + assert_eq!(requests.len(), 2); + assert_eq!(requests[0].body.len(), maximum); + assert_eq!(requests[1].body.len(), maximum + 1); +} diff --git a/backend/crates/gateway-host/tests/outbound/network.rs b/backend/crates/gateway-host/tests/outbound/network.rs index f51d3e415..58fd36472 100644 --- a/backend/crates/gateway-host/tests/outbound/network.rs +++ b/backend/crates/gateway-host/tests/outbound/network.rs @@ -98,6 +98,7 @@ fn request(url: String) -> HttpRequest { url, headers: vec![], body: vec![], + maximum_body_bytes: 0, } } diff --git a/backend/crates/gateway-host/tests/outbound/websocket.rs b/backend/crates/gateway-host/tests/outbound/websocket.rs index 58e92f92e..c2b13c89d 100644 --- a/backend/crates/gateway-host/tests/outbound/websocket.rs +++ b/backend/crates/gateway-host/tests/outbound/websocket.rs @@ -16,6 +16,7 @@ fn request(url: String) -> HttpRequest { url, headers: vec![], body: vec![], + maximum_body_bytes: 0, } } fn network() -> NetworkPolicy { diff --git a/backend/crates/gateway-plugin/runtime/src/adapter/middleware/request.rs b/backend/crates/gateway-plugin/runtime/src/adapter/middleware/request.rs index aed53e177..d38d12e66 100644 --- a/backend/crates/gateway-plugin/runtime/src/adapter/middleware/request.rs +++ b/backend/crates/gateway-plugin/runtime/src/adapter/middleware/request.rs @@ -86,6 +86,7 @@ pub(super) async fn invoke( context.extension_scope().clone(), context.plan(), context.cancellation().clone(), + context.request_body_limit(), ) .map_err(|_| MiddlewareError::Fault)?; let _binding = invocation_ports.callbacks.bind_middleware( diff --git a/backend/crates/gateway-plugin/runtime/src/adapter/policy/retry.rs b/backend/crates/gateway-plugin/runtime/src/adapter/policy/retry.rs index 8b44ca61f..e5117583f 100644 --- a/backend/crates/gateway-plugin/runtime/src/adapter/policy/retry.rs +++ b/backend/crates/gateway-plugin/runtime/src/adapter/policy/retry.rs @@ -91,6 +91,7 @@ async fn call( input.extension_scope.clone(), None, gateway_core::lifecycle::CancellationToken::new(), + input.request_body_limit(), ) .map_err(|_| ())?; let reply = invocation diff --git a/backend/crates/gateway-plugin/runtime/src/adapter/policy/route_schedule.rs b/backend/crates/gateway-plugin/runtime/src/adapter/policy/route_schedule.rs index 62e6cdec7..8bcd264ed 100644 --- a/backend/crates/gateway-plugin/runtime/src/adapter/policy/route_schedule.rs +++ b/backend/crates/gateway-plugin/runtime/src/adapter/policy/route_schedule.rs @@ -137,6 +137,7 @@ async fn route_with_entry( input.extension_scope().clone(), None, gateway_core::lifecycle::CancellationToken::new(), + input.request_body_limit(), ) .map_err(|_| ())?; let reply = invocation @@ -212,6 +213,7 @@ async fn schedule_with_entry( input.extension_scope().clone(), None, gateway_core::lifecycle::CancellationToken::new(), + input.request_body_limit(), ) .map_err(|_| ())?; let reply = invocation diff --git a/backend/crates/gateway-plugin/runtime/src/adapter/upstream_adapter/execution.rs b/backend/crates/gateway-plugin/runtime/src/adapter/upstream_adapter/execution.rs index a0120839f..bfb60f44e 100644 --- a/backend/crates/gateway-plugin/runtime/src/adapter/upstream_adapter/execution.rs +++ b/backend/crates/gateway-plugin/runtime/src/adapter/upstream_adapter/execution.rs @@ -212,10 +212,21 @@ impl Execution { .extension_scope() .extending(self.adapter.instance_id.clone()) .ok_or_else(invalid)?; + let request_body_limit = invocation + .context + .request_body_limit() + .map(Ok) + .unwrap_or_else(|| self.adapter.callbacks.request_body_limit()) + .map_err(|error| fault_error(error, UpstreamSendState::NotSent))?; let scope = self .adapter .callbacks - .prepare_upstream(&context, Arc::clone(&managed), extension_scope) + .prepare_upstream( + &context, + Arc::clone(&managed), + extension_scope, + request_body_limit, + ) .map_err(|error| fault_error(error, UpstreamSendState::NotSent))?; let request = UpstreamAdapterRequest { adapter_id: declaration.id.clone(), @@ -239,14 +250,18 @@ impl Execution { .collect(), continuation: previous.map(|state| state.continuation), }; + let body = invocation + .operation + .middleware_body() + .map_err(|source| invalid().with_source(source))?; + if body.len() > gateway_plugin_sdk::call::model::request_payload_limit(request_body_limit) { + return Err(fault_error( + crate::callback::request_limits::too_large(), + UpstreamSendState::NotSent, + )); + } let payload = request - .encode( - invocation - .operation - .middleware_body() - .map_err(|source| invalid().with_source(source))? - .to_vec(), - ) + .encode(body.to_vec()) .map_err(|source| invalid().with_source(source))?; invocation.context.trace().record("plugin.upstream", serde_json::json!({ "instanceId": self.adapter.instance_id, "adapterId": declaration.id, "generation": context.generation, "transport": declaration.transport.as_str() })); let mut stream = tokio::select! { diff --git a/backend/crates/gateway-plugin/runtime/src/callback/http.rs b/backend/crates/gateway-plugin/runtime/src/callback/http.rs index 5970057f7..24a950353 100644 --- a/backend/crates/gateway-plugin/runtime/src/callback/http.rs +++ b/backend/crates/gateway-plugin/runtime/src/callback/http.rs @@ -17,9 +17,18 @@ pub(super) struct HttpCallbacks<'a> { pub(super) scope: &'a Arc, pub(super) call: &'a CallResources, pub(super) maximum_payload: usize, + pub(super) request_limits: &'a super::request_limits::RequestLimits, } impl HttpCallbacks<'_> { + pub(super) fn body_limit(&self, body: &[u8]) -> Result { + // 空正文不占请求预算，也允许尚未发布设置的注册阶段执行无正文查询 + if body.is_empty() { + return Ok(0); + } + self.request_limits.body_bytes(self.call) + } + pub(super) async fn send( &self, request: HttpRequest, @@ -32,7 +41,11 @@ impl HttpCallbacks<'_> { scope, call, maximum_payload, + .. } = *self; + if request.body.len() > request.maximum_body_bytes { + return Err(super::request_limits::too_large()); + } let timeout = call.timeout()?; let attempt = call.start_upstream(upstream_purpose)?; let response = client @@ -116,6 +129,7 @@ impl HttpCallbacks<'_> { .into_iter() .map(|(name, value)| (name, value.into_bytes())) .collect(), + maximum_body_bytes: self.body_limit(&payload)?, body: payload, }, method == "host.http.do_stream", diff --git a/backend/crates/gateway-plugin/runtime/src/callback/mod.rs b/backend/crates/gateway-plugin/runtime/src/callback/mod.rs index 722ea0e85..2359e5c89 100644 --- a/backend/crates/gateway-plugin/runtime/src/callback/mod.rs +++ b/backend/crates/gateway-plugin/runtime/src/callback/mod.rs @@ -15,6 +15,7 @@ mod log; mod middleware; mod model; pub(crate) mod private_state; +pub(crate) mod request_limits; mod resources; mod scope; pub(crate) mod services; @@ -80,6 +81,7 @@ impl PluginCallbackPorts { } pub(crate) struct PluginCallbacks { + pub(crate) registration_body_limit: Option, services: Arc, service_actor: gateway_admin::model::usage_data::UsageActor, accounts: Arc, @@ -96,6 +98,7 @@ pub(crate) struct PluginCallbacks { pending_middleware: Mutex>>, calls: Mutex>>, maximum_payload: usize, + request_limits: Arc, } impl PluginCallbacks { @@ -106,8 +109,10 @@ impl PluginCallbacks { log_slots: Arc, private_state: Arc, ports: PluginCallbackPorts, + request_limits: Arc, ) -> Result { Ok(Self { + registration_body_limit: None, services: ports.services, service_actor: gateway_admin::model::usage_data::UsageActor::Plugin { publisher: manifest.publisher.clone(), @@ -126,7 +131,11 @@ impl PluginCallbacks { log::PluginLog::new(manifest, log_slots) .map_err(|_| AdminError::invalid("插件清单身份不合法"))?, ), - models: Arc::new(model::PluginModels::new(ports.models, maximum_payload)), + models: Arc::new(model::PluginModels::new( + ports.models, + request_limits.clone(), + maximum_payload, + )), private_state, http: ports.http, network: Arc::new(ports.network), @@ -134,6 +143,7 @@ impl PluginCallbacks { pending_middleware: Mutex::new(BTreeMap::new()), calls: Mutex::new(BTreeMap::new()), maximum_payload, + request_limits, }) } @@ -181,6 +191,7 @@ impl PluginCallbacks { extensions: gateway_core::engine::extensions::ExtensionCallScope, plan: Option, cancellation: CancellationToken, + request_body_limit: Option, ) -> Result, PluginFault> { if !matches!( context.stage, @@ -197,6 +208,7 @@ impl PluginCallbacks { scope = scope.with_execution_effects(effects); } scope.extension_scope = extensions; + scope.request_body_limit = request_body_limit; scope.origin = plan.map(|plan| InvocationContext { request_id: context .request_id @@ -212,11 +224,16 @@ impl PluginCallbacks { self.register_scope(context, scope) } + pub(crate) fn request_body_limit(&self) -> Result { + self.request_limits.current_body_bytes() + } + pub(crate) fn prepare_upstream( &self, context: &CallContext, managed: Arc, extension_scope: gateway_core::engine::extensions::ExtensionCallScope, + request_body_limit: usize, ) -> Result, PluginFault> { if context.stage != gateway_plugin_sdk::Stage::Upstream || context.account_id.as_deref() != Some(managed.account.account_id().as_str()) @@ -227,6 +244,7 @@ impl PluginCallbacks { let mut scope = CallbackScope::new(context, managed.account.outbound_proxy().cloned()) .with_upstream(managed); scope.extension_scope = extension_scope; + scope.request_body_limit = Some(request_body_limit); self.register_scope(context, scope) } @@ -352,7 +370,13 @@ impl CallbackHandler for PluginCallbacks { .unwrap_or_else(std::sync::PoisonError::into_inner) .get(&context.resource_scope_id) .and_then(Weak::upgrade) - .unwrap_or_else(|| Arc::new(CallbackScope::for_call(context))); + .unwrap_or_else(|| { + let mut scope = CallbackScope::for_call(context); + if context.stage == gateway_plugin_sdk::Stage::Registration { + scope.request_body_limit = self.registration_body_limit; + } + Arc::new(scope) + }); let middleware = self .pending_middleware .lock() @@ -401,6 +425,7 @@ impl CallbackHandler for PluginCallbacks { let data = self.data.clone(); let keys = self.keys.clone(); let models = self.models.clone(); + let request_limits = self.request_limits.clone(); let affinity = self.affinity.clone(); let services = self.services.clone(); let service_actor = self.service_actor.clone(); @@ -514,6 +539,7 @@ impl CallbackHandler for PluginCallbacks { scope, call: &call, maximum_payload, + request_limits: &request_limits, }; if method.starts_with("host.upstream.") { return upstream::dispatch(&http, &method, params, payload).await; diff --git a/backend/crates/gateway-plugin/runtime/src/callback/model.rs b/backend/crates/gateway-plugin/runtime/src/callback/model.rs index 99fe13581..168263db4 100644 --- a/backend/crates/gateway-plugin/runtime/src/callback/model.rs +++ b/backend/crates/gateway-plugin/runtime/src/callback/model.rs @@ -63,13 +63,19 @@ impl PluginModelPortSlot { pub(super) struct PluginModels { slot: Arc, + request_limits: Arc, maximum_payload: usize, } impl PluginModels { - pub(super) fn new(slot: Arc, maximum_payload: usize) -> Self { + pub(super) fn new( + slot: Arc, + request_limits: Arc, + maximum_payload: usize, + ) -> Self { Self { slot, + request_limits, maximum_payload, } } @@ -121,17 +127,27 @@ impl PluginModels { "host.model.execute" | "host.model.execute_stream" => { let mut request: ModelExecuteRequest = serde_json::from_value(params).map_err(|_| invalid())?; - if payload.is_empty() || payload.len() > self.maximum_payload { + if payload.is_empty() { return Err(invalid()); } + let check_size = |limit| { + if payload.len() > gateway_plugin_sdk::call::model::request_payload_limit(limit) + { + Err(super::request_limits::too_large()) + } else { + Ok(()) + } + }; let port = self.slot.upgrade()?; let stream = method == "host.model.execute_stream"; let started = if let Some(client_key_id) = request.client_key_id.take() { let bound = self.binding(&port, context, call, client_key_id).await?; + check_size(bound.request_body_limit())?; port.start_bound(self.bound_request(request, payload, stream, bound)?) .await .map_err(super::error::gateway)? } else { + check_size(self.request_limits.body_bytes(call)?)?; let request = self.request(context, request, payload, stream)?; port.start(request).await.map_err(super::error::gateway)? }; diff --git a/backend/crates/gateway-plugin/runtime/src/callback/request_limits.rs b/backend/crates/gateway-plugin/runtime/src/callback/request_limits.rs new file mode 100644 index 000000000..6fb5de551 --- /dev/null +++ b/backend/crates/gateway-plugin/runtime/src/callback/request_limits.rs @@ -0,0 +1,49 @@ +//! 从当前请求或已发布设置读取正文预算，不持有发布集合的强引用 + +use super::CallResources; +use gateway_core::runtime::{RuntimeSnapshotHandle, WeakRuntimeSnapshotHandle}; +use gateway_plugin_sdk::{ErrorCode, PluginFault}; +use std::sync::OnceLock; + +#[derive(Default)] +pub(crate) struct RequestLimits { + snapshots: OnceLock, +} + +impl RequestLimits { + pub(crate) fn bind( + &self, + snapshots: &RuntimeSnapshotHandle, + ) -> Result<(), gateway_admin::model::AdminError> { + self.snapshots + .set(snapshots.downgrade()) + .map_err(|_| gateway_admin::model::AdminError::conflict("插件设置读取端口已经绑定")) + } + + pub(super) fn body_bytes(&self, call: &CallResources) -> Result { + if let Some(settings) = call.request_settings() { + return Ok(settings.snapshot().responses_max_decompressed_body_bytes()); + } + if let Some(bytes) = call.scope.request_body_limit { + return Ok(bytes); + } + self.current_body_bytes() + } + + pub(crate) fn current_body_bytes(&self) -> Result { + self.snapshots + .get() + .and_then(|snapshots| snapshots.acquire().ok()) + .map(|snapshot| snapshot.responses_max_decompressed_body_bytes()) + .ok_or_else(|| PluginFault::new(ErrorCode::Fault, "request body limit is unavailable")) + } +} + +pub(crate) fn too_large() -> PluginFault { + let mut fault = PluginFault::new( + ErrorCode::InvalidInput, + "request body exceeds the configured limit", + ); + fault.http_status = Some(413); + fault +} diff --git a/backend/crates/gateway-plugin/runtime/src/callback/scope.rs b/backend/crates/gateway-plugin/runtime/src/callback/scope.rs index a9dae78e2..475847cae 100644 --- a/backend/crates/gateway-plugin/runtime/src/callback/scope.rs +++ b/backend/crates/gateway-plugin/runtime/src/callback/scope.rs @@ -19,6 +19,7 @@ pub(crate) struct InvocationContext { /// 一次回调的调用链、账号与网络事实，由操作持有到完成 pub(crate) struct CallbackScope { pub(super) origin: Option, + pub(super) request_body_limit: Option, stage: Stage, account_id: Option, credential_revision: Option, @@ -61,6 +62,7 @@ impl CallbackScope { pub(super) fn new(context: &CallContext, proxy: Option) -> Self { Self { origin: None, + request_body_limit: None, stage: context.stage, account_id: context.account_id.clone(), credential_revision: context.credential_revision, diff --git a/backend/crates/gateway-plugin/runtime/src/callback/upstream/mod.rs b/backend/crates/gateway-plugin/runtime/src/callback/upstream/mod.rs index c030a0a46..66154e733 100644 --- a/backend/crates/gateway-plugin/runtime/src/callback/upstream/mod.rs +++ b/backend/crates/gateway-plugin/runtime/src/callback/upstream/mod.rs @@ -133,6 +133,7 @@ impl ManagedUpstream { &self, request: adapter::UpstreamHttpRequest, body: Vec, + maximum_body_bytes: usize, ) -> Result<(HttpRequest, UpstreamPathPurpose), PluginFault> { let (target, purpose) = self.target.resolve(&request.path, &request.query)?; if request.headers.len() > 128 { @@ -170,6 +171,7 @@ impl ManagedUpstream { url: target.into(), headers, body, + maximum_body_bytes, }, purpose, )) @@ -188,6 +190,7 @@ pub(super) async fn dispatch( scope, call, maximum_payload, + .. } = *http; let managed = scope.upstream.as_ref().ok_or_else(denied)?; let timeout = || { @@ -197,7 +200,8 @@ pub(super) async fn dispatch( match method { "host.upstream.http.do" | "host.upstream.http.do_stream" => { let request = serde_json::from_value(params).map_err(|_| invalid())?; - let (request, purpose) = managed.request(request, payload)?; + let maximum_body_bytes = http.body_limit(&payload)?; + let (request, purpose) = managed.request(request, payload, maximum_body_bytes)?; http.send(request, method.ends_with("do_stream"), Some(purpose)) .await } @@ -227,6 +231,7 @@ pub(super) async fn dispatch( headers: request.headers, }, vec![], + 0, )?; if purpose != UpstreamPathPurpose::Inference { return Err(denied()); diff --git a/backend/crates/gateway-plugin/runtime/src/generation/prepare/instance.rs b/backend/crates/gateway-plugin/runtime/src/generation/prepare/instance.rs index cddbe610d..1d3b8c3ed 100644 --- a/backend/crates/gateway-plugin/runtime/src/generation/prepare/instance.rs +++ b/backend/crates/gateway-plugin/runtime/src/generation/prepare/instance.rs @@ -21,6 +21,7 @@ impl PluginRuntime { &self, instance: PluginInstance, target_revision: Revision, + request_body_limit: Option, ) -> Result, AdminError> { let mut observer_entries = Vec::new(); let mut commands = Vec::new(); @@ -151,7 +152,7 @@ impl PluginRuntime { configuration, contributes: manifest.contributes.clone(), }; - let callbacks = Arc::new(PluginCallbacks::new( + let mut callbacks = PluginCallbacks::new( &instance, self.config.rpc_limits.maximum_buffered_body_bytes, &manifest, @@ -166,7 +167,10 @@ impl PluginRuntime { self.affinity_ports.clone(), self.resource_ports.clone(), ), - )?); + self.request_limits.clone(), + )?; + callbacks.registration_body_limit = request_body_limit; + let callbacks = Arc::new(callbacks); let session = Arc::new( RpcSession::start_supervised( package.clone(), diff --git a/backend/crates/gateway-plugin/runtime/src/generation/prepare/management.rs b/backend/crates/gateway-plugin/runtime/src/generation/prepare/management.rs index 2a18fddec..1364e12a1 100644 --- a/backend/crates/gateway-plugin/runtime/src/generation/prepare/management.rs +++ b/backend/crates/gateway-plugin/runtime/src/generation/prepare/management.rs @@ -13,7 +13,7 @@ impl PluginRuntime { .load_instances() .await .map_err(|_| AdminError::unavailable("无法读取已安装插件"))?; - let reference = self.prepare_snapshot(snapshot, None).await?; + let reference = self.prepare_snapshot(snapshot, None, None).await?; let set = self.prepared_set(&reference).await?; Ok(crate::PluginCommandSession::new( reference, diff --git a/backend/crates/gateway-plugin/runtime/src/generation/prepare/runtime.rs b/backend/crates/gateway-plugin/runtime/src/generation/prepare/runtime.rs index 1e1187b3c..3ded63376 100644 --- a/backend/crates/gateway-plugin/runtime/src/generation/prepare/runtime.rs +++ b/backend/crates/gateway-plugin/runtime/src/generation/prepare/runtime.rs @@ -52,6 +52,7 @@ pub struct PluginRuntimeConfig { /// 只索引有主人的候选；当前代次仅由 Core 的发布视图持有 pub struct PluginRuntime { + pub(super) request_limits: Arc, pub(super) service_ports: Arc, pub(super) store: Arc, pub(super) state: Arc, @@ -109,6 +110,7 @@ impl PluginRuntime { processes, validators: Arc::new(Semaphore::new(MAXIMUM_PARALLEL_PREPARATIONS)), configuration_schemas: Arc::default(), + request_limits: Arc::default(), service_ports: Arc::new(crate::callback::services::ServicePorts::default()), account_ports: Arc::new(PluginAccountPortSlot::new()), client_key_ports: Arc::new(PluginClientKeyPortSlot::new()), @@ -138,6 +140,14 @@ impl PluginRuntime { self.service_ports.bind_http(dispatcher) } + /// 读取当前发布设置，同时避免回调反向保活自身所在的插件集合 + pub fn bind_snapshots( + &self, + snapshots: &gateway_core::runtime::RuntimeSnapshotHandle, + ) -> Result<(), AdminError> { + self.request_limits.bind(snapshots) + } + pub fn bind_services( &self, registry: &Arc, diff --git a/backend/crates/gateway-plugin/runtime/src/generation/prepare/snapshot.rs b/backend/crates/gateway-plugin/runtime/src/generation/prepare/snapshot.rs index 2dc464a06..9391e2c0c 100644 --- a/backend/crates/gateway-plugin/runtime/src/generation/prepare/snapshot.rs +++ b/backend/crates/gateway-plugin/runtime/src/generation/prepare/snapshot.rs @@ -35,10 +35,13 @@ impl PluginRuntime { &self, snapshot: PluginInstanceSnapshot, required_revision: Option, + request_body_limit: Option, ) -> Result { let target_revision = snapshot.config_revision.get(); self.record_preparation(target_revision, PreparationDiagnostic::Preparing); - let result = self.prepare_inner(snapshot, required_revision).await; + let result = self + .prepare_inner(snapshot, required_revision, request_body_limit) + .await; match &result { Ok(reference) => self.record_preparation( target_revision, @@ -58,6 +61,7 @@ impl PluginRuntime { &self, snapshot: PluginInstanceSnapshot, required_revision: Option, + request_body_limit: Option, ) -> Result { if self.shutting_down.load(Ordering::Acquire) { return Err(AdminError::unavailable("插件运行时正在关闭")); @@ -124,7 +128,11 @@ impl PluginRuntime { tokio::time::timeout( // 沿用准备阶段原有的 30 秒预算；打包校验与能力恢复也在预算内 Duration::from_secs(30), - self.prepare_instance(instance.clone(), snapshot.config_revision), + self.prepare_instance( + instance.clone(), + snapshot.config_revision, + request_body_limit, + ), ) .await .unwrap_or_else(|_| Err(AdminError::unavailable("插件启动超时"))) @@ -363,15 +371,33 @@ impl PluginPreparation for PluginRuntime { snapshot: PluginInstanceSnapshot, ) -> Result { let required_revision = snapshot.config_revision; - self.prepare_snapshot(snapshot, Some(required_revision)) + self.prepare_snapshot(snapshot, Some(required_revision), None) .await } } impl ExtensionPreparationPort for PluginRuntime { + fn prepare_with_request_body_limit( + &self, + revision: ConfigRevision, + request_body_limit: usize, + ) -> BoxFuture<'_, Result> { + self.prepare_revision(revision, Some(request_body_limit)) + } + fn prepare( &self, revision: ConfigRevision, + ) -> BoxFuture<'_, Result> { + self.prepare_revision(revision, None) + } +} + +impl PluginRuntime { + fn prepare_revision( + &self, + revision: ConfigRevision, + request_body_limit: Option, ) -> BoxFuture<'_, Result> { Box::pin(async move { let snapshot = self @@ -382,7 +408,7 @@ impl ExtensionPreparationPort for PluginRuntime { if snapshot.config_revision.get() != revision.get() { return Err(ExtensionPreparationError); } - self.prepare_snapshot(snapshot, None) + self.prepare_snapshot(snapshot, None, request_body_limit) .await .map_err(|_| ExtensionPreparationError) }) diff --git a/backend/crates/gateway-plugin/runtime/tests/adapter/command_line/execution.rs b/backend/crates/gateway-plugin/runtime/tests/adapter/command_line/execution.rs index 7bfbc479e..14a5f7d77 100644 --- a/backend/crates/gateway-plugin/runtime/tests/adapter/command_line/execution.rs +++ b/backend/crates/gateway-plugin/runtime/tests/adapter/command_line/execution.rs @@ -6,6 +6,116 @@ use serde_json::json; use crate::support::environment::Environment; +#[tokio::test] +async fn model_requests_have_a_body_limit_independent_of_the_response_buffer() { + use gateway_plugin_sdk::call::model::request_payload_limit; + + for stream in [false, true] { + let Some(mut environment) = Environment::create_command().await else { + eprintln!("SKIP: plugin integration environment absent"); + return; + }; + let account = environment.account(None).await; + let key_id = format!("key_{}", uuid::Uuid::new_v4().simple()); + environment + .client_key(&key_id, "sk-large-body-fixture") + .await; + let body_path = environment.directory.path().join("model-body.json"); + let mut body = serde_json::to_vec(&json!({ + "model":crate::support::native::MODEL, + "input":"x".repeat(2 * 1024 * 1024), + })) + .unwrap(); + let raw_limit = 2 * 1024 * 1024; + body.resize(request_payload_limit(raw_limit), b' '); + std::fs::write(&body_path, &body).unwrap(); + let marker = environment.directory.path().join("model-results.jsonl"); + environment + .install_plugin(json!({ + "plugin_id":"test.large-model-request", + "command_registration":{"commands":[{"name":"run","description":"执行模型"}]}, + "command_nested_model_fixture":{ + "request":{ + "client_key_id":key_id, + "model":crate::support::native::MODEL, + "protocol":"openai", + "operation":"generate", + "provider":"openai", + "account_id":account.as_str(), + }, + "body_file":body_path, + "stream":stream, + "expect_error":true, + }, + "command_nested_model_marker":marker, + "command_result":{"stdout":"ok\n","stderr":"","exit_code":0}, + })) + .await; + let (runtime, core) = environment.runtime().await; + let snapshots = core.snapshots(); + let configure = |bytes: usize| { + let snapshot = snapshots.acquire().unwrap(); + let settings = snapshot + .settings() + .clone() + .with_responses_max_decompressed_body_bytes(bytes as u64); + let updated = gateway_core::routing::request_settings::RequestSettings::new(snapshot) + .replace(settings, "fixture") + .unwrap(); + snapshots.publish((*updated.snapshot()).clone()); + }; + configure(raw_limit); + environment.store.start_command_line_writes().unwrap(); + let instances = environment + .store + .admin_ports() + .plugins() + .load_instances() + .await + .unwrap(); + let commands = runtime.prepare_command_line().await.unwrap(); + let instance = &instances.instances[0]; + commands.execute(&instance.id, "run", &[]).await.unwrap(); + environment.wait_for_bound_model_requests(&key_id, 1).await; + let first: serde_json::Value = serde_json::from_str( + std::fs::read_to_string(&marker) + .unwrap() + .lines() + .next() + .unwrap(), + ) + .unwrap(); + assert!(first["request_id"].is_string(), "{first}"); + assert!(first["events"].as_u64().unwrap() > 0); + + // 合法 JSON 仅多一个空白字节，超限必须在进入 Core 和账本前拒绝 + let mut oversized = body.clone(); + oversized.push(b'\n'); + std::fs::write(&body_path, oversized).unwrap(); + commands.execute(&instance.id, "run", &[]).await.unwrap(); + let records = std::fs::read_to_string(&marker).unwrap(); + let rejected: serde_json::Value = + serde_json::from_str(records.lines().last().unwrap()).unwrap(); + assert_eq!(rejected["error"], "invalid_input"); + assert_eq!(rejected["http_status"], 413); + assert_eq!(environment.bound_model_requests(&key_id).await.len(), 1); + + // 不重新准备插件实例：发布更大的配置后，同一个请求必须恢复成功 + configure(raw_limit + 1); + commands.execute(&instance.id, "run", &[]).await.unwrap(); + environment.wait_for_bound_model_requests(&key_id, 2).await; + commands.shutdown().await; + environment + .store + .shutdown_command_line_writes() + .await + .unwrap(); + drop(core); + drop(runtime); + environment.close().await; + } +} + #[tokio::test] async fn command_selects_key_at_call_time_and_uses_normal_model_ledger() { let Some(mut environment) = Environment::create_command().await else { @@ -277,3 +387,131 @@ async fn command_accounts_domain_does_not_need_separate_login_or_write_grants() drop(runtime); environment.close().await; } + +#[tokio::test] +async fn bound_model_body_budget_stays_frozen_until_the_parent_call_ends() { + use gateway_plugin_sdk::call::model::request_payload_limit; + + for stream in [false, true] { + for growing in [false, true] { + let Some(mut environment) = Environment::create_command().await else { + eprintln!("SKIP: plugin integration environment absent"); + return; + }; + let account = environment.account(None).await; + let key_id = format!("key_{}", uuid::Uuid::new_v4().simple()); + environment + .client_key(&key_id, "sk-frozen-body-fixture") + .await; + let old_limit = if growing { 64 * 1024 } else { 128 * 1024 }; + let new_limit = if growing { 128 * 1024 } else { 64 * 1024 }; + let body_path = environment.directory.path().join("model-body.json"); + let body = json!({"model":crate::support::native::MODEL,"input":"hello"}); + let mut large_body = serde_json::to_vec(&body).unwrap(); + large_body.resize( + request_payload_limit(old_limit) + usize::from(growing), + b' ', + ); + std::fs::write(&body_path, large_body).unwrap(); + let marker = environment.directory.path().join("model-results.jsonl"); + let gate = environment.directory.path().join("resume-model"); + let request = json!({ + "client_key_id":key_id,"model":crate::support::native::MODEL, + "protocol":"openai","operation":"generate","provider":"openai", + "account_id":account.as_str(), + }); + environment + .install_plugin(json!({ + "plugin_id":"test.frozen-model-request", + "command_registration":{"commands":[{"name":"run","description":"执行模型"}]}, + "command_nested_model_fixture":[ + {"request":request,"body":body,"stream":stream}, + {"request":request,"body_file":body_path,"stream":stream, + "expect_error":true,"wait_for_file":gate}, + ], + "command_nested_model_marker":marker, + "command_result":{"stdout":"ok\n","stderr":"","exit_code":0}, + })) + .await; + let (runtime, core) = environment.runtime().await; + let snapshots = core.snapshots(); + let configure = |bytes: usize| { + let snapshot = snapshots.acquire().unwrap(); + let settings = snapshot + .settings() + .clone() + .with_responses_max_decompressed_body_bytes(bytes as u64); + let updated = + gateway_core::routing::request_settings::RequestSettings::new(snapshot) + .replace(settings, "fixture") + .unwrap(); + snapshots.publish((*updated.snapshot()).clone()); + }; + configure(old_limit); + environment.store.start_command_line_writes().unwrap(); + let instances = environment + .store + .admin_ports() + .plugins() + .load_instances() + .await + .unwrap(); + let commands = runtime.prepare_command_line().await.unwrap(); + let instance = &instances.instances[0]; + let update = async { + tokio::time::timeout(std::time::Duration::from_secs(10), async { + while std::fs::read_to_string(&marker) + .unwrap_or_default() + .lines() + .count() + < 1 + { + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + }) + .await + .unwrap(); + configure(new_limit); + std::fs::write(&gate, b"resume").unwrap(); + }; + let (result, ()) = tokio::join!(commands.execute(&instance.id, "run", &[]), update); + result.unwrap(); + let expected = if growing { 1 } else { 2 }; + environment + .wait_for_bound_model_requests(&key_id, expected) + .await; + assert_eq!( + environment.bound_model_requests(&key_id).await.len(), + expected + ); + + // 父调用结束后重新绑定，下一次调用才使用新发布的预算 + commands.execute(&instance.id, "run", &[]).await.unwrap(); + environment.wait_for_bound_model_requests(&key_id, 3).await; + let records = std::fs::read_to_string(&marker) + .unwrap() + .lines() + .map(|line| serde_json::from_str::(line).unwrap()) + .collect::>(); + assert_eq!(records.len(), 4); + for (index, rejected) in [(1, growing), (3, !growing)] { + if rejected { + assert_eq!(records[index]["http_status"], 413, "{records:?}"); + assert_eq!(records[index]["error"], "invalid_input"); + } else { + assert!(records[index]["request_id"].is_string(), "{records:?}"); + } + } + assert_eq!(environment.bound_model_requests(&key_id).await.len(), 3); + commands.shutdown().await; + environment + .store + .shutdown_command_line_writes() + .await + .unwrap(); + drop(core); + drop(runtime); + environment.close().await; + } + } +} diff --git a/backend/crates/gateway-plugin/runtime/tests/generation/policy/mounts.rs b/backend/crates/gateway-plugin/runtime/tests/generation/policy/mounts.rs index fe8f8b90b..2c9de8cc5 100644 --- a/backend/crates/gateway-plugin/runtime/tests/generation/policy/mounts.rs +++ b/backend/crates/gateway-plugin/runtime/tests/generation/policy/mounts.rs @@ -187,3 +187,164 @@ async fn mixed_mounts_keep_independent_order_and_fresh_continuations() { drop(generation); runtime.shutdown().await; } + +#[tokio::test] +async fn data_plane_http_callbacks_keep_the_frozen_body_budget_across_publications() { + use gateway_core::{ + engine::policy::RetryFacts, + error::ProviderErrorKind, + policy::ClientPolicy, + routing::{RuntimeSnapshot, request_settings::RequestSettings}, + runtime::RuntimeSnapshotHandle, + settings::SettingsValues, + upstream::UpstreamSendState, + }; + + let snapshot = |limit: u64, revision| { + RuntimeSnapshot::new( + ConfigRevision::new(revision).unwrap(), + SettingsValues::new(1, "smart", Default::default(), None, None) + .with_responses_max_decompressed_body_bytes(limit), + vec![], + vec![], + vec![], + ) + .unwrap() + }; + for body_bytes in [128, 129] { + let server = MockServer::start().await; + Mock::given(any()) + .respond_with(ResponseTemplate::new(200)) + .mount(&server) + .await; + let records = tempfile::tempdir().unwrap(); + let marker = records.path().join("http-results.jsonl"); + let (cache, runtime) = setup(vec![InstanceFixture { + id: "frozen-limits", + configuration: serde_json::json!({ + "http_probe_methods":["middleware.handle","policy.route_model","policy.schedule_account","policy.retry_decision"], + "http_probe_url":server.uri(),"http_probe_body_bytes":body_bytes,"http_probe_marker":marker, + "retry_decision":{"decision":"delegate"}, + "route_decision":{"decision":"unhandled"}, + "schedule_decision":{"decision":"delegate"}, + }), + bindings: vec![ + binding(MIDDLEWARE_CONTRIBUTION,"request",0,PluginFailurePolicy::Reject), + binding(MIDDLEWARE_CONTRIBUTION,"attempt",0,PluginFailurePolicy::Reject), + binding(MODEL_ROUTER_CONTRIBUTION,"routing",0,PluginFailurePolicy::Reject), + binding(SCHEDULER_CONTRIBUTION,"scheduling",0,PluginFailurePolicy::Reject), + binding("test.example.retryPolicy","retry",0,PluginFailurePolicy::Delegate), + ], + }]).await; + let snapshots = RuntimeSnapshotHandle::new(snapshot(128, 1)); + runtime.bind_snapshots(&snapshots).unwrap(); + let settings = RequestSettings::new(snapshots.acquire().unwrap()).with_execution( + &ClientPolicy::new( + ClientApiKeyId::new("key-middleware").unwrap(), + gateway_core::policy::PlaintextClientApiKey::new("frozen-budget-fixture").unwrap(), + Arc::new(gateway_core::routing::FrozenAccountScope::new( + Arc::default(), + gateway_core::routing::ClientRoutingScope::all_accounts(), + )), + true, + Default::default(), + ), + Some(60_000), + ); + let generation = prepare(&runtime).await; + let plan = runtime + .execution_registry() + .middleware(&generation) + .unwrap(); + let policy = policy_context(&runtime, generation.clone(), "req_frozen_limits") + .with_request_body_limit(128); + for (revision, current_limit) in [(2, 64), (3, 256)] { + snapshots.publish(snapshot(current_limit, revision)); + for attempt in [false, true] { + let mut request = MiddlewareRequest::new( + "openai", + middleware_headers(), + Bytes::from_static(br#"{"input":"hello"}"#), + ); + let context = if attempt { + attempt_middleware_context(Arc::new(ExecutionEffects::default())) + .with_request_body_limit(Some(128)) + } else { + request = request.with_settings(settings.clone()); + middleware_context(ClientTransport::HttpJson) + }; + let response = plan + .handle( + context, + request, + Downstream { + calls: Arc::default(), + closes: Arc::default(), + reads: Arc::default(), + error: false, + } + .into_next(), + ) + .await + .unwrap(); + let (_, _, _, mut body, _) = response.into_parts(); + while body.next_frame().await.unwrap().is_some() {} + body.close().await; + } + policy + .route_model( + operation(), + PublicModelId::new("public-a").unwrap(), + BTreeSet::from([ProviderKind::new("example").unwrap()]), + ) + .await + .unwrap(); + policy + .select_account( + NonZeroU32::MIN, + &ProviderKind::new("example").unwrap(), + None, + &[candidate("acct_high", 100)], + &mut selection_context(), + ) + .await + .unwrap(); + policy + .retry_decision(RetryFacts { + attempt_index: NonZeroU32::MIN, + provider: ProviderKind::new("example").unwrap(), + model: None, + error_kind: ProviderErrorKind::RateLimited, + upstream_status: Some(429), + send_state: UpstreamSendState::NotSent, + remaining_routing_attempts: 1, + remaining_deadline: Duration::from_secs(3), + retry_allowed: true, + }) + .await; + } + let records = std::fs::read_to_string(marker) + .unwrap() + .lines() + .map(|line| serde_json::from_str::(line).unwrap()) + .collect::>(); + assert_eq!(records.len(), 10, "{records:?}"); + let expected = if body_bytes == 128 { 200 } else { 413 }; + assert!( + records.iter().all(|record| record["status"] == expected), + "{records:?}" + ); + let received = server.received_requests().await.unwrap(); + assert_eq!(received.len(), if body_bytes == 128 { 10 } else { 0 }); + assert!( + received + .iter() + .all(|request| request.body == vec![b'x'; body_bytes]) + ); + drop(policy); + drop(plan); + drop(generation); + runtime.shutdown().await; + super::super::wait_until_empty(cache.path()).await; + } +} diff --git a/backend/crates/gateway-plugin/runtime/tests/generation/prepare/snapshot.rs b/backend/crates/gateway-plugin/runtime/tests/generation/prepare/snapshot.rs index c5041941e..61863a4f1 100644 --- a/backend/crates/gateway-plugin/runtime/tests/generation/prepare/snapshot.rs +++ b/backend/crates/gateway-plugin/runtime/tests/generation/prepare/snapshot.rs @@ -9,6 +9,44 @@ use gateway_plugin_sdk::Capability; use gateway_plugin_sdk::Contributions; use gateway_plugin_sdk::Stage; +#[tokio::test] +async fn initial_registration_http_uses_the_prepared_settings_without_a_publication() { + use wiremock::{Mock, MockServer, ResponseTemplate, matchers::method}; + for limit in [63, 64] { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200)) + .expect(u64::from(limit == 64)) + .mount(&server) + .await; + let (cache, store, runtime) = + super::super::setup_with_contributions(Contributions::from([ + crate::support::contribution( + Capability::ModelCatalog, + vec![Stage::Registration], + vec![], + vec![], + ), + ])) + .await; + store.snapshot.lock().unwrap().instances[0].configuration = serde_json::json!({ + "registration_http": {"params": {"method":"POST", "url":server.uri(), "headers":[]}, "body":"x".repeat(64)}, + "model_catalog":{"models":[{"id":"fixture-alias","provider":"openai","model":"fixture-upstream"}]} + }); + let prepared = ExtensionPreparationPort::prepare_with_request_body_limit( + &runtime, + ConfigRevision::new(1).unwrap(), + limit, + ) + .await + .unwrap(); + assert_eq!(prepared.model_aliases().len(), usize::from(limit == 64)); + drop(prepared); + runtime.shutdown().await; + super::super::wait_until_empty(cache.path()).await; + } +} + #[tokio::test] async fn model_catalog_is_frozen_without_entering_the_request_chain() { let (cache, store, runtime) = super::super::setup_with_contributions(Contributions::from([ diff --git a/backend/crates/gateway-plugin/runtime/tests/generation/upstream_adapter.rs b/backend/crates/gateway-plugin/runtime/tests/generation/upstream_adapter.rs index 8384da019..3c1ce5ba5 100644 --- a/backend/crates/gateway-plugin/runtime/tests/generation/upstream_adapter.rs +++ b/backend/crates/gateway-plugin/runtime/tests/generation/upstream_adapter.rs @@ -159,6 +159,16 @@ fn context_with_fast_mode( generation: &ExtensionSetReference, key: &str, mode: gateway_core::account::FastMode, +) -> AttemptContext { + context_with_body_limit(runtime, generation, key, mode, 64 * 1024 * 1024) +} + +fn context_with_body_limit( + runtime: &gateway_plugin_runtime::PluginRuntime, + generation: &ExtensionSetReference, + key: &str, + mode: gateway_core::account::FastMode, + bytes: usize, ) -> AttemptContext { AttemptContext::new( RequestAttemptContext::new( @@ -166,7 +176,8 @@ fn context_with_fast_mode( ClientApiKeyId::new(key).unwrap(), ) .with_upstream_adapters(runtime.execution_registry().upstream_adapters(generation)) - .with_fast_mode(mode), + .with_fast_mode(mode) + .with_request_body_limit(bytes), NonZeroU32::new(1).unwrap(), SystemTime::now() + Duration::from_secs(5), AccountSelectionPolicy::new( @@ -1225,3 +1236,120 @@ async fn malformed_adapter_events_fail_only_the_current_stream() { super::wait_until_empty(cache.path()).await; } } + +#[tokio::test] +async fn large_adapter_requests_use_the_frozen_budget_for_both_http_capabilities() { + for streaming in [false, true] { + for upstream in [false, true] { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_string(if streaming { + "data: {}\n\n" + } else { + "{}" + })) + .expect(if upstream { 2 } else { 4 }) + .mount(&server) + .await; + let mut config = configuration(&server.uri(), streaming); + if !upstream { + let mut inference = config["upstream_callbacks"][0].clone(); + inference["body"] = json!("{}"); + config["upstream_callbacks"] + .as_array_mut() + .unwrap() + .push(inference); + config["upstream_callbacks"][0]["method"] = json!(if streaming { + "host.http.do_stream" + } else { + "host.http.do" + }); + config["upstream_callbacks"][0]["params"] = json!({"method":"POST", "url":format!("{}/auxiliary", server.uri()), "headers":[]}); + } + let (cache, _, runtime) = setup(config).await; + let generation = + ExtensionPreparationPort::prepare(&runtime, ConfigRevision::new(1).unwrap()) + .await + .unwrap(); + let account = Account::new(); + let operation = Operation::Generate(GenerateRequest::from_protocol_payload( + ProtocolPayload::json_object( + "openai", + json!({"model":"native-model","input":"x".repeat(17 * 1024 * 1024)}) + .as_object() + .unwrap() + .clone(), + ) + .unwrap(), + )); + let expected = operation.middleware_body().unwrap(); + for accepted in [true, false, true] { + let context = context_with_body_limit( + &runtime, + &generation, + "key-one", + Default::default(), + expected.len() - usize::from(!accepted), + ); + let provider = ProviderKind::new("openai").unwrap(); + let model = UpstreamModelId::new("native-model").unwrap(); + let adapter = context + .upstream_adapter(&provider, &model) + .unwrap() + .unwrap(); + let metadata = ProviderCallMetadata::new( + provider, + model, + account.id.clone(), + UpstreamTransport::new(adapter.transport()).unwrap(), + ); + let mut stream = adapter.execute(UpstreamAdapterInvocation { + operation: operation.clone(), + headers: vec![], + context, + metadata, + account: account.clone(), + }); + let mut completed = false; + let mut failure = None; + while let Some(event) = stream.next().await { + match event { + Ok(event) => { + completed |= event + .canonical_facts() + .iter() + .any(|fact| matches!(fact, GatewayEvent::Completed(_))) + } + Err(error) => { + failure = Some(error); + break; + } + } + } + if accepted { + assert!( + failure.is_none(), + "streaming={streaming} upstream={upstream}: {failure:?}" + ); + assert!(completed); + } else { + assert_eq!(failure.unwrap().send_state(), UpstreamSendState::NotSent); + } + } + let requests = server + .received_requests() + .await + .unwrap() + .into_iter() + .filter(|request| request.body.len() > 1024 * 1024) + .collect::>(); + assert_eq!(requests.len(), 2); + for request in requests { + assert_eq!(request.body.as_slice(), expected.as_ref()); + } + drop(generation); + runtime.shutdown().await; + super::wait_until_empty(cache.path()).await; + } + } +} diff --git a/backend/crates/gateway-plugin/runtime/tests/support/environment.rs b/backend/crates/gateway-plugin/runtime/tests/support/environment.rs index 52f74e974..3147aa83d 100644 --- a/backend/crates/gateway-plugin/runtime/tests/support/environment.rs +++ b/backend/crates/gateway-plugin/runtime/tests/support/environment.rs @@ -474,6 +474,7 @@ impl Environment { Some(runtime.execution_registry()), Some(runtime.frontend_authentication_registry()), ); + runtime.bind_snapshots(&core.snapshots()).unwrap(); let access = gateway_admin::initialize_plugin_accounts( super::native::admin_registry(), self.store.admin_ports().accounts(), @@ -509,6 +510,7 @@ impl Environment { .activate() .await .unwrap(); + runtime.bind_snapshots(&core.snapshots()).unwrap(); runtime .bind_model_ports( &core.nested_model_execution_port(), diff --git a/backend/crates/gateway-plugin/runtime/tests/support/worker/mod.rs b/backend/crates/gateway-plugin/runtime/tests/support/worker/mod.rs index 38b8e562c..4a4245edf 100644 --- a/backend/crates/gateway-plugin/runtime/tests/support/worker/mod.rs +++ b/backend/crates/gateway-plugin/runtime/tests/support/worker/mod.rs @@ -284,15 +284,51 @@ impl Peer { let Some(fixture) = self.configuration.get(fixture_field) else { return Ok(()); }; + for fixture in fixture + .as_array() + .map_or(std::slice::from_ref(fixture), Vec::as_slice) + { + if let Some(gate) = fixture["wait_for_file"].as_str() { + while !std::path::Path::new(gate).exists() { + tokio::time::sleep(Duration::from_millis(5)).await; + } + } + self.execute_nested_model_fixture(parent, fixture, marker_field) + .await?; + } + Ok(()) + } + + async fn execute_nested_model_fixture( + &self, + parent: u64, + fixture: &Value, + marker_field: &str, + ) -> Result<(), PluginFault> { + let body = if let Some(path) = fixture["body_file"].as_str() { + std::fs::read(path).unwrap() + } else { + serde_json::to_vec(&fixture["body"]).unwrap() + }; if fixture["stream"] == true { - let (result, payload) = self + let response = self .callback_payload( parent, "host.model.execute_stream", fixture["request"].clone(), - serde_json::to_vec(&fixture["body"]).unwrap(), + body, ) - .await?; + .await; + let (result, payload) = match response { + Err(error) if fixture["expect_error"] == true => { + self.append_observation_marker( + marker_field, + &json!({"error":error.code,"http_status":error.http_status}), + ); + return Ok(()); + } + result => result?, + }; assert!(payload.is_empty()); let stream: gateway_plugin_sdk::call::host::ModelStreamResult = serde_json::from_value(result).unwrap(); @@ -348,7 +384,7 @@ impl Peer { parent, "host.model.execute", fixture["request"].clone(), - serde_json::to_vec(&fixture["body"]).unwrap(), + body, ) .await; match callback { @@ -366,7 +402,10 @@ impl Peer { Ok(()) } Err(error) if fixture["expect_error"] == true => { - self.append_observation_marker(marker_field, &json!({"error":error.code})); + self.append_observation_marker( + marker_field, + &json!({"error":error.code,"http_status":error.http_status}), + ); Ok(()) } Err(error) => { @@ -439,6 +478,37 @@ impl Peer { self.send(Message::End { id, error: None }, vec![]).await; return; } + if self.configuration["http_probe_methods"] + .as_array() + .is_some_and(|methods| { + methods + .iter() + .any(|probe| probe.as_str() == Some(method.as_str())) + }) + { + let body = vec![ + b'x'; + self.configuration["http_probe_body_bytes"] + .as_u64() + .unwrap() as usize + ]; + let response = self + .callback_payload( + id, + "host.http.do", + json!({"method":"POST", "url":self.configuration["http_probe_url"]}), + body, + ) + .await; + let status = match response { + Ok((result, _)) => result["status"].as_u64().unwrap(), + Err(error) => u64::from(error.http_status.unwrap()), + }; + self.append_observation_marker( + "http_probe_marker", + &json!({"method":method,"status":status}), + ); + } match method.as_str() { "upstream_adapter.register" => { self.send( @@ -1304,6 +1374,20 @@ impl Peer { return; } "plugin.register" => { + if let Some(request) = self.configuration.get("registration_http") { + let result = self + .callback_payload( + id, + "host.http.do", + request["params"].clone(), + request["body"].as_str().unwrap().as_bytes().to_vec(), + ) + .await; + if let Err(error) = result { + self.send(Message::Error { id, error }, vec![]).await; + return; + } + } if self.configuration["exit_during_registration"] == true { std::process::exit(31); } @@ -1598,13 +1682,19 @@ impl Peer { ) })) .await?; - for (result, payload) in results { + for (callback, (result, payload)) in callbacks.iter().zip(results) { if let Some(stream) = result["stream"].as_str() { + response_body.clear(); + let read_method = if callback["method"] == "host.http.do_stream" { + "host.http.stream_read" + } else { + "host.upstream.http.stream_read" + }; loop { let (result, chunk) = self .callback_payload( id, - "host.upstream.http.stream_read", + read_method, json!({"stream":stream,"maximum_bytes":65536}), vec![], ) diff --git a/backend/crates/gateway-plugin/sdk/docs/capabilities.md b/backend/crates/gateway-plugin/sdk/docs/capabilities.md index ac4aa612c..43f9aeeae 100644 --- a/backend/crates/gateway-plugin/sdk/docs/capabilities.md +++ b/backend/crates/gateway-plugin/sdk/docs/capabilities.md @@ -209,6 +209,17 @@ SDK 不自动重试。超时或断连不能证明写入未提交；重试上限 - `host.model.execute[_stream]`：交给 Core 的路由、准入、租约、重试、账本和计费链路执行 - `host.model.stream_read/close`：读取或关闭当前父调用创建的有界流 +模型请求正文使用独立的二进制载荷，不共用响应缓冲预算。宿主从有效设置 +`responses_max_decompressed_body_bytes` 取得原始正文预算 `L`，通过 +`call::model::request_payload_limit(L)` 为内部请求预留 JSON 字符串转义和元数据空间 +（`2 × L + 64 KiB`，受请求编码的 32 位分段长度约束）。超限返回带 HTTP 413 的 +`invalid_input`，不启动模型执行。上游适配器的请求 `GPAQ` 也使用请求预算，响应事件仍受 +原有 8 MiB 编码上限、Runtime 缓冲预算及流控窗口限制 + +HTTP 和 request 中间件中的模型回调沿用有效请求设置；attempt、路由、调度、重试和上游适配器 +沿用该次模型请求冻结的正文预算。显式 Key 模型调用的预检与执行使用同一绑定快照， +同一父调用内复用 Key 时不受后续配置发布影响；父调用结束后重新绑定才读取新配置，无需重新启用插件 + `ModelExecuteRequest.client_key_id` 可显式选择执行 Key，不与触发插件的父请求 Key 求交。省略时继承现有模型父请求身份；没有模型父请求的调用必须提供 Key。所选 Key 的启用状态、模型规则、账号组、并发和预算继续由 Core 执行 模型事件使用 `call::model::ExecutionEvent` 的 `GPE2` 封套，将 canonical 事实、宿主事实和原始 wire 分段编码。`host` 包含费用及定价明细、完整上游观测、会话状态和改写来源；当前 wire 与原始 wire 都保留二进制字节。读取或回传快照不触发第二次计费、续接提交或发送状态更新。 @@ -227,6 +238,13 @@ SDK 不自动重试。超时或断连不能证明写入未提交；重试上限 账号上游使用同一正文对象，使用方式见[受管上游](upstream-adapters.md) `host.http.do/do_stream` 和流读取／关闭方法提供受管 HTTP。正文使用独立二进制载荷，Host 统一执行代理、超时和背压，不按插件身份限制目标地址段。DNS 由宿主解析，连接保留目标 Host 与 TLS 域名校验；重定向不会自动跟随。 +受管 HTTP（包括 `host.upstream.http.do[_stream]`）的请求正文按有效设置 +`responses_max_decompressed_body_bytes` 检查。HTTP/request 中间件使用有效请求设置； +attempt、路由、调度、重试和上游适配器使用执行冻结预算，其余调用读取当前发布配置。 +超限返回带 HTTP 413 的 `invalid_input` 和 `not_sent`，不向目标发送请求。 +注册阶段沿用正在编译的配置预算，首次发布前也可发送非空正文。 +非空正文无法取得有效配置时拒绝调用；无正文查询不消耗正文预算。 +这里检查实际出站正文，内部模型请求的转义余量不增加出站预算。响应缓冲及流窗口不随请求配置扩大 普通调用使用父调用剩余期限；已建立的受管 HTTP 流或 WebSocket 会话按每次网络操作计时，仍受父资源取消约束 期限耗尽返回 `timeout`，解析、连接或响应读取失败返回 `upstream`。错误消息提供分类提示，响应正文仍由插件读取；协议故障不应直接展示内部消息 diff --git a/backend/crates/gateway-plugin/sdk/src/call/model/codec.rs b/backend/crates/gateway-plugin/sdk/src/call/model/codec.rs index 481ecc25a..4c1228d7f 100644 --- a/backend/crates/gateway-plugin/sdk/src/call/model/codec.rs +++ b/backend/crates/gateway-plugin/sdk/src/call/model/codec.rs @@ -45,29 +45,44 @@ pub(in crate::call) fn pack( prefix: [u8; 4], metadata: &T, payloads: [Vec; N], +) -> Result, ExecutionEncodingError> { + pack_with_limit(prefix, metadata, payloads, MAX_EXECUTION_PAYLOAD_BYTES) +} + +pub(in crate::call) fn pack_with_limit( + prefix: [u8; 4], + metadata: &T, + payloads: [Vec; N], + maximum: usize, ) -> Result, ExecutionEncodingError> { let header_bytes = 4 + 4 * (N + 1); let raw_bytes = payloads .iter() - .fold(0usize, |total, bytes| total.saturating_add(bytes.len())); - if raw_bytes > MAX_EXECUTION_PAYLOAD_BYTES { + .try_fold(0usize, |total, bytes| total.checked_add(bytes.len())) + .ok_or(ExecutionEncodingError)?; + if raw_bytes > maximum { return Err(ExecutionEncodingError); } let metadata = serde_json::to_vec(metadata).map_err(|_| ExecutionEncodingError)?; let total = header_bytes - .saturating_add(metadata.len()) - .saturating_add(raw_bytes); - if total > MAX_EXECUTION_PAYLOAD_BYTES { + .checked_add(metadata.len()) + .and_then(|total| total.checked_add(raw_bytes)) + .ok_or(ExecutionEncodingError)?; + let lengths = std::iter::once(metadata.len()) + .chain(payloads.iter().map(Vec::len)) + .map(u32::try_from) + .collect::, _>>() + .map_err(|_| ExecutionEncodingError)?; + if total > maximum { return Err(ExecutionEncodingError); } - let mut result = Vec::with_capacity(total); + let mut result = Vec::new(); + result + .try_reserve_exact(total) + .map_err(|_| ExecutionEncodingError)?; result.extend_from_slice(&prefix); - for length in std::iter::once(metadata.len()).chain(payloads.iter().map(Vec::len)) { - result.extend_from_slice( - &u32::try_from(length) - .map_err(|_| ExecutionEncodingError)? - .to_be_bytes(), - ); + for length in lengths { + result.extend_from_slice(&length.to_be_bytes()); } result.extend_from_slice(&metadata); for bytes in payloads { @@ -79,12 +94,17 @@ pub(in crate::call) fn pack( pub(in crate::call) fn unpack( prefix: [u8; 4], bytes: &[u8], +) -> Result<(T, [&[u8]; N]), ExecutionEncodingError> { + unpack_with_limit(prefix, bytes, MAX_EXECUTION_PAYLOAD_BYTES) +} + +pub(in crate::call) fn unpack_with_limit( + prefix: [u8; 4], + bytes: &[u8], + maximum: usize, ) -> Result<(T, [&[u8]; N]), ExecutionEncodingError> { let header_bytes = 4 + 4 * (N + 1); - if bytes.len() < header_bytes - || bytes.len() > MAX_EXECUTION_PAYLOAD_BYTES - || bytes[..4] != prefix - { + if bytes.len() < header_bytes || bytes.len() > maximum || bytes[..4] != prefix { return Err(ExecutionEncodingError); } let mut cursor = header_bytes; diff --git a/backend/crates/gateway-plugin/sdk/src/call/model/mod.rs b/backend/crates/gateway-plugin/sdk/src/call/model/mod.rs index 7171bb751..0013e6e97 100644 --- a/backend/crates/gateway-plugin/sdk/src/call/model/mod.rs +++ b/backend/crates/gateway-plugin/sdk/src/call/model/mod.rs @@ -6,6 +6,15 @@ pub(in crate::call) mod codec; pub mod facts; pub use codec::{ExecutionEncodingError, MAX_EXECUTION_PAYLOAD_BYTES}; +/// 内部请求允许 JSON 字符串转义与元数据开销，出站正文仍按原始配置上限检查 +#[must_use] +pub fn request_payload_limit(body_bytes: usize) -> usize { + body_bytes + .saturating_mul(2) + .saturating_add(64 * 1024) + .min(u32::MAX as usize) +} + #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum ContentKind { diff --git a/backend/crates/gateway-plugin/sdk/src/call/upstream_adapter/codec.rs b/backend/crates/gateway-plugin/sdk/src/call/upstream_adapter/codec.rs index 111eb0c77..973e5425b 100644 --- a/backend/crates/gateway-plugin/sdk/src/call/upstream_adapter/codec.rs +++ b/backend/crates/gateway-plugin/sdk/src/call/upstream_adapter/codec.rs @@ -5,7 +5,7 @@ use serde::{Deserialize, Serialize}; use super::{UpstreamAdapterEvent, UpstreamAdapterRequest, UpstreamContinuation, UpstreamFailure}; use crate::call::model::{ ExecutionEncodingError, ExecutionEvent, - codec::{pack, unpack}, + codec::{pack, pack_with_limit, unpack, unpack_with_limit}, }; #[derive(Serialize, Deserialize)] @@ -38,12 +38,13 @@ pub(super) fn encode_request( request: UpstreamAdapterRequest, body: Vec, ) -> Result, ExecutionEncodingError> { - pack(*b"GPAQ", &request, [body]) + // 请求预算由宿主配置检查，线协议只约束 u32 长度；响应事件仍受独立预算限制 + pack_with_limit(*b"GPAQ", &request, [body], usize::MAX) } pub(super) fn decode_request( bytes: &[u8], ) -> Result<(UpstreamAdapterRequest, Vec), ExecutionEncodingError> { - let (request, [body]) = unpack(*b"GPAQ", bytes)?; + let (request, [body]) = unpack_with_limit(*b"GPAQ", bytes, usize::MAX)?; Ok((request, body.to_vec())) } diff --git a/backend/crates/gateway-plugin/sdk/tests/call/mod.rs b/backend/crates/gateway-plugin/sdk/tests/call/mod.rs index 03301607e..67d11919a 100644 --- a/backend/crates/gateway-plugin/sdk/tests/call/mod.rs +++ b/backend/crates/gateway-plugin/sdk/tests/call/mod.rs @@ -8,3 +8,4 @@ mod management; mod middleware; mod observation; mod policy; +mod upstream_adapter; diff --git a/backend/crates/gateway-plugin/sdk/tests/call/upstream_adapter/codec.rs b/backend/crates/gateway-plugin/sdk/tests/call/upstream_adapter/codec.rs new file mode 100644 index 000000000..8e012a396 --- /dev/null +++ b/backend/crates/gateway-plugin/sdk/tests/call/upstream_adapter/codec.rs @@ -0,0 +1,46 @@ +//! 验证上游请求的大正文往返与响应事件预算保持独立 + +use gateway_plugin_sdk::call::{ + model::{ExecutionEvent, MAX_EXECUTION_PAYLOAD_BYTES, WireEvent, WirePayload}, + upstream_adapter::UpstreamAdapterRequest, +}; +use serde_json::json; + +fn request() -> UpstreamAdapterRequest { + serde_json::from_value(json!({ + "adapter_id":"example", "provider":"openai", "upstream_model":"fixture", + "client_key_id":"key", "account_id":"account", "credential_revision":1, + "protocol":"openai", "client_transport":"http_json", "fast_mode":"default", + "headers":[["x-fixture",[97,98]]], "continuation":null + })) + .unwrap() +} + +#[test] +fn large_request_preserves_bytes_and_rejects_malformed_lengths() { + let body = (0..MAX_EXECUTION_PAYLOAD_BYTES + 1) + .map(|i| (i % 251) as u8) + .collect::>(); + let encoded = request().encode(body.clone()).unwrap(); + let (decoded, actual) = UpstreamAdapterRequest::decode(&encoded).unwrap(); + assert_eq!(actual, body); + assert_eq!(decoded.headers, request().headers); + assert!(UpstreamAdapterRequest::decode(&encoded[..encoded.len() - 1]).is_err()); + let mut corrupted = encoded; + corrupted[8..12].copy_from_slice(&u32::MAX.to_be_bytes()); + assert!(UpstreamAdapterRequest::decode(&corrupted).is_err()); +} + +#[test] +fn large_requests_do_not_expand_response_event_budget() { + let event = ExecutionEvent { + wire: Some(WireEvent { + protocol: "openai".into(), + payload: WirePayload::RawBody { + body: vec![b'x'; MAX_EXECUTION_PAYLOAD_BYTES], + }, + }), + ..Default::default() + }; + assert!(event.encode().is_err()); +} diff --git a/backend/crates/gateway-plugin/sdk/tests/call/upstream_adapter/mod.rs b/backend/crates/gateway-plugin/sdk/tests/call/upstream_adapter/mod.rs new file mode 100644 index 000000000..d22e18864 --- /dev/null +++ b/backend/crates/gateway-plugin/sdk/tests/call/upstream_adapter/mod.rs @@ -0,0 +1,3 @@ +//! 上游适配器调用合同的测试入口 + +mod codec; diff --git a/docs/api.md b/docs/api.md index 5059f6e41..5a97ed017 100644 --- a/docs/api.md +++ b/docs/api.md @@ -265,7 +265,8 @@ Codex 的 review 等子代理请求仍使用 `/v1/responses`，并通过 `x-open 多帧连续解码，整体展开结果受运行设置 `responsesMaxDecompressedBodyBytes` 约束（默认 64 MiB）， 超限在继续展开前返回 `400 request_too_large`，错误信息包含当前请求的上限字节数。zstd 回溯窗口独立固定为最多 64 MiB，不能满足该限制的帧按解码失败处理。这个限制保护入站解压资源，不是 -模型上下文或 Token 上限；未压缩正文不受此长度限制。 +模型上下文或 Token 上限；未压缩正文不经过此解压检查。插件受管出站仍受同一设置约束， +见[请求解压上限](#请求解压上限)。 不支持的编码、逗号分隔的叠加编码和重复 `Content-Encoding` 头返回 `400 unsupported_content_encoding`；压缩正文损坏、截断或解压后不是合法 JSON 返回 `400 invalid_json`。本地错误不包含原始正文或解压库细节。WebSocket 文本帧不经过这条解压路径 @@ -1627,10 +1628,15 @@ Guardian 以 `subagent_kind` 或 `client_metadata.x-openai-subagent` 值 `guardi ### 请求解压上限 -`responsesMaxDecompressedBodyBytes` 是压缩 Responses HTTP 请求的解压输出上限，单位字节，默认 -67108864（64 MiB）。必须为正整数，且可表示为进程平台的 `isize`；管理端以整数 MiB 编辑。 -保存并发布成功后，新请求使用新值；已鉴权请求沿用原快照，无需重启。调高上限会增加大请求的内存占用， -它不代表整个进程的内存预算 +`responsesMaxDecompressedBodyBytes` 限制压缩 Responses HTTP 请求的解压输出，以及插件受管 HTTP +的实际出站正文（包括账号上游请求，无论是否压缩），单位字节，默认 67108864（64 MiB）。 +必须为正整数，且可表示为进程平台的 `isize`；管理端以整数 MiB 编辑。 +插件模型调用从同一设置派生内部请求编码预算，具体边界见 [SDK 模型请求](../backend/crates/gateway-plugin/sdk/docs/capabilities.md#key模型与模型调用) + +保存并发布成功后，新请求使用新值；在途模型请求及其路由、调度、attempt 和重试回调沿用冻结预算。 +显式 Key 的模型调用沿用当前父调用内绑定的快照，父调用结束后重新绑定时使用新值；无模型请求上下文的 +受管 HTTP 调用读取当前发布值，无需重启插件。调小该值可能使插件上传或转发返回 413，且不发送出站请求。 +调高上限会增加大请求的内存占用，不代表扩大模型 Token 上限、响应缓冲或整个进程的内存预算 ### 智能调度 diff --git a/docs/architecture.md b/docs/architecture.md index 79f54a966..e73769202 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -197,6 +197,10 @@ flowchart LR 子调用保留显式改写，在目标 Key 作用域重算默认值；Key 限额、Fast 和模型超时的覆盖绑定原 Key，同级调用互不改写。 原生 Responses WebSocket 每轮从当前快照重新认证，保留握手的显式改写；已经开始的执行不随全局发布改变。具体字段与来源合同见 [SDK 中间件](../backend/crates/gateway-plugin/sdk/docs/capabilities.md#http-入口) +- **正文预算随执行冻结**：Core 将有效正文预算传给路由、调度、重试、attempt 和上游适配器；显式 Key 的模型预检与执行共用绑定快照。 + Runtime 优先读取有效请求设置或执行预算，其余调用读取当前发布配置，注册准备使用编译中的预算。 + 组合根在激活 Core 前绑定弱快照读取句柄，不延长发布集合寿命；Host 仅检查调用方传入的实际出站正文预算 + #### 调用资源与状态 - **资源跟随调用**：账号、HTTP、模型、状态和日志回调绑定有效父调用，取消随调用传播。HTTP 与模型 request/attempt 中间件在首个流式响应头交付后，由正文 owner 管理取消与终态，不沿用初始处理期限；网络操作与模型执行各自计时。