From ac36f5bc1367e5e4b0fd464ce3f44b989dcd0635 Mon Sep 17 00:00:00 2001 From: Zyy <571092077@qq.com> Date: Thu, 15 Oct 2026 15:17:00 +0800 Subject: [PATCH] fix(plugins): separate execution capacity from downstream waits (#434) * fix(plugins): delegate HTTP requests on RPC admission exhaustion Co-authored-by: Codex * fix(plugins): align default RPC capacity at 64 Raise host and SDK call and callback limits together, exercise 64 in-flight HTTP calls, and document plugin rebuild requirements. Co-authored-by: Codex * fix(plugins): preserve middleware execution under RPC saturation Remove capacity-triggered delegation, retain the 64-call defaults, and verify that saturation never bypasses a healthy HTTP plugin. Co-authored-by: Codex * fix(plugins): separate execution capacity from downstream waits Bound initial admission and total in-flight resources while preserving middleware continuations. Keep queue waits independent of processing deadlines and report overload without bypassing plugins. Co-authored-by: Codex --------- Co-authored-by: Zyy Co-authored-by: Codex --- .../crates/gateway-api/src/middleware/mod.rs | 8 +- .../crates/gateway-api/src/openai/error.rs | 5 + .../gateway-api/tests/middleware/mod.rs | 29 ++ .../crates/gateway-api/tests/openai/error.rs | 12 + backend/crates/gateway-core/src/error/mod.rs | 3 + .../src/adapter/command_line/execution.rs | 2 +- .../runtime/src/adapter/middleware/request.rs | 1 + .../runtime/src/callback/error.rs | 12 +- .../src/generation/prepare/snapshot.rs | 5 +- .../runtime/src/rpc/admission.rs | 50 +++ .../runtime/src/rpc/dispatch.rs | 8 +- .../gateway-plugin/runtime/src/rpc/mod.rs | 1 + .../gateway-plugin/runtime/src/rpc/session.rs | 144 +++++--- .../runtime/src/stream/bridge.rs | 2 +- .../runtime/tests/callback/log.rs | 11 +- .../tests/generation/policy/http/capacity.rs | 162 +++++++++ .../tests/generation/policy/http/mod.rs | 21 +- .../tests/generation/policy/lifecycle.rs | 76 ++++ .../runtime/tests/rpc/admission.rs | 328 ++++++++++++++++++ .../gateway-plugin/runtime/tests/rpc/mod.rs | 1 + .../runtime/tests/rpc/session.rs | 8 +- backend/crates/gateway-plugin/sdk/README.md | 6 + .../gateway-plugin/sdk/docs/capabilities.md | 4 +- .../sdk/src/client/session/config.rs | 4 +- .../sdk/src/client/session/driver.rs | 6 +- .../sdk/src/client/session/registry.rs | 24 +- .../sdk/tests/client/session/mod.rs | 43 ++- .../sdk/tests/client/session/registry.rs | 54 +++ docs/architecture.md | 4 +- docs/plugins.md | 8 + 30 files changed, 959 insertions(+), 83 deletions(-) create mode 100644 backend/crates/gateway-plugin/runtime/src/rpc/admission.rs create mode 100644 backend/crates/gateway-plugin/runtime/tests/generation/policy/http/capacity.rs create mode 100644 backend/crates/gateway-plugin/runtime/tests/rpc/admission.rs diff --git a/backend/crates/gateway-api/src/middleware/mod.rs b/backend/crates/gateway-api/src/middleware/mod.rs index aa99343b6..f3e0968f2 100644 --- a/backend/crates/gateway-api/src/middleware/mod.rs +++ b/backend/crates/gateway-api/src/middleware/mod.rs @@ -17,9 +17,10 @@ use axum::{ http::{HeaderName, HeaderValue, Request, StatusCode}, response::{IntoResponse as _, Response}, }; -use gateway_core::engine::middleware::FrozenMiddlewarePlan; use gateway_core::engine::middleware::http as contract; use gateway_core::engine::middleware::http::Settings; +use gateway_core::engine::middleware::{FrozenMiddlewarePlan, MiddlewareError}; +use gateway_core::error::GatewayErrorKind; use gateway_core::lifecycle::CancellationToken; use gateway_core::middleware::compose; use http_body::{Body as HttpBody, Frame, SizeHint}; @@ -270,6 +271,11 @@ async fn handle( } response.map(|body| Body::new(ResponseBody::new(body, guard))) } + Err(MiddlewareError::Gateway(error)) + if error.kind() == GatewayErrorKind::MiddlewareCapacityUnavailable => + { + crate::openai::error::gateway_error_response(&error) + } Err(error) => { let status = if error.is_rejected() { StatusCode::FORBIDDEN diff --git a/backend/crates/gateway-api/src/openai/error.rs b/backend/crates/gateway-api/src/openai/error.rs index ff28b3443..2c8313b6a 100644 --- a/backend/crates/gateway-api/src/openai/error.rs +++ b/backend/crates/gateway-api/src/openai/error.rs @@ -279,6 +279,11 @@ pub const fn gateway_error_contract( "server_error", "provider_infrastructure_unavailable", ), + GatewayErrorKind::MiddlewareCapacityUnavailable => ( + StatusCode::SERVICE_UNAVAILABLE, + "server_error", + "middleware_capacity_unavailable", + ), GatewayErrorKind::ConcurrencyQueueFull => ( StatusCode::TOO_MANY_REQUESTS, "rate_limit_error", diff --git a/backend/crates/gateway-api/tests/middleware/mod.rs b/backend/crates/gateway-api/tests/middleware/mod.rs index d5b318f66..fbd025aa5 100644 --- a/backend/crates/gateway-api/tests/middleware/mod.rs +++ b/backend/crates/gateway-api/tests/middleware/mod.rs @@ -35,6 +35,7 @@ use tower::ServiceExt as _; #[derive(Clone, Copy, Debug)] enum Mode { + Overloaded, Rewrite, Upgrade { reject: bool, @@ -142,6 +143,12 @@ impl MiddlewarePlan for Plan { let mode = self.mode; Box::pin(async move { match mode { + Mode::Overloaded => Err(MiddlewareError::Gateway( + gateway_core::error::GatewayError::new( + gateway_core::error::GatewayErrorKind::MiddlewareCapacityUnavailable, + "Request middleware is temporarily busy", + ), + )), Mode::Service => unreachable!("service-only plan bypasses HTTP layers"), Mode::Upgrade { reject, idle, .. } => { let upgrade = request @@ -659,3 +666,25 @@ async fn plugin_http_dispatch_uses_host_identity_without_client_credentials() { .unwrap(); assert_eq!(response.status(), StatusCode::UNAUTHORIZED); } + +#[tokio::test] +async fn http_middleware_overload_is_an_explicit_503_before_native_routing() { + let (router, plan) = router(Mode::Overloaded, None).await; + for path in [ + "/healthz", + "/assets/index.js", + "/api/settings", + "/v1/responses", + ] { + let response = router + .clone() + .oneshot(Request::builder().uri(path).body(Body::empty()).unwrap()) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + let body = response.into_body().collect().await.unwrap().to_bytes(); + let body: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(body["error"]["code"], "middleware_capacity_unavailable"); + } + assert_eq!(plan.calls.load(Ordering::Acquire), 4); +} diff --git a/backend/crates/gateway-api/tests/openai/error.rs b/backend/crates/gateway-api/tests/openai/error.rs index d6288b00c..65cb183f2 100644 --- a/backend/crates/gateway-api/tests/openai/error.rs +++ b/backend/crates/gateway-api/tests/openai/error.rs @@ -263,6 +263,18 @@ fn rate_limited_error_should_map_to_openai_retryable_status() { ); } +#[test] +fn middleware_capacity_is_service_unavailable() { + assert_eq!( + gateway_error_contract(GatewayErrorKind::MiddlewareCapacityUnavailable), + ( + StatusCode::SERVICE_UNAVAILABLE, + "server_error", + "middleware_capacity_unavailable" + ), + ); +} + #[test] fn queue_rejections_keep_distinct_retryable_http_codes() { for (kind, code) in [ diff --git a/backend/crates/gateway-core/src/error/mod.rs b/backend/crates/gateway-core/src/error/mod.rs index d4fd5aae2..c5595a051 100644 --- a/backend/crates/gateway-core/src/error/mod.rs +++ b/backend/crates/gateway-core/src/error/mod.rs @@ -1067,6 +1067,8 @@ pub enum GatewayErrorKind { NoAvailableProvider, /// 符合条件的上游账号暂时没有调度容量 AccountCapacityUnavailable, + /// 中间件在途资源或执行等待队列已满 + MiddlewareCapacityUnavailable, ConcurrencyQueueFull, ConcurrencyQueueTimeout, /// Provider 的本地账号基础设施不可用 @@ -1098,6 +1100,7 @@ impl GatewayErrorKind { Self::ModelNotFound => "model_not_found", Self::NoAvailableProvider => "no_available_provider", Self::AccountCapacityUnavailable => "account_capacity_unavailable", + Self::MiddlewareCapacityUnavailable => "middleware_capacity_unavailable", Self::ConcurrencyQueueFull => "concurrency_queue_full", Self::ConcurrencyQueueTimeout => "concurrency_queue_timeout", Self::ProviderInfrastructureUnavailable => "provider_infrastructure_unavailable", diff --git a/backend/crates/gateway-plugin/runtime/src/adapter/command_line/execution.rs b/backend/crates/gateway-plugin/runtime/src/adapter/command_line/execution.rs index fb224f993..8c6151e9c 100644 --- a/backend/crates/gateway-plugin/runtime/src/adapter/command_line/execution.rs +++ b/backend/crates/gateway-plugin/runtime/src/adapter/command_line/execution.rs @@ -139,7 +139,7 @@ fn rpc_failure(error: RpcError) -> &'static str { RpcError::Handshake(_) | RpcError::Protocol(_) | RpcError::InvalidResponse(..) => { "插件协议错误" } - RpcError::Capacity => "插件调用容量不足", + RpcError::Capacity | RpcError::Overloaded => "插件调用容量不足", RpcError::Context(_) => "调用上下文无效", RpcError::Remote(_) => "插件拒绝命令", } 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 4da90ed99..5c7ee4f5d 100644 --- a/backend/crates/gateway-plugin/runtime/src/adapter/middleware/request.rs +++ b/backend/crates/gateway-plugin/runtime/src/adapter/middleware/request.rs @@ -221,6 +221,7 @@ async fn recover_rpc( RpcError::Remote(fault) if fault.code == ErrorCode::Rejected ); let delegate_safe = !rejected + && !error.is_overloaded() && !matches!( &error, RpcError::Remote(fault) if fault.send_state != SendState::NotSent diff --git a/backend/crates/gateway-plugin/runtime/src/callback/error.rs b/backend/crates/gateway-plugin/runtime/src/callback/error.rs index b3371e3f3..047b72733 100644 --- a/backend/crates/gateway-plugin/runtime/src/callback/error.rs +++ b/backend/crates/gateway-plugin/runtime/src/callback/error.rs @@ -20,6 +20,7 @@ pub(super) fn gateway(error: GatewayError) -> PluginFault { GatewayErrorKind::Unauthorized => ErrorCode::PermissionDenied, GatewayErrorKind::PolicyDenied => ErrorCode::Rejected, GatewayErrorKind::AccountCapacityUnavailable + | GatewayErrorKind::MiddlewareCapacityUnavailable | GatewayErrorKind::ConcurrencyQueueFull | GatewayErrorKind::ConcurrencyQueueTimeout | GatewayErrorKind::RateLimited @@ -142,13 +143,22 @@ pub(crate) fn rpc(error: &crate::RpcError) -> PluginFault { RpcError::Remote(fault) => return fault.clone(), RpcError::Timeout => ErrorCode::Timeout, RpcError::Cancelled => ErrorCode::Cancelled, - RpcError::Capacity => ErrorCode::Capacity, + RpcError::Capacity | RpcError::Overloaded => ErrorCode::Capacity, _ => ErrorCode::Fault, }; PluginFault::new(code, error.to_string()) } pub(crate) fn rpc_middleware(error: crate::RpcError) -> MiddlewareError { + if error.is_overloaded() { + return MiddlewareError::Gateway( + GatewayError::new( + GatewayErrorKind::MiddlewareCapacityUnavailable, + "Request middleware is temporarily busy", + ) + .with_source(gateway_core::error::ErrorSource::new(error)), + ); + } let rejected = matches!(&error, crate::RpcError::Remote(fault) if fault.code == ErrorCode::Rejected); MiddlewareError::Remote { 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 ecc02445a..d233a68e3 100644 --- a/backend/crates/gateway-plugin/runtime/src/generation/prepare/snapshot.rs +++ b/backend/crates/gateway-plugin/runtime/src/generation/prepare/snapshot.rs @@ -194,7 +194,10 @@ impl PluginRuntime { .map_err(|_| AdminError::internal("插件集合 ID 无效"))?; let observers = crate::adapter::observer::PluginObserverPlan::compile( observer_entries, - self.config.rpc_limits.maximum_calls, + self.config + .rpc_limits + .maximum_calls + .min(self.config.rpc_limits.maximum_executing_calls), self.config.rpc_limits.maximum_call_timeout, self.config.rpc_limits.maximum_buffered_body_bytes, ) diff --git a/backend/crates/gateway-plugin/runtime/src/rpc/admission.rs b/backend/crates/gateway-plugin/runtime/src/rpc/admission.rs new file mode 100644 index 000000000..209aad4a8 --- /dev/null +++ b/backend/crates/gateway-plugin/runtime/src/rpc/admission.rs @@ -0,0 +1,50 @@ +//! 区分插件执行名额与初次准入队列，恢复中的调用沿用在途资源预算 + +use std::{sync::Arc, time::Duration}; + +use tokio::sync::{OwnedSemaphorePermit, Semaphore}; + +use super::RpcError; + +pub(super) struct ExecutionCapacity { + active: Arc, + queued: Arc, + wait_timeout: Duration, +} + +impl ExecutionCapacity { + pub(super) fn new(active: usize, queued: usize, wait_timeout: Duration) -> Self { + Self { + active: Arc::new(Semaphore::new(active)), + queued: Arc::new(Semaphore::new(queued)), + wait_timeout, + } + } + + pub(super) async fn enter(&self) -> Result { + if let Ok(permit) = self.active.clone().try_acquire_owned() { + return Ok(permit); + } + let _queued = self.queued.try_acquire().map_err(|error| { + if matches!(error, tokio::sync::TryAcquireError::Closed) { + RpcError::Closed(None) + } else { + RpcError::Overloaded + } + })?; + self.resume().await + } + + /// 恢复的等待者已持有在途名额，不再竞争初次准入队列 + pub(super) async fn resume(&self) -> Result { + tokio::time::timeout(self.wait_timeout, self.active.clone().acquire_owned()) + .await + .map_err(|_| RpcError::Overloaded)? + .map_err(|_| RpcError::Closed(None)) + } + + pub(super) fn close(&self) { + self.active.close(); + self.queued.close(); + } +} diff --git a/backend/crates/gateway-plugin/runtime/src/rpc/dispatch.rs b/backend/crates/gateway-plugin/runtime/src/rpc/dispatch.rs index a26f3fcab..9fbfe3aa5 100644 --- a/backend/crates/gateway-plugin/runtime/src/rpc/dispatch.rs +++ b/backend/crates/gateway-plugin/runtime/src/rpc/dispatch.rs @@ -125,7 +125,7 @@ async fn read_loop( Err(RpcError::Protocol(None)) } else if let Some(context) = shared.context(parent_id) { last_callback = id; - if let Some(permit) = shared.try_callback_slot() { + if let Some(permit) = shared.try_callback_slot(&method) { let handler = Arc::clone(&callbacks); let response = Arc::clone(&shared); let (ready, started) = oneshot::channel(); @@ -138,6 +138,12 @@ async fn read_loop( return; } let result = handler.call(context, method, params, frame.payload).await; + if pause.is_some() + && let Err(error) = response.resume_next(parent_id).await + { + response.cancel(parent_id, error); + return; + } drop(pause); let mut reply = match result { Ok(reply) => Frame { diff --git a/backend/crates/gateway-plugin/runtime/src/rpc/mod.rs b/backend/crates/gateway-plugin/runtime/src/rpc/mod.rs index 11f48e43d..85acef174 100644 --- a/backend/crates/gateway-plugin/runtime/src/rpc/mod.rs +++ b/backend/crates/gateway-plugin/runtime/src/rpc/mod.rs @@ -1,5 +1,6 @@ //! 插件双向 RPC 会话、调用结果与生命周期的公共入口 +mod admission; mod dispatch; mod session; diff --git a/backend/crates/gateway-plugin/runtime/src/rpc/session.rs b/backend/crates/gateway-plugin/runtime/src/rpc/session.rs index 59993b85d..b7785e77d 100644 --- a/backend/crates/gateway-plugin/runtime/src/rpc/session.rs +++ b/backend/crates/gateway-plugin/runtime/src/rpc/session.rs @@ -15,17 +15,21 @@ use gateway_plugin_sdk::{ CallContext, Frame, Handshake, Message, PROTOCOL_VERSION, PluginFault, Stage, client::{CallDeadline, CallDeadlinePause, read_frame, validate_frame, write_frame}, }; -use tokio::sync::{Mutex as AsyncMutex, Semaphore, mpsc, oneshot, watch}; +use tokio::sync::{Mutex as AsyncMutex, OwnedSemaphorePermit, Semaphore, mpsc, oneshot, watch}; use crate::{PreparedPackage, RpcStream, stream::StreamIngress}; -use super::dispatch; +use super::{admission::ExecutionCapacity, dispatch}; #[derive(Debug, Clone, Copy)] pub struct RpcLimits { /// 受管回调与观察队列的缓冲预算，不是请求正文或 IPC 消息总量上限 pub maximum_buffered_body_bytes: usize, + /// 包含初次排队、下游等待、取消确认和未消费流的总在途上限 pub maximum_calls: usize, + pub maximum_executing_calls: usize, + pub maximum_queued_calls: usize, + pub maximum_queue_timeout: Duration, pub maximum_callbacks: usize, pub handshake_timeout: Duration, pub maximum_call_timeout: Duration, @@ -36,8 +40,11 @@ impl Default for RpcLimits { fn default() -> Self { Self { maximum_buffered_body_bytes: 1024 * 1024, - maximum_calls: 16, - maximum_callbacks: 16, + maximum_calls: 256, + maximum_executing_calls: 64, + maximum_queued_calls: 64, + maximum_queue_timeout: Duration::from_secs(30), + maximum_callbacks: 64, handshake_timeout: Duration::from_secs(5), maximum_call_timeout: Duration::from_secs(120), maximum_stderr_bytes: 64 * 1024, @@ -63,6 +70,8 @@ pub enum RpcError { Cancelled, #[error("plugin call capacity is exhausted")] Capacity, + #[error("plugin admission capacity is exhausted")] + Overloaded, #[error("plugin call context is invalid")] Context(#[source] Option), #[error("plugin returned an error")] @@ -70,6 +79,11 @@ pub enum RpcError { } impl RpcError { + pub(crate) fn is_overloaded(&self) -> bool { + matches!(self, Self::Overloaded) + || matches!(self, Self::Remote(fault) if fault.code == gateway_plugin_sdk::ErrorCode::Capacity) + } + /// 会话状态与请求错误共用安全分类，不展开插件返回正文 pub(crate) fn diagnostic(&self) -> (&'static str, &'static str) { match self { @@ -101,6 +115,7 @@ impl RpcError { RpcError::Timeout => ("timeout", "插件调用超时"), RpcError::Cancelled => ("cancelled", "插件会话已取消"), RpcError::Capacity => ("capacity", "插件调用容量已耗尽"), + RpcError::Overloaded => ("overloaded", "插件暂时繁忙"), RpcError::Context(_) => ("context", "插件调用上下文无效"), RpcError::Remote(_) => ("remote", "插件返回错误"), } @@ -216,6 +231,8 @@ pub(crate) struct Shared { pub process: ProcessControl, pub stopped: watch::Sender, callback_slots: Arc, + next_slots: Arc, + execution: ExecutionCapacity, } struct State { @@ -224,6 +241,8 @@ struct State { } struct PendingCall { + _slot: Arc, + execution: Option, deadline: CallDeadline, handler: Arc, transmitted: bool, @@ -274,6 +293,7 @@ impl Shared { .flatten(); (std::mem::take(&mut state.calls), observer) }; + self.execution.close(); for call in calls.into_values() { call.finish(Err(error.clone())); } @@ -303,6 +323,7 @@ impl Shared { call.deadline.clear(); } let initial = call.result.take().ok_or(RpcError::Protocol(None))?; + call.execution.take(); call.handler.streaming(&call.context); let _ = initial.send(result); } else { @@ -384,11 +405,40 @@ impl Shared { } pub fn pause_next(&self, id: u64) -> Option { - self.state() + let mut state = self.state(); + let call = state .calls - .get(&id) + .get_mut(&id) + .filter(|call| call.cancellation.is_none())?; + let pause = call.deadline.pause()?; + call.execution.take(); + Some(pause) + } + + pub async fn resume_next(&self, id: u64) -> Result<(), RpcError> { + { + let state = self.state(); + let call = state + .calls + .get(&id) + .filter(|call| call.cancellation.is_none()) + .ok_or(RpcError::Cancelled)?; + // 已交付响应头的流只占在途资源，不重新占用首结果执行名额 + if call.result.is_none() { + return Ok(()); + } + } + let permit = self.execution.resume().await?; + let mut state = self.state(); + let call = state + .calls + .get_mut(&id) .filter(|call| call.cancellation.is_none()) - .and_then(|call| call.deadline.pause()) + .ok_or(RpcError::Cancelled)?; + if call.result.is_some() { + call.execution = Some(permit); + } + Ok(()) } pub fn is_cancelling(&self, id: u64) -> bool { @@ -495,21 +545,23 @@ impl Shared { } } - pub(super) fn try_callback_slot(&self) -> Option { - self.callback_slots.clone().try_acquire_owned().ok() + pub(super) fn try_callback_slot(&self, method: &str) -> Option { + let slots = if method == gateway_plugin_sdk::call::middleware::NEXT_METHOD { + &self.next_slots + } else { + &self.callback_slots + }; + slots.clone().try_acquire_owned().ok() } - async fn wait_for_callbacks(&self, maximum_callbacks: usize) { - let Ok(maximum_callbacks) = u32::try_from(maximum_callbacks) else { - return; - }; - if let Ok(permits) = self - .callback_slots - .clone() - .acquire_many_owned(maximum_callbacks) - .await - { - drop(permits); + async fn wait_for_callbacks(&self, limits: RpcLimits) { + for (slots, maximum) in [ + (&self.callback_slots, limits.maximum_callbacks), + (&self.next_slots, limits.maximum_calls), + ] { + if let Ok(permits) = slots.clone().acquire_many_owned(maximum as u32).await { + drop(permits); + } } } } @@ -558,6 +610,10 @@ impl RpcSession { ) -> Result { if limits.maximum_calls == 0 || limits.maximum_calls > 256 + || limits.maximum_executing_calls == 0 + || limits.maximum_executing_calls > 256 + || limits.maximum_queued_calls > 256 + || limits.maximum_queue_timeout.is_zero() || limits.maximum_callbacks == 0 || limits.maximum_callbacks > 256 || limits.maximum_buffered_body_bytes < 1024 @@ -657,7 +713,7 @@ impl RpcSession { } let (data, data_receiver) = mpsc::channel(limits.maximum_calls); let (control, control_receiver) = - mpsc::channel(limits.maximum_callbacks + limits.maximum_calls + 8); + mpsc::channel(limits.maximum_callbacks + 2 * limits.maximum_calls + 8); let (stopped, _) = watch::channel(false); let callback_slots = Arc::new(Semaphore::new(limits.maximum_callbacks)); let shared = Arc::new(Shared { @@ -674,6 +730,12 @@ impl RpcSession { process, stopped, callback_slots, + next_slots: Arc::new(Semaphore::new(limits.maximum_calls)), + execution: ExecutionCapacity::new( + limits.maximum_executing_calls, + limits.maximum_queued_calls, + limits.maximum_queue_timeout, + ), }); dispatch::start( input, @@ -754,11 +816,7 @@ impl RpcSession { } } - fn call_deadline( - &self, - method: &str, - context: &CallContext, - ) -> Result { + fn call_timeout(&self, method: &str, context: &CallContext) -> Result { if context.instance_id != self.handshake.instance_id || context.generation != self.handshake.generation || context.incarnation != self.handshake.incarnation @@ -769,7 +827,7 @@ impl RpcSession { { return Err(RpcError::Context(None)); } - Ok(tokio::time::Instant::now() + Duration::from_millis(context.timeout_ms)) + Ok(Duration::from_millis(context.timeout_ms)) } fn begin_call( @@ -777,6 +835,8 @@ impl RpcSession { context: CallContext, deadline: tokio::time::Instant, stream: Option, + slot: Arc, + execution: OwnedSemaphorePermit, ) -> Result { let id = context.call_id; let (result, received) = oneshot::channel(); @@ -784,14 +844,13 @@ impl RpcSession { if let Some(error) = &state.failure { return Err(error.clone()); } - if state.calls.len() >= self.limits.maximum_calls { - return Err(RpcError::Capacity); - } let deadline = CallDeadline::new(deadline); self.callbacks.begin(&context, &deadline); state.calls.insert( id, PendingCall { + _slot: slot, + execution: Some(execution), deadline: deadline.clone(), handler: self.callbacks.clone(), transmitted: false, @@ -818,7 +877,16 @@ impl RpcSession { stream: Option, credit: Option<(u32, u32)>, ) -> Result { - let deadline = self.call_deadline(method, &context)?; + let timeout = self.call_timeout(method, &context)?; + let slot = Arc::new(self.slots.clone().try_acquire_owned().map_err(|error| { + if matches!(error, tokio::sync::TryAcquireError::Closed) { + RpcError::Closed(None) + } else { + RpcError::Overloaded + } + })?); + let execution = self.shared.execution.enter().await?; + let deadline = tokio::time::Instant::now() + timeout; // FIFO 锁同时拥有 ID 与 data 入队顺序；Call/初始 Credit 入队后必须立即释放， // 不能把插件回复或重入回调纳入串行区，也不能为等待锁重置调用期限 let mut next_call = tokio::time::timeout_at(deadline, self.next_call.lock()) @@ -841,7 +909,7 @@ impl RpcSession { validate_frame(&frame).map_err(|source| { RpcError::Context(Some(gateway_core::error::ErrorSource::new(source))) })?; - let started = self.begin_call(context, deadline, stream)?; + let started = self.begin_call(context, deadline, stream, slot.clone(), execution)?; let mut guard = CallGuard { shared: Arc::clone(&self.shared), id: started.id, @@ -906,6 +974,7 @@ impl RpcSession { } }; Ok(CompletedCall { + slot, id: started.id, deadline: started.deadline, reply, @@ -919,7 +988,6 @@ impl RpcSession { params: serde_json::Value, payload: Vec, ) -> Result { - let _slot = self.slots.try_acquire().map_err(|_| RpcError::Capacity)?; self.send_call(method, context, params, payload, None, None) .await .map(|completed| completed.reply) @@ -932,9 +1000,6 @@ impl RpcSession { params: serde_json::Value, payload: Vec, ) -> Result { - let slot = Arc::clone(&self.slots) - .try_acquire_owned() - .map_err(|_| RpcError::Capacity)?; let bytes = self.limits.maximum_buffered_body_bytes.min(256 * 1024) as u32; let frames = 32; let (stream, chunks, terminal) = StreamIngress::new(bytes, frames); @@ -955,7 +1020,7 @@ impl RpcSession { deadline: completed.deadline, id: completed.id, shared: Arc::clone(&self.shared), - _slot: slot, + _slot: completed.slot, _session: Arc::clone(self), }) } @@ -988,9 +1053,7 @@ impl RpcSession { self.shared.process.exited().await; // 不等待监控任务调度；先撤销父调用并中止其回调，再确认回调 future 已析构 self.shared.fail(RpcError::Closed(None)); - self.shared - .wait_for_callbacks(self.limits.maximum_callbacks) - .await; + self.shared.wait_for_callbacks(self.limits).await; } } @@ -1022,6 +1085,7 @@ struct StartedCall { } struct CompletedCall { + slot: Arc, id: u64, deadline: CallDeadline, reply: RpcReply, diff --git a/backend/crates/gateway-plugin/runtime/src/stream/bridge.rs b/backend/crates/gateway-plugin/runtime/src/stream/bridge.rs index 2ce6a6a79..98df3029d 100644 --- a/backend/crates/gateway-plugin/runtime/src/stream/bridge.rs +++ b/backend/crates/gateway-plugin/runtime/src/stream/bridge.rs @@ -62,7 +62,7 @@ pub struct RpcStream { pub(crate) deadline: CallDeadline, pub(crate) id: u64, pub(crate) shared: Arc, - pub(crate) _slot: OwnedSemaphorePermit, + pub(crate) _slot: Arc, pub(crate) _session: Arc, } diff --git a/backend/crates/gateway-plugin/runtime/tests/callback/log.rs b/backend/crates/gateway-plugin/runtime/tests/callback/log.rs index b473f6e94..6f4073907 100644 --- a/backend/crates/gateway-plugin/runtime/tests/callback/log.rs +++ b/backend/crates/gateway-plugin/runtime/tests/callback/log.rs @@ -99,6 +99,8 @@ impl tracing::Subscriber for Capture { #[tokio::test] async fn real_worker_logs_are_correlated_sanitized_validated_and_bounded() { + // 低于每秒日志上限，单独验证阻塞写入的槽位及跨代保留行为 + const MAXIMUM_CALLBACKS: usize = 16; let capture = Capture::default(); tracing::subscriber::set_global_default(capture.clone()).unwrap(); let directory = tempfile::tempdir().unwrap(); @@ -155,7 +157,10 @@ async fn real_worker_logs_are_correlated_sanitized_validated_and_bounded() { cache_directory: directory.path().join("cache"), host_version: "1.0.0".parse().unwrap(), package_limits: PackageLimits::default(), - rpc_limits: RpcLimits::default(), + rpc_limits: RpcLimits { + maximum_callbacks: MAXIMUM_CALLBACKS, + ..RpcLimits::default() + }, restart_circuit: Default::default(), }, Arc::new(gateway_host::outbound::HttpClient::new().unwrap()), @@ -270,7 +275,7 @@ async fn real_worker_logs_are_correlated_sanitized_validated_and_bounded() { blocked.config_revision = Revision::new(9).unwrap(); blocked.instances[0].id = "log-blocked".into(); blocked.instances[0].configuration = json!({ - "log_method":"plugin.register", "log_entries":[{"params":{"event":"fixture.blocked"},"repeat":32}], + "log_method":"plugin.register", "log_entries":[{"params":{"event":"fixture.blocked"},"repeat":MAXIMUM_CALLBACKS * 2}], "log_marker":blocked_marker, }); // 日志后端停住时，注册的业务结果和父调用结束仍须及时完成 @@ -288,7 +293,7 @@ async fn real_worker_logs_are_correlated_sanitized_validated_and_bounded() { .iter() .filter(|result| result["recorded"] == true) .count(), - RpcLimits::default().maximum_callbacks + MAXIMUM_CALLBACKS ); assert!( blocked_results diff --git a/backend/crates/gateway-plugin/runtime/tests/generation/policy/http/capacity.rs b/backend/crates/gateway-plugin/runtime/tests/generation/policy/http/capacity.rs new file mode 100644 index 000000000..49ee4ebee --- /dev/null +++ b/backend/crates/gateway-plugin/runtime/tests/generation/policy/http/capacity.rs @@ -0,0 +1,162 @@ +//! 验证 next 等待不占执行名额，硬上限拒绝请求且不跳过插件 + +use super::*; +use gateway_core::error::GatewayErrorKind; + +fn context() -> core::Context { + core::Context { + plugin_instance_id: None, + call_id: uuid::Uuid::new_v4().to_string(), + parent_call_id: None, + extensions: Default::default(), + plan: None, + request_id: uuid::Uuid::new_v4().to_string(), + cancellation: CancellationToken::new(), + } +} + +fn plugin_request() -> core::Request { + ::http::Request::builder() + .uri("/v1/responses") + .header("authorization", "Bearer fixture-secret") + .body(core::empty_body()) + .unwrap() +} + +#[tokio::test] +async fn http_waiting_next_releases_execution_for_delegate_plugin() { + assert_capacity(PluginFailurePolicy::Delegate, RpcLimits::default(), false).await; +} + +#[tokio::test] +async fn http_waiting_next_releases_execution_for_reject_plugin() { + assert_capacity(PluginFailurePolicy::Reject, RpcLimits::default(), false).await; +} + +#[tokio::test] +async fn http_in_flight_limit_never_delegates_and_recovers_after_cancellation() { + for policy in [PluginFailurePolicy::Delegate, PluginFailurePolicy::Reject] { + assert_capacity( + policy, + RpcLimits { + maximum_calls: 64, + ..Default::default() + }, + true, + ) + .await; + } +} + +async fn assert_capacity(policy: PluginFailurePolicy, limits: RpcLimits, saturated: bool) { + const CONCURRENT_CALLS: usize = 64; + let (cache, runtime, generation, plan) = + http_plugin_with_policy("passthrough", limits, policy).await; + let release = Arc::new(tokio::sync::Semaphore::new(0)); + let (started, mut entered) = tokio::sync::mpsc::unbounded_channel(); + let mut slow_calls = tokio::task::JoinSet::new(); + for index in 0..CONCURRENT_CALLS { + let plan = plan.clone(); + let release = release.clone(); + let started = started.clone(); + slow_calls.spawn(async move { + plan.handle_http( + context(), + plugin_request(), + compose(Vec::new(), move |request: core::Request| { + assert_eq!(request.uri(), "/rewritten?encoded=%2F"); + let release = release.clone(); + Box::pin(async move { + started.send(()).unwrap(); + release.acquire().await.unwrap().forget(); + Ok(core::Response::new(core::empty_body())) + }) + }), + ) + .await + }); + tokio::time::timeout(Duration::from_secs(5), entered.recv()) + .await + .unwrap_or_else(|_| panic!("第 {} 个调用未能进入下游", index + 1)) + .unwrap(); + } + + for uri in [ + "/healthz", + "/assets/index.js", + "/api/settings", + "/v1/responses", + ] { + let request = ::http::Request::builder() + .uri(uri) + .header("authorization", "Bearer fixture-secret") + .body(core::empty_body()) + .unwrap(); + let result = tokio::time::timeout( + Duration::from_secs(3), + plan.handle_http( + context(), + request, + compose(Vec::new(), move |request: core::Request| { + assert!(!saturated, "容量耗尽时不得执行下游"); + assert_eq!(request.uri(), "/rewritten?encoded=%2F"); + Box::pin(async { Ok(core::Response::new(core::empty_body())) }) + }), + ), + ) + .await + .expect("下游等待不应阻塞其它请求"); + if saturated { + assert!(matches!(result, Err(MiddlewareError::Gateway(error)) + if error.kind() == GatewayErrorKind::MiddlewareCapacityUnavailable)); + } else { + assert_eq!(result.unwrap().headers()["x-http-plugin"], "active"); + } + } + + if saturated { + slow_calls.abort_all(); + while let Some(result) = slow_calls.join_next().await { + assert!(matches!(result, Err(error) if error.is_cancelled())); + } + } + release.add_permits(CONCURRENT_CALLS); + while let Some(result) = slow_calls.join_next().await { + let response = result.unwrap().unwrap(); + assert_eq!(response.headers()["x-http-plugin"], "active"); + } + // 取消确认到达前仍持有在途名额，探测只允许遇到明确的暂时过载 + let response = tokio::time::timeout(Duration::from_secs(2), async { + loop { + let result = plan + .handle_http( + context(), + plugin_request(), + compose(Vec::new(), |request: core::Request| { + Box::pin(async move { + assert_eq!(request.uri(), "/rewritten?encoded=%2F"); + Ok(core::Response::new(core::empty_body())) + }) + }), + ) + .await; + match result { + Ok(response) => break response, + Err(MiddlewareError::Gateway(error)) + if error.kind() == GatewayErrorKind::MiddlewareCapacityUnavailable => + { + tokio::task::yield_now().await + } + Err(error) => panic!("unexpected error: {error}"), + } + } + }) + .await + .unwrap(); + assert_eq!(response.headers()["x-http-plugin"], "active"); + drop(response); + drop(plan); + drop(generation); + runtime.shutdown().await; + crate::generation::wait_until_empty(cache.path()).await; +} diff --git a/backend/crates/gateway-plugin/runtime/tests/generation/policy/http/mod.rs b/backend/crates/gateway-plugin/runtime/tests/generation/policy/http/mod.rs index 497edc06e..2bc02f9cd 100644 --- a/backend/crates/gateway-plugin/runtime/tests/generation/policy/http/mod.rs +++ b/backend/crates/gateway-plugin/runtime/tests/generation/policy/http/mod.rs @@ -1,5 +1,6 @@ //! 验证真实插件 HTTP 中间件的请求改写、流式转换与短路响应 +mod capacity; mod dispatch; mod websocket; @@ -27,6 +28,19 @@ async fn http_plugin_with_limits( PluginRuntime, ExtensionSetReference, gateway_core::engine::middleware::FrozenMiddlewarePlan, +) { + http_plugin_with_policy(mode, limits, PluginFailurePolicy::Reject).await +} + +async fn http_plugin_with_policy( + mode: &str, + limits: RpcLimits, + failure_policy: PluginFailurePolicy, +) -> ( + tempfile::TempDir, + PluginRuntime, + ExtensionSetReference, + gateway_core::engine::middleware::FrozenMiddlewarePlan, ) { let worker = std::fs::read(env!("CARGO_BIN_EXE_gateway-plugin-test-middleware")).unwrap(); let package = crate::support::package_with_contributions( @@ -42,12 +56,7 @@ async fn http_plugin_with_limits( vec![InstanceFixture { id: "http-plugin", configuration: serde_json::json!({"http":true,"mode":mode}), - bindings: vec![binding( - MIDDLEWARE_CONTRIBUTION, - "http", - 0, - PluginFailurePolicy::Reject, - )], + bindings: vec![binding(MIDDLEWARE_CONTRIBUTION, "http", 0, failure_policy)], }], package, limits, diff --git a/backend/crates/gateway-plugin/runtime/tests/generation/policy/lifecycle.rs b/backend/crates/gateway-plugin/runtime/tests/generation/policy/lifecycle.rs index be91fbe14..7db8e01a4 100644 --- a/backend/crates/gateway-plugin/runtime/tests/generation/policy/lifecycle.rs +++ b/backend/crates/gateway-plugin/runtime/tests/generation/policy/lifecycle.rs @@ -327,3 +327,79 @@ async fn sdk_waiting_next_obeys_request_deadline_and_client_cancellation() { super::super::wait_until_empty(cache.path()).await; } } + +#[tokio::test] +async fn request_middleware_never_delegates_when_total_in_flight_capacity_is_full() { + for policy in [PluginFailurePolicy::Delegate, PluginFailurePolicy::Reject] { + let package = crate::support::package_with_contributions( + crate::support::worker(), + Contributions::from([crate::support::contribution( + Capability::Middleware, + vec![Stage::Request], + vec!["openai".into()], + vec!["openai".into()], + )]), + ); + let (cache, runtime) = setup_package_with_limits( + vec![InstanceFixture { + id: "busy-middleware", + configuration: serde_json::json!({}), + bindings: vec![binding(MIDDLEWARE_CONTRIBUTION, "request", 0, policy)], + }], + package, + RpcLimits { + maximum_calls: 1, + ..Default::default() + }, + ) + .await; + let generation = prepare(&runtime).await; + let plan = runtime + .execution_registry() + .middleware(&generation) + .unwrap(); + let (entered, waiting) = tokio::sync::oneshot::channel(); + let first = tokio::spawn({ + let plan = plan.clone(); + async move { + plan.handle( + middleware_context(ClientTransport::HttpJson), + MiddlewareRequest::new( + "openai", + middleware_headers(), + Bytes::from_static(b"{}"), + ), + gateway_core::middleware::compose(Vec::new(), move |_| { + Box::pin(async move { + entered.send(()).unwrap(); + std::future::pending().await + }) + }), + ) + .await + } + }); + tokio::time::timeout(Duration::from_secs(3), waiting) + .await + .unwrap() + .unwrap(); + let result = plan + .handle( + middleware_context(ClientTransport::HttpJson), + MiddlewareRequest::new("openai", middleware_headers(), Bytes::from_static(b"{}")), + gateway_core::middleware::compose(Vec::new(), |_| { + Box::pin(async { panic!("RPC 过载不能跳过请求中间件") }) + }), + ) + .await; + assert!( + matches!(result, Err(MiddlewareError::Gateway(error)) if error.kind() == gateway_core::error::GatewayErrorKind::MiddlewareCapacityUnavailable) + ); + first.abort(); + assert!(matches!(first.await, Err(error) if error.is_cancelled())); + drop(plan); + drop(generation); + runtime.shutdown().await; + super::super::wait_until_empty(cache.path()).await; + } +} diff --git a/backend/crates/gateway-plugin/runtime/tests/rpc/admission.rs b/backend/crates/gateway-plugin/runtime/tests/rpc/admission.rs new file mode 100644 index 000000000..b83416e39 --- /dev/null +++ b/backend/crates/gateway-plugin/runtime/tests/rpc/admission.rs @@ -0,0 +1,328 @@ +//! 验证执行排队、next 恢复、流式在途预算与取消回收 + +use std::{ + sync::{Arc, atomic::Ordering}, + time::Duration, +}; + +use super::session::{Callbacks, LifecycleCallbacks, prepared_session}; +use futures::future::BoxFuture; +use gateway_plugin_runtime::{CallbackHandler, RpcError, RpcLimits, RpcReply, RpcSession}; +use gateway_plugin_sdk::{CallContext, PluginFault, Stage, call::middleware::NEXT_METHOD}; +use serde_json::json; +use tokio::sync::{Semaphore, mpsc}; + +async fn limited( + callbacks: Arc, + limits: RpcLimits, +) -> (tempfile::TempDir, Arc) { + let (cache, prepared, handshake, processes) = prepared_session(); + let session = RpcSession::start(prepared, handshake, limits, &processes, callbacks) + .await + .unwrap(); + (cache, Arc::new(session)) +} + +fn limits() -> RpcLimits { + RpcLimits { + maximum_calls: 4, + maximum_executing_calls: 1, + maximum_queued_calls: 1, + maximum_queue_timeout: Duration::from_millis(200), + maximum_callbacks: 1, + ..Default::default() + } +} + +async fn wait_begun(callbacks: &LifecycleCallbacks, count: usize) { + tokio::time::timeout(Duration::from_secs(2), async { + while callbacks.begun.load(Ordering::Acquire) < count { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); +} + +fn call(session: &RpcSession, method: &str) -> impl Future> { + session.call( + method, + session.context(Stage::Request, Duration::from_secs(2)), + json!({}), + Vec::new(), + ) +} + +#[tokio::test] +async fn bounded_initial_queue_waits_and_releases_cancelled_or_expired_waiters() { + let callbacks = Arc::new(LifecycleCallbacks::default()); + let (_cache, session) = limited(callbacks.clone(), limits()).await; + let active = tokio::spawn({ + let session = session.clone(); + async move { call(&session, "hang").await } + }); + wait_begun(&callbacks, 1).await; + + let mut queued = Box::pin(call(&session, "echo")); + assert!(futures::poll!(&mut queued).is_pending()); + assert!(matches!( + call(&session, "echo").await, + Err(RpcError::Overloaded) + )); + assert_eq!(callbacks.begun.load(Ordering::Acquire), 1); + drop(queued); + + let expired = session + .call( + "echo", + session.context(Stage::Request, Duration::from_millis(30)), + json!({}), + Vec::new(), + ) + .await; + assert!(matches!(expired, Err(RpcError::Overloaded))); + assert_eq!(callbacks.begun.load(Ordering::Acquire), 1); + + let mut queued = Box::pin(call(&session, "echo")); + assert!(futures::poll!(&mut queued).is_pending()); + active.abort(); + assert!(matches!(active.await, Err(error) if error.is_cancelled())); + assert!(queued.await.is_ok()); + assert_eq!(callbacks.begun.load(Ordering::Acquire), 2); + assert!(session.is_ready()); + session.shutdown(Duration::from_secs(1)).await; +} + +struct NextCallbacks { + entered: mpsc::UnboundedSender<()>, + release: Arc, +} + +impl CallbackHandler for NextCallbacks { + fn call( + &self, + _context: CallContext, + method: String, + params: serde_json::Value, + payload: Vec, + ) -> BoxFuture<'static, Result> { + let release = self.release.clone(); + self.entered.send(()).unwrap(); + Box::pin(async move { + if method == NEXT_METHOD { + release.acquire().await.unwrap().forget(); + } + Ok(RpcReply { + result: params, + payload, + }) + }) + } +} + +#[tokio::test] +async fn next_resumption_does_not_need_a_vacant_initial_queue_slot() { + let (entered, mut started) = mpsc::unbounded_channel(); + let release = Arc::new(Semaphore::new(0)); + let (_cache, session) = limited( + Arc::new(NextCallbacks { + entered, + release: release.clone(), + }), + limits(), + ) + .await; + let waiting = tokio::spawn({ + let session = session.clone(); + async move { + session + .call( + "callback_method", + session.context(Stage::Request, Duration::from_secs(2)), + json!({"method": NEXT_METHOD}), + Vec::new(), + ) + .await + } + }); + started.recv().await.unwrap(); + // 等待 next 不挤占唯一的普通回调额度 + let mut occupying = Box::pin(call(&session, "callback")); + assert!(futures::poll!(&mut occupying).is_pending()); + started.recv().await.unwrap(); + occupying.await.unwrap(); + let mut active = Box::pin(call(&session, "hang")); + assert!(futures::poll!(&mut active).is_pending()); + let mut queued = Box::pin(call(&session, "echo")); + assert!(futures::poll!(&mut queued).is_pending()); + assert!(matches!( + call(&session, "echo").await, + Err(RpcError::Overloaded) + )); + release.add_permits(1); + // 恢复等待不能立即因初次队列已满而拒绝，也不能绕过执行名额完成 + tokio::time::sleep(Duration::from_millis(30)).await; + assert!(!waiting.is_finished()); + drop(active); + assert!(queued.await.is_ok()); + assert!(waiting.await.unwrap().is_ok()); + assert!(session.is_ready()); + session.shutdown(Duration::from_secs(1)).await; +} + +#[tokio::test] +async fn response_stream_releases_execution_but_retains_total_capacity_until_drop() { + let callbacks = Arc::new(Callbacks::default()); + let (_cache, session) = limited( + callbacks, + RpcLimits { + maximum_calls: 2, + ..limits() + }, + ) + .await; + let mut streams = Vec::new(); + for _ in 0..2 { + streams.push( + session + .call_stream( + "stream", + session.context(Stage::Request, Duration::from_secs(5)), + json!({}), + Vec::new(), + ) + .await + .unwrap(), + ); + } + assert!(matches!( + call(&session, "echo").await, + Err(RpcError::Overloaded) + )); + drop(streams.pop()); + tokio::time::timeout(Duration::from_secs(1), async { + loop { + match call(&session, "echo").await { + Ok(_) => break, + Err(RpcError::Overloaded) => tokio::task::yield_now().await, + Err(error) => panic!("unexpected error: {error}"), + } + } + }) + .await + .unwrap(); + assert!(session.is_ready()); + drop(streams); + session.shutdown(Duration::from_secs(1)).await; +} + +#[tokio::test] +async fn queued_resumption_is_bounded_and_cancellable_without_failing_session() { + for cancelled in [false, true] { + let (entered, mut started) = mpsc::unbounded_channel(); + let release = Arc::new(Semaphore::new(0)); + let (_cache, session) = limited( + Arc::new(NextCallbacks { + entered, + release: release.clone(), + }), + limits(), + ) + .await; + let waiting = tokio::spawn({ + let session = session.clone(); + async move { + session + .call( + "callback_method", + session.context(Stage::Request, Duration::from_millis(200)), + json!({"method": NEXT_METHOD}), + Vec::new(), + ) + .await + } + }); + started.recv().await.unwrap(); + let mut active = Box::pin(call(&session, "hang")); + assert!(futures::poll!(&mut active).is_pending()); + release.add_permits(1); + if cancelled { + tokio::time::sleep(Duration::from_millis(30)).await; + assert!(!waiting.is_finished()); + waiting.abort(); + assert!(matches!(waiting.await, Err(error) if error.is_cancelled())); + } else { + assert!(matches!(waiting.await.unwrap(), Err(RpcError::Overloaded))); + } + drop(active); + assert!(call(&session, "echo").await.is_ok()); + assert!(session.is_ready()); + session.shutdown(Duration::from_secs(1)).await; + } +} + +#[tokio::test] +async fn initial_queue_wait_does_not_consume_the_plugin_processing_budget() { + let (_cache, session) = limited( + Arc::new(Callbacks::default()), + RpcLimits { + maximum_queue_timeout: Duration::from_secs(2), + ..limits() + }, + ) + .await; + let mut active = Box::pin(call(&session, "hang")); + assert!(futures::poll!(&mut active).is_pending()); + let mut queued = Box::pin(session.call( + "echo", + session.context(Stage::Request, Duration::from_millis(200)), + json!({}), + Vec::new(), + )); + assert!(futures::poll!(&mut queued).is_pending()); + tokio::time::sleep(Duration::from_millis(350)).await; + drop(active); + assert!(queued.await.is_ok()); + assert!(session.is_ready()); + session.shutdown(Duration::from_secs(1)).await; +} + +#[tokio::test] +async fn resumption_queue_wait_preserves_the_remaining_plugin_processing_budget() { + let (entered, mut started) = mpsc::unbounded_channel(); + let release = Arc::new(Semaphore::new(0)); + let (_cache, session) = limited( + Arc::new(NextCallbacks { + entered, + release: release.clone(), + }), + RpcLimits { + maximum_queue_timeout: Duration::from_secs(2), + ..limits() + }, + ) + .await; + let waiting = tokio::spawn({ + let session = session.clone(); + async move { + session + .call( + "callback_method", + session.context(Stage::Request, Duration::from_millis(200)), + json!({"method": NEXT_METHOD}), + Vec::new(), + ) + .await + } + }); + started.recv().await.unwrap(); + let mut active = Box::pin(call(&session, "hang")); + assert!(futures::poll!(&mut active).is_pending()); + release.add_permits(1); + tokio::time::sleep(Duration::from_millis(350)).await; + assert!(!waiting.is_finished()); + drop(active); + assert!(waiting.await.unwrap().is_ok()); + assert!(session.is_ready()); + session.shutdown(Duration::from_secs(1)).await; +} diff --git a/backend/crates/gateway-plugin/runtime/tests/rpc/mod.rs b/backend/crates/gateway-plugin/runtime/tests/rpc/mod.rs index 746ef525f..43a5aadf1 100644 --- a/backend/crates/gateway-plugin/runtime/tests/rpc/mod.rs +++ b/backend/crates/gateway-plugin/runtime/tests/rpc/mod.rs @@ -1,4 +1,5 @@ //! 插件双向 RPC 分派与会话生命周期的测试入口 +mod admission; mod dispatch; mod session; diff --git a/backend/crates/gateway-plugin/runtime/tests/rpc/session.rs b/backend/crates/gateway-plugin/runtime/tests/rpc/session.rs index 4d7266077..fe9dcf67e 100644 --- a/backend/crates/gateway-plugin/runtime/tests/rpc/session.rs +++ b/backend/crates/gateway-plugin/runtime/tests/rpc/session.rs @@ -17,7 +17,7 @@ use serde_json::json; use tokio::sync::{Barrier, Notify}; #[derive(Default)] -struct Callbacks { +pub(super) struct Callbacks { called: AtomicUsize, nested: Mutex>>, } @@ -29,8 +29,8 @@ struct BlockingCallbacks { } #[derive(Default)] -struct LifecycleCallbacks { - begun: AtomicUsize, +pub(super) struct LifecycleCallbacks { + pub(super) begun: AtomicUsize, finished: AtomicUsize, } @@ -193,7 +193,7 @@ where (cache, Arc::new(session)) } -fn prepared_session() -> ( +pub(super) fn prepared_session() -> ( tempfile::TempDir, Arc, Handshake, diff --git a/backend/crates/gateway-plugin/sdk/README.md b/backend/crates/gateway-plugin/sdk/README.md index f539781bc..1f88b50a9 100644 --- a/backend/crates/gateway-plugin/sdk/README.md +++ b/backend/crates/gateway-plugin/sdk/README.md @@ -52,6 +52,12 @@ use gateway_plugin_sdk::client::{read_frame, write_frame}; `PluginSession::accept/run` 和 `PluginHandler`。组合插件从作者清单建立类型化方法，SDK 自动生成 `plugin.register`；作者不需要复制贡献项或手写方法名、阶段和敏感载荷编解码。 会话管理握手、有界并发、父调用关联、回调、流信用、取消、排空与关闭。 + +`SessionConfig.maximum_calls` 默认 256，限制包含下游等待和响应流在内的在途调用；`maximum_callbacks` 默认 64， +限制普通宿主回调。`next` 使用独立回调预算，上限为 `maximum_calls`，不挤占普通回调容量。 +宿主负责执行与准入调度，等待与处理独立计时；默认预算与过载行为见[插件使用](../../../../docs/plugins.md#故障处理)。 +插件显式覆盖配置时需与宿主协调在途预算；较小的插件侧上限会提前拒绝新调用 + `ResponseStream::channel` 提供有界流生产入口，`ResponseStream::pull` 按消费进度驱动生产者；`PluginCall` 的宿主客户端只代表当前父调用， 不能把一个调用的资源误用到另一个调用。插件程序自行选择运行时和启动入口，不会启动网关。 `SessionConfig.maximum_stream_chunk_bytes` 默认 1 MiB，只约束单个业务流分块； diff --git a/backend/crates/gateway-plugin/sdk/docs/capabilities.md b/backend/crates/gateway-plugin/sdk/docs/capabilities.md index 9b1055e72..a62f2642a 100644 --- a/backend/crates/gateway-plugin/sdk/docs/capabilities.md +++ b/backend/crates/gateway-plugin/sdk/docs/capabilities.md @@ -243,7 +243,9 @@ SDK 不自动重试。超时或断连不能证明写入未提交；重试上限 `next.run(input)` 消费自身，最多执行一次；结果按相反顺序经过外层。重试或独立子请求通过宿主执行服务发起 调用上下文的 `timeout_ms` 限制插件自身处理时间，默认上限为 120 秒。等待宿主 `next` 时暂停这份预算， -收到结果或错误后恢复剩余时间；入站处理与返回处理合计使用同一预算，普通宿主回调不会暂停计时。 +宿主等待期间归还执行名额，返回后先重新取得名额再交付结果或错误，插件继续同一次调用。 +初次准入和恢复排队各有独立等待期限，默认 30 秒，不扣插件处理预算；恢复等待受总在途上限约束，不占初次准入队列。 +入站处理与返回处理合计使用同一预算，普通宿主回调不会暂停计时。 请求总期限、下游操作超时和取消信号独立生效，客户端取消或插件退出仍会结束等待并回收资源。 宿主与插件 SDK 必须同时支持此语义；仍使用固定绝对期限的 SDK 构建的插件需重新构建，单独升级宿主不能解除插件侧期限 diff --git a/backend/crates/gateway-plugin/sdk/src/client/session/config.rs b/backend/crates/gateway-plugin/sdk/src/client/session/config.rs index 6c880ad38..29b1ba2d7 100644 --- a/backend/crates/gateway-plugin/sdk/src/client/session/config.rs +++ b/backend/crates/gateway-plugin/sdk/src/client/session/config.rs @@ -25,8 +25,8 @@ impl Default for SessionConfig { Self { // 流分块还需匹配宿主授予的信用窗口 maximum_stream_chunk_bytes: 1024 * 1024, - maximum_calls: 32, - maximum_callbacks: 32, + maximum_calls: 256, + maximum_callbacks: 64, maximum_buffered_stream_chunks: 1_024, handshake_timeout: Duration::from_secs(5), maximum_call_timeout: Duration::from_secs(120), diff --git a/backend/crates/gateway-plugin/sdk/src/client/session/driver.rs b/backend/crates/gateway-plugin/sdk/src/client/session/driver.rs index 5a09df6d8..d4d60868d 100644 --- a/backend/crates/gateway-plugin/sdk/src/client/session/driver.rs +++ b/backend/crates/gateway-plugin/sdk/src/client/session/driver.rs @@ -91,6 +91,7 @@ where .config .maximum_calls .checked_add(self.config.maximum_callbacks) + .and_then(|capacity| capacity.checked_add(self.config.maximum_calls)) .ok_or(SessionError::Configuration)?; let control_reserve = self .config @@ -105,7 +106,10 @@ where sender: outbound, data_slots: Arc::new(Semaphore::new(data_capacity)), }; - let callbacks = Arc::new(CallbackRegistry::new(self.config.maximum_callbacks)); + let callbacks = Arc::new(CallbackRegistry::new( + self.config.maximum_callbacks, + self.config.maximum_calls, + )); let handler: Arc = Arc::new(handler); let lifecycle = LifecycleHooks::new(Arc::clone(&handler), control_reserve)?; write_frame( diff --git a/backend/crates/gateway-plugin/sdk/src/client/session/registry.rs b/backend/crates/gateway-plugin/sdk/src/client/session/registry.rs index 5d5c9166b..06f9fc62f 100644 --- a/backend/crates/gateway-plugin/sdk/src/client/session/registry.rs +++ b/backend/crates/gateway-plugin/sdk/src/client/session/registry.rs @@ -67,7 +67,7 @@ impl HostClient { ordered = self.callbacks.order.lock() => ordered, }; let managed_next = method == NEXT_METHOD; - let (id, received) = self.callbacks.begin(self.parent_id)?; + let (id, received) = self.callbacks.begin(self.parent_id, managed_next)?; let frame = Frame { message: Message::Callback { id, @@ -114,6 +114,7 @@ pub(super) struct CallbackRegistry { state: Mutex, order: tokio::sync::Mutex<()>, maximum: usize, + maximum_next: usize, maximum_retired: usize, } @@ -126,12 +127,13 @@ struct CallbackState { struct PendingCallback { parent_id: u64, + managed_next: bool, pause: Option, result: oneshot::Sender>, } impl CallbackRegistry { - pub(super) fn new(maximum: usize) -> Self { + pub(super) fn new(maximum: usize, maximum_next: usize) -> Self { Self { state: Mutex::new(CallbackState { next_id: 2, @@ -141,19 +143,32 @@ impl CallbackRegistry { }), order: tokio::sync::Mutex::new(()), maximum, - maximum_retired: maximum.saturating_mul(2), + maximum_next, + maximum_retired: (maximum + maximum_next).saturating_mul(2), } } fn begin( &self, parent_id: u64, + managed_next: bool, ) -> Result<(u64, oneshot::Receiver>), SessionError> { let mut state = self .state .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - if state.pending.len() >= self.maximum { + let maximum = if managed_next { + self.maximum_next + } else { + self.maximum + }; + if state + .pending + .values() + .filter(|pending| pending.managed_next == managed_next) + .count() + >= maximum + { return Err(SessionError::Capacity); } let id = state.next_id; @@ -163,6 +178,7 @@ impl CallbackRegistry { id, PendingCallback { parent_id, + managed_next, result, pause: None, }, diff --git a/backend/crates/gateway-plugin/sdk/tests/client/session/mod.rs b/backend/crates/gateway-plugin/sdk/tests/client/session/mod.rs index ad35ddbb1..cb6a29727 100644 --- a/backend/crates/gateway-plugin/sdk/tests/client/session/mod.rs +++ b/backend/crates/gateway-plugin/sdk/tests/client/session/mod.rs @@ -140,10 +140,15 @@ impl PluginHandler for TestHandler { tokio::time::sleep(Duration::from_millis(80)).await; Ok(CallReply::unary(call.params, call.payload)) } - "callback" => { + "callback" | "next_callback" => { + let method = if call.method == "next_callback" { + NEXT_METHOD + } else { + "host.log" + }; let reply = call .host - .call("host.log", call.params, call.payload) + .call(method, call.params, call.payload) .await .map_err(SessionError::into_plugin_fault)?; Ok(CallReply::unary(reply.result, reply.payload)) @@ -248,23 +253,31 @@ pub(super) async fn start_session( async fn start_session_with_capacity( handler: H, capacity: usize, +) -> (HostPeer, JoinHandle>) { + start_configured_session( + handler, + capacity, + SessionConfig { + maximum_stream_chunk_bytes: MAXIMUM_STREAM_CHUNK_BYTES, + maximum_calls: 4, + maximum_callbacks: 4, + maximum_buffered_stream_chunks: 16, + handshake_timeout: Duration::from_secs(1), + maximum_call_timeout: Duration::from_secs(2), + }, + ) + .await +} + +async fn start_configured_session( + handler: H, + capacity: usize, + config: SessionConfig, ) -> (HostPeer, JoinHandle>) { let (host, plugin) = tokio::io::duplex(capacity); let (plugin_reader, plugin_writer) = tokio::io::split(plugin); let task = tokio::spawn(async move { - let session = PluginSession::accept( - plugin_reader, - plugin_writer, - SessionConfig { - maximum_stream_chunk_bytes: MAXIMUM_STREAM_CHUNK_BYTES, - maximum_calls: 4, - maximum_callbacks: 4, - maximum_buffered_stream_chunks: 16, - handshake_timeout: Duration::from_secs(1), - maximum_call_timeout: Duration::from_secs(2), - }, - ) - .await?; + let session = PluginSession::accept(plugin_reader, plugin_writer, config).await?; session.run(handler).await }); let (mut reader, mut writer) = tokio::io::split(host); diff --git a/backend/crates/gateway-plugin/sdk/tests/client/session/registry.rs b/backend/crates/gateway-plugin/sdk/tests/client/session/registry.rs index cda12ef26..473867923 100644 --- a/backend/crates/gateway-plugin/sdk/tests/client/session/registry.rs +++ b/backend/crates/gateway-plugin/sdk/tests/client/session/registry.rs @@ -297,3 +297,57 @@ async fn typed_key_and_quota_calls_keep_payloads_and_do_not_retry_failures() { } } } + +#[tokio::test] +async fn managed_next_does_not_consume_the_bounded_ordinary_callback_budget() { + let (mut host, task) = start_configured_session( + TestHandler::default(), + MAXIMUM_STREAM_CHUNK_BYTES * 2, + SessionConfig { + maximum_calls: 4, + maximum_callbacks: 1, + ..Default::default() + }, + ) + .await; + send_call(&mut host, 1, "next_callback", json!({}), Vec::new()).await; + let Message::Callback { + id: next, + parent_id: 1, + method, + .. + } = receive(&mut host).await.message + else { + panic!("expected next callback") + }; + assert_eq!(method, NEXT_METHOD); + send_call(&mut host, 3, "callback", json!({}), Vec::new()).await; + let Message::Callback { + id: ordinary, + parent_id: 3, + method, + .. + } = receive(&mut host).await.message + else { + panic!("next must not occupy ordinary callback capacity") + }; + assert_eq!(method, "host.log"); + send_call(&mut host, 5, "callback", json!({}), Vec::new()).await; + assert!( + matches!(receive(&mut host).await.message, Message::Error { id: 5, error } if error.code == ErrorCode::Capacity) + ); + for (callback, parent) in [(ordinary, 3), (next, 1)] { + send_control( + &mut host, + Message::Result { + id: callback, + result: json!({}), + }, + ) + .await; + assert!( + matches!(receive(&mut host).await.message, Message::Result { id, .. } if id == parent) + ); + } + shutdown(&mut host, task).await; +} diff --git a/docs/architecture.md b/docs/architecture.md index f2a8e5ab0..508de53a3 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -197,7 +197,9 @@ flowchart LR #### 调用资源与状态 - **资源跟随调用**：账号、HTTP、模型、状态和日志回调绑定有效父调用，取消随调用传播。受管流建立后不沿用握手的绝对期限，网络操作与模型执行各自计时。 - Runtime 与 SDK 的中间件处理预算在宿主 `next` 等待期间暂停，返回后恢复剩余预算；调用等待与回调有效期共用各端的预算状态，请求总期限独立生效。 + Runtime 与 SDK 的中间件处理预算在宿主 `next` 等待期间暂停；Runtime 同时归还执行名额，返回后取得名额再以剩余处理预算恢复原调用。 + 执行调度由 Runtime 统一负责，准入和恢复等待使用独立期限；初次排队、总在途上下文和普通回调各有上限；`next` 回调按在途上限独立计数。 + 流、取消确认与消费方共同持有同一在途租约，执行名额只覆盖首结果前的处理；调用等待与回调有效期共用各端的预算状态，请求总期限独立生效。 显式 Key 的身份快照随调用保留，每个模型执行拥有独立子调用图与结算；省略 Key 时继承现有模型父请求。页面 Responses 桥复核精确实例目标和 revision `frontend_authentication` 使用 principal 到 Key 的映射；公开登录回调的一次性票据用于关联流程，不限制插件回调能力。接口合同见 [SDK 回调](../backend/crates/gateway-plugin/sdk/docs/capabilities.md#宿主资源回调) - **观察合同统一**：`observer` 通过一个类型化 RPC 入口接收完成与实际上游 WebSocket 事件，发布时按事件类型编译独立作用域。完成事件直接携带终态、用量、费用和耗时，不按能力声明裁剪字段；异步投递的并发、队列和载荷有界，观察失败不改变业务结果 diff --git a/docs/plugins.md b/docs/plugins.md index 52e4d9f74..f2d35b7b8 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -124,6 +124,14 @@ flowchart LR 上游 429、用户请求错误和插件主动拒绝请求等业务结果不会直接停止插件。其他插件、原生转发和管理端继续运行， 不因单个插件故障撤销整个宿主的运行配置。命中故障插件绑定的请求仍遵循既有拒绝或委托策略，入口认证插件失败不会绕过认证 +宿主每个插件会话默认允许 64 个执行中的 RPC 调用、64 个初次排队调用和 256 个总在途上下文。 +等待 `next` 时释放执行名额，返回后重新取得名额并继续原调用；已返回响应头的流继续占用在途预算，不占首结果执行名额。 +初次排队和恢复等待各限 30 秒，不扣插件自身默认上限 120 秒的处理预算。达到容量或等待上限时，中间件明确返回过载错误，HTTP 响应为 503（`middleware_capacity_unavailable`），不跳过插件、不重放请求。 +这些数值是有界资源预算的初值，实际容量需结合插件耗时、正文和流的资源占用评估 + +普通宿主回调默认上限 64，`next` 使用总在途上限对应的独立回调预算。实际容量同时受插件 SDK 配置约束； +使用较小上限或未区分 `next` 回调预算的插件需由作者调整并重新构建，升级宿主不会改变已安装插件的容量 + ### 停用与删除 | 操作 | 删除什么 | 保留什么 |