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 单独维护重放屏障, 仍检查错误分类、续接归属、发送状态、副作用及预算。中间件链不启用此前导例外