From 94c769fc9bce7e56173ff4bc57d25531e0390a24 Mon Sep 17 00:00:00 2001 From: Zyy <571092077@qq.com> Date: Mon, 12 Oct 2026 11:40:01 +0800 Subject: [PATCH] fix(plugins): exclude managed next waits from RPC budgets (#397) Pause Runtime and SDK processing budgets while middleware delegates to host next, then resume the remaining budget. Preserve request deadlines and cancellation, share callback deadline state, and add timeout diagnostics and regression coverage. Fixes #392 Co-authored-by: Codex --- .../runtime/src/adapter/middleware/request.rs | 12 +- .../runtime/src/callback/call_resources.rs | 29 +-- .../runtime/src/callback/mod.rs | 20 +- .../runtime/src/rpc/dispatch.rs | 5 + .../gateway-plugin/runtime/src/rpc/session.rs | 80 ++++++- .../runtime/src/stream/bridge.rs | 16 +- .../tests/generation/policy/lifecycle.rs | 205 +++++++++++++++++- .../runtime/tests/generation/policy/mod.rs | 12 +- .../runtime/tests/rpc/session.rs | 54 ++++- backend/crates/gateway-plugin/sdk/Cargo.toml | 2 +- .../gateway-plugin/sdk/docs/capabilities.md | 5 + .../gateway-plugin/sdk/src/client/mod.rs | 6 +- .../sdk/src/client/session/credit.rs | 12 +- .../sdk/src/client/session/deadline.rs | 106 +++++++++ .../sdk/src/client/session/driver.rs | 43 ++-- .../sdk/src/client/session/lifecycle.rs | 21 +- .../sdk/src/client/session/mod.rs | 2 + .../sdk/src/client/session/output.rs | 7 +- .../sdk/src/client/session/registry.rs | 49 ++++- .../crates/gateway-plugin/sdk/src/context.rs | 1 + .../sdk/tests/client/session/deadline.rs | 52 +++++ .../sdk/tests/client/session/lifecycle.rs | 200 +++++++++++++++++ .../sdk/tests/client/session/mod.rs | 1 + docs/architecture.md | 5 + 24 files changed, 815 insertions(+), 130 deletions(-) create mode 100644 backend/crates/gateway-plugin/sdk/src/client/session/deadline.rs create mode 100644 backend/crates/gateway-plugin/sdk/tests/client/session/deadline.rs 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 943dac135..4da90ed99 100644 --- a/backend/crates/gateway-plugin/runtime/src/adapter/middleware/request.rs +++ b/backend/crates/gateway-plugin/runtime/src/adapter/middleware/request.rs @@ -48,8 +48,10 @@ pub(super) async fn invoke( Err(MiddlewareError::Fault) }; }; - let remaining = context.deadline().bounded(maximum_timeout); - if remaining.is_zero() || context.cancellation().is_cancelled() { + if maximum_timeout.is_zero() + || context.deadline().is_elapsed() + || context.cancellation().is_cancelled() + { return Err(MiddlewareError::Fault); } let invocation = MiddlewareInvocation::new(&context, request, next, entry.instance_id.clone()); @@ -65,7 +67,7 @@ pub(super) async fn invoke( MiddlewareMount::Request => Stage::Request, MiddlewareMount::Attempt => Stage::Attempt, }, - remaining, + maximum_timeout, ); call_context.request_id = Some(context.request_id().as_str().to_owned()); call_context.attempt_id = context @@ -129,6 +131,10 @@ pub(super) async fn invoke( let stream = tokio::select! { biased; () = context.cancellation().cancelled() => return Err(MiddlewareError::Fault), + () = context.deadline().wait() => return Err(gateway_core::error::GatewayError::new( + gateway_core::error::GatewayErrorKind::Timeout, + "request deadline elapsed", + ).into()), result = invocation_ports.session.call_stream( HANDLE_METHOD, call_context, diff --git a/backend/crates/gateway-plugin/runtime/src/callback/call_resources.rs b/backend/crates/gateway-plugin/runtime/src/callback/call_resources.rs index 308023a36..9a3cc0657 100644 --- a/backend/crates/gateway-plugin/runtime/src/callback/call_resources.rs +++ b/backend/crates/gateway-plugin/runtime/src/callback/call_resources.rs @@ -3,16 +3,15 @@ use super::{CallbackScope, MiddlewareCallback, denied, http_middleware, model}; use gateway_core::lifecycle::CancellationToken; use gateway_host::outbound::HttpBody; -use gateway_plugin_sdk::{CallContext, ErrorCode, PluginFault}; +use gateway_plugin_sdk::{CallContext, ErrorCode, PluginFault, client::CallDeadline}; use std::{ collections::BTreeMap, sync::{Arc, Mutex}, time::Duration, }; -use tokio::time::Instant; pub(super) struct CallResources { - pub(super) deadline: Instant, + pub(super) deadline: CallDeadline, operation_timeout: Duration, pub(super) cancellation: CancellationToken, pub(super) scope: Arc, @@ -26,11 +25,12 @@ pub(super) struct CallResources { impl CallResources { pub(super) fn new( context: &CallContext, + deadline: &CallDeadline, scope: Arc, middleware: Option>, ) -> Self { Self { - deadline: Instant::now() + Duration::from_millis(context.timeout_ms), + deadline: deadline.clone(), operation_timeout: Duration::from_millis(context.timeout_ms), cancellation: CancellationToken::new(), scope, @@ -46,13 +46,6 @@ impl CallResources { } } - pub(super) fn set_streaming(&self, streaming: bool) { - self.state - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .resource_stream = streaming; - } - pub(super) fn close(&self) { self.cancellation.cancel(); let mut state = self @@ -94,20 +87,20 @@ impl CallResources { if state.closed { return Err(denied()); } - if state.resource_stream { - return Ok(self.operation_timeout); + let remaining = self.deadline.remaining().unwrap_or(self.operation_timeout); + if remaining.is_zero() { + return Err(PluginFault::new( + ErrorCode::Timeout, + "callback deadline elapsed", + )); } - self.deadline - .checked_duration_since(Instant::now()) - .filter(|timeout| !timeout.is_zero()) - .ok_or_else(|| PluginFault::new(ErrorCode::Timeout, "callback deadline elapsed")) + Ok(remaining) } } #[derive(Default)] pub(super) struct CallState { pub(super) closed: bool, - pub(super) resource_stream: bool, pub(super) middleware: Option>, pub(super) streams: BTreeMap>, pub(super) model_streams: BTreeMap>, diff --git a/backend/crates/gateway-plugin/runtime/src/callback/mod.rs b/backend/crates/gateway-plugin/runtime/src/callback/mod.rs index 82a654ba1..841dbcefe 100644 --- a/backend/crates/gateway-plugin/runtime/src/callback/mod.rs +++ b/backend/crates/gateway-plugin/runtime/src/callback/mod.rs @@ -30,8 +30,7 @@ use futures::future::BoxFuture; use gateway_admin::model::AdminError; use gateway_core::{account::OutboundProxy, lifecycle::CancellationToken}; use gateway_host::outbound::{HttpClient, NetworkPolicy}; -use gateway_plugin_sdk::{CallContext, ErrorCode, PluginFault}; -use tokio::time::Instant; +use gateway_plugin_sdk::{CallContext, ErrorCode, PluginFault, client::CallDeadline}; use crate::{CallbackHandler, RpcReply}; pub(crate) use accounts::PluginAccountPortSlot; @@ -329,7 +328,7 @@ impl PluginCallbacks { } impl CallbackHandler for PluginCallbacks { - fn begin(&self, context: &CallContext) { + fn begin(&self, context: &CallContext, deadline: &CallDeadline) { let scope = self .scopes .lock() @@ -347,21 +346,10 @@ impl CallbackHandler for PluginCallbacks { .unwrap_or_else(std::sync::PoisonError::into_inner) .insert( context.call_id, - Arc::new(CallResources::new(context, scope, middleware)), + Arc::new(CallResources::new(context, deadline, scope, middleware)), ); } - fn streaming(&self, context: &CallContext) { - if let Some(call) = self - .calls - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .get(&context.call_id) - { - call.set_streaming(context.resource_stream); - } - } - fn finish(&self, context: &CallContext) { if let Some(call) = self .calls @@ -406,7 +394,7 @@ impl CallbackHandler for PluginCallbacks { .state .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - if state.closed || (!state.resource_stream && Instant::now() >= call.deadline) { + if state.closed || call.deadline.is_elapsed() { return Err(denied()); } if method == "host.log" { diff --git a/backend/crates/gateway-plugin/runtime/src/rpc/dispatch.rs b/backend/crates/gateway-plugin/runtime/src/rpc/dispatch.rs index b5b30217e..a26f3fcab 100644 --- a/backend/crates/gateway-plugin/runtime/src/rpc/dispatch.rs +++ b/backend/crates/gateway-plugin/runtime/src/rpc/dispatch.rs @@ -4,6 +4,7 @@ use std::sync::Arc; use gateway_plugin_sdk::{ ErrorCode, Frame, Message, PluginFault, + call::middleware::NEXT_METHOD, client::{read_frame, validate_frame, write_frame}, }; use tokio::{ @@ -128,12 +129,16 @@ async fn read_loop( let handler = Arc::clone(&callbacks); let response = Arc::clone(&shared); let (ready, started) = oneshot::channel(); + let pause = (method == NEXT_METHOD) + .then(|| shared.pause_next(parent_id)) + .flatten(); let task = tokio::spawn(async move { // 先把任务归属登记到父调用，再允许执行宿主操作 if started.await.is_err() { return; } let result = handler.call(context, method, params, frame.payload).await; + drop(pause); let mut reply = match result { Ok(reply) => Frame { message: Message::Result { diff --git a/backend/crates/gateway-plugin/runtime/src/rpc/session.rs b/backend/crates/gateway-plugin/runtime/src/rpc/session.rs index 26bdc9769..59993b85d 100644 --- a/backend/crates/gateway-plugin/runtime/src/rpc/session.rs +++ b/backend/crates/gateway-plugin/runtime/src/rpc/session.rs @@ -13,7 +13,7 @@ use futures::future::BoxFuture; use gateway_host::process::{ProcessControl, ProcessSpec, ProcessStartError, ProcessSupervisor}; use gateway_plugin_sdk::{ CallContext, Frame, Handshake, Message, PROTOCOL_VERSION, PluginFault, Stage, - client::{read_frame, validate_frame, write_frame}, + client::{CallDeadline, CallDeadlinePause, read_frame, validate_frame, write_frame}, }; use tokio::sync::{Mutex as AsyncMutex, Semaphore, mpsc, oneshot, watch}; @@ -155,7 +155,7 @@ pub struct RpcReply { /// 具体回调在领域适配器校验账号、出站与执行身份；上下文只取自宿主在途调用表 pub trait CallbackHandler: Send + Sync { /// 在 Call 可见之前建立资源范围；不得执行 I/O 或调用 RPC - fn begin(&self, _context: &CallContext) {} + fn begin(&self, _context: &CallContext, _deadline: &CallDeadline) {} /// 调用结束或取消时立即撤销其句柄；实现必须幂等且不能阻塞 fn finish(&self, _context: &CallContext) {} @@ -224,6 +224,7 @@ struct State { } struct PendingCall { + deadline: CallDeadline, handler: Arc, transmitted: bool, cancellation: Option, @@ -293,7 +294,14 @@ impl Shared { if call.cancellation.is_some() { return Ok(()); } + if matches!(&result, Err(RpcError::Remote(fault)) if fault.code == gateway_plugin_sdk::ErrorCode::Timeout) + { + log_timeout(&call.context, "plugin", "initial_reply", "remote_timeout"); + } if call.stream.is_some() && result.is_ok() { + if call.context.resource_stream { + call.deadline.clear(); + } let initial = call.result.take().ok_or(RpcError::Protocol(None))?; call.handler.streaming(&call.context); let _ = initial.send(result); @@ -375,6 +383,14 @@ impl Shared { .map(|call| call.context.clone()) } + pub fn pause_next(&self, id: u64) -> Option { + self.state() + .calls + .get(&id) + .filter(|call| call.cancellation.is_none()) + .and_then(|call| call.deadline.pause()) + } + pub fn is_cancelling(&self, id: u64) -> bool { self.state() .calls @@ -403,6 +419,18 @@ impl Shared { if call.cancellation.is_some() { return; } + if matches!(error, RpcError::Timeout) { + log_timeout( + &call.context, + "runtime", + if call.result.is_some() { + "initial_reply" + } else { + "response_stream" + }, + "processing_budget_exhausted", + ); + } if !call.transmitted { if let Some(call) = state.calls.remove(&id) { call.finish(Err(error)); @@ -759,10 +787,12 @@ impl RpcSession { if state.calls.len() >= self.limits.maximum_calls { return Err(RpcError::Capacity); } - self.callbacks.begin(&context); + let deadline = CallDeadline::new(deadline); + self.callbacks.begin(&context, &deadline); state.calls.insert( id, PendingCall { + deadline: deadline.clone(), handler: self.callbacks.clone(), transmitted: false, cancellation: None, @@ -817,7 +847,7 @@ impl RpcSession { id: started.id, armed: true, }; - let sent = tokio::time::timeout_at(started.deadline, async { + let send = async { self.data .send(frame) .await @@ -834,8 +864,12 @@ impl RpcSession { .map_err(|_| RpcError::Closed(None))?; } Ok::<(), RpcError>(()) - }) - .await; + }; + let sent = tokio::select! { + biased; + () = started.deadline.wait() => Err(()), + result = send => Ok(result), + }; drop(next_call); match sent { Ok(Ok(())) => {} @@ -850,7 +884,11 @@ impl RpcSession { return Err(RpcError::Timeout); } } - let result = tokio::time::timeout_at(started.deadline, started.received).await; + let result = tokio::select! { + biased; + () = started.deadline.wait() => Err(()), + result = started.received => Ok(result), + }; let reply = match result { Ok(Ok(result)) => { guard.armed = false; @@ -900,7 +938,6 @@ impl RpcSession { let bytes = self.limits.maximum_buffered_body_bytes.min(256 * 1024) as u32; let frames = 32; let (stream, chunks, terminal) = StreamIngress::new(bytes, frames); - let resource_stream = context.resource_stream; let completed = self .send_call( method, @@ -915,7 +952,7 @@ impl RpcSession { initial: completed.reply, chunks, terminal: Some(terminal), - deadline: (!resource_stream).then_some(completed.deadline), + deadline: completed.deadline, id: completed.id, shared: Arc::clone(&self.shared), _slot: slot, @@ -980,12 +1017,33 @@ impl Drop for CallGuard { struct StartedCall { id: u64, - deadline: tokio::time::Instant, + deadline: CallDeadline, received: oneshot::Receiver>, } struct CompletedCall { id: u64, - deadline: tokio::time::Instant, + deadline: CallDeadline, reply: RpcReply, } + +fn log_timeout( + context: &CallContext, + owner: &'static str, + phase: &'static str, + reason: &'static str, +) { + tracing::warn!( + event = "plugin.rpc.timeout", + owner, + phase, + instance_id = context.instance_id, + call_id = context.call_id, + request_id = context.request_id, + attempt_id = context.attempt_id, + stage = ?context.stage, + budget_ms = context.timeout_ms, + reason, + "插件 RPC 调用超时" + ); +} diff --git a/backend/crates/gateway-plugin/runtime/src/stream/bridge.rs b/backend/crates/gateway-plugin/runtime/src/stream/bridge.rs index 500431236..2ce6a6a79 100644 --- a/backend/crates/gateway-plugin/runtime/src/stream/bridge.rs +++ b/backend/crates/gateway-plugin/runtime/src/stream/bridge.rs @@ -2,10 +2,9 @@ use std::sync::Arc; -use tokio::{ - sync::{OwnedSemaphorePermit, mpsc, oneshot}, - time::Instant, -}; +use tokio::sync::{OwnedSemaphorePermit, mpsc, oneshot}; + +use gateway_plugin_sdk::client::CallDeadline; use crate::{RpcError, RpcReply, RpcSession, rpc::Shared}; @@ -60,7 +59,7 @@ pub struct RpcStream { pub initial: RpcReply, pub(crate) chunks: mpsc::Receiver>, pub(crate) terminal: Option>>, - pub(crate) deadline: Option, + pub(crate) deadline: CallDeadline, pub(crate) id: u64, pub(crate) shared: Arc, pub(crate) _slot: OwnedSemaphorePermit, @@ -72,9 +71,10 @@ impl RpcStream { if self.terminal.is_none() { return Ok(None); } - let result = match self.deadline { - Some(deadline) => tokio::time::timeout_at(deadline, self.chunks.recv()).await, - None => Ok(self.chunks.recv().await), + let result = tokio::select! { + biased; + () = self.deadline.wait() => Err(()), + chunk = self.chunks.recv() => Ok(chunk), }; match result { Ok(Some(chunk)) => { 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 7fd9be922..be91fbe14 100644 --- a/backend/crates/gateway-plugin/runtime/tests/generation/policy/lifecycle.rs +++ b/backend/crates/gateway-plugin/runtime/tests/generation/policy/lifecycle.rs @@ -1,4 +1,4 @@ -//! 验证跨集合复用实例时旧请求和宿主回调仍持有各自的冻结计划 +//! 验证中间件续体的调用预算、取消回收及跨集合冻结计划 use super::*; use gateway_admin::ports::plugins::PluginPreparation; @@ -124,3 +124,206 @@ async fn partial_update_preserves_inflight_middleware_and_uses_new_configuration drop(new); super::super::wait_until_empty(cache.path()).await; } + +async fn sdk_budget_runtime(stage: Stage, count: usize) -> (tempfile::TempDir, PluginRuntime) { + let worker = std::fs::read(env!("CARGO_BIN_EXE_gateway-plugin-test-middleware")).unwrap(); + let package = crate::support::package_with_contributions( + &worker, + Contributions::from([crate::support::contribution( + Capability::Middleware, + vec![stage], + vec!["openai".into()], + vec!["openai".into()], + )]), + ); + let mount = if stage == Stage::Attempt { + "attempt" + } else { + "request" + }; + let instances = ["budget-outer", "budget-inner"] + .into_iter() + .take(count) + .enumerate() + .map(|(order, id)| InstanceFixture { + id, + configuration: serde_json::json!({"mode":"map", "attempt":stage == Stage::Attempt}), + bindings: vec![binding( + MIDDLEWARE_CONTRIBUTION, + mount, + order as i32, + PluginFailurePolicy::Reject, + )], + }) + .collect(); + setup_package_with_limits( + instances, + package, + RpcLimits { + maximum_call_timeout: Duration::from_millis(300), + ..RpcLimits::default() + }, + ) + .await +} + +fn budget_request() -> MiddlewareRequest { + MiddlewareRequest::new( + "openai", + middleware_headers(), + Bytes::from_static(br#"{"input":"hello"}"#), + ) +} + +#[tokio::test] +async fn nested_sdk_middleware_waits_beyond_budget_and_can_read_response_callbacks() { + for stage in [Stage::Request, Stage::Attempt] { + let (cache, runtime) = sdk_budget_runtime(stage, 2).await; + let generation = prepare(&runtime).await; + let plan = runtime + .execution_registry() + .middleware(&generation) + .unwrap(); + let context = if stage == Stage::Attempt { + attempt_middleware_context(Arc::default()) + } else { + middleware_context_with_deadline(ClientTransport::HttpJson, Default::default()) + }; + let calls = Arc::new(AtomicUsize::new(0)); + let downstream = Downstream { + calls: calls.clone(), + closes: Arc::default(), + reads: Arc::default(), + error: false, + }; + let response = tokio::time::timeout( + Duration::from_secs(3), + plan.handle( + context, + budget_request(), + gateway_core::middleware::compose(Vec::new(), move |request| { + Box::pin(async move { + tokio::time::sleep(Duration::from_millis(900)).await; + downstream.into_next().run(request).await + }) + }), + ), + ) + .await + .unwrap() + .expect("host next must not consume either plugin's processing budget"); + let (_, status, _, mut body, _) = response.into_parts(); + assert_eq!(status, 200); + // 正文改写还会发起新的 body_read，证明 callback 有效期也随暂停恢复 + let frame = body.next_frame().await.unwrap().unwrap(); + assert_eq!( + frame.into_bytes(), + Bytes::from_static(b"{\"downstream\":true} ") + ); + assert!(body.next_frame().await.unwrap().is_none()); + body.close().await; + assert_eq!(calls.load(Ordering::Relaxed), 1); + assert!(generation.is_ready()); + drop(plan); + drop(generation); + super::super::wait_until_empty(cache.path()).await; + } +} + +struct NotifyOnDrop(Option>); + +impl Drop for NotifyOnDrop { + fn drop(&mut self) { + if let Some(sender) = self.0.take() { + let _ = sender.send(()); + } + } +} + +#[tokio::test] +async fn sdk_waiting_next_obeys_request_deadline_and_client_cancellation() { + for cancel_client in [false, true] { + let (cache, runtime) = sdk_budget_runtime(Stage::Request, 1).await; + let generation = prepare(&runtime).await; + let plan = runtime + .execution_registry() + .middleware(&generation) + .unwrap(); + let deadline = if cancel_client { + Default::default() + } else { + (SystemTime::now() + Duration::from_millis(900)).into() + }; + let context = middleware_context_with_deadline(ClientTransport::HttpJson, deadline); + let cancellation = context.cancellation().clone(); + let (started, entered) = tokio::sync::oneshot::channel(); + let (closed, dropped) = tokio::sync::oneshot::channel(); + let pending = tokio::spawn(async move { + plan.handle( + context, + budget_request(), + gateway_core::middleware::compose(Vec::new(), move |_| { + Box::pin(async move { + let _closed = NotifyOnDrop(Some(closed)); + let _ = started.send(()); + futures::future::pending().await + }) + }), + ) + .await + }); + tokio::time::timeout(Duration::from_secs(2), entered) + .await + .unwrap() + .unwrap(); + tokio::time::sleep(Duration::from_millis(600)).await; + assert!( + !pending.is_finished(), + "plugin budget must stay paused while waiting for next" + ); + if cancel_client { + cancellation.cancel(); + } + let result = tokio::time::timeout(Duration::from_secs(2), pending) + .await + .unwrap() + .unwrap(); + match result { + Err(MiddlewareError::Fault) if cancel_client => {} + Err(MiddlewareError::Gateway(error)) if !cancel_client => { + assert_eq!(error.kind(), gateway_core::error::GatewayErrorKind::Timeout); + } + _ => panic!("request termination must end the suspended middleware"), + } + tokio::time::timeout(Duration::from_secs(2), dropped) + .await + .unwrap() + .unwrap(); + // 取消单次调用后，同一进程仍能完成新的中间件及正文回调 + let plan = runtime + .execution_registry() + .middleware(&generation) + .unwrap(); + let response = plan + .handle( + middleware_context(ClientTransport::HttpJson), + budget_request(), + Downstream { + calls: Arc::default(), + closes: Arc::default(), + reads: Arc::default(), + error: false, + } + .into_next(), + ) + .await + .unwrap(); + let (_, _, _, mut body, _) = response.into_parts(); + assert!(body.next_frame().await.unwrap().is_some()); + body.close().await; + assert!(generation.is_ready()); + drop(plan); + drop(generation); + super::super::wait_until_empty(cache.path()).await; + } +} diff --git a/backend/crates/gateway-plugin/runtime/tests/generation/policy/mod.rs b/backend/crates/gateway-plugin/runtime/tests/generation/policy/mod.rs index 69707f303..9c31d7164 100644 --- a/backend/crates/gateway-plugin/runtime/tests/generation/policy/mod.rs +++ b/backend/crates/gateway-plugin/runtime/tests/generation/policy/mod.rs @@ -681,6 +681,16 @@ impl Downstream { } fn middleware_context(transport: ClientTransport) -> MiddlewareContext { + middleware_context_with_deadline( + transport, + (SystemTime::now() + Duration::from_secs(5)).into(), + ) +} + +fn middleware_context_with_deadline( + transport: ClientTransport, + deadline: gateway_core::lifecycle::Deadline, +) -> MiddlewareContext { MiddlewareContext::new( MiddlewareTarget { request_id: ModelRequestId::new("req_middleware").unwrap(), @@ -697,7 +707,7 @@ fn middleware_context(transport: ClientTransport) -> MiddlewareContext { client_key_id: ClientApiKeyId::new("key-middleware").unwrap(), account_group_ids: Arc::<[AccountGroupId]>::from([]), cancellation: CancellationToken::new(), - deadline: (SystemTime::now() + Duration::from_secs(5)).into(), + deadline, extension_scope: ExtensionCallScope::default(), execution_effects: None, }, diff --git a/backend/crates/gateway-plugin/runtime/tests/rpc/session.rs b/backend/crates/gateway-plugin/runtime/tests/rpc/session.rs index 2ca74a003..4d7266077 100644 --- a/backend/crates/gateway-plugin/runtime/tests/rpc/session.rs +++ b/backend/crates/gateway-plugin/runtime/tests/rpc/session.rs @@ -97,7 +97,7 @@ impl CallbackHandler for BlockingCallbacks { } impl CallbackHandler for LifecycleCallbacks { - fn begin(&self, _context: &CallContext) { + fn begin(&self, _context: &CallContext, _deadline: &gateway_plugin_sdk::client::CallDeadline) { self.begun.fetch_add(1, Ordering::Relaxed); } @@ -122,7 +122,7 @@ impl CallbackHandler for LifecycleCallbacks { } impl CallbackHandler for GatedFirstBegin { - fn begin(&self, context: &CallContext) { + fn begin(&self, context: &CallContext, _deadline: &gateway_plugin_sdk::client::CallDeadline) { if context.call_id == 1 { self.entered.notify_one(); let released = self.released.lock().unwrap(); @@ -865,3 +865,53 @@ async fn acknowledged_timeout_does_not_interrupt_another_call() { assert!(session.is_ready()); session.shutdown(Duration::from_secs(1)).await; } + +struct SlowNextCallbacks; + +impl CallbackHandler for SlowNextCallbacks { + fn call( + &self, + _context: CallContext, + method: String, + params: serde_json::Value, + payload: Vec, + ) -> BoxFuture<'static, Result> { + Box::pin(async move { + assert_eq!(method, gateway_plugin_sdk::call::middleware::NEXT_METHOD); + tokio::time::sleep(Duration::from_millis(600)).await; + Ok(RpcReply { + result: params, + payload, + }) + }) + } +} + +#[tokio::test] +async fn managed_next_suspends_only_its_parent_and_not_a_concurrent_call() { + let (_cache, session) = session(Arc::new(SlowNextCallbacks)).await; + let delegated = session.call( + "callback_method", + session.context(Stage::Request, Duration::from_millis(200)), + json!({"method": gateway_plugin_sdk::call::middleware::NEXT_METHOD}), + Vec::new(), + ); + let hanging = session.call( + "hang", + session.context(Stage::Request, Duration::from_millis(100)), + json!({}), + Vec::new(), + ); + let (delegated, hanging) = tokio::time::timeout(Duration::from_secs(2), async { + tokio::join!(delegated, hanging) + }) + .await + .unwrap(); + assert!( + delegated.is_ok(), + "host waiting for next must outlive its processing budget" + ); + assert!(matches!(hanging, Err(RpcError::Timeout))); + assert!(session.is_ready()); + session.shutdown(Duration::from_secs(1)).await; +} diff --git a/backend/crates/gateway-plugin/sdk/Cargo.toml b/backend/crates/gateway-plugin/sdk/Cargo.toml index 0d5bd92f5..4900ed5d7 100644 --- a/backend/crates/gateway-plugin/sdk/Cargo.toml +++ b/backend/crates/gateway-plugin/sdk/Cargo.toml @@ -19,7 +19,7 @@ thiserror = "2.0.18" tokio = { version = "1.52.3", optional = true, default-features = false, features = ["io-util", "macros", "rt", "sync", "time"] } [dev-dependencies] -tokio = { version = "1.52.3", default-features = false, features = ["io-util", "macros", "rt", "sync", "time"] } +tokio = { version = "1.52.3", default-features = false, features = ["io-util", "macros", "rt", "sync", "time", "test-util"] } [[test]] name = "main" diff --git a/backend/crates/gateway-plugin/sdk/docs/capabilities.md b/backend/crates/gateway-plugin/sdk/docs/capabilities.md index 2ceb02f75..9b1055e72 100644 --- a/backend/crates/gateway-plugin/sdk/docs/capabilities.md +++ b/backend/crates/gateway-plugin/sdk/docs/capabilities.md @@ -242,6 +242,11 @@ SDK 不自动重试。超时或断连不能证明写入未提交；重试上限 `PluginBuilder::middleware` 与 `MiddlewarePlugin::new` 共用 `MiddlewareInput` / `MiddlewareOutput`。 `next.run(input)` 消费自身，最多执行一次；结果按相反顺序经过外层。重试或独立子请求通过宿主执行服务发起 +调用上下文的 `timeout_ms` 限制插件自身处理时间，默认上限为 120 秒。等待宿主 `next` 时暂停这份预算， +收到结果或错误后恢复剩余时间；入站处理与返回处理合计使用同一预算，普通宿主回调不会暂停计时。 +请求总期限、下游操作超时和取消信号独立生效，客户端取消或插件退出仍会结束等待并回收资源。 +宿主与插件 SDK 必须同时支持此语义；仍使用固定绝对期限的 SDK 构建的插件需重新构建，单独升级宿主不能解除插件侧期限 + HTTP 总入口统一包裹原生路由，`dispatch_http` 复用同一路由与组合能力；无需为每个 HTTP 接口登记新的挂载点。 `service` 用于不经过 HTTP 的类型化主动调用，具体公开范围见下节 diff --git a/backend/crates/gateway-plugin/sdk/src/client/mod.rs b/backend/crates/gateway-plugin/sdk/src/client/mod.rs index 53ba30375..4c63b16a6 100644 --- a/backend/crates/gateway-plugin/sdk/src/client/mod.rs +++ b/backend/crates/gateway-plugin/sdk/src/client/mod.rs @@ -28,9 +28,9 @@ pub use plugin::{ AuthorError, ComposedPlugin, Empty, Method, PluginBuilder, TypedCall, TypedReply, methods, }; pub use session::{ - CallCancellation, CallFuture, CallReply, HostClient, HostReply, PluginCall, PluginHandler, - PluginSession, PullResponseFuture, PullResponseStream, ResponseStream, SessionConfig, - SessionError, StreamSender, + CallCancellation, CallDeadline, CallDeadlinePause, CallFuture, CallReply, HostClient, + HostReply, PluginCall, PluginHandler, PluginSession, PullResponseFuture, PullResponseStream, + ResponseStream, SessionConfig, SessionError, StreamSender, }; pub use upstream_adapter::{UpstreamWebSocket, UpstreamWebSocketUpgrade}; diff --git a/backend/crates/gateway-plugin/sdk/src/client/session/credit.rs b/backend/crates/gateway-plugin/sdk/src/client/session/credit.rs index 8e6810364..520edf1fb 100644 --- a/backend/crates/gateway-plugin/sdk/src/client/session/credit.rs +++ b/backend/crates/gateway-plugin/sdk/src/client/session/credit.rs @@ -2,9 +2,7 @@ use std::sync::Mutex; -use tokio::time::Instant; - -use super::{CallCancellation, SessionError, lifecycle::wait_deadline}; +use super::{CallCancellation, CallDeadline, SessionError}; #[derive(Default)] pub(super) struct CreditWindow { @@ -53,7 +51,7 @@ impl CreditWindow { pub(super) async fn initial_window( &self, - deadline: Option, + deadline: &CallDeadline, cancellation: &CallCancellation, ) -> Result { loop { @@ -70,7 +68,7 @@ impl CreditWindow { tokio::select! { biased; () = cancellation.cancelled() => return Err(SessionError::Cancelled), - () = wait_deadline(deadline) => return Err(SessionError::Timeout), + () = deadline.wait() => return Err(SessionError::Timeout), () = changed => {} } } @@ -79,7 +77,7 @@ impl CreditWindow { pub(super) async fn take( &self, bytes: u64, - deadline: Option, + deadline: &CallDeadline, cancellation: &CallCancellation, ) -> Result<(), SessionError> { loop { @@ -101,7 +99,7 @@ impl CreditWindow { tokio::select! { biased; () = cancellation.cancelled() => return Err(SessionError::Cancelled), - () = wait_deadline(deadline) => return Err(SessionError::Timeout), + () = deadline.wait() => return Err(SessionError::Timeout), () = changed => {} } } diff --git a/backend/crates/gateway-plugin/sdk/src/client/session/deadline.rs b/backend/crates/gateway-plugin/sdk/src/client/session/deadline.rs new file mode 100644 index 000000000..b7e6ea324 --- /dev/null +++ b/backend/crates/gateway-plugin/sdk/src/client/session/deadline.rs @@ -0,0 +1,106 @@ +//! 宿主与插件共用的调用处理预算，受管续体等待期间暂停并保留剩余额度 + +use std::time::Duration; + +use tokio::{sync::watch, time::Instant}; + +/// 同一次调用的所有等待点共享此期限，避免回调恢复后仍使用旧截止时间 +#[derive(Clone)] +pub struct CallDeadline { + state: watch::Sender, +} + +#[derive(Clone, Copy)] +enum State { + Running(Instant), + Paused(Duration), + Unlimited, +} + +/// 随受管续体释放而恢复计时；取消和错误出口同样归还暂停状态 +pub struct CallDeadlinePause { + deadline: CallDeadline, +} + +impl CallDeadline { + #[must_use] + pub fn new(deadline: Instant) -> Self { + let (state, _) = watch::channel(State::Running(deadline)); + Self { state } + } + + #[must_use] + pub fn remaining(&self) -> Option { + match *self.state.borrow() { + State::Running(at) => Some(at.saturating_duration_since(Instant::now())), + State::Paused(remaining) => Some(remaining), + State::Unlimited => None, + } + } + + #[must_use] + pub fn is_elapsed(&self) -> bool { + self.remaining() + .is_some_and(|remaining| remaining.is_zero()) + } + + /// 只有尚未耗尽的运行预算可以暂停；重复 next 不能重置或复活预算 + #[must_use] + pub fn pause(&self) -> Option { + self.state + .send_if_modified(|state| { + let State::Running(at) = *state else { + return false; + }; + let remaining = at.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return false; + } + *state = State::Paused(remaining); + true + }) + .then(|| CallDeadlinePause { + deadline: self.clone(), + }) + } + + /// 初始资源流响应已交付，后续期限由资源 owner 持有 + pub fn clear(&self) { + self.state.send_replace(State::Unlimited); + } + + pub async fn wait(&self) { + let mut state = self.state.subscribe(); + loop { + let current = *state.borrow_and_update(); + let elapsed = async { + match current { + State::Running(at) => tokio::time::sleep_until(at).await, + State::Paused(_) | State::Unlimited => std::future::pending().await, + } + }; + tokio::select! { + biased; + _ = state.changed() => {} + () = elapsed => { + // 暂停可与旧计时器唤醒并发，返回前重新核对当前预算 + if self.is_elapsed() { + return; + } + } + } + } + } +} + +impl Drop for CallDeadlinePause { + fn drop(&mut self) { + self.deadline.state.send_if_modified(|state| { + let State::Paused(remaining) = *state else { + return false; + }; + *state = State::Running(Instant::now() + remaining); + true + }); + } +} 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 0c850465b..5a09df6d8 100644 --- a/backend/crates/gateway-plugin/sdk/src/client/session/driver.rs +++ b/backend/crates/gateway-plugin/sdk/src/client/session/driver.rs @@ -16,9 +16,10 @@ use crate::{ use super::super::frame::{read_frame, validate_frame, write_frame}; use super::{ - CallCancellation, HostClient, PluginCall, PluginHandler, SessionConfig, SessionError, + CallCancellation, CallDeadline, HostClient, PluginCall, PluginHandler, SessionConfig, + SessionError, credit::CreditWindow, - lifecycle::{CancellationSource, LifecycleHooks, wait_deadline}, + lifecycle::{CancellationSource, LifecycleHooks}, output::{Output, writer_loop}, registry::CallbackRegistry, stream::validate_stream_chunk, @@ -340,13 +341,12 @@ impl SessionDriver { let deadline = Instant::now() .checked_add(Duration::from_millis(context.timeout_ms)) .ok_or(SessionError::Protocol)?; - let deadline = Some(deadline); - let (deadline_source, callback_deadline) = tokio::sync::watch::channel(deadline); + let deadline = CallDeadline::new(deadline); let credits = Arc::new(CreditWindow::default()); let (cancellation, cancellation_signal) = CancellationSource::new(); let host = HostClient { parent_id: id, - deadline: callback_deadline, + deadline: deadline.clone(), cancellation: cancellation_signal.clone(), callbacks: Arc::clone(&self.callbacks), output: self.output.clone(), @@ -379,7 +379,6 @@ impl SessionDriver { output, lifecycle, deadline, - deadline_source, maximum_stream_chunk_bytes, maximum_chunks, ) @@ -453,8 +452,7 @@ async fn run_call( credits: Arc, output: Output, lifecycle: LifecycleHooks, - deadline: Option, - deadline_source: tokio::sync::watch::Sender>, + deadline: CallDeadline, maximum_stream_chunk_bytes: usize, maximum_chunks: usize, ) -> Result<(), SessionError> { @@ -463,7 +461,7 @@ async fn run_call( let response = tokio::select! { biased; () = cancellation.cancelled() => return Ok(()), - () = wait_deadline(deadline) => { + () = deadline.wait() => { lifecycle.cancel(context.clone()); output.send_control(fault_frame( id, @@ -481,7 +479,7 @@ async fn run_call( bounded_fault_frame(id, error), &context, &lifecycle, - deadline, + &deadline, &cancellation, ) .await?; @@ -511,13 +509,13 @@ async fn run_call( initial, &context, &lifecycle, - deadline, + &deadline, &cancellation, ) .await?; return Ok(()); }; - let window_bytes = match credits.initial_window(deadline, &cancellation).await { + let window_bytes = match credits.initial_window(&deadline, &cancellation).await { Ok(window) => window, Err(SessionError::Timeout) => { lifecycle.cancel(context.clone()); @@ -541,25 +539,22 @@ async fn run_call( initial, &context, &lifecycle, - deadline, + &deadline, &cancellation, ) .await? { return Ok(()); } - let deadline = if context.resource_stream { - None - } else { - deadline - }; - deadline_source.send_replace(deadline); + if context.resource_stream { + deadline.clear(); + } let mut sequence = 0_u64; loop { let item = tokio::select! { biased; () = cancellation.cancelled() => return Ok(()), - () = wait_deadline(deadline) => { + () = deadline.wait() => { lifecycle.cancel(context.clone()); output.send_control(end_frame( id, @@ -587,7 +582,7 @@ async fn run_call( return Ok(()); } match credits - .take(payload.len() as u64, deadline, &cancellation) + .take(payload.len() as u64, &deadline, &cancellation) .await { Ok(()) => {} @@ -615,7 +610,7 @@ async fn run_call( id, &context, &lifecycle, - deadline, + &deadline, &cancellation, ) .await? @@ -631,7 +626,7 @@ async fn send_initial( frame: Frame, context: &CallContext, lifecycle: &LifecycleHooks, - deadline: Option, + deadline: &CallDeadline, cancellation: &CallCancellation, ) -> Result { match output.send_data_until(frame, deadline, cancellation).await { @@ -655,7 +650,7 @@ async fn send_stream_data( id: u64, context: &CallContext, lifecycle: &LifecycleHooks, - deadline: Option, + deadline: &CallDeadline, cancellation: &CallCancellation, ) -> Result { match output.send_data_until(frame, deadline, cancellation).await { diff --git a/backend/crates/gateway-plugin/sdk/src/client/session/lifecycle.rs b/backend/crates/gateway-plugin/sdk/src/client/session/lifecycle.rs index fefc65cdb..99a11ccd0 100644 --- a/backend/crates/gateway-plugin/sdk/src/client/session/lifecycle.rs +++ b/backend/crates/gateway-plugin/sdk/src/client/session/lifecycle.rs @@ -2,7 +2,7 @@ use std::sync::Arc; -use tokio::{sync::watch, time::Instant}; +use tokio::sync::watch; use crate::CallContext; @@ -95,22 +95,3 @@ impl LifecycleHooks { let _ = self.events.try_send(LifecycleEvent::Shutdown); } } - -pub(super) async fn wait_deadline(deadline: Option) { - match deadline { - Some(deadline) => tokio::time::sleep_until(deadline).await, - None => std::future::pending().await, - } -} - -pub(super) async fn callback_deadline(mut deadline: tokio::sync::watch::Receiver>) { - loop { - let current = *deadline.borrow_and_update(); - tokio::select! { - () = wait_deadline(current) => return, - changed = deadline.changed() => { - if changed.is_err() { return wait_deadline(current).await; } - } - } - } -} diff --git a/backend/crates/gateway-plugin/sdk/src/client/session/mod.rs b/backend/crates/gateway-plugin/sdk/src/client/session/mod.rs index 7e7b5f62b..d400e0794 100644 --- a/backend/crates/gateway-plugin/sdk/src/client/session/mod.rs +++ b/backend/crates/gateway-plugin/sdk/src/client/session/mod.rs @@ -3,6 +3,7 @@ mod call; mod config; mod credit; +mod deadline; mod driver; mod error; mod lifecycle; @@ -12,6 +13,7 @@ mod stream; pub use call::{CallFuture, CallReply, PluginCall, PluginHandler}; pub use config::SessionConfig; +pub use deadline::{CallDeadline, CallDeadlinePause}; pub use driver::PluginSession; pub use error::SessionError; pub use lifecycle::CallCancellation; diff --git a/backend/crates/gateway-plugin/sdk/src/client/session/output.rs b/backend/crates/gateway-plugin/sdk/src/client/session/output.rs index aa2a7cffb..9486238ae 100644 --- a/backend/crates/gateway-plugin/sdk/src/client/session/output.rs +++ b/backend/crates/gateway-plugin/sdk/src/client/session/output.rs @@ -5,13 +5,12 @@ use std::sync::Arc; use tokio::{ io::AsyncWrite, sync::{OwnedSemaphorePermit, Semaphore, mpsc}, - time::Instant, }; use crate::Frame; use super::super::frame::{validate_frame, write_frame}; -use super::{CallCancellation, SessionError, lifecycle::wait_deadline}; +use super::{CallCancellation, CallDeadline, SessionError}; pub(super) struct Output { pub(super) sender: mpsc::Sender, @@ -46,13 +45,13 @@ impl Output { pub(super) async fn send_data_until( &self, frame: Frame, - deadline: Option, + deadline: &CallDeadline, cancellation: &CallCancellation, ) -> Result<(), SessionError> { tokio::select! { biased; () = cancellation.cancelled() => Err(SessionError::Cancelled), - () = wait_deadline(deadline) => Err(SessionError::Timeout), + () = deadline.wait() => Err(SessionError::Timeout), result = self.send_data(frame) => result, } } 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 cfe11e86e..5d5c9166b 100644 --- a/backend/crates/gateway-plugin/sdk/src/client/session/registry.rs +++ b/backend/crates/gateway-plugin/sdk/src/client/session/registry.rs @@ -6,13 +6,13 @@ use std::{ }; use serde_json::Value; -use tokio::{sync::oneshot, time::Instant}; +use tokio::sync::oneshot; -use crate::{Frame, Message}; +use crate::{Frame, Message, call::middleware::NEXT_METHOD}; use super::super::frame::validate_frame; use super::{ - CallCancellation, SessionError, credit::CreditWindow, lifecycle::callback_deadline, + CallCancellation, CallDeadline, CallDeadlinePause, SessionError, credit::CreditWindow, output::Output, }; @@ -26,7 +26,7 @@ pub struct HostReply { #[derive(Clone)] pub struct HostClient { pub(super) parent_id: u64, - pub(super) deadline: tokio::sync::watch::Receiver>, + pub(super) deadline: CallDeadline, pub(super) cancellation: CallCancellation, pub(super) callbacks: Arc, pub(super) output: Output, @@ -60,7 +60,13 @@ impl HostClient { if method.is_empty() || method.len() > 128 { return Err(SessionError::Protocol); } - let ordered = self.callbacks.order.lock().await; + let ordered = tokio::select! { + biased; + () = self.cancellation.cancelled() => return Err(SessionError::Cancelled), + () = self.deadline.wait() => return Err(SessionError::Timeout), + ordered = self.callbacks.order.lock() => ordered, + }; + let managed_next = method == NEXT_METHOD; let (id, received) = self.callbacks.begin(self.parent_id)?; let frame = Frame { message: Message::Callback { @@ -78,13 +84,16 @@ impl HostClient { let sent = tokio::select! { biased; () = self.cancellation.cancelled() => Err(SessionError::Cancelled), - () = callback_deadline(self.deadline.clone()) => Err(SessionError::Timeout), + () = self.deadline.wait() => Err(SessionError::Timeout), result = self.output.send_data(frame) => result, }; if let Err(error) = sent { self.callbacks.retire(id); return Err(error); } + if managed_next { + self.callbacks.pause_next(id, &self.deadline); + } drop(ordered); tokio::select! { biased; @@ -92,7 +101,7 @@ impl HostClient { self.callbacks.retire(id); Err(SessionError::Cancelled) } - () = callback_deadline(self.deadline.clone()) => { + () = self.deadline.wait() => { self.callbacks.retire(id); Err(SessionError::Timeout) } @@ -117,6 +126,7 @@ struct CallbackState { struct PendingCallback { parent_id: u64, + pause: Option, result: oneshot::Sender>, } @@ -149,12 +159,27 @@ impl CallbackRegistry { let id = state.next_id; state.next_id = state.next_id.checked_add(2).ok_or(SessionError::Capacity)?; let (result, received) = oneshot::channel(); - state - .pending - .insert(id, PendingCallback { parent_id, result }); + state.pending.insert( + id, + PendingCallback { + parent_id, + result, + pause: None, + }, + ); Ok((id, received)) } + fn pause_next(&self, id: u64, deadline: &CallDeadline) { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(pending) = state.pending.get_mut(&id) { + pending.pause = deadline.pause(); + } + } + pub(super) fn resolve(&self, frame: Frame) -> Result<(), SessionError> { let id = match &frame.message { Message::Result { id, .. } | Message::Error { id, .. } => *id, @@ -174,7 +199,7 @@ impl CallbackRegistry { return Err(SessionError::Protocol); } }; - let Some(pending) = pending else { + let Some(mut pending) = pending else { return Ok(()); }; let result = match frame.message { @@ -187,6 +212,8 @@ impl CallbackRegistry { } _ => return Err(SessionError::Protocol), }; + // 收到宿主结果就恢复预算，不依赖插件再次轮询 next future + pending.pause.take(); let _ = pending.result.send(result); Ok(()) } diff --git a/backend/crates/gateway-plugin/sdk/src/context.rs b/backend/crates/gateway-plugin/sdk/src/context.rs index 4417f9ca4..62df486ca 100644 --- a/backend/crates/gateway-plugin/sdk/src/context.rs +++ b/backend/crates/gateway-plugin/sdk/src/context.rs @@ -41,6 +41,7 @@ pub struct CallContext { pub generation: u64, pub incarnation: String, pub stage: Stage, + /// 插件处理预算；等待宿主 middleware next 时暂停，请求总期限由宿主单独执行 pub timeout_ms: u64, /// 首个流式结果返回后由资源 owner 控制取消，不再沿用初始调用期限 pub resource_stream: bool, diff --git a/backend/crates/gateway-plugin/sdk/tests/client/session/deadline.rs b/backend/crates/gateway-plugin/sdk/tests/client/session/deadline.rs new file mode 100644 index 000000000..8aaafc6cd --- /dev/null +++ b/backend/crates/gateway-plugin/sdk/tests/client/session/deadline.rs @@ -0,0 +1,52 @@ +//! 验证续体等待不计入预算，恢复、到期和资源流切换不会重置已消耗时间 + +use std::time::Duration; + +use gateway_plugin_sdk::client::CallDeadline; +use tokio::time::{Instant, advance}; + +#[tokio::test(start_paused = true)] +async fn suspension_preserves_remaining_budget_and_wakes_existing_waiters() { + let deadline = CallDeadline::new(Instant::now() + Duration::from_secs(10)); + let waiter = tokio::spawn({ + let deadline = deadline.clone(); + async move { deadline.wait().await } + }); + tokio::task::yield_now().await; + advance(Duration::from_secs(3)).await; + let pause = deadline.pause().unwrap(); + advance(Duration::from_secs(30)).await; + assert!(!waiter.is_finished()); + assert_eq!(deadline.remaining(), Some(Duration::from_secs(7))); + drop(pause); + advance(Duration::from_secs(6)).await; + assert!(!waiter.is_finished()); + advance(Duration::from_secs(1)).await; + waiter.await.unwrap(); + assert!(deadline.is_elapsed()); +} + +#[tokio::test(start_paused = true)] +async fn exhausted_budget_cannot_be_suspended_or_revived() { + let deadline = CallDeadline::new(Instant::now() + Duration::from_secs(1)); + advance(Duration::from_secs(1)).await; + assert!(deadline.pause().is_none()); + deadline.wait().await; + assert!(deadline.is_elapsed()); +} + +#[tokio::test(start_paused = true)] +async fn resource_stream_clear_survives_an_outstanding_pause_guard() { + let deadline = CallDeadline::new(Instant::now() + Duration::from_secs(1)); + let pause = deadline.pause().unwrap(); + assert!(deadline.pause().is_none()); + deadline.clear(); + drop(pause); + advance(Duration::from_secs(10)).await; + assert_eq!(deadline.remaining(), None); + assert!( + tokio::time::timeout(Duration::from_secs(1), deadline.wait()) + .await + .is_err() + ); +} diff --git a/backend/crates/gateway-plugin/sdk/tests/client/session/lifecycle.rs b/backend/crates/gateway-plugin/sdk/tests/client/session/lifecycle.rs index edfd2e860..6cae0d80c 100644 --- a/backend/crates/gateway-plugin/sdk/tests/client/session/lifecycle.rs +++ b/backend/crates/gateway-plugin/sdk/tests/client/session/lifecycle.rs @@ -110,3 +110,203 @@ async fn quiesce_rejects_new_calls_and_shutdown_closes_the_session() { shutdown(&mut host, task).await; } + +async fn send_middleware_call(host: &mut HostPeer, id: u64, resource_stream: bool) { + let mut context = middleware_context(id); + context.resource_stream = resource_stream; + write_frame( + &mut host.writer, + &Frame { + message: Message::Call { + id, + method: HANDLE_METHOD.into(), + context, + params: serde_json::to_value(middleware_request()).unwrap(), + }, + payload: Vec::new(), + }, + ) + .await + .unwrap(); + send_control( + host, + Message::Credit { + id, + bytes: 1024, + frames: 2, + }, + ) + .await; +} + +async fn receive_next(host: &mut HostPeer, parent: u64) -> u64 { + let Message::Callback { + id, + parent_id, + method, + .. + } = receive(host).await.message + else { + panic!("expected middleware next") + }; + assert_eq!(parent_id, parent); + assert_eq!(method, NEXT_METHOD); + id +} + +async fn reply_next(host: &mut HostPeer, id: u64) { + send_control( + host, + Message::Result { + id, + result: serde_json::to_value(MiddlewareNextResponse { + metadata: None, + response: format!("response-{id}"), + protocol: "openai".into(), + status: 200, + headers: Vec::new(), + body: None, + }) + .unwrap(), + }, + ) + .await; +} + +#[tokio::test(start_paused = true)] +async fn managed_next_can_outlive_the_initial_budget_with_or_without_resource_stream() { + for resource_stream in [false, true] { + let (mut host, task) = start_session(middleware_plugin(false)).await; + send_middleware_call(&mut host, 1, resource_stream).await; + let next = receive_next(&mut host, 1).await; + tokio::time::advance(Duration::from_secs(10)).await; + reply_next(&mut host, next).await; + assert!(matches!( + receive(&mut host).await.message, + Message::Result { id: 1, .. } + )); + assert!(matches!( + receive(&mut host).await.message, + Message::End { id: 1, error: None } + )); + shutdown(&mut host, task).await; + } +} + +#[tokio::test(start_paused = true)] +async fn response_processing_uses_only_the_budget_left_before_next() { + let plugin = MiddlewarePlugin::new( + &middleware_contributions(), + |call: RequestCall| async move { + tokio::time::sleep(Duration::from_millis(200)).await; + let response = call.next.run(call.request).await?; + tokio::time::sleep(Duration::from_millis(850)).await; + Ok(response) + }, + ) + .unwrap(); + let (mut host, task) = start_session(plugin).await; + send_middleware_call(&mut host, 1, false).await; + let next = receive_next(&mut host, 1).await; + tokio::time::advance(Duration::from_secs(10)).await; + reply_next(&mut host, next).await; + assert!(matches!( + receive(&mut host).await.message, + Message::Error { + id: 1, + error: PluginFault { + code: ErrorCode::Timeout, + .. + } + } + )); + shutdown(&mut host, task).await; +} + +#[tokio::test(start_paused = true)] +async fn waiting_next_does_not_suspend_an_independent_handlers_budget() { + let plugin = MiddlewarePlugin::new( + &middleware_contributions(), + |call: RequestCall| async move { + if call.context.call_id == 3 { + return pending().await; + } + call.next.run(call.request).await + }, + ) + .unwrap(); + let (mut host, task) = start_session(plugin).await; + send_middleware_call(&mut host, 1, false).await; + let next = receive_next(&mut host, 1).await; + send_middleware_call(&mut host, 3, false).await; + // 先让第二个 Call 进入 SDK，再推进到其预算之外 + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_secs(2)).await; + assert!(matches!( + receive(&mut host).await.message, + Message::Error { + id: 3, + error: PluginFault { + code: ErrorCode::Timeout, + .. + } + } + )); + reply_next(&mut host, next).await; + assert!(matches!( + receive(&mut host).await.message, + Message::Result { id: 1, .. } + )); + shutdown(&mut host, task).await; +} + +#[tokio::test(start_paused = true)] +async fn cancelling_managed_next_retires_its_late_reply_and_keeps_session_available() { + let (mut host, task) = start_session(middleware_plugin(false)).await; + send_middleware_call(&mut host, 1, false).await; + let next = receive_next(&mut host, 1).await; + tokio::time::advance(Duration::from_secs(10)).await; + send_control(&mut host, Message::Cancel { id: 1 }).await; + assert!(matches!( + receive(&mut host).await.message, + Message::Cancelled { id: 1 } + )); + reply_next(&mut host, next).await; + send_middleware_call(&mut host, 3, false).await; + let next = receive_next(&mut host, 3).await; + reply_next(&mut host, next).await; + assert!(matches!( + receive(&mut host).await.message, + Message::Result { id: 3, .. } + )); + shutdown(&mut host, task).await; +} + +#[tokio::test(start_paused = true)] +async fn ordinary_callbacks_still_consume_the_parent_budget() { + let (mut host, task) = start_session(TestHandler::default()).await; + send_call_with_timeout( + &mut host, + 1, + "callback", + json!({}), + Vec::new(), + Duration::from_millis(100), + ) + .await; + assert!(matches!( + receive(&mut host).await.message, + Message::Callback { parent_id: 1, .. } + )); + assert!(matches!( + receive(&mut host).await.message, + Message::Error { + id: 1, + error: PluginFault { + code: ErrorCode::Timeout, + .. + } + } + )); + shutdown(&mut host, task).await; +} 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 5b86d9bec..ad35ddbb1 100644 --- a/backend/crates/gateway-plugin/sdk/tests/client/session/mod.rs +++ b/backend/crates/gateway-plugin/sdk/tests/client/session/mod.rs @@ -1,6 +1,7 @@ //! 插件会话测试入口与双工传输、处理器和宿主回调夹具 mod credit; +mod deadline; mod driver; mod lifecycle; mod registry; diff --git a/docs/architecture.md b/docs/architecture.md index 73c254cfc..f8d577414 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -197,6 +197,7 @@ flowchart LR #### 调用资源与状态 - **资源跟随调用**：账号、HTTP、模型、状态和日志回调绑定有效父调用，取消随调用传播。受管流建立后不沿用握手的绝对期限，网络操作与模型执行各自计时。 + Runtime 与 SDK 的中间件处理预算在宿主 `next` 等待期间暂停，返回后恢复剩余预算；调用等待与回调有效期共用各端的预算状态，请求总期限独立生效。 显式 Key 的身份快照随调用保留，每个模型执行拥有独立子调用图与结算；省略 Key 时继承现有模型父请求。页面 Responses 桥复核精确实例目标和 revision `frontend_authentication` 使用 principal 到 Key 的映射；公开登录回调的一次性票据用于关联流程，不限制插件回调能力。接口合同见 [SDK 回调](../backend/crates/gateway-plugin/sdk/docs/capabilities.md#宿主资源回调) - **观察合同统一**：`observer` 通过一个类型化 RPC 入口接收完成与实际上游 WebSocket 事件，发布时按事件类型编译独立作用域。完成事件直接携带终态、用量、费用和耗时，不按能力声明裁剪字段；异步投递的并发、队列和载荷有界，观察失败不改变业务结果 @@ -375,7 +376,11 @@ Client Key 鉴权完成后，API adapter 从有界请求头识别客户端，Cor - 一个客户端请求对应一条 `model_requests`；attempt 是请求内事实，不建立第二张权威表 - Provider 的一次 `execute` 只选择一个 credential 并返回一个冷流；换号、重试和 fallback 由 Core 决定 +- OpenAI 的自动恢复限于明确识别的账号、容量、续接和传输场景；普通上游拒绝保留响应与重试提示， + 交由客户端决策。WS opening 尚未发送业务 payload 也不扩大该恢复范围，明确的 `426` 仍允许 HTTP 回退 - `not_sent`、`sent`、`ambiguous` 是单调的上游发送边界；结果不明确时不能假定上游未收到请求 +- 明确未发送的瞬时建连失败（含 TLS 握手 EOF）共享最多 4 次 opening、30 秒的请求内预算； + 证书、TLS 协议及配置错误不进入同账号建连恢复，取消和执行 deadline 优先终止等待 - downstream commit 记录不可撤回的首次交付；不可重放事件交付后禁止换号、重试和 fallback 原生 Responses Lite WebSocket 及时交付正常事件，仅 Provider 明确标记的可替换前导保留重试资格；Core 单独维护重放屏障， 仍检查错误分类、续接归属、发送状态、副作用及预算。中间件链不启用此前导例外