starred/codex-proxy-rs
1
0
Fork 0
mirror of https://github.com/zyycn/codex-proxy-rs.git synced 2026-11-06 08:06:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(plugins): exclude managed next waits from RPC budgets (#397)

Browse Source 
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 <noreply@openai.com>
This commit is contained in:
Zyy
2026-10-12 11:40:01 +08:00
committed by GitHub
co-authored by Codex
parent eea16f5275
commit 94c769fc9b
24 changed files with 815 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/crates/gateway-plugin/runtime/src/adapter/middleware/request.rs
+9 -3
View File
@@ -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,
backend/crates/gateway-plugin/runtime/src/callback/call_resources.rs
+11 -18
View File
@@ -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<CallbackScope>,
@@ -26,11 +25,12 @@ pub(super) struct CallResources {
impl CallResources {
pub(super) fn new(
context: &CallContext,
deadline: &CallDeadline,
scope: Arc<CallbackScope>,
middleware: Option<Arc<dyn MiddlewareCallback>>,
) -> 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<Arc<dyn MiddlewareCallback>>,
pub(super) streams: BTreeMap<String, Arc<HttpStream>>,
pub(super) model_streams: BTreeMap<String, Arc<model::ModelStream>>,
backend/crates/gateway-plugin/runtime/src/callback/mod.rs
+4 -16
View File
@@ -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" {
backend/crates/gateway-plugin/runtime/src/rpc/dispatch.rs
+5
View File
@@ -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<R: AsyncRead + Unpin>(
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 {
backend/crates/gateway-plugin/runtime/src/rpc/session.rs
+69 -11
View File
@@ -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<dyn CallbackHandler>,
transmitted: bool,
cancellation: Option<RpcError>,
@@ -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<CallDeadlinePause> {
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<Result<RpcReply, RpcError>>,
}
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 调用超时"
);
}
backend/crates/gateway-plugin/runtime/src/stream/bridge.rs
+8 -8
View File
@@ -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<Vec<u8>>,
pub(crate) terminal: Option<oneshot::Receiver<Result<(), RpcError>>>,
pub(crate) deadline: Option<Instant>,
pub(crate) deadline: CallDeadline,
pub(crate) id: u64,
pub(crate) shared: Arc<Shared>,
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)) => {
backend/crates/gateway-plugin/runtime/tests/generation/policy/lifecycle.rs
+204 -1
View File
@@ -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<tokio::sync::oneshot::Sender<()>>);
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;
}
}
backend/crates/gateway-plugin/runtime/tests/generation/policy/mod.rs
+11 -1
View File
@@ -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,
},
backend/crates/gateway-plugin/runtime/tests/rpc/session.rs
+52 -2
View File
@@ -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<u8>,
) -> BoxFuture<'static, Result<RpcReply, PluginFault>> {
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;
}
backend/crates/gateway-plugin/sdk/Cargo.toml
+1 -1
View File
@@ -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"
backend/crates/gateway-plugin/sdk/docs/capabilities.md
+5
View File
@@ -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 的类型化主动调用，具体公开范围见下节
backend/crates/gateway-plugin/sdk/src/client/mod.rs
+3 -3
View File
@@ -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};
backend/crates/gateway-plugin/sdk/src/client/session/credit.rs
+5 -7
View File
@@ -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<Instant>,
deadline: &CallDeadline,
cancellation: &CallCancellation,
) -> Result<u64, SessionError> {
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<Instant>,
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 => {}
}
}
backend/crates/gateway-plugin/sdk/src/client/session/deadline.rs
+106
View File
@@ -0,0 +1,106 @@
//! 宿主与插件共用的调用处理预算，受管续体等待期间暂停并保留剩余额度
use std::time::Duration;
use tokio::{sync::watch, time::Instant};
/// 同一次调用的所有等待点共享此期限，避免回调恢复后仍使用旧截止时间
#[derive(Clone)]
pub struct CallDeadline {
state: watch::Sender<State>,
}
#[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<Duration> {
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<CallDeadlinePause> {
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
});
}
}
backend/crates/gateway-plugin/sdk/src/client/session/driver.rs
+19 -24
View File
@@ -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<CreditWindow>,
output: Output,
lifecycle: LifecycleHooks,
deadline: Option<Instant>,
deadline_source: tokio::sync::watch::Sender<Option<Instant>>,
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<Instant>,
deadline: &CallDeadline,
cancellation: &CallCancellation,
) -> Result<bool, SessionError> {
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<Instant>,
deadline: &CallDeadline,
cancellation: &CallCancellation,
) -> Result<bool, SessionError> {
match output.send_data_until(frame, deadline, cancellation).await {
backend/crates/gateway-plugin/sdk/src/client/session/lifecycle.rs
+1 -20
View File
@@ -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<Instant>) {
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<Option<Instant>>) {
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; }
}
}
}
}
backend/crates/gateway-plugin/sdk/src/client/session/mod.rs
+2
View File
@@ -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;
backend/crates/gateway-plugin/sdk/src/client/session/output.rs
+3 -4
View File
@@ -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<Outbound>,
@@ -46,13 +45,13 @@ impl Output {
pub(super) async fn send_data_until(
&self,
frame: Frame,
deadline: Option<Instant>,
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,
}
}
backend/crates/gateway-plugin/sdk/src/client/session/registry.rs
+38 -11
View File
@@ -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<Option<Instant>>,
pub(super) deadline: CallDeadline,
pub(super) cancellation: CallCancellation,
pub(super) callbacks: Arc<CallbackRegistry>,
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<CallDeadlinePause>,
result: oneshot::Sender<Result<HostReply, SessionError>>,
}
@@ -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(())
}
backend/crates/gateway-plugin/sdk/src/context.rs
+1
View File
@@ -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,
backend/crates/gateway-plugin/sdk/tests/client/session/deadline.rs
+52
View File
@@ -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()
);
}
backend/crates/gateway-plugin/sdk/tests/client/session/lifecycle.rs
+200
View File
@@ -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;
}
backend/crates/gateway-plugin/sdk/tests/client/session/mod.rs
+1
View File
@@ -1,6 +1,7 @@
//! 插件会话测试入口与双工传输、处理器和宿主回调夹具
mod credit;
mod deadline;
mod driver;
mod lifecycle;
mod registry;
docs/architecture.md
+5
View File
@@ -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 单独维护重放屏障，
仍检查错误分类、续接归属、发送状态、副作用及预算。中间件链不启用此前导例外