From af9790ac375e645c372b172031228bbf28fa7aa9 Mon Sep 17 00:00:00 2001 From: Ke Wang Date: Tue, 27 Oct 2026 13:12:42 +0800 Subject: [PATCH] fix(plugin): prove replay safety for managed HTTP 429 rejections (#532) Seal managed callback observations before Core retry decisions and synchronize RPC resource revocation with stream teardown. Co-authored-by: Codex --- .../src/adapter/upstream_adapter/event.rs | 5 + .../src/adapter/upstream_adapter/execution.rs | 24 +- .../runtime/src/callback/dispatch.rs | 20 +- .../runtime/src/callback/http.rs | 3 +- .../runtime/src/callback/mod.rs | 1 + .../runtime/src/callback/replay.rs | 59 +++ .../runtime/src/callback/scope.rs | 25 +- .../runtime/src/callback/upstream/mod.rs | 2 + .../gateway-plugin/runtime/src/rpc/calls.rs | 17 +- .../tests/generation/upstream_adapter.rs | 357 ++++++++++++++++++ .../gateway-plugin/runtime/tests/rpc/calls.rs | 115 +++++- .../runtime/tests/support/worker/mod.rs | 40 +- .../sdk/docs/upstream-adapters.md | 7 + 13 files changed, 664 insertions(+), 11 deletions(-) create mode 100644 backend/crates/gateway-plugin/runtime/src/callback/replay.rs diff --git a/backend/crates/gateway-plugin/runtime/src/adapter/upstream_adapter/event.rs b/backend/crates/gateway-plugin/runtime/src/adapter/upstream_adapter/event.rs index e3117c651..1e57d035c 100644 --- a/backend/crates/gateway-plugin/runtime/src/adapter/upstream_adapter/event.rs +++ b/backend/crates/gateway-plugin/runtime/src/adapter/upstream_adapter/event.rs @@ -28,6 +28,7 @@ pub(super) struct EventDecoder { usage: Usage, contents: usize, pub(super) completed: bool, + pub(super) replay_blocked: bool, } pub(super) struct DecodedEvent { @@ -50,6 +51,10 @@ impl EventDecoder { if self.completed || message.event.facts.len() > 64 { return Err(invalid(sent)); } + // 即使与失败同批携带，也不能把 wire、工具或其他业务事实丢弃后重放 + self.replay_blocked |= !message.event.facts.is_empty() + || message.event.wire.is_some() + || message.continuation.is_some(); let wire = message .event .wire diff --git a/backend/crates/gateway-plugin/runtime/src/adapter/upstream_adapter/execution.rs b/backend/crates/gateway-plugin/runtime/src/adapter/upstream_adapter/execution.rs index ee36e65f9..70bd7de60 100644 --- a/backend/crates/gateway-plugin/runtime/src/adapter/upstream_adapter/execution.rs +++ b/backend/crates/gateway-plugin/runtime/src/adapter/upstream_adapter/execution.rs @@ -75,12 +75,34 @@ pub(super) fn execute( let result = state.next().await; match result { Ok(event) => Ok(Some((event, state))), - Err(error) => Err(state.invocation.account.record_failure(error).await), + Err(error) => { + let error = state.finish_error(error); + Err(state.invocation.account.record_failure(error).await) + } } })) } impl Execution { + fn finish_error(&mut self, mut error: ProviderError) -> ProviderError { + if let Some(active) = self.active.take() { + let managed = Arc::clone(&active.managed); + // RpcStream 析构同步撤销回调并封存未完成发送，证明不能早于这一步 + drop(active); + if error.kind() == ProviderErrorKind::RateLimited + && error.upstream_status() == Some(429) + && error.send_state() == UpstreamSendState::Sent + && managed.send_state.get() == UpstreamSendState::Sent + && managed.replay.rejected_once() + && !self.decoder.replay_blocked + && !self.invocation.context.cancellation().is_cancelled() + { + error = error.with_replay_safe(); + } + } + error + } + async fn next(&mut self) -> Result { if self.active.is_none() { self.active = Some(self.begin().await?); diff --git a/backend/crates/gateway-plugin/runtime/src/callback/dispatch.rs b/backend/crates/gateway-plugin/runtime/src/callback/dispatch.rs index c440d020d..515ec5982 100644 --- a/backend/crates/gateway-plugin/runtime/src/callback/dispatch.rs +++ b/backend/crates/gateway-plugin/runtime/src/callback/dispatch.rs @@ -94,6 +94,18 @@ impl CallbackHandler for PluginCallbacks { if method == host::LOG { return log.record(&context, params, &payload); } + // 只证明受管 HTTP 拒绝；其他能力即使只读，也不推测其业务副作用 + if let Some(managed) = &scope.upstream + && !matches!( + method.as_str(), + upstream_wire::HTTP_DO + | upstream_wire::HTTP_DO_STREAM + | upstream_wire::HTTP_STREAM_READ + | upstream_wire::HTTP_STREAM_CLOSE + ) + { + managed.replay.invalidate(); + } let middleware = if matches!( method.as_str(), middleware_wire::NEXT_METHOD @@ -225,7 +237,13 @@ impl CallbackHandler for PluginCallbacks { | upstream_wire::WEBSOCKET_READ | upstream_wire::WEBSOCKET_CLOSE ) { - return upstream::dispatch(&http, &method, params, payload).await; + return upstream::dispatch(&http, &method, params, payload) + .await + .inspect_err(|_| { + if let Some(managed) = &scope.upstream { + managed.replay.invalidate(); + } + }); } if matches!( method.as_str(), diff --git a/backend/crates/gateway-plugin/runtime/src/callback/http.rs b/backend/crates/gateway-plugin/runtime/src/callback/http.rs index 121eba2f2..66c821c6a 100644 --- a/backend/crates/gateway-plugin/runtime/src/callback/http.rs +++ b/backend/crates/gateway-plugin/runtime/src/callback/http.rs @@ -75,10 +75,11 @@ impl HttpCallbacks<'_> { let timeout = call.timeout()?; let attempt = call.start_upstream(upstream_purpose)?; let response = client.open(request, proxy, network, timeout).await; - attempt.finish( + attempt.finish_http( response .as_ref() .map_or_else(|error| error.send_state, |_| UpstreamSendState::Sent), + response.as_ref().ok().map(|response| response.status), ); let response = response.map_err(http_error)?; let mut body = response.body; diff --git a/backend/crates/gateway-plugin/runtime/src/callback/mod.rs b/backend/crates/gateway-plugin/runtime/src/callback/mod.rs index 0271616b1..46a990a59 100644 --- a/backend/crates/gateway-plugin/runtime/src/callback/mod.rs +++ b/backend/crates/gateway-plugin/runtime/src/callback/mod.rs @@ -20,6 +20,7 @@ mod model; pub(crate) mod outbound; mod ports; pub(crate) mod private_state; +mod replay; pub(crate) mod request_limits; mod resources; mod scope; diff --git a/backend/crates/gateway-plugin/runtime/src/callback/replay.rs b/backend/crates/gateway-plugin/runtime/src/callback/replay.rs new file mode 100644 index 000000000..1c5536f4f --- /dev/null +++ b/backend/crates/gateway-plugin/runtime/src/callback/replay.rs @@ -0,0 +1,59 @@ +//! 记录受管单次 HTTP 拒绝，其他调用或不明结果只会收紧重放资格 + +use std::sync::Mutex; + +#[derive(Default)] +pub(crate) struct HttpReplay(Mutex); + +#[derive(Default)] +enum State { + #[default] + Untouched, + Pending, + Rejected, + Ineligible, +} + +impl HttpReplay { + pub(super) fn start(&self, inference: bool) { + let mut state = self + .0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + *state = if inference && matches!(*state, State::Untouched) { + State::Pending + } else { + State::Ineligible + }; + } + + pub(super) fn finish(&self, status: Option) { + let mut state = self + .0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + *state = if status == Some(429) && matches!(*state, State::Pending) { + State::Rejected + } else { + State::Ineligible + }; + } + + pub(super) fn invalidate(&self) { + *self + .0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = State::Ineligible; + } + + /// 调用撤销后读取，防止检查通过后仍有新回调发送 + pub(crate) fn rejected_once(&self) -> bool { + matches!( + *self + .0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner), + State::Rejected + ) + } +} diff --git a/backend/crates/gateway-plugin/runtime/src/callback/scope.rs b/backend/crates/gateway-plugin/runtime/src/callback/scope.rs index ef240f2b8..b3ef2b5ad 100644 --- a/backend/crates/gateway-plugin/runtime/src/callback/scope.rs +++ b/backend/crates/gateway-plugin/runtime/src/callback/scope.rs @@ -98,10 +98,13 @@ impl CallbackScope { &self, purpose: Option, ) -> HttpAttempt { + let inference = purpose + == Some(gateway_plugin_sdk::call::upstream_adapter::UpstreamPathPurpose::Inference); + if let Some(managed) = &self.upstream { + managed.replay.start(inference); + } HttpAttempt(Arc::new(HttpAttemptObservation { - effects: if purpose - == Some(gateway_plugin_sdk::call::upstream_adapter::UpstreamPathPurpose::Inference) - { + effects: if inference { None } else { self.execution_effects.clone() @@ -111,6 +114,10 @@ impl CallbackScope { .as_ref() .map(|managed| Arc::clone(&managed.send_state)), completed: Mutex::new(false), + replay: self + .upstream + .as_ref() + .map(|managed| Arc::clone(&managed.replay)), })) } } @@ -121,6 +128,7 @@ pub(super) struct HttpAttemptObservation { effects: Option>, completed: Mutex, upstream: Option>, + replay: Option>, } impl HttpAttempt { @@ -131,10 +139,18 @@ impl HttpAttempt { pub(super) fn finish(self, observed: UpstreamSendState) { self.0.finish(observed); } + + pub(super) fn finish_http(self, observed: UpstreamSendState, status: Option) { + self.0.finish_http(observed, status); + } } impl HttpAttemptObservation { pub(super) fn finish(&self, observed: UpstreamSendState) { + self.finish_http(observed, None); + } + + fn finish_http(&self, observed: UpstreamSendState, status: Option) { let mut completed = self .completed .lock() @@ -145,6 +161,9 @@ impl HttpAttemptObservation { if let Some(upstream) = &self.upstream { upstream.observe(observed); } + if let Some(replay) = &self.replay { + replay.finish(status.filter(|_| observed == UpstreamSendState::Sent)); + } if observed != UpstreamSendState::NotSent && let Some(effects) = &self.effects { diff --git a/backend/crates/gateway-plugin/runtime/src/callback/upstream/mod.rs b/backend/crates/gateway-plugin/runtime/src/callback/upstream/mod.rs index 513025bbc..c76c0a69d 100644 --- a/backend/crates/gateway-plugin/runtime/src/callback/upstream/mod.rs +++ b/backend/crates/gateway-plugin/runtime/src/callback/upstream/mod.rs @@ -55,6 +55,7 @@ impl SendWatermark { pub(crate) struct ManagedUpstream { pub(crate) account: Arc, pub(crate) send_state: Arc, + pub(crate) replay: Arc, pub(crate) effects: Option>, target: Arc, socket: tokio::sync::Mutex)>>, @@ -85,6 +86,7 @@ impl ManagedUpstream { owner, websocket_allowed, send_state: Arc::default(), + replay: Arc::default(), socket: tokio::sync::Mutex::new(socket), })) } diff --git a/backend/crates/gateway-plugin/runtime/src/rpc/calls.rs b/backend/crates/gateway-plugin/runtime/src/rpc/calls.rs index 50014faa6..1079d2f18 100644 --- a/backend/crates/gateway-plugin/runtime/src/rpc/calls.rs +++ b/backend/crates/gateway-plugin/runtime/src/rpc/calls.rs @@ -44,6 +44,7 @@ struct PendingCall { execution: Option, deadline: CallDeadline, handler: Arc, + revoked: bool, transmitted: bool, cancellation: Option, context: CallContext, @@ -84,6 +85,10 @@ impl Drop for CallbackTask { impl PendingCall { /// 关闭资源与封存副作用必须先于任何结果通知；取消确认只延续容量租约 fn revoke(&mut self) { + if self.revoked { + return; + } + self.revoked = true; self.handler.finish(&self.context); self.callbacks.clear(); self.stop_resumption(); @@ -201,6 +206,7 @@ impl Shared { execution: Some(execution), deadline: deadline.clone(), handler, + revoked: false, transmitted: false, cancellation: None, context, @@ -256,6 +262,10 @@ impl Shared { let observer = (!self.expected_stop.load(Ordering::Acquire)) .then(|| self.failure_observer.clone()) .flatten(); + // 消费者可能正在析构已有流；调用对查询者不可见前必须完成同步撤销 + for call in state.calls.values_mut() { + call.revoke(); + } (std::mem::take(&mut state.calls), observer) }; self.execution.close(); @@ -298,7 +308,8 @@ impl Shared { drop(state); let _ = initial.send(result); } else { - let pending = state.calls.remove(&id).ok_or(RpcError::Protocol(None))?; + let mut pending = state.calls.remove(&id).ok_or(RpcError::Protocol(None))?; + pending.revoke(); drop(state); pending.finish(result); } @@ -329,7 +340,8 @@ impl Shared { { return Ok(()); } - let call = state.calls.remove(&id).ok_or(RpcError::Protocol(None))?; + let mut call = state.calls.remove(&id).ok_or(RpcError::Protocol(None))?; + call.revoke(); drop(state); if call.stream.is_none() || call.result.is_some() { call.finish(Err(RpcError::Protocol(None))); @@ -416,6 +428,7 @@ impl Shared { ); } if !call.transmitted { + call.revoke(); let call = state.calls.remove(&id); drop(state); if let Some(call) = call { diff --git a/backend/crates/gateway-plugin/runtime/tests/generation/upstream_adapter.rs b/backend/crates/gateway-plugin/runtime/tests/generation/upstream_adapter.rs index a95244141..0f828e173 100644 --- a/backend/crates/gateway-plugin/runtime/tests/generation/upstream_adapter.rs +++ b/backend/crates/gateway-plugin/runtime/tests/generation/upstream_adapter.rs @@ -848,6 +848,10 @@ async fn adapter_failure_preserves_upstream_diagnostics_and_native_feedback() { assert_eq!(error.kind(), ProviderErrorKind::RateLimited); assert_eq!(error.send_state(), UpstreamSendState::Sent); assert_eq!(error.upstream_status(), Some(429)); + assert!( + error.replay_is_safe(), + "宿主观测的单次 HTTP429 应允许 Core 判断重试" + ); assert_eq!( error.upstream_code().unwrap().as_str(), "rate_limit_exceeded" @@ -876,6 +880,197 @@ async fn adapter_failure_preserves_upstream_diagnostics_and_native_feedback() { super::wait_until_empty(cache.path()).await; } +#[tokio::test] +async fn adapter_replay_requires_a_single_managed_http_rejection_before_output() { + for scenario in [ + "json", + "stream", + "no_request", + "http_200", + "http_503", + "wrong_status", + "wrong_kind", + "sequential", + "concurrent", + "auxiliary", + "unmanaged", + "started", + "text", + "tool", + "failure_wire", + ] { + let server = MockServer::start().await; + let status = match scenario { + "http_200" => 200, + "http_503" => 503, + _ => 429, + }; + let requests = match scenario { + "no_request" => 0, + "sequential" | "concurrent" => 2, + _ => 1, + }; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(status).set_body_string("{}")) + .expect(requests) + .mount(&server) + .await; + let mut config = configuration(&server.uri(), scenario == "stream"); + let callback = config["upstream_callbacks"][0].clone(); + match scenario { + "no_request" => config["upstream_callbacks"] = json!([]), + "sequential" => config["upstream_callbacks"] = json!([callback, callback]), + "concurrent" => config["upstream_callbacks"] = json!([[callback, callback]]), + "auxiliary" => { + config["upstream_registration"]["adapters"][0]["paths"][0]["purpose"] = + json!("auxiliary"); + } + "unmanaged" => { + config["upstream_callbacks"] = json!([{ + "method":"host.http.do", "params":{"method":"POST","url":server.uri()}, + "body":"request" + }]); + } + _ => {} + } + let mut events = Vec::new(); + if matches!(scenario, "started" | "text" | "tool") { + events.push(json!({"event":{"facts":[{"type":"started","id":"response-one"}]}})); + } + if matches!(scenario, "text" | "tool") { + let kind = if scenario == "text" { + "text" + } else { + "tool_call" + }; + events + .push(json!({"event":{"facts":[{"type":"content_added","index":0,"kind":kind}]}})); + let fact = if scenario == "text" { + json!({"type":"text_delta","index":0,"text":"partial"}) + } else { + json!({"type":"tool_call_delta","index":0,"id":"call-one","name":"write_file","arguments":"{}"}) + }; + events.push(json!({"event":{"facts":[fact]}})); + } + let mut failure = json!({ + "event":{"facts":[]}, + "failure":{"kind":"rate_limited","status":429,"message":"fixture rejection"} + }); + if scenario == "wrong_status" { + failure["failure"]["status"] = json!(503); + } + if scenario == "wrong_kind" { + failure["failure"]["kind"] = json!("unavailable"); + } + if scenario == "failure_wire" { + failure["wire"] = json!("http"); + } + events.push(failure); + config["upstream_events"] = json!(events); + let (cache, _, runtime) = setup(config).await; + let generation = + ExtensionPreparationPort::prepare(&runtime, ConfigRevision::new(1).unwrap()) + .await + .unwrap(); + let account = Account::new(); + let mut stream = execute(&runtime, &generation, Arc::clone(&account), "key-one", None); + let error = loop { + match stream.next().await.expect("fixture must report a failure") { + Ok(_) => {} + Err(error) => break error, + } + }; + assert_eq!( + error.kind(), + if scenario == "wrong_kind" { + ProviderErrorKind::Unavailable + } else { + ProviderErrorKind::RateLimited + }, + "{scenario}" + ); + assert_eq!( + error.replay_is_safe(), + matches!(scenario, "json" | "stream"), + "{scenario}" + ); + assert_eq!( + error.send_state(), + if requests == 0 { + UpstreamSendState::NotSent + } else { + UpstreamSendState::Sent + }, + "{scenario}" + ); + drop(stream); + assert_eq!(Arc::strong_count(&account), 1, "{scenario}"); + drop(generation); + runtime.shutdown().await; + super::wait_until_empty(cache.path()).await; + } +} + +#[tokio::test] +async fn adapter_rejection_with_pending_http_revokes_the_call_without_replay() { + use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; + + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(429).set_body_string("{}")) + .expect(1) + .mount(&server) + .await; + let directory = tempfile::tempdir().unwrap(); + let marker = directory.path().join("pending-http-received"); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let pending_url = format!("http://{}/pending", listener.local_addr().unwrap()); + let received = marker.clone(); + let pending = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.unwrap(); + let mut bytes = [0; 4096]; + assert_ne!(socket.read(&mut bytes).await.unwrap(), 0); + std::fs::write(received, b"received").unwrap(); + // 不返回响应头；父调用结束必须释放尚未确定结果的 HTTP 请求 + while socket.read(&mut bytes).await.unwrap() != 0 {} + let _ = socket.shutdown().await; + }); + let mut config = configuration(&server.uri(), false); + let first = config["upstream_callbacks"][0].clone(); + config["upstream_callbacks"] = json!([first, { + "method":"host.upstream.http.do", + "params":{"method":"POST","path":pending_url,"headers":[]}, + "pending_marker":marker + }]); + config["upstream_events"] = json!([{ + "event":{"facts":[]}, + "failure":{"kind":"rate_limited","status":429,"message":"fixture rejection"} + }]); + let (cache, _, runtime) = setup(config).await; + let generation = ExtensionPreparationPort::prepare(&runtime, ConfigRevision::new(1).unwrap()) + .await + .unwrap(); + let account = Account::new(); + let mut stream = execute(&runtime, &generation, Arc::clone(&account), "key-one", None); + let error = tokio::time::timeout(Duration::from_secs(10), stream.next()) + .await + .unwrap() + .unwrap() + .unwrap_err(); + assert_eq!(error.kind(), ProviderErrorKind::RateLimited); + assert!(!error.replay_is_safe()); + assert_ne!(error.send_state(), UpstreamSendState::NotSent); + tokio::time::timeout(Duration::from_secs(5), pending) + .await + .unwrap() + .unwrap(); + drop(stream); + drop(generation); + runtime.shutdown().await; + assert_eq!(Arc::strong_count(&account), 1); + super::wait_until_empty(cache.path()).await; +} + #[tokio::test] async fn adapter_overrides_auth_headers_and_rejects_stale_continuation() { let server = MockServer::start().await; @@ -1381,6 +1576,168 @@ async fn upstream_adapter_usage_uses_native_account_pricing() { runtime.shutdown().await; } +struct RotatingAdapterProvider { + accounts: Vec>, + selected: std::sync::Mutex>, +} + +#[async_trait::async_trait] +impl gateway_core::engine::provider::Provider for RotatingAdapterProvider { + fn name(&self) -> &'static str { + "openai" + } + + fn catalog_generation(&self) -> gateway_core::routing::ProviderCatalogGeneration { + Default::default() + } + + async fn query_model_capabilities( + &self, + ) -> Result, ProviderError> { + Ok(vec![gateway_core::routing::ProviderModelCapabilities::new( + UpstreamModelId::new("native-model").unwrap(), + gateway_core::routing::ModelCapabilities::new( + std::collections::BTreeSet::from([ + gateway_core::operation::OperationKind::Generate, + ]), + None, + ), + )]) + } + + async fn execute( + self: Arc, + request: gateway_core::engine::provider::ProviderRequest, + context: AttemptContext, + ) -> Result { + let account = self + .accounts + .iter() + .find(|account| !context.excluded_accounts().contains(&account.id)) + .cloned() + .ok_or_else(|| { + ProviderError::new( + ProviderErrorKind::NoEligibleAccount, + UpstreamSendState::NotSent, + ) + })?; + self.selected.lock().unwrap().push(account.id.clone()); + let candidate = request.candidate(); + let model = candidate.upstream_model().unwrap(); + let adapter = context + .upstream_adapter_for_account(candidate.provider(), model, &account.id)? + .unwrap(); + let metadata = ProviderCallMetadata::new( + candidate.provider().clone(), + model.clone(), + account.id.clone(), + UpstreamTransport::new(adapter.transport()).unwrap(), + ); + let events = adapter.execute(UpstreamAdapterInvocation { + operation: request.operation().clone(), + headers: vec![], + context, + metadata: metadata.clone(), + account, + }); + Ok(gateway_core::engine::provider::ProviderStream::new( + metadata, + events, + (), + )) + } +} + +#[tokio::test] +async fn core_rotates_after_a_managed_adapter_http_rejection() { + use gateway_core::engine::{ + execution::{ClientTransport, ExecutionRequestMetadata, StartExecution}, + provider::{Provider, ProviderRegistry}, + }; + let Some(environment) = crate::support::environment::Environment::create().await else { + eprintln!("SKIP: plugin integration environment absent"); + return; + }; + let server = MockServer::start().await; + let calls = Arc::new(AtomicUsize::new(0)); + let observed = Arc::clone(&calls); + Mock::given(method("POST")) + .respond_with(move |_: &wiremock::Request| { + let status = if observed.fetch_add(1, Ordering::SeqCst) == 0 { + 429 + } else { + 200 + }; + ResponseTemplate::new(status).set_body_string("{}") + }) + .expect(2) + .mount(&server) + .await; + let first = environment.account(None).await; + let second = environment.account(None).await; + let provider = Arc::new(RotatingAdapterProvider { + accounts: vec![ + Account::with_identity(first.as_str(), 1), + Account::with_identity(second.as_str(), 1), + ], + selected: Default::default(), + }); + let mut config = configuration(&server.uri(), false); + config["upstream_registration"]["adapters"][0]["account_ids"] = + json!([first.as_str(), second.as_str()]); + config["upstream_account_events"] = json!({first.as_str(): [{ + "event":{"facts":[]}, + "failure":{"kind":"rate_limited","status":429,"message":"fixture rejection"} + }]}); + environment.install_plugin(config).await; + let key = "sk-adapter-rotation-fixture"; + environment.client_key("key_adapter_rotation", key).await; + let (runtime, core) = environment + .runtime_with_providers( + Default::default(), + ProviderRegistry::new([provider.clone() as Arc]).unwrap(), + ) + .await; + let execution = core.execution_service(); + let operation = Operation::Generate(GenerateRequest::from_protocol_payload( + ProtocolPayload::json_object( + "openai", + json!({"model":"native-model","input":"hello"}) + .as_object() + .unwrap() + .clone(), + ) + .unwrap(), + )); + let mut started = execution + .start(StartExecution { + client: execution.authenticate(key).unwrap(), + public_model: gateway_core::routing::PublicModelId::new("native-model").unwrap(), + operation, + metadata: ExecutionRequestMetadata { + protocol: "openai".into(), + endpoint: "/v1/responses".into(), + transport: ClientTransport::HttpJson, + stream: false, + client_ip: None, + user_agent: None, + previous_response_id: None, + }, + }) + .await + .unwrap(); + started.session.collect_uncommitted().await.unwrap(); + started.session.commit_downstream(Some(200)).await.unwrap(); + started.session.detach_finalize().await; + assert_eq!(*provider.selected.lock().unwrap(), vec![first, second]); + assert_eq!(calls.load(Ordering::SeqCst), 2); + environment.release_plugin_accounts(&runtime); + drop(execution); + drop(core); + runtime.shutdown().await; + environment.close().await; +} + #[tokio::test] async fn core_published_adapter_uses_normal_attempt_usage_and_cost_ledger() { use crate::support::environment::Environment; diff --git a/backend/crates/gateway-plugin/runtime/tests/rpc/calls.rs b/backend/crates/gateway-plugin/runtime/tests/rpc/calls.rs index dd50df34e..e42bd801d 100644 --- a/backend/crates/gateway-plugin/runtime/tests/rpc/calls.rs +++ b/backend/crates/gateway-plugin/runtime/tests/rpc/calls.rs @@ -1,7 +1,10 @@ //! 验证调用表在成功、远端错误和超时的结果交付前同步撤销回调资源 use std::{ - sync::{Arc, atomic::Ordering}, + sync::{ + Arc, Condvar, Mutex, + atomic::{AtomicBool, Ordering}, + }, time::Duration, }; @@ -11,6 +14,116 @@ use serde_json::json; use super::session::{LifecycleCallbacks, prepared_session}; +struct GatedFinish { + ready: Arc, + entered: std::sync::mpsc::Sender<()>, + started: AtomicBool, + finished: AtomicBool, + released: Mutex, + release: Condvar, +} + +impl gateway_plugin_runtime::CallbackHandler for GatedFinish { + fn finish(&self, _: &gateway_plugin_sdk::CallContext) { + if self.started.swap(true, Ordering::AcqRel) { + return; + } + self.entered.send(()).unwrap(); + // 测试屏障模拟撤销处理中被调度器暂停，不依赖恰好命中的线程交错 + let released = self.released.lock().unwrap(); + let (released, timeout) = self + .release + .wait_timeout_while(released, Duration::from_secs(5), |released| !*released) + .unwrap(); + assert!(!timeout.timed_out() && *released); + self.finished.store(true, Ordering::Release); + } + + fn call( + &self, + _: gateway_plugin_sdk::CallContext, + method: String, + _: serde_json::Value, + _: Vec, + ) -> futures::future::BoxFuture< + 'static, + Result, + > { + let ready = Arc::clone(&self.ready); + Box::pin(async move { + assert_eq!(method, "test.stream_ready"); + ready.notified().await; + Ok(gateway_plugin_runtime::RpcReply { + result: json!({}), + payload: vec![], + }) + }) + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn dropping_stream_waits_for_resource_revocation_even_after_end_or_process_failure() { + for method in ["stream_single", "stream_single_crash"] { + let (_cache, prepared, handshake, processes) = prepared_session(); + let (entered, entry) = std::sync::mpsc::channel(); + let callbacks = Arc::new(GatedFinish { + ready: Arc::new(tokio::sync::Notify::new()), + entered, + started: AtomicBool::new(false), + finished: AtomicBool::new(false), + released: Mutex::new(false), + release: Condvar::new(), + }); + let session = Arc::new( + RpcSession::start( + prepared, + handshake, + RpcLimits::default(), + &processes, + callbacks.clone(), + ) + .await + .unwrap(), + ); + let stream = session + .call_stream( + method, + session.context(Stage::Request, Duration::from_secs(5)), + json!({}), + vec![], + ) + .await + .unwrap(); + // 拿到流后才允许对端结束；整个屏障控制不依赖被暂停的 I/O 调度器 + callbacks.ready.notify_one(); + entry.recv_timeout(Duration::from_secs(5)).unwrap(); + let observed = Arc::clone(&callbacks); + let (started, received) = std::sync::mpsc::channel(); + let (completed, completion) = std::sync::mpsc::channel(); + let dropped = std::thread::spawn(move || { + started.send(()).unwrap(); + drop(stream); + completed + .send(observed.finished.load(Ordering::Acquire)) + .unwrap(); + }); + received.recv_timeout(Duration::from_secs(1)).unwrap(); + // 撤销屏障会暂停 I/O 调度线程，不能用同一 runtime 的计时器释放它 + let early = completion.recv_timeout(Duration::from_millis(100)); + *callbacks.released.lock().unwrap() = true; + callbacks.release.notify_one(); + let was_early = early.is_ok(); + let revoked = + early.unwrap_or_else(|_| completion.recv_timeout(Duration::from_secs(1)).unwrap()); + dropped.join().unwrap(); + session.shutdown(Duration::from_secs(1)).await; + assert!( + !was_early && revoked, + "{method}: stream drop must not outrun callback revocation" + ); + } +} + #[tokio::test] async fn terminal_results_revoke_callback_resources_before_waking_callers() { for method in ["echo", "deny", "hang"] { diff --git a/backend/crates/gateway-plugin/runtime/tests/support/worker/mod.rs b/backend/crates/gateway-plugin/runtime/tests/support/worker/mod.rs index 140a33007..aaaf857e5 100644 --- a/backend/crates/gateway-plugin/runtime/tests/support/worker/mod.rs +++ b/backend/crates/gateway-plugin/runtime/tests/support/worker/mod.rs @@ -489,7 +489,15 @@ impl Peer { ) .await; let credits = self.streams.lock().await.get(&id).unwrap().clone(); - for sequence in 0..128_u64 { + let chunks = if method.starts_with("stream_single") { + self.callback(id, "test.stream_ready", json!({})) + .await + .unwrap(); + 1 + } else { + 128 + }; + for sequence in 0..chunks { credits.take(8192).await; let payload = if method == "stream_overflow" { vec![b'x'; 256 * 1024 + 1] @@ -499,6 +507,9 @@ impl Peer { let sequence = sequence + u64::from(method == "stream_sequence"); self.send(Message::Stream { id, sequence }, payload).await; } + if method == "stream_single_crash" { + std::process::exit(7); + } self.send(Message::End { id, error: None }, vec![]).await; return; } @@ -1737,6 +1748,27 @@ impl Peer { .into_iter() .flatten() { + if let Some(marker) = callback["pending_marker"].as_str() { + // 等真实服务器确认收到字节，再报告失败，避免依赖回调调度顺序 + let _pending = self + .send_callback( + id, + callback["method"].as_str().unwrap(), + callback["params"].clone(), + body.clone(), + ) + .await; + tokio::time::timeout(Duration::from_secs(5), async { + while !std::path::Path::new(marker).exists() { + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .map_err(|_| { + PluginFault::new(ErrorCode::Timeout, "pending HTTP fixture did not start") + })?; + continue; + } let callbacks = callback .as_array() .map_or_else(|| std::slice::from_ref(callback), Vec::as_slice); @@ -1787,7 +1819,11 @@ impl Peer { } } } - let events = self.configuration["upstream_events"].as_array().unwrap(); + let events = self.configuration["upstream_account_events"] + .get(&request.account_id) + .unwrap_or(&self.configuration["upstream_events"]) + .as_array() + .unwrap(); let credits = self.streams.lock().await.get(&id).unwrap().clone(); for (sequence, event) in events.iter().enumerate() { let mut message = UpstreamAdapterEvent::new( diff --git a/backend/crates/gateway-plugin/sdk/docs/upstream-adapters.md b/backend/crates/gateway-plugin/sdk/docs/upstream-adapters.md index a040d58b4..ffaff0291 100644 --- a/backend/crates/gateway-plugin/sdk/docs/upstream-adapters.md +++ b/backend/crates/gateway-plugin/sdk/docs/upstream-adapters.md @@ -94,6 +94,13 @@ let (message, ()) = tokio::try_join!( 包含标准事实与可选原始 wire；事件需符合开始、内容、工具、用量和唯一结束的顺序。 原始 JSON / SSE / WebSocket 字节与标准事实分开传递，错误使用 `UpstreamFailure`；宿主网络观测决定发送状态，插件不能将其降为未发送 +宿主只对一次受管 `inference` HTTP 调用明确返回的 429 提供有限重放证明：插件必须报告 +`rate_limited` 和 `status: 429`，且此前及失败事件中均无业务事实、wire 或续接材料。 +调用结束时先撤销回调、回收资源并封存未完成发送，再将错误交给 Core；发送状态仍为 `sent`。 +多次或并发请求、辅助请求、其他宿主能力调用（日志除外）、受管回调失败以及取消都不能取得该证明。 +观测事件和该次 HTTP 的读流、关流不单独阻断；HTTP200 流内报错、WebSocket 拒绝及其他状态码不适用。 +Core 仍复核续接归属、外部副作用、交付与重试预算；这不保证一定换号，也不证明完整信任插件绕过宿主执行的操作可重放 + 初始 RPC 回复受插件处理预算约束，默认上限为 120 秒；持续接收与解析应放在返回的流中。 初始流式回复后，流的寿命遵循模型请求期限与取消，受管 HTTP 正文每次读取单独计时。 持续有数据的流不会因初始 RPC 预算耗尽而结束，单次读取超时、客户端取消和流结束仍会回收资源