diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 95db01b61..fdbc20507 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -3255,6 +3255,7 @@ dependencies = [ "tokio-rustls", "tokio-tungstenite 0.28.0", "tokio-util", + "tower-layer", "tower-service", "tracing", "tracing-subscriber", diff --git a/backend/crates/gateway-api/tests/openai/error.rs b/backend/crates/gateway-api/tests/openai/error.rs index ca123284a..f7044548b 100644 --- a/backend/crates/gateway-api/tests/openai/error.rs +++ b/backend/crates/gateway-api/tests/openai/error.rs @@ -639,18 +639,14 @@ async fn continuation_recovery_should_preserve_the_official_retry_signal() { mod model_routing { use std::collections::BTreeMap; use std::sync::Arc; - use std::time::{Duration, SystemTime}; use axum::{ body::{Body, to_bytes}, http::{Request, StatusCode}, }; - use futures::future::BoxFuture; - use gateway_core::engine::execution::{ - DefaultExecutionService, ProviderCircuitDecision, ProviderCircuitError, ProviderCircuitPort, - }; + use gateway_core::engine::execution::DefaultExecutionService; use gateway_core::engine::provider::ProviderRegistry; - use gateway_core::routing::{ProviderKind, RuntimeSnapshot}; + use gateway_core::routing::RuntimeSnapshot; use gateway_core::runtime::RuntimeSnapshotHandle; use serde_json::{Value, json}; use tower::ServiceExt; @@ -659,53 +655,16 @@ mod model_routing { IgnoredClientApiKeyUsage, UnusedAdmissions, UnusedContinuation, UnusedExecutionStore, }; - struct TestCircuits { - blocked: bool, - } - - impl ProviderCircuitPort for TestCircuits { - fn decision<'a>( - &'a self, - _: &'a ProviderKind, - ) -> BoxFuture<'a, Result> { - Box::pin(async move { - Ok(if self.blocked { - ProviderCircuitDecision::BlockedUntil( - SystemTime::now() + Duration::from_secs(30), - ) - } else { - ProviderCircuitDecision::Allow - }) - }) - } - - fn observe_failure<'a>( - &'a self, - _: &'a ProviderKind, - ) -> BoxFuture<'a, Result<(), ProviderCircuitError>> { - Box::pin(async { unreachable!("routing rejection must not execute a provider") }) - } - - fn observe_success<'a>( - &'a self, - _: &'a ProviderKind, - ) -> BoxFuture<'a, Result<(), ProviderCircuitError>> { - Box::pin(async { unreachable!("routing rejection must not execute a provider") }) - } - } - async fn request_model( snapshot: RuntimeSnapshot, model: &str, stream: bool, - blocked: bool, ) -> (StatusCode, Value) { let execution = DefaultExecutionService::new( RuntimeSnapshotHandle::new(snapshot), Arc::new(UnusedExecutionStore), ProviderRegistry::default(), Arc::new(UnusedAdmissions), - Arc::new(TestCircuits { blocked }), Arc::new(UnusedContinuation), Arc::new(IgnoredClientApiKeyUsage), ); @@ -746,7 +705,7 @@ mod model_routing { for stream in [false, true] { let snapshot = crate::openai::snapshot("sk_model_routing", "openai") .with_model_mappings(mappings.clone()); - let response = request_model(snapshot, model, stream, false).await; + let response = request_model(snapshot, model, stream).await; assert_eq!( response, @@ -763,25 +722,6 @@ mod model_routing { } } } - - #[tokio::test] - async fn an_existing_model_with_a_blocked_provider_should_remain_service_unavailable() { - let response = request_model( - crate::openai::snapshot("sk_model_routing", "openai"), - "model-a", - true, - true, - ) - .await; - - assert_eq!( - (response.0, response.1["error"]["code"].as_str()), - ( - StatusCode::SERVICE_UNAVAILABLE, - Some("no_available_provider") - ), - ); - } } #[tokio::test] @@ -813,3 +753,27 @@ async fn quota_recovery_http_response_uses_projection_without_replacing_upstream "usage_limit_reached" ); } + +#[test] +fn final_connection_failures_and_local_capacity_have_distinct_http_statuses() { + for (kind, expected) in [ + (ProviderErrorKind::Transport, StatusCode::BAD_GATEWAY), + (ProviderErrorKind::Timeout, StatusCode::GATEWAY_TIMEOUT), + ( + ProviderErrorKind::ProviderInfrastructureUnavailable, + StatusCode::SERVICE_UNAVAILABLE, + ), + ] { + let error = EngineError::Provider(ProviderError::new(kind, UpstreamSendState::NotSent)); + assert_eq!(engine_error_response(&error).status(), expected); + } + for kind in [ + GatewayErrorKind::NoAvailableProvider, + GatewayErrorKind::AccountCapacityUnavailable, + ] { + assert_eq!( + gateway_error_response(&GatewayError::new(kind, "temporarily unavailable")).status(), + StatusCode::SERVICE_UNAVAILABLE + ); + } +} diff --git a/backend/crates/gateway-api/tests/openai/mod.rs b/backend/crates/gateway-api/tests/openai/mod.rs index bb1f8dbb8..63f176599 100644 --- a/backend/crates/gateway-api/tests/openai/mod.rs +++ b/backend/crates/gateway-api/tests/openai/mod.rs @@ -26,7 +26,6 @@ use gateway_core::engine::continuation::{ }; use gateway_core::engine::execution::{ AuthenticatedClient, ClientApiKeyUsageSink, DefaultExecutionService, ExecutionService, - ProviderCircuitDecision, ProviderCircuitError, ProviderCircuitPort, }; use gateway_core::engine::provider::ProviderRegistry; use gateway_core::engine::{ @@ -89,7 +88,6 @@ pub(super) fn api_router_with_admin_and_client( Arc::new(UnusedExecutionStore), ProviderRegistry::default(), Arc::new(UnusedAdmissions), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(IgnoredClientApiKeyUsage), )); @@ -105,7 +103,6 @@ pub(super) fn api_router_with_config( Arc::new(UnusedExecutionStore), ProviderRegistry::default(), Arc::new(UnusedAdmissions), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(IgnoredClientApiKeyUsage), )); @@ -194,7 +191,6 @@ pub(super) fn authenticated_client_for_provider_with_limit( Arc::new(UnusedExecutionStore), ProviderRegistry::default(), Arc::new(UnusedAdmissions), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(IgnoredClientApiKeyUsage), ); @@ -218,7 +214,6 @@ pub(super) fn authenticated_client_with_min_versions( Arc::new(UnusedExecutionStore), ProviderRegistry::default(), Arc::new(UnusedAdmissions), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(IgnoredClientApiKeyUsage), ); @@ -395,31 +390,6 @@ impl ClientAdmissionPort for UnusedAdmissions { } } -struct UnusedCircuits; - -impl ProviderCircuitPort for UnusedCircuits { - fn decision<'a>( - &'a self, - _: &'a ProviderKind, - ) -> BoxFuture<'a, Result> { - Box::pin(async { unreachable!("authentication fixture does not execute") }) - } - - fn observe_failure<'a>( - &'a self, - _: &'a ProviderKind, - ) -> BoxFuture<'a, Result<(), ProviderCircuitError>> { - Box::pin(async { unreachable!("authentication fixture does not execute") }) - } - - fn observe_success<'a>( - &'a self, - _: &'a ProviderKind, - ) -> BoxFuture<'a, Result<(), ProviderCircuitError>> { - Box::pin(async { unreachable!("authentication fixture does not execute") }) - } -} - struct UnusedContinuation; impl NativeContinuationPort for UnusedContinuation { diff --git a/backend/crates/gateway-api/tests/openai/responses/websocket/mod.rs b/backend/crates/gateway-api/tests/openai/responses/websocket/mod.rs index 02d31ed91..4e7025cd7 100644 --- a/backend/crates/gateway-api/tests/openai/responses/websocket/mod.rs +++ b/backend/crates/gateway-api/tests/openai/responses/websocket/mod.rs @@ -27,8 +27,7 @@ use gateway_core::engine::admission::{ use gateway_core::engine::budget::{ClientBudgetCharge, ClientBudgetError, ClientBudgetPort}; use gateway_core::engine::execution::{ AuthenticatedClient, ClientAuthenticationError, DefaultExecutionService, ExecutionService, - ExecutionSession, ProviderCircuitDecision, ProviderCircuitError, ProviderCircuitPort, - StartExecution, StartProviderExecution, StartedExecution, + ExecutionSession, StartExecution, StartProviderExecution, StartedExecution, }; use gateway_core::engine::provider::{ Provider, ProviderCallMetadata, ProviderRegistry, ProviderRequest, ProviderStream, @@ -48,9 +47,7 @@ use gateway_core::lifecycle::{ use gateway_core::metering::{Decimal, ProviderReportedCost}; use gateway_core::operation::Operation; use gateway_core::policy::ClientApiKeyId; -use gateway_core::routing::{ - ProviderCatalogGeneration, ProviderKind, ProviderModelCapabilities, PublicModelId, -}; +use gateway_core::routing::{ProviderCatalogGeneration, ProviderModelCapabilities, PublicModelId}; use gateway_core::runtime::RuntimeSnapshotHandle; use gateway_core::upstream::{UpstreamSendState, UpstreamTransport}; use gateway_protocol::openai::codex_responses_request_semantics; @@ -984,29 +981,6 @@ impl ClientBudgetPort for SettlementPorts { } } -impl ProviderCircuitPort for SettlementPorts { - fn decision<'a>( - &'a self, - _: &'a ProviderKind, - ) -> BoxFuture<'a, Result> { - Box::pin(async { Ok(ProviderCircuitDecision::Allow) }) - } - - fn observe_failure<'a>( - &'a self, - _: &'a ProviderKind, - ) -> BoxFuture<'a, Result<(), ProviderCircuitError>> { - Box::pin(async { Ok(()) }) - } - - fn observe_success<'a>( - &'a self, - _: &'a ProviderKind, - ) -> BoxFuture<'a, Result<(), ProviderCircuitError>> { - Box::pin(async { Ok(()) }) - } -} - struct ChargedWebSocketProvider; fn websocket_charge() -> Decimal { @@ -1114,7 +1088,6 @@ async fn websocket_disconnect_during_core_settlement_finishes_charge_before_rele ProviderRegistry::new([Arc::new(ChargedWebSocketProvider) as Arc]) .unwrap(), ports.clone(), - ports.clone(), Arc::new(UnusedContinuation), Arc::new(IgnoredClientApiKeyUsage), ) diff --git a/backend/crates/gateway-core/src/engine/connection.rs b/backend/crates/gateway-core/src/engine/connection.rs new file mode 100644 index 000000000..92c697938 --- /dev/null +++ b/backend/crates/gateway-core/src/engine/connection.rs @@ -0,0 +1,108 @@ +//! 请求内共享的发送前连接预算；不投影账号或 Provider 健康状态。 + +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +const MAX_OPENINGS: u32 = 4; +const STARTUP_TIMEOUT: Duration = Duration::from_secs(30); + +#[derive(Debug, Default)] +struct State { + started: Option, + openings: u32, + completed: bool, +} + +/// 换账号、换传输和内部重试共享同一预算；成功发送后的协议恢复不重置它。 +#[derive(Debug, Clone, Default)] +pub struct ConnectionBudget(Arc>); + +#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)] +#[error("connection recovery budget exhausted")] +pub struct ConnectionBudgetExhausted; + +impl ConnectionBudget { + /// 在可能建立连接前预约一次 opening；返回本次连接允许使用的剩余时间。 + /// HTTP 连接池隐藏了是否冷建连，因此一次 HTTP opening 也计入次数上限。 + pub fn begin(&self) -> Result, ConnectionBudgetExhausted> { + let mut state = self + .0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if state.completed { + return Ok(None); + } + let started = *state.started.get_or_insert_with(Instant::now); + let remaining = STARTUP_TIMEOUT.saturating_sub(started.elapsed()); + if state.openings >= MAX_OPENINGS || remaining.is_zero() { + return Err(ConnectionBudgetExhausted); + } + state.openings += 1; + Ok(Some(remaining)) + } + + /// 仅连接成功不足以清零；确认业务发送后退出新增的 NotSent 恢复策略。 + pub fn complete(&self) { + self.0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .completed = true; + } + + #[must_use] + pub fn remaining(&self) -> Option { + let state = self + .0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if state.completed || state.openings >= MAX_OPENINGS { + return None; + } + let remaining = state.started.map_or(STARTUP_TIMEOUT, |started| { + STARTUP_TIMEOUT.saturating_sub(started.elapsed()) + }); + (!remaining.is_zero()).then_some(remaining) + } + + #[must_use] + pub fn exhausted(&self) -> bool { + let state = self + .0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + !state.completed + && (state.openings >= MAX_OPENINGS + || state + .started + .is_some_and(|started| started.elapsed() >= STARTUP_TIMEOUT)) + } + + /// 首次建连后的等待也计入恢复窗口；不缩短已发送请求的生成时间。 + #[must_use] + pub fn startup_remaining(&self) -> Option { + let state = self + .0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if state.completed { + return None; + } + state + .started + .map(|started| STARTUP_TIMEOUT.saturating_sub(started.elapsed())) + } + + /// 额外 attempt 最多三个；抖动由请求 ID 派生，重试不依赖运行时随机源。 + #[must_use] + pub fn retry_delay(&self, retries: u32, request_id: &str) -> Option { + if retries >= MAX_OPENINGS - 1 { + return None; + } + let remaining = self.remaining()?; + let hash = request_id.bytes().fold(u64::from(retries), |hash, byte| { + hash.wrapping_mul(31).wrapping_add(u64::from(byte)) + }); + let delay = Duration::from_millis((500_u64 << retries) * (80 + hash % 41) / 100); + (delay < remaining).then_some(delay) + } +} diff --git a/backend/crates/gateway-core/src/engine/coordinator.rs b/backend/crates/gateway-core/src/engine/coordinator.rs index 36b9c72eb..cb4784dfe 100644 --- a/backend/crates/gateway-core/src/engine/coordinator.rs +++ b/backend/crates/gateway-core/src/engine/coordinator.rs @@ -279,6 +279,8 @@ where request_id, client_api_key_ref, concurrency_wait_budget: ConcurrencyWaitBudget::default(), + connection_budget: super::connection::ConnectionBudget::default(), + connection_retries: 0, observation: ResponseObservation::new(timing_started_at), request_observation, budget_prior_attempts_usd: Decimal::ZERO, @@ -395,6 +397,8 @@ pub struct ResponseExecutionSession { request_id: ModelRequestId, client_api_key_ref: crate::policy::ClientApiKeyId, concurrency_wait_budget: ConcurrencyWaitBudget, + connection_budget: super::connection::ConnectionBudget, + connection_retries: u32, observation: ResponseObservation, request_observation: Option, budget_prior_attempts_usd: Decimal, @@ -801,7 +805,7 @@ where PollBoundary::Deadline => { // 会话 deadline 是网关自身的请求预算，不是上游超时； // 真正的上游超时会作为流错误进入 `handle_stream_error` 记账。 - // 这里不写 provider 失败，避免长流集中到期误触 provider 熔断。 + // 这里不写 provider 失败，避免把本地预算到期归因为上游故障。 self.finish_interruption(&EngineError::Deadline).await?; return Err(EngineError::Deadline); } @@ -885,7 +889,13 @@ where if let Some(recovery) = pending_retry.as_ref() && !recovery.delay.is_zero() { - match poll_retry_delay(recovery.delay, self.cancellation.clone(), self.deadline).await { + let deadline = self + .connection_budget + .startup_remaining() + .map_or(self.deadline, |remaining| { + self.deadline.min(SystemTime::now() + remaining) + }); + match poll_retry_delay(recovery.delay, self.cancellation.clone(), deadline).await { RetryDelayBoundary::Elapsed => {} RetryDelayBoundary::Cancelled => { self.finish_interruption(&EngineError::Cancelled).await?; @@ -991,6 +1001,7 @@ where .with_pricing(self.plan.pricing()) .with_request_location(self.plan.request_location().cloned()) .with_concurrency_wait_budget(self.concurrency_wait_budget.clone()) + .with_connection_budget(self.connection_budget.clone()) .with_timing_started_at(self.observation.timing_started_at) .with_request_policy(self.request_policy.clone()) .with_execution_effects(self.execution_effects.as_ref().map(Arc::clone)) @@ -1043,9 +1054,7 @@ where return Err(EngineError::Cancelled); } ProviderBoundary::Deadline => { - // 网关预算到期同样不是候选 Provider 的上游超时，不计入熔断； - // Provider 自身的握手/传输超时会以 `ProviderErrorKind::Timeout` - // 错误返回并在下方 `Result` 分支记账。 + // 网关预算到期是本请求的终态，不推定候选 Provider 不可用。 self.finish_interruption(&EngineError::Deadline).await?; return Err(EngineError::Deadline); } @@ -1375,6 +1384,10 @@ where } else { error.send_state() }; + if attempt_send_state != UpstreamSendState::NotSent { + // 已发送的容量拒绝、凭据恢复沿用原策略，不再受首次建连窗口限制。 + self.connection_budget.complete(); + } let execution_effect_observed = self.execution_effect_observed(); let send_state = self.raise_send_watermark(attempt_send_state); if self.request_persisted { @@ -1406,11 +1419,34 @@ where Some(crate::error::PreDeliveryRetry::AccountRotation) ) && self.routing_attempts < self.plan.max_attempts().get(); + // 只有尚未发送的逻辑请求进入新增策略；已有发送后安全恢复保持原有路由规则。 + let connection_retry_requested = self.current_send_state() == UpstreamSendState::NotSent + && matches!( + error.pre_delivery_retry(), + Some(crate::error::PreDeliveryRetry::SameAccountConnectionRetry { .. }) + ); let transport_recovery = match error.pre_delivery_retry() { + Some(crate::error::PreDeliveryRetry::SameAccountConnectionRetry { transport }) + if !execution_effect_observed + && !self.connection_budget.exhausted() + && self.downstream_committed_at.is_none() + && !self.delivery_pending + && attempt_send_state == UpstreamSendState::NotSent + && self.current_send_state() == UpstreamSendState::NotSent + && self.continuation_attempt == ContinuationAttempt::None => + { + self.connection_budget + .retry_delay(self.connection_retries, self.request_id.as_str()) + .map(|delay| { + self.connection_retries += 1; + (transport, delay) + }) + } Some(crate::error::PreDeliveryRetry::SameAccountTransportRetry { retry_index, delay, }) if !execution_effect_observed + && !self.connection_budget.exhausted() && self.downstream_committed_at.is_none() && !self.delivery_pending && attempt_send_state != UpstreamSendState::Ambiguous => @@ -1419,6 +1455,7 @@ where } Some(crate::error::PreDeliveryRetry::SameAccountTransportFallback) if !execution_effect_observed + && !self.connection_budget.exhausted() && self.downstream_committed_at.is_none() && !self.delivery_pending && attempt_send_state != UpstreamSendState::Ambiguous => @@ -1427,7 +1464,9 @@ where } _ => None, }; - let ordinary_retry = !execution_effect_observed + let ordinary_retry = !self.connection_budget.exhausted() + && !connection_retry_requested + && !execution_effect_observed && self.account_selection.required_account().is_none() && self.continuation_attempt == ContinuationAttempt::None && self.downstream_committed_at.is_none() @@ -1477,6 +1516,9 @@ where "sameAccountRetry": same_account_retry, "accountRotationRetry": account_rotation_retry, "ordinaryRetry": ordinary_retry, "transportRecovery": transport_recovery.is_some(), "transientRetry": transient_retry.is_some(), + "connectionRetry": connection_retry_requested, + "connectionRetries": self.connection_retries, + "connectionBudgetRemainingMs": self.connection_budget.remaining().map(duration_ms), "executionEffectObserved": execution_effect_observed, "delayMs": transient_retry.or(transport_recovery.map(|(_, delay)| delay)).map(duration_ms), "downstreamCommitted": self.downstream_committed_at.is_some(), diff --git a/backend/crates/gateway-core/src/engine/execution.rs b/backend/crates/gateway-core/src/engine/execution.rs index 5ad87f403..755c6962e 100644 --- a/backend/crates/gateway-core/src/engine/execution.rs +++ b/backend/crates/gateway-core/src/engine/execution.rs @@ -1,9 +1,8 @@ -//! 数据面执行用例：认证、准入、路由、continuation、circuit 与会话生命周期。 +//! 数据面执行用例：认证、准入、路由、continuation 与会话生命周期。 use std::collections::{BTreeMap, BTreeSet}; use std::fmt; use std::net::IpAddr; -use std::num::NonZeroU32; use std::sync::{ Arc, Mutex, Weak, atomic::{AtomicUsize, Ordering}, @@ -52,9 +51,9 @@ use crate::engine::provider::ProviderRegistry; use crate::engine::{ AttemptCoordinator, AttemptRecord, CoordinatedEvent, EngineError, ExecutionStore, GatewayEngine, IntermediateFailure, ModelRequestFinalization, ModelRequestId, NewModelRequest, - ProbeFailure, ProviderAccountId, ProviderAttemptOutcome, RecoveryReport, UpstreamSendState, + ProbeFailure, ProviderAccountId, RecoveryReport, UpstreamSendState, }; -use crate::error::{GatewayError, GatewayErrorKind, ProviderErrorKind, StoreError}; +use crate::error::{GatewayError, GatewayErrorKind, StoreError}; use crate::event::{GatewayEvent, ProviderEvent, ProviderResponseHeader}; use crate::identity::ProviderKind; use crate::lifecycle::CancellationToken; @@ -649,47 +648,6 @@ pub trait ClientApiKeyUsageSink: Send + Sync { fn record_used(&self, key_id: &ClientApiKeyId); } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum ProviderCircuitDecision { - Allow, - BlockedUntil(SystemTime), -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)] -#[error("provider circuit store is unavailable")] -pub struct ProviderCircuitError; - -/// Provider circuit 的可重建协调策略；由 Core 拥有并交给 Store adapter 执行。 -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct ProviderCircuitPolicy { - pub failure_threshold: NonZeroU32, - pub open_duration: Duration, -} - -impl Default for ProviderCircuitPolicy { - fn default() -> Self { - Self { - failure_threshold: NonZeroU32::new(3).unwrap_or(NonZeroU32::MIN), - open_duration: Duration::from_secs(30), - } - } -} - -pub trait ProviderCircuitPort: Send + Sync { - fn decision<'a>( - &'a self, - provider_kind: &'a ProviderKind, - ) -> BoxFuture<'a, Result>; - fn observe_failure<'a>( - &'a self, - provider_kind: &'a ProviderKind, - ) -> BoxFuture<'a, Result<(), ProviderCircuitError>>; - fn observe_success<'a>( - &'a self, - provider_kind: &'a ProviderKind, - ) -> BoxFuture<'a, Result<(), ProviderCircuitError>>; -} - pub struct DefaultExecutionService { snapshots: RuntimeSnapshotHandle, /// probe 自身走 transient store，探测失败仍写入持久 store 的 ops_events。 @@ -697,7 +655,6 @@ pub struct DefaultExecutionService { providers: ProviderRegistry, admissions: Arc, admission_waiting: ConcurrencyWaitQueue, - circuits: Arc, continuation: Arc, client_api_key_usage: Arc, budget: Option>, @@ -717,7 +674,6 @@ impl DefaultExecutionService { execution: Arc, providers: ProviderRegistry, admissions: Arc, - circuits: Arc, continuation: Arc, client_api_key_usage: Arc, ) -> Self { @@ -727,7 +683,6 @@ impl DefaultExecutionService { providers, admissions, admission_waiting: ConcurrencyWaitQueue::default(), - circuits, continuation, client_api_key_usage, budget: None, @@ -1029,14 +984,7 @@ impl DefaultExecutionService { }); // 路由插件可以调用 host.model/host.affinity；仅这条扩展路径必须在首次 RPC // 前取得父 Key 准入。无策略的原生路由仍保持“先路由、后准入”的既有顺序。 - let mut start_guard = if request_policy.is_some() { - Some( - self.prepare_execution_start(&request, &request_id, &mut authorization) - .await?, - ) - } else { - None - }; + let mut start_guard = None; let request_observation = self.request_observers.as_ref().and_then(|observers| { let generation = request.client.snapshot.extensions()?.clone(); let plan = observers.resolve(&generation)?; @@ -1055,10 +1003,12 @@ impl DefaultExecutionService { )) }); let budget_key_id = request.client.policy.key_id().clone(); + let mut entered_execution = false; let result = async { - let mut routing_context = self - .route_context(authorization.account_scope.provider_kinds()) - .await?; + if request_policy.is_some() { + start_guard = Some(self.prepare_execution_start(&request, &request_id, &mut authorization).await?); + } + let mut routing_context = RoutingContext::default(); if let Some(provider) = authorization.required_provider.clone() { if !authorization.account_scope.provider_kinds().contains(&provider) { return Err(GatewayError::new( @@ -1217,6 +1167,7 @@ impl DefaultExecutionService { request.metadata.transport, ) .with_extension_scope(authorization.extension_scope.clone()); + entered_execution = true; self.start_without_continuation( request, PreparedExecutionStart { @@ -1233,6 +1184,27 @@ impl DefaultExecutionService { .await } .await; + if !entered_execution && let Err(error) = &result { + tracing::warn!( + request_id = request_id.as_str(), + key_id = budget_key_id.as_str(), + failure_kind = error.kind().as_str(), + "请求在路由或准入阶段被拒绝" + ); + let rejection = super::EntryRejection { + request_id: request_id.clone(), + client_key_id: budget_key_id.clone(), + error: error.clone(), + latency: started_at.elapsed().unwrap_or_default(), + }; + let write = self.observations.record_entry_rejection(rejection).fuse(); + let timeout = Delay::new(COORDINATION_TIMEOUT).fuse(); + pin_mut!(write, timeout); + select_biased! { + result = write => { if result.is_err() { tracing::warn!("入口拒绝观测写入失败"); } }, + _ = timeout => tracing::warn!("入口拒绝观测写入超时"), + } + } if let (Some(observation), Err(error)) = (&request_observation, &result) { observation.reject(error); } @@ -1430,7 +1402,6 @@ impl DefaultExecutionService { core, admission, active_request, - Arc::clone(&self.circuits), Arc::clone(&self.continuation), self.budget.clone(), )), @@ -1856,48 +1827,6 @@ impl DefaultExecutionService { } } - async fn route_context( - &self, - provider_kinds: &BTreeSet, - ) -> Result { - let decisions = futures::future::join_all(provider_kinds.iter().map(|provider_kind| { - let circuits = Arc::clone(&self.circuits); - async move { - let decision = circuits.decision(provider_kind).fuse(); - let timeout = Delay::new(COORDINATION_TIMEOUT).fuse(); - pin_mut!(decision, timeout); - let decision = select_biased! { - result = decision => Some(result), - _ = timeout => None, - }; - (provider_kind, decision) - } - })) - .await; - let mut blocked_providers = BTreeSet::new(); - for (provider_kind, decision) in decisions { - match decision { - Some(Ok(ProviderCircuitDecision::BlockedUntil(_))) => { - blocked_providers.insert(provider_kind.clone()); - } - Some(Err(error)) => tracing::warn!( - provider = provider_kind.as_str(), - %error, - "Provider circuit 读取失败，按可重建协调状态 fail-open" - ), - None => tracing::warn!( - provider = provider_kind.as_str(), - "Provider circuit 读取超时，按可重建协调状态 fail-open" - ), - Some(Ok(ProviderCircuitDecision::Allow)) => {} - } - } - Ok(RoutingContext { - required_provider: None, - blocked_providers, - }) - } - async fn probe_inner( &self, request: AccountProbeRequest, @@ -1988,11 +1917,6 @@ impl DefaultExecutionService { } }; let events = session.collect_uncommitted().await; - publish_provider_attempt_outcomes( - self.circuits.as_ref(), - session.provider_attempt_outcomes(), - ) - .await; let events = match events { Ok(events) => events, Err(error) => { @@ -2510,9 +2434,7 @@ struct DefaultExecutionSession { admission: Option, active_request: Option, cleanup: Option>, - circuits: Arc, continuation: Arc, - observed_provider_outcomes: usize, continuation_recorded: bool, budget: Option>, } @@ -2522,7 +2444,6 @@ impl DefaultExecutionSession { core: ResponseExecutionSession, admission: ExecutionAdmission, active_request: Option, - circuits: Arc, continuation: Arc, budget: Option>, ) -> Self { @@ -2531,9 +2452,7 @@ impl DefaultExecutionSession { admission: Some(admission), active_request, cleanup: None, - circuits, continuation, - observed_provider_outcomes: 0, continuation_recorded: false, budget, } @@ -2563,16 +2482,6 @@ impl DefaultExecutionSession { } } - async fn observe_provider_outcomes(&mut self) { - let outcomes = self.core.provider_attempt_outcomes(); - let new_outcomes = outcomes - .get(self.observed_provider_outcomes..) - .unwrap_or_default() - .to_vec(); - self.observed_provider_outcomes = outcomes.len(); - publish_provider_attempt_outcomes(self.circuits.as_ref(), &new_outcomes).await; - } - async fn record_continuation(&mut self, state: Option<&ProviderSessionState>) { if self.continuation_recorded { return; @@ -2591,7 +2500,6 @@ impl DefaultExecutionSession { if let Err(error) = self.core.cancel_and_finalize().await { tracing::warn!(%error, "Detached execution 终态收敛失败"); } - self.observe_provider_outcomes().await; self.settle_if_finalized().await; } } @@ -2613,7 +2521,6 @@ impl ExecutionSession for DefaultExecutionSession { if let Ok(Some(event)) = result.as_ref() { self.record_continuation(event.session_update()).await; } - self.observe_provider_outcomes().await; self.settle_if_finalized().await; result }) @@ -2626,7 +2533,6 @@ impl ExecutionSession for DefaultExecutionSession { let state = events.iter().find_map(ProviderEvent::session_update); self.record_continuation(state).await; } - self.observe_provider_outcomes().await; self.settle_if_finalized().await; result }) @@ -2650,7 +2556,6 @@ impl ExecutionSession for DefaultExecutionSession { ) -> BoxFuture<'_, Result<(), EngineError>> { Box::pin(async move { let result = self.core.commit_downstream(client_status_code).await; - self.observe_provider_outcomes().await; self.settle_if_finalized().await; result }) @@ -2662,7 +2567,6 @@ impl ExecutionSession for DefaultExecutionSession { ) -> BoxFuture<'_, Result<(), EngineError>> { Box::pin(async move { let result = self.core.record_client_status(client_status_code).await; - self.observe_provider_outcomes().await; self.settle_if_finalized().await; result }) @@ -2684,39 +2588,6 @@ impl ExecutionSession for DefaultExecutionSession { } } -#[must_use] -pub const fn provider_failure_affects_circuit(error_kind: ProviderErrorKind) -> bool { - matches!( - error_kind, - ProviderErrorKind::Timeout - | ProviderErrorKind::Transport - | ProviderErrorKind::Protocol - | ProviderErrorKind::Unavailable - ) -} - -async fn publish_provider_attempt_outcomes( - circuits: &dyn ProviderCircuitPort, - outcomes: &[ProviderAttemptOutcome], -) { - for outcome in outcomes { - let result = match outcome.error_kind() { - None => circuits.observe_success(outcome.provider_kind()).await, - Some(kind) if provider_failure_affects_circuit(kind) => { - circuits.observe_failure(outcome.provider_kind()).await - } - Some(_) => continue, - }; - if let Err(error) = result { - tracing::warn!( - provider = outcome.provider_kind().as_str(), - %error, - "Provider circuit feedback 写入失败，数据面不受影响" - ); - } - } -} - async fn record_native_continuation( continuation: &dyn NativeContinuationPort, pin: NativeContinuationPin, diff --git a/backend/crates/gateway-core/src/engine/mod.rs b/backend/crates/gateway-core/src/engine/mod.rs index 5ae8b40a7..c68db365c 100644 --- a/backend/crates/gateway-core/src/engine/mod.rs +++ b/backend/crates/gateway-core/src/engine/mod.rs @@ -3,6 +3,7 @@ pub mod admission; pub mod authentication; pub mod budget; +pub mod connection; pub mod continuation; pub mod coordinator; pub mod execution; @@ -120,9 +121,9 @@ impl AttemptTrigger { } } -/// 一次实际上游调用对 Provider 健康度产生的事实。 +/// 一次实际上游调用的诊断结果。 /// -/// 该事实只描述调用结果，不在 Core 中定义 circuit 策略或持久化方式。 +/// 该事实只描述调用结果，不参与跨请求的路由屏蔽。 #[derive(Debug, Clone, PartialEq, Eq)] pub enum ProviderAttemptOutcome { /// 上游流自然完成且通过 canonical event 序列校验。 @@ -331,6 +332,7 @@ pub struct RequestAttemptContext { timing_started_at: Instant, trace: crate::diagnostics::TraceContext, concurrency_wait_budget: crate::concurrency::ConcurrencyWaitBudget, + connection_budget: connection::ConnectionBudget, request_policy: Option, execution_effects: Option>, middleware: Option, @@ -383,6 +385,7 @@ impl RequestAttemptContext { timing_started_at: Instant::now(), trace: crate::diagnostics::TraceContext::default(), concurrency_wait_budget: crate::concurrency::ConcurrencyWaitBudget::default(), + connection_budget: connection::ConnectionBudget::default(), request_policy: None, execution_effects: None, middleware: None, @@ -409,6 +412,13 @@ impl RequestAttemptContext { self } + /// 传递请求内共享的连接恢复预算。 + #[must_use] + pub fn with_connection_budget(mut self, budget: connection::ConnectionBudget) -> Self { + self.connection_budget = budget; + self + } + /// 覆盖本次请求的单调计时原点。 #[must_use] pub fn with_timing_started_at(mut self, timing_started_at: Instant) -> Self { @@ -725,6 +735,11 @@ impl AttemptContext { self.continuation_attempt } + #[must_use] + pub const fn connection_budget(&self) -> &connection::ConnectionBudget { + &self.request.connection_budget + } + /// 返回本次 attempt 的 Provider 传输档位。 #[must_use] pub const fn transport(&self) -> AttemptTransport { @@ -817,6 +832,15 @@ pub struct IntermediateFailure { pub latency: Duration, } +/// 已认证有效请求在执行会话建立前的拒绝；不伪造模型执行或上游 attempt。 +#[derive(Debug)] +pub struct EntryRejection { + pub request_id: ModelRequestId, + pub client_key_id: ClientApiKeyId, + pub error: GatewayError, + pub latency: Duration, +} + /// 不属于任何 `model_requests` 行的管理端账号探测失败。 #[derive(Debug)] pub struct ProbeFailure { @@ -927,6 +951,11 @@ pub trait ExecutionStore: Send + Sync { &self, failure: IntermediateFailure, ) -> Result<(), StoreError>; + /// 记录路由或准入拒绝；可恢复观测失败不改变客户端结果。 + async fn record_entry_rejection(&self, _rejection: EntryRejection) -> Result<(), StoreError> { + Ok(()) + } + /// 记录不挂在 `model_requests` 上的账号探测失败；默认丢弃。 async fn record_probe_failure(&self, _failure: ProbeFailure) -> Result<(), StoreError> { Ok(()) diff --git a/backend/crates/gateway-core/src/error.rs b/backend/crates/gateway-core/src/error.rs index 07d9f3514..3487dafd0 100644 --- a/backend/crates/gateway-core/src/error.rs +++ b/backend/crates/gateway-core/src/error.rs @@ -135,6 +135,11 @@ pub enum PreDeliveryRetry { /// 发起下一次传输尝试前的退避时长。 delay: Duration, }, + /// 可靠 NotSent 的建连恢复；次数与时间由 Core 的请求级预算裁决。 + SameAccountConnectionRetry { + /// Provider 明确指定实际失败的传输，避免 HTTP 恢复重新进入 WS。 + transport: crate::engine::AttemptTransport, + }, /// 固定本次账号，并要求 Provider 使用备用传输。 SameAccountTransportFallback, } @@ -625,6 +630,14 @@ impl ProviderError { })); } + #[must_use] + pub fn with_connection_retry(mut self, transport: crate::engine::AttemptTransport) -> Self { + self.pre_delivery_retry = Some(Box::new(PreDeliveryRetry::SameAccountConnectionRetry { + transport, + })); + self + } + /// 要求 Core 在账号凭据已恢复后仅对原账号重放一次。 #[must_use] pub const fn with_same_account_retry(mut self) -> Self { diff --git a/backend/crates/gateway-core/src/lib.rs b/backend/crates/gateway-core/src/lib.rs index 06e14e1eb..980ad70f4 100644 --- a/backend/crates/gateway-core/src/lib.rs +++ b/backend/crates/gateway-core/src/lib.rs @@ -32,7 +32,6 @@ use engine::admission::{ use engine::continuation::NativeContinuationPort; use engine::execution::{ ClientApiKeyUsageSink, ClientKeyVerifier, DefaultExecutionService, ExecutionService, - ProviderCircuitPort, }; use engine::nested::{AffinityLookupPort, NestedModelExecutionPort}; use engine::probe::AccountProbe; @@ -51,7 +50,6 @@ pub struct CoreStorePorts { execution: Arc, admissions: Arc, admission_recovery: Arc, - circuits: Arc, continuation: Arc, snapshots: Arc, snapshot_subscriptions: Arc, @@ -68,7 +66,6 @@ impl CoreStorePorts { Arc, Arc, ), - circuits: Arc, continuation: Arc, (snapshots, snapshot_subscriptions): ( Arc, @@ -80,7 +77,6 @@ impl CoreStorePorts { execution, admissions, admission_recovery, - circuits, continuation, snapshots, snapshot_subscriptions, @@ -198,7 +194,6 @@ fn build_execution_service( Arc::clone(&ports.execution), providers, Arc::clone(&ports.admissions), - Arc::clone(&ports.circuits), Arc::clone(&ports.continuation), Arc::clone(&ports.client_api_key_usage), ); diff --git a/backend/crates/gateway-core/src/routing/snapshot.rs b/backend/crates/gateway-core/src/routing/snapshot.rs index 8562721aa..b7abd23a8 100644 --- a/backend/crates/gateway-core/src/routing/snapshot.rs +++ b/backend/crates/gateway-core/src/routing/snapshot.rs @@ -1046,7 +1046,7 @@ impl RuntimeSnapshot { self.client_policies.get(id) } - /// 返回当前 Key 范围、本代次注册表与 circuit 条件共同允许的 Provider。 + /// 返回当前 Key 范围、本代次注册表与请求路由限制共同允许的 Provider。 #[must_use] pub fn available_providers( &self, @@ -1167,7 +1167,7 @@ impl RuntimeSnapshot { } if candidates.is_empty() { - // 模型存在性必须包含被熔断的 Provider；目录未知时不能断言模型不存在。 + // 模型存在性必须包含被请求路由限制排除的 Provider；目录未知时不能断言模型不存在。 let mut scoped_providers = providers.intersection(&self.providers).peekable(); if scoped_providers.peek().is_some() && scoped_providers.all(|provider| { diff --git a/backend/crates/gateway-core/tests/engine/connection.rs b/backend/crates/gateway-core/tests/engine/connection.rs new file mode 100644 index 000000000..130540c1a --- /dev/null +++ b/backend/crates/gateway-core/tests/engine/connection.rs @@ -0,0 +1,26 @@ +use gateway_core::engine::connection::ConnectionBudget; + +#[test] +fn all_transports_share_one_opening_budget_but_other_requests_do_not() { + let request = ConnectionBudget::default(); + let http = request.clone(); + let websocket = request.clone(); + for transport in [&http, &websocket, &http, &websocket] { + assert!(transport.begin().unwrap().is_some()); + } + assert!(request.exhausted()); + assert!(http.begin().is_err()); + assert!(websocket.retry_delay(0, "req_shared").is_none()); + assert!(ConnectionBudget::default().begin().is_ok()); +} + +#[test] +fn successful_opening_does_not_limit_later_generation_or_protocol_recovery() { + let request = ConnectionBudget::default(); + request.begin().unwrap(); + request.complete(); + assert!(request.startup_remaining().is_none()); + assert!(request.retry_delay(0, "req_sent").is_none()); + assert!(!request.exhausted()); + assert!(request.begin().unwrap().is_none()); +} diff --git a/backend/crates/gateway-core/tests/engine/coordinator.rs b/backend/crates/gateway-core/tests/engine/coordinator.rs index 807235571..66db6fa9f 100644 --- a/backend/crates/gateway-core/tests/engine/coordinator.rs +++ b/backend/crates/gateway-core/tests/engine/coordinator.rs @@ -4738,3 +4738,199 @@ fn provider_default_profile_is_applied_once_and_frozen_across_account_retries() ); assert_eq!(contexts[1].request_profile(), Some(first)); } + +#[test] +fn connection_recovery_keeps_one_account_and_http_transport_until_success() { + let operation = generate_operation(); + let route_plan = plan(&operation); + let failure = || Script::Stream { + account_id: "acct_only", + items: vec![Err(ProviderError::new( + ProviderErrorKind::Transport, + UpstreamSendState::NotSent, + ) + .with_connection_retry(AttemptTransport::Fallback))], + }; + let (coordinator, store, provider) = coordinator(vec![ + failure(), + failure(), + failure(), + Script::Stream { + account_id: "acct_only", + items: complete_stream(None), + }, + ]); + let mut session = block_on(coordinator.start( + model_request(&operation, SystemTime::now() + Duration::from_secs(30)), + operation, + route_plan, + None, + None, + CancellationToken::new(), + )) + .unwrap(); + block_on(session.collect_uncommitted()).unwrap(); + block_on(session.commit_downstream(Some(200))).unwrap(); + let contexts = provider.contexts.lock().unwrap(); + assert_eq!(contexts.len(), 4); + let account = ProviderAccountId::new("acct_only").unwrap(); + for context in contexts.iter().skip(1) { + assert_eq!(context.required_account(), Some(&account)); + assert_eq!(context.transport(), AttemptTransport::Fallback); + assert!(!context.excluded_accounts().contains(&account)); + } + let state = store.state.lock().unwrap(); + assert_eq!(state.intermediate_failures, 3); + assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Succeeded); + assert_eq!(provider.released_leases.load(Ordering::SeqCst), 3); + drop(session); + assert_eq!(provider.released_leases.load(Ordering::SeqCst), 4); +} + +#[test] +fn connection_recovery_exhaustion_preserves_transport_failure() { + let operation = generate_operation(); + let route_plan = plan(&operation); + let (coordinator, store, provider) = coordinator( + (0..4) + .map(|_| Script::Stream { + account_id: "acct_only", + items: vec![Err(ProviderError::new( + ProviderErrorKind::Transport, + UpstreamSendState::NotSent, + ) + .with_connection_retry(AttemptTransport::Fallback))], + }) + .collect(), + ); + let mut session = block_on(coordinator.start( + model_request(&operation, SystemTime::now() + Duration::from_secs(30)), + operation, + route_plan, + None, + None, + CancellationToken::new(), + )) + .unwrap(); + let error = block_on(session.collect_uncommitted()).unwrap_err(); + assert!( + matches!(error, EngineError::Provider(ref error) if error.kind() == ProviderErrorKind::Transport) + ); + assert_eq!(provider.contexts.lock().unwrap().len(), 4); + assert_eq!( + store.state.lock().unwrap().finalizations[0].attempt_count, + 4 + ); + assert_eq!(provider.released_leases.load(Ordering::SeqCst), 3); + drop(session); + assert_eq!(provider.released_leases.load(Ordering::SeqCst), 4); +} + +#[test] +fn connection_recovery_never_replays_sent_or_ambiguous_payloads() { + for send_state in [UpstreamSendState::Sent, UpstreamSendState::Ambiguous] { + let operation = generate_operation(); + let route_plan = plan(&operation); + let (coordinator, store, provider) = coordinator(vec![Script::Stream { + account_id: "acct_only", + items: vec![Err(ProviderError::new( + ProviderErrorKind::Transport, + send_state, + ) + .with_connection_retry(AttemptTransport::Fallback))], + }]); + let mut session = block_on(coordinator.start( + model_request(&operation, SystemTime::now() + Duration::from_secs(30)), + operation, + route_plan, + None, + None, + CancellationToken::new(), + )) + .unwrap(); + assert!(block_on(session.collect_uncommitted()).is_err()); + assert_eq!(provider.contexts.lock().unwrap().len(), 1); + assert_eq!(store.state.lock().unwrap().intermediate_failures, 0); + } +} + +#[test] +fn connection_backoff_cancellation_releases_account_without_another_attempt() { + let operation = generate_operation(); + let route_plan = plan(&operation); + let (coordinator, _, provider) = coordinator(vec![Script::Stream { + account_id: "acct_first", + items: vec![Err(ProviderError::new( + ProviderErrorKind::Transport, + UpstreamSendState::NotSent, + ) + .with_connection_retry(AttemptTransport::Fallback))], + }]); + let cancellation = CancellationToken::new(); + let mut session = block_on(coordinator.start( + model_request(&operation, SystemTime::now() + Duration::from_secs(30)), + operation, + route_plan, + None, + None, + cancellation.clone(), + )) + .expect("start execution"); + let cancel = std::thread::spawn(move || { + std::thread::sleep(Duration::from_millis(50)); + cancellation.cancel(); + }); + assert!(matches!( + block_on(session.collect_uncommitted()), + Err(EngineError::Cancelled) + )); + cancel.join().expect("cancel task"); + assert_eq!(provider.contexts.lock().expect("contexts").len(), 1); + assert_eq!(provider.released_leases.load(Ordering::SeqCst), 1); +} + +#[test] +fn connection_failure_after_a_replay_safe_rejection_keeps_existing_account_rotation() { + let operation = generate_operation(); + let route_plan = plan(&operation); + let (coordinator, _, provider) = coordinator(vec![ + Script::Stream { + account_id: "acct_first", + items: vec![Err(ProviderError::new( + ProviderErrorKind::RateLimited, + UpstreamSendState::Sent, + ) + .with_replay_safe())], + }, + Script::Stream { + account_id: "acct_second", + items: vec![Err(ProviderError::new( + ProviderErrorKind::Transport, + UpstreamSendState::NotSent, + ) + .with_connection_retry(AttemptTransport::Fallback))], + }, + Script::Stream { + account_id: "acct_other", + items: complete_stream(None), + }, + ]); + let mut session = block_on(coordinator.start( + model_request(&operation, SystemTime::now() + Duration::from_secs(30)), + operation, + route_plan, + None, + None, + CancellationToken::new(), + )) + .unwrap(); + block_on(session.collect_uncommitted()).unwrap(); + let contexts = provider.contexts.lock().unwrap(); + assert_eq!(contexts.len(), 3); + assert!(contexts[2].required_account().is_none()); + assert!( + contexts[2] + .excluded_accounts() + .contains(&ProviderAccountId::new("acct_second").unwrap()) + ); +} diff --git a/backend/crates/gateway-core/tests/engine/execution.rs b/backend/crates/gateway-core/tests/engine/execution.rs index 5b0665fc8..06618f38d 100644 --- a/backend/crates/gateway-core/tests/engine/execution.rs +++ b/backend/crates/gateway-core/tests/engine/execution.rs @@ -111,7 +111,6 @@ fn service(admissions: Arc, budget: Arc) -> DefaultExecution Arc::new(TrackingExecutionStore::default()), ProviderRegistry::default(), admissions, - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ) @@ -280,7 +279,6 @@ fn reused_client_uses_updated_limits_for_each_execution() { Arc::new(TrackingExecutionStore::default()), ProviderRegistry::default(), admissions.clone(), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ); @@ -312,7 +310,6 @@ fn reused_client_cannot_start_after_key_disable_or_snapshot_suspension() { Arc::new(TrackingExecutionStore::default()), ProviderRegistry::default(), admissions.clone(), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ); @@ -416,7 +413,6 @@ fn reused_client_is_reauthenticated_by_the_current_frontend_plan_without_identit Arc::new(TrackingExecutionStore::default()), ProviderRegistry::default(), admissions.clone(), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ) @@ -508,7 +504,6 @@ fn early_failure_service( store, ProviderRegistry::new([Arc::new(LocalFailingProvider) as Arc]).unwrap(), admissions, - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ) @@ -972,7 +967,6 @@ fn completed_parent_rejects_new_nested_execution_and_cancels_an_active_child() { Arc::new(TrackingExecutionStore::default()), provider_index.clone(), Arc::new(Admissions::default()), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ) @@ -1164,7 +1158,6 @@ fn native_response_processing_uses_the_real_translation_boundary() { Arc::new(TrackingExecutionStore::default()), provider_index.clone(), Arc::new(Admissions::default()), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ) @@ -1315,7 +1308,6 @@ fn response_translation_zero_output_preserves_canonical_usage_and_finalization() store.clone(), provider_index.clone(), Arc::new(Admissions::default()), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ); @@ -1469,7 +1461,6 @@ fn native_response_failure_after_downstream_commit_is_not_replayed() { Arc::new(TrackingExecutionStore::default()), provider_index.clone(), Arc::new(Admissions::default()), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ); @@ -1542,7 +1533,6 @@ fn model_routing_policy_selects_native_provider() { Arc::new(TrackingExecutionStore::default()), provider_index.clone(), Arc::new(Admissions::default()), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ) @@ -1596,7 +1586,6 @@ fn model_routing_reject_stops_before_provider_execution() { Arc::new(TrackingExecutionStore::default()), provider_index.clone(), Arc::new(Admissions::default()), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ) @@ -1681,7 +1670,6 @@ fn model_routing_cannot_expand_the_frozen_key_model_scope() { Arc::new(TrackingExecutionStore::default()), provider_index.clone(), Arc::new(Admissions::default()), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ) @@ -1828,7 +1816,6 @@ fn scheduler_reject_is_a_terminal_policy_rejection_before_upstream_send() { Arc::new(TrackingExecutionStore::default()), provider_index.clone(), Arc::new(Admissions::default()), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ) @@ -2103,7 +2090,6 @@ fn observation_service( Arc::new(TrackingExecutionStore::default()), providers, admissions, - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ) @@ -2141,7 +2127,6 @@ fn fallback_exhaustion_observation_uses_the_last_actual_provider_not_the_candida Arc::new(TrackingExecutionStore::default()), providers, Arc::new(Admissions::default()), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ) @@ -2206,7 +2191,6 @@ fn routing_external_effect_stops_a_not_sent_provider_retry() { Arc::new(TrackingExecutionStore::default()), providers, Arc::new(Admissions::default()), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ) @@ -2408,7 +2392,6 @@ fn charged_service( }) as Arc]) .unwrap(), admissions, - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ); @@ -2686,9 +2669,8 @@ use gateway_core::engine::continuation::{ }; use gateway_core::engine::execution::{ ClientApiKeyUsageSink, ClientKeyVerifier, ClientTransport, DefaultExecutionService, - ExecutionRequestMetadata, ExecutionService, ExecutionSession, ProviderCircuitDecision, - ProviderCircuitError, ProviderCircuitPort, StartExecution, StartProviderExecution, - provider_failure_affects_circuit, + ExecutionRequestMetadata, ExecutionService, ExecutionSession, StartExecution, + StartProviderExecution, }; use gateway_core::engine::nested::NestedModelExecutionRequest; use gateway_core::engine::observation::{ @@ -2735,30 +2717,6 @@ use gateway_core::runtime::{ use gateway_core::upstream::{UpstreamSendState, UpstreamTransport}; use serde_json::{Value, json}; -#[test] -fn only_provider_attributable_failures_should_affect_circuit() { - assert!(provider_failure_affects_circuit(ProviderErrorKind::Timeout)); - assert!(provider_failure_affects_circuit( - ProviderErrorKind::Transport - )); - assert!(!provider_failure_affects_circuit( - ProviderErrorKind::RateLimited - )); - assert!(!provider_failure_affects_circuit( - ProviderErrorKind::InvalidRequest - )); - assert!(!provider_failure_affects_circuit( - ProviderErrorKind::ContinuationRecoveryRequired - )); - assert!(!provider_failure_affects_circuit( - ProviderErrorKind::UpstreamCapacityUnavailable - )); - // 上游 close 1009（message too big）是请求自身的问题：不得熔断 provider。 - assert!(!provider_failure_affects_circuit( - ProviderErrorKind::MessageTooBig - )); -} - #[test] fn account_probe_should_not_write_to_the_persistent_execution_store() { let store = Arc::new(TrackingExecutionStore::default()); @@ -2767,7 +2725,6 @@ fn account_probe_should_not_write_to_the_persistent_execution_store() { store.clone(), ProviderRegistry::default(), Arc::new(UnusedAdmissions), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ); @@ -2799,7 +2756,6 @@ fn probe_failures_should_be_observable_without_a_model_request_row() { store.clone(), providers, Arc::new(UnusedAdmissions), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ); @@ -2845,7 +2801,6 @@ fn provider_local_probe_failure_should_remain_distinct_from_upstream() { store, providers, Arc::new(UnusedAdmissions), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ); @@ -2888,7 +2843,6 @@ fn diagnostic_probe_does_not_apply_data_plane_account_model_policy() { ProviderRegistry::new([Arc::new(LocalFailingProvider) as Arc]) .expect("providers"), Arc::new(UnusedAdmissions), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ); @@ -2920,7 +2874,6 @@ fn probe_observation_store_failure_preserves_the_provider_error() { store.clone(), providers, Arc::new(UnusedAdmissions), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ); @@ -3015,7 +2968,6 @@ fn client_catalog_forwards_scope_maps_whole_objects_and_omits_unroutable_models( ProviderRegistry::new([Arc::new(NativeCatalogProvider { fail }) as Arc]) .expect("registry"), Arc::new(UnusedAdmissions), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ); @@ -3167,7 +3119,6 @@ fn successful_authentication_should_record_client_key_usage() { Arc::new(TrackingExecutionStore::default()), ProviderRegistry::default(), Arc::new(UnusedAdmissions), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), usage.clone(), ); @@ -3187,7 +3138,6 @@ fn client_key_verification_should_not_record_client_key_usage() { Arc::new(TrackingExecutionStore::default()), ProviderRegistry::default(), Arc::new(UnusedAdmissions), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), usage.clone(), ); @@ -3209,7 +3159,6 @@ fn request_verification_should_apply_entry_authentication_without_recording_key_ Arc::new(TrackingExecutionStore::default()), ProviderRegistry::default(), Arc::new(UnusedAdmissions), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), usage.clone(), ); @@ -3243,7 +3192,6 @@ fn assert_provider_endpoint_observation(model: Option<&str>) { }) as Arc]) .expect("provider registry"), Arc::new(UnusedAdmissions), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ); @@ -3289,84 +3237,14 @@ fn assert_provider_endpoint_observation(model: Option<&str>) { assert!(upstream_models.iter().all(Option::is_none)); } -#[test] -fn circuit_store_failure_should_fail_open_during_request_start() { - let service = DefaultExecutionService::new( - RuntimeSnapshotHandle::new(start_snapshot()), - Arc::new(TrackingExecutionStore::default()), - ProviderRegistry::default(), - Arc::new(UnusedAdmissions), - Arc::new(FailingDecisionCircuits), - Arc::new(UnusedContinuation), - Arc::new(RecordingClientApiKeyUsage::default()), - ); - let client = service - .authenticate("sk_start_test") - .expect("authenticated client"); - - let started = block_on(service.start(StartExecution { - client, - public_model: PublicModelId::new("gpt-start").expect("public model"), - operation: start_operation(), - metadata: ExecutionRequestMetadata { - protocol: "openai".to_owned(), - endpoint: "/v1/responses".to_owned(), - transport: ClientTransport::HttpJson, - stream: false, - client_ip: None, - user_agent: None, - previous_response_id: None, - }, - })) - .expect("recoverable circuit state must not reject the request"); - - assert!(!started.session.is_finalized()); -} - -#[test] -fn slow_circuit_store_should_time_out_and_fail_open_during_request_start() { - let service = DefaultExecutionService::new( - RuntimeSnapshotHandle::new(start_snapshot()), - Arc::new(TrackingExecutionStore::default()), - ProviderRegistry::default(), - Arc::new(UnusedAdmissions), - Arc::new(PendingDecisionCircuits), - Arc::new(UnusedContinuation), - Arc::new(RecordingClientApiKeyUsage::default()), - ); - let client = service - .authenticate("sk_start_test") - .expect("authenticated client"); - let started_at = Instant::now(); - - let started = block_on(service.start(StartExecution { - client, - public_model: PublicModelId::new("gpt-start").expect("public model"), - operation: start_operation(), - metadata: ExecutionRequestMetadata { - protocol: "openai".to_owned(), - endpoint: "/v1/responses".to_owned(), - transport: ClientTransport::HttpJson, - stream: false, - client_ip: None, - user_agent: None, - previous_response_id: None, - }, - })) - .expect("slow recoverable circuit state must not reject the request"); - - assert!(started_at.elapsed() < Duration::from_secs(2)); - assert!(!started.session.is_finalized()); -} - #[test] fn known_catalog_should_reject_a_model_that_the_provider_did_not_publish() { + let store = Arc::new(TrackingExecutionStore::default()); let service = DefaultExecutionService::new( RuntimeSnapshotHandle::new(start_snapshot()), - Arc::new(TrackingExecutionStore::default()), + store.clone(), ProviderRegistry::default(), Arc::new(UnusedAdmissions), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ); @@ -3400,6 +3278,10 @@ fn known_catalog_should_reject_a_model_that_the_provider_did_not_publish() { "the requested model was not found in the provider catalogs available to this API key; check the model name", ) ); + let rejections = store.entry_rejections.lock().unwrap(); + assert_eq!(rejections.len(), 1); + assert_eq!(rejections[0].error.kind(), GatewayErrorKind::ModelNotFound); + assert!(store.requests.lock().unwrap().is_empty()); } #[test] @@ -3409,7 +3291,6 @@ fn continuation_owned_by_another_client_api_key_should_fail_closed() { Arc::new(TrackingExecutionStore::default()), ProviderRegistry::default(), Arc::new(UnusedAdmissions), - Arc::new(UnusedCircuits), Arc::new(RejectedContinuation::OwnershipMismatch), Arc::new(RecordingClientApiKeyUsage::default()), ); @@ -3437,7 +3318,6 @@ fn invalid_continuation_record_should_not_be_forwarded_as_an_external_handle() { Arc::new(TrackingExecutionStore::default()), ProviderRegistry::default(), Arc::new(UnusedAdmissions), - Arc::new(UnusedCircuits), Arc::new(RejectedContinuation::InvalidData), Arc::new(RecordingClientApiKeyUsage::default()), ); @@ -3494,6 +3374,7 @@ impl ClientApiKeyUsageSink for RecordingClientApiKeyUsage { struct TrackingExecutionStore { touched: AtomicBool, probe_failures: Mutex>, + entry_rejections: Mutex>, requests: Mutex>, attempts: Mutex>, finalizations: Mutex>, @@ -3556,6 +3437,14 @@ impl TrackingExecutionStore { #[async_trait] impl ExecutionStore for TrackingExecutionStore { + async fn record_entry_rejection( + &self, + rejection: gateway_core::engine::EntryRejection, + ) -> Result<(), StoreError> { + self.entry_rejections.lock().unwrap().push(rejection); + Ok(()) + } + async fn create_model_request(&self, request: NewModelRequest) -> Result<(), StoreError> { self.touch(); self.creates.fetch_add(1, Ordering::SeqCst); @@ -3673,81 +3562,6 @@ impl ClientAdmissionPort for UnusedAdmissions { } } -struct UnusedCircuits; - -impl ProviderCircuitPort for UnusedCircuits { - fn decision<'a>( - &'a self, - _: &'a ProviderKind, - ) -> BoxFuture<'a, Result> { - Box::pin(async { Ok(ProviderCircuitDecision::Allow) }) - } - - fn observe_failure<'a>( - &'a self, - _: &'a ProviderKind, - ) -> BoxFuture<'a, Result<(), ProviderCircuitError>> { - Box::pin(async { Ok(()) }) - } - - fn observe_success<'a>( - &'a self, - _: &'a ProviderKind, - ) -> BoxFuture<'a, Result<(), ProviderCircuitError>> { - Box::pin(async { Ok(()) }) - } -} - -struct FailingDecisionCircuits; - -impl ProviderCircuitPort for FailingDecisionCircuits { - fn decision<'a>( - &'a self, - _: &'a ProviderKind, - ) -> BoxFuture<'a, Result> { - Box::pin(async { Err(ProviderCircuitError) }) - } - - fn observe_failure<'a>( - &'a self, - _: &'a ProviderKind, - ) -> BoxFuture<'a, Result<(), ProviderCircuitError>> { - Box::pin(async { Ok(()) }) - } - - fn observe_success<'a>( - &'a self, - _: &'a ProviderKind, - ) -> BoxFuture<'a, Result<(), ProviderCircuitError>> { - Box::pin(async { Ok(()) }) - } -} - -struct PendingDecisionCircuits; - -impl ProviderCircuitPort for PendingDecisionCircuits { - fn decision<'a>( - &'a self, - _: &'a ProviderKind, - ) -> BoxFuture<'a, Result> { - Box::pin(futures::future::pending()) - } - - fn observe_failure<'a>( - &'a self, - _: &'a ProviderKind, - ) -> BoxFuture<'a, Result<(), ProviderCircuitError>> { - Box::pin(async { Ok(()) }) - } - - fn observe_success<'a>( - &'a self, - _: &'a ProviderKind, - ) -> BoxFuture<'a, Result<(), ProviderCircuitError>> { - Box::pin(async { Ok(()) }) - } -} - struct UnusedContinuation; impl NativeContinuationPort for UnusedContinuation { @@ -3993,7 +3807,6 @@ fn account_wait_inherits_the_budget_spent_during_client_admission() { ) .unwrap(), admissions.clone(), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ); @@ -4101,7 +3914,6 @@ fn queue_service( Arc::new(TrackingExecutionStore::default()), ProviderRegistry::default(), admissions.clone(), - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ); @@ -4226,7 +4038,6 @@ fn reused_websocket_client_gets_group_fast_policy_from_each_new_request_snapshot Arc::new(TrackingExecutionStore::default()), ProviderRegistry::new([provider.clone() as Arc]).unwrap(), admissions, - Arc::new(UnusedCircuits), Arc::new(UnusedContinuation), Arc::new(RecordingClientApiKeyUsage::default()), ); @@ -4246,3 +4057,54 @@ fn reused_websocket_client_gets_group_fast_policy_from_each_new_request_snapshot } assert_eq!(*provider.policies.lock().unwrap(), vec![true, false]); } + +#[test] +fn repeated_connection_failures_never_block_later_requests_for_the_provider() { + let store = Arc::new(TrackingExecutionStore::default()); + let service = DefaultExecutionService::new( + RuntimeSnapshotHandle::new(client_snapshot()), + store.clone(), + ProviderRegistry::new([Arc::new(ColdFailingProvider { + requested_model: None, + }) as Arc]) + .expect("provider registry"), + Arc::new(UnusedAdmissions), + Arc::new(UnusedContinuation), + Arc::new(RecordingClientApiKeyUsage::default()), + ); + for _ in 0..5 { + let client = service + .authenticate("sk_usage_test") + .expect("authenticated client"); + let operation = Operation::GenerateImage(ImageRequest::from_raw_json( + ImageRequestKind::Generation, + RawJsonPayload::new( + "openai", + Bytes::from_static(br#"{"model":"gpt-image-2","prompt":"hello"}"#), + ) + .expect("image payload"), + )); + + let mut started = block_on(service.start_provider_endpoint(StartProviderExecution { + client, + provider: ProviderKind::new("openai").expect("provider"), + upstream_model: None, + operation, + metadata: ExecutionRequestMetadata { + protocol: "openai".to_owned(), + endpoint: "/v1/images/generations".to_owned(), + transport: ClientTransport::HttpJson, + stream: false, + client_ip: None, + user_agent: None, + previous_response_id: None, + }, + })) + .expect("provider endpoint request should start without a text catalog entry"); + + let _error = block_on(started.session.collect_uncommitted()) + .expect_err("the cold provider stops execution after persistence"); + } + assert_eq!(store.requests.lock().unwrap().len(), 5); + assert!(store.entry_rejections.lock().unwrap().is_empty()); +} diff --git a/backend/crates/gateway-core/tests/engine/mod.rs b/backend/crates/gateway-core/tests/engine/mod.rs index 77f636625..a1edf0e2c 100644 --- a/backend/crates/gateway-core/tests/engine/mod.rs +++ b/backend/crates/gateway-core/tests/engine/mod.rs @@ -1,4 +1,5 @@ mod admission; +mod connection; mod continuation; mod coordinator; mod execution; diff --git a/backend/crates/gateway-store/src/bundle.rs b/backend/crates/gateway-store/src/bundle.rs index 721e597aa..842ae8ac0 100644 --- a/backend/crates/gateway-store/src/bundle.rs +++ b/backend/crates/gateway-store/src/bundle.rs @@ -57,7 +57,7 @@ impl StoreBundle { Ok(()) } - /// 命令成功、失败或取消后有界排空账本、Key 使用、准入释放与 circuit 反馈。 + /// 命令成功、失败或取消后有界排空账本、Key 使用与准入释放。 pub async fn shutdown_command_line_writes(&mut self) -> Result<(), CommandStoreDrainError> { match self.command_drain.take() { Some(drain) => drain.shutdown().await, @@ -192,12 +192,6 @@ async fn connect( let admissions: Arc = Arc::new( redis::RedisClientAdmissionRepository::new(redis_connection.clone(), REDIS_NAMESPACE)?, ); - let circuits: Arc = - Arc::new(redis::RedisProviderCircuitRepository::new( - redis_connection.clone(), - REDIS_NAMESPACE, - gateway_core::engine::execution::ProviderCircuitPolicy::default(), - )?); // Continuation affinity 是下一轮请求的路由事实，Core 必须直接等待 Redis 确认。 let continuation: Arc = Arc::new(redis::RedisNativeContinuationRepository::new( @@ -206,7 +200,6 @@ async fn connect( )?); let (admissions, admission_release_writer) = redis::BufferedClientAdmissionPort::new(admissions); - let (circuits, circuit_feedback_writer) = redis::BufferedProviderCircuitPort::new(circuits); let core_ports = CoreStorePorts::new( execution, ( @@ -215,7 +208,6 @@ async fn connect( pool.clone(), )), ), - Arc::new(circuits), continuation, ( Arc::new(postgres::PgRuntimeSnapshotRepository::new(pool.clone())), @@ -263,7 +255,6 @@ async fn connect( execution_writer, client_key_usage_writer, admission_release_writer, - circuit_feedback_writer, retention, )?, None, @@ -274,7 +265,6 @@ async fn connect( execution: execution_writer, client_key_usage: client_key_usage_writer, admission_release: admission_release_writer, - circuit_feedback: circuit_feedback_writer, }), ), }; diff --git a/backend/crates/gateway-store/src/postgres/execution.rs b/backend/crates/gateway-store/src/postgres/execution.rs index 09725f420..ecfbd7851 100644 --- a/backend/crates/gateway-store/src/postgres/execution.rs +++ b/backend/crates/gateway-store/src/postgres/execution.rs @@ -1108,6 +1108,48 @@ impl ExecutionStore for PgExecutionStore { .map_err(core_store_error) } + async fn record_entry_rejection( + &self, + rejection: gateway_core::engine::EntryRejection, + ) -> Result<(), CoreStoreError> { + let error = rejection.error; + super::OpsEventRepository::append_ops_event( + &super::PgOpsEventRepository::new(self.pool.clone()), + super::OpsEvent { + id: Uuid::now_v7().to_string(), + model_request_id: None, + attempt_index: None, + level: super::OpsEventLevel::Warning, + component: "request_entry".to_owned(), + operation: "reject".to_owned(), + provider_kind: None, + provider_account_id: None, + provider_account_ref: None, + upstream_model_id: None, + failure_kind: error.kind().as_str().to_owned(), + upstream_send_state: None, + raw_upstream_error: None, + status_code: None, + provider_error_code: error.client_error_code().map(str::to_owned), + retry_after_ms: error + .retry_after() + .and_then(|delay| u64::try_from(delay.as_millis()).ok()), + upstream_request_id: None, + latency_ms: u64::try_from(rejection.latency.as_millis()).ok(), + // 入口尚无 model_requests 行，关联 ID 放在安全消息中，不能伪造外键。 + message: serde_json::json!({ + "requestId": rejection.request_id.as_str(), + "clientKeyId": rejection.client_key_id.as_str(), + "message": error.client_message(), + }) + .to_string(), + created_at: Utc::now(), + }, + ) + .await + .map_err(core_store_error) + } + async fn record_probe_failure(&self, failure: ProbeFailure) -> Result<(), CoreStoreError> { let error = failure.error; let retry_after_ms = error diff --git a/backend/crates/gateway-store/src/postgres/execution_buffer.rs b/backend/crates/gateway-store/src/postgres/execution_buffer.rs index 0107671ee..b8b3997fc 100644 --- a/backend/crates/gateway-store/src/postgres/execution_buffer.rs +++ b/backend/crates/gateway-store/src/postgres/execution_buffer.rs @@ -9,8 +9,8 @@ use std::time::{Duration, Instant, SystemTime}; use async_trait::async_trait; use futures::future::try_join_all; use gateway_core::engine::{ - AttemptRecord, ExecutionStore, IntermediateFailure, ModelRequestFinalization, ModelRequestId, - NewModelRequest, ProbeFailure, RecoveryReport, + AttemptRecord, EntryRejection, ExecutionStore, IntermediateFailure, ModelRequestFinalization, + ModelRequestId, NewModelRequest, ProbeFailure, RecoveryReport, }; use gateway_core::error::{ProviderError, StoreError}; use gateway_core::lifecycle::CancellationToken; @@ -358,6 +358,13 @@ where Ok(()) } + async fn record_entry_rejection(&self, rejection: EntryRejection) -> Result<(), StoreError> { + self.enqueue(ExecutionObservationWrite::EntryRejection(Box::new( + rejection, + ))); + Ok(()) + } + async fn record_probe_failure(&self, failure: ProbeFailure) -> Result<(), StoreError> { self.enqueue(ExecutionObservationWrite::ProbeFailure(Box::new(failure))); Ok(()) @@ -677,6 +684,7 @@ enum ExecutionObservationWrite { }, IntermediateFailure(Box), ProbeFailure(Box), + EntryRejection(Box), Finalize(Box), } @@ -691,6 +699,7 @@ impl ExecutionObservationWrite { Self::RecordClientStatus { .. } => "record_client_status", Self::IntermediateFailure(_) => "record_intermediate_failure", Self::ProbeFailure(_) => "record_probe_failure", + Self::EntryRejection(_) => "record_entry_rejection", Self::Finalize(_) => "finalize_model_request", } } @@ -704,7 +713,7 @@ impl ExecutionObservationWrite { | Self::MarkDownstreamCommitted { request_id, .. } | Self::RecordClientStatus { request_id, .. } => Some(request_id.as_str()), Self::IntermediateFailure(failure) => Some(failure.request_id.as_str()), - Self::ProbeFailure(_) => None, + Self::ProbeFailure(_) | Self::EntryRejection(_) => None, Self::Finalize(finalization) => Some(finalization.request_id.as_str()), } } @@ -730,6 +739,14 @@ impl ExecutionObservationWrite { failure.upstream_request_id.as_deref(), ]) .saturating_add(provider_error_bytes(&failure.error)), + Self::EntryRejection(rejection) => text_bytes([ + Some(rejection.request_id.as_str()), + Some(rejection.client_key_id.as_str()), + Some(rejection.error.client_message()), + rejection.error.client_error_code(), + rejection.error.client_error_type(), + ]) + .saturating_add(size_of::()), Self::ProbeFailure(failure) => text_bytes([ Some(failure.provider_kind.as_str()), Some(failure.account_id.as_str()), @@ -799,6 +816,7 @@ impl ExecutionObservationWrite { } Self::IntermediateFailure(failure) => store.record_intermediate_failure(*failure).await, Self::ProbeFailure(failure) => store.record_probe_failure(*failure).await, + Self::EntryRejection(rejection) => store.record_entry_rejection(*rejection).await, Self::Finalize(finalization) => store.finalize_model_request(*finalization).await, } } diff --git a/backend/crates/gateway-store/src/redis/coordination_buffer.rs b/backend/crates/gateway-store/src/redis/coordination_buffer.rs index ab0334969..8bed5699a 100644 --- a/backend/crates/gateway-store/src/redis/coordination_buffer.rs +++ b/backend/crates/gateway-store/src/redis/coordination_buffer.rs @@ -15,12 +15,8 @@ use gateway_core::engine::admission::{ ClientAdmissionDecision, ClientAdmissionError, ClientAdmissionPort, ClientAdmissionRecovery, ClientAdmissionRequest, ClientAdmissionRestoreResult, }; -use gateway_core::engine::execution::{ - ProviderCircuitDecision, ProviderCircuitError, ProviderCircuitPort, -}; use gateway_core::lifecycle::CancellationToken; use gateway_core::policy::ClientApiKeyId; -use gateway_core::routing::ProviderKind; use gateway_core::task::{DaemonTask, WorkerTaskError}; use tokio::sync::{Mutex, mpsc}; @@ -154,123 +150,6 @@ impl DaemonTask for ClientAdmissionReleaseWriter { } } -/// Circuit decision 仍直读 Redis；success/failure feedback 只做有界入队。 -#[derive(Clone)] -pub struct BufferedProviderCircuitPort { - inner: Arc, - sender: mpsc::Sender, -} - -impl BufferedProviderCircuitPort { - #[must_use] - pub fn new(inner: Arc) -> (Self, ProviderCircuitFeedbackWriter) { - Self::with_capacity( - inner, - NonZeroUsize::new(DEFAULT_QUEUE_CAPACITY).expect("queue capacity is non-zero"), - ) - } - - #[must_use] - pub fn with_capacity( - inner: Arc, - capacity: NonZeroUsize, - ) -> (Self, ProviderCircuitFeedbackWriter) { - let (sender, receiver) = mpsc::channel(capacity.get()); - ( - Self { - inner: Arc::clone(&inner), - sender, - }, - ProviderCircuitFeedbackWriter { - inner, - receiver: Mutex::new(receiver), - }, - ) - } - - fn enqueue(&self, feedback: CircuitFeedback) { - if let Err(error) = self.sender.try_send(feedback) { - let reason = match error { - mpsc::error::TrySendError::Full(_) => "full", - mpsc::error::TrySendError::Closed(_) => "closed", - }; - tracing::warn!( - operation = "record_provider_circuit_feedback", - reason, - "Redis 协调队列不可用，已丢弃本次 circuit feedback" - ); - } - } -} - -impl ProviderCircuitPort for BufferedProviderCircuitPort { - fn decision<'a>( - &'a self, - provider_kind: &'a ProviderKind, - ) -> BoxFuture<'a, Result> { - self.inner.decision(provider_kind) - } - - fn observe_failure<'a>( - &'a self, - provider_kind: &'a ProviderKind, - ) -> BoxFuture<'a, Result<(), ProviderCircuitError>> { - self.enqueue(CircuitFeedback::Failure(provider_kind.clone())); - Box::pin(ready(Ok(()))) - } - - fn observe_success<'a>( - &'a self, - provider_kind: &'a ProviderKind, - ) -> BoxFuture<'a, Result<(), ProviderCircuitError>> { - self.enqueue(CircuitFeedback::Success(provider_kind.clone())); - Box::pin(ready(Ok(()))) - } -} - -enum CircuitFeedback { - Failure(ProviderKind), - Success(ProviderKind), -} - -pub struct ProviderCircuitFeedbackWriter { - inner: Arc, - receiver: Mutex>, -} - -impl DaemonTask for ProviderCircuitFeedbackWriter { - fn run(&self, cancellation: CancellationToken) -> BoxFuture<'_, Result<(), WorkerTaskError>> { - Box::pin(async move { - let mut receiver = self.receiver.lock().await; - loop { - let feedback = tokio::select! { - () = cancellation.cancelled() => { - drain_circuit_feedback(&mut receiver, self.inner.as_ref()).await; - return Ok(()); - }, - feedback = receiver.recv() => feedback, - }; - let Some(feedback) = feedback else { - return Err(WorkerTaskError::safe( - "provider circuit feedback queue closed", - )); - }; - let result = match feedback { - CircuitFeedback::Failure(provider) => { - self.inner.observe_failure(&provider).await - } - CircuitFeedback::Success(provider) => { - self.inner.observe_success(&provider).await - } - }; - if let Err(error) = result { - tracing::warn!(%error, "Provider circuit feedback 后台写入失败"); - } - } - }) - } -} - async fn drain_admission_releases( receiver: &mut mpsc::Receiver, port: &dyn ClientAdmissionPort, @@ -292,30 +171,3 @@ async fn drain_admission_releases( } } } - -async fn drain_circuit_feedback( - receiver: &mut mpsc::Receiver, - port: &dyn ProviderCircuitPort, -) { - receiver.close(); - let started_at = Instant::now(); - while let Some(feedback) = receiver.recv().await { - let remaining = SHUTDOWN_DRAIN_TIMEOUT.saturating_sub(started_at.elapsed()); - if remaining.is_zero() { - tracing::warn!("Provider circuit 关闭排空超时，剩余反馈已丢弃"); - break; - } - let result = match feedback { - CircuitFeedback::Failure(provider) => { - tokio::time::timeout(remaining, port.observe_failure(&provider)).await - } - CircuitFeedback::Success(provider) => { - tokio::time::timeout(remaining, port.observe_success(&provider)).await - } - }; - if result.is_err() { - tracing::warn!("Provider circuit 关闭排空超时，剩余反馈已丢弃"); - break; - } - } -} diff --git a/backend/crates/gateway-store/src/redis/mod.rs b/backend/crates/gateway-store/src/redis/mod.rs index 2ba2f36c8..5c71dad53 100644 --- a/backend/crates/gateway-store/src/redis/mod.rs +++ b/backend/crates/gateway-store/src/redis/mod.rs @@ -12,7 +12,6 @@ mod credential_leases; mod credential_state; mod native_continuation; mod oauth_pending; -mod provider_circuit; mod provider_session_affinity; mod provider_session_exclusion; mod runtime_change; @@ -28,7 +27,6 @@ pub use credential_leases::*; pub use credential_state::*; pub use native_continuation::*; pub use oauth_pending::*; -pub use provider_circuit::*; pub use provider_session_affinity::*; pub use provider_session_exclusion::*; pub use runtime_change::*; diff --git a/backend/crates/gateway-store/src/redis/provider_circuit.rs b/backend/crates/gateway-store/src/redis/provider_circuit.rs deleted file mode 100644 index 1e6006254..000000000 --- a/backend/crates/gateway-store/src/redis/provider_circuit.rs +++ /dev/null @@ -1,208 +0,0 @@ -//! Provider 的可重建 Redis circuit。 - -use async_trait::async_trait; -use chrono::{DateTime, Utc}; -use gateway_core::engine::execution::{ - ProviderCircuitDecision as CoreCircuitDecision, ProviderCircuitError, ProviderCircuitPolicy, - ProviderCircuitPort, -}; -use gateway_core::routing::ProviderKind; -use redis::{Script, aio::ConnectionManager}; - -use crate::{StoreError, StoreResult, redis_unavailable, require_nonempty}; - -use super::{namespace, resource_fingerprint}; - -const FAILURE_SCRIPT: &str = r#" -local clock = redis.call('TIME') -local now_ms = (tonumber(clock[1]) * 1000) + math.floor(tonumber(clock[2]) / 1000) -local failures = redis.call('HINCRBY', KEYS[1], 'failures', 1) -local open_until = tonumber(redis.call('HGET', KEYS[1], 'open_until_ms') or '0') -if failures >= tonumber(ARGV[1]) then - open_until = now_ms + tonumber(ARGV[2]) - redis.call('HSET', KEYS[1], 'open_until_ms', tostring(open_until)) -end -redis.call('HSET', KEYS[1], 'observed_at_ms', tostring(now_ms)) -redis.call('PEXPIRE', KEYS[1], math.max(86400000, tonumber(ARGV[2]) * 2)) -return {tostring(failures), tostring(open_until)} -"#; - -const DECISION_SCRIPT: &str = r#" -local clock = redis.call('TIME') -local now_ms = (tonumber(clock[1]) * 1000) + math.floor(tonumber(clock[2]) / 1000) -local open_until = tonumber(redis.call('HGET', KEYS[1], 'open_until_ms') or '0') -if open_until > now_ms then return {0, tostring(open_until)} end -return {1, '0'} -"#; - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum ProviderCircuitDecision { - Allow, - BlockedUntil(DateTime), -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct ProviderCircuitObservation { - pub failure_count: u32, - pub open_until: Option>, -} - -#[async_trait] -pub trait ProviderCircuitRepository: Send + Sync { - async fn provider_circuit_decision( - &self, - provider_kind: &str, - ) -> StoreResult; - async fn observe_provider_failure( - &self, - provider_kind: &str, - ) -> StoreResult; - async fn observe_provider_success(&self, provider_kind: &str) -> StoreResult<()>; -} - -#[derive(Clone)] -pub struct RedisProviderCircuitRepository { - connection: ConnectionManager, - namespace: String, - policy: ProviderCircuitPolicy, -} - -impl RedisProviderCircuitRepository { - pub fn new( - connection: ConnectionManager, - key_namespace: &str, - policy: ProviderCircuitPolicy, - ) -> StoreResult { - if policy.open_duration.is_zero() { - return Err(invalid("open duration must be positive")); - } - Ok(Self { - connection, - namespace: namespace(key_namespace)?, - policy, - }) - } - - fn key(&self, provider_kind: &str) -> StoreResult { - let fingerprint = resource_fingerprint("provider circuit", provider_kind)?; - Ok(format!("{}:provider:{fingerprint}:circuit", self.namespace)) - } -} - -#[async_trait] -impl ProviderCircuitRepository for RedisProviderCircuitRepository { - async fn provider_circuit_decision( - &self, - provider_kind: &str, - ) -> StoreResult { - require_nonempty("provider circuit", "provider_kind", provider_kind)?; - let mut connection = self.connection.clone(); - let (allow, until): (i64, String) = Script::new(DECISION_SCRIPT) - .key(self.key(provider_kind)?) - .invoke_async(&mut connection) - .await - .map_err(|_| redis_unavailable("read provider circuit"))?; - if allow == 1 { - Ok(ProviderCircuitDecision::Allow) - } else { - Ok(ProviderCircuitDecision::BlockedUntil(timestamp(&until)?)) - } - } - - async fn observe_provider_failure( - &self, - provider_kind: &str, - ) -> StoreResult { - require_nonempty("provider circuit", "provider_kind", provider_kind)?; - let duration_ms = u64::try_from(self.policy.open_duration.as_millis()) - .map_err(|_| invalid("open duration is too large"))?; - let mut connection = self.connection.clone(); - let (failures, until): (String, String) = Script::new(FAILURE_SCRIPT) - .key(self.key(provider_kind)?) - .arg(self.policy.failure_threshold.get()) - .arg(duration_ms) - .invoke_async(&mut connection) - .await - .map_err(|_| redis_unavailable("observe provider failure"))?; - let failure_count = failures - .parse() - .map_err(|_| invalid("Redis returned an invalid failure count"))?; - let open_until = if until == "0" { - None - } else { - Some(timestamp(&until)?) - }; - Ok(ProviderCircuitObservation { - failure_count, - open_until, - }) - } - - async fn observe_provider_success(&self, provider_kind: &str) -> StoreResult<()> { - require_nonempty("provider circuit", "provider_kind", provider_kind)?; - let mut connection = self.connection.clone(); - redis::cmd("DEL") - .arg(self.key(provider_kind)?) - .query_async::(&mut connection) - .await - .map_err(|_| redis_unavailable("reset provider circuit"))?; - Ok(()) - } -} - -impl ProviderCircuitPort for RedisProviderCircuitRepository { - fn decision<'a>( - &'a self, - provider_kind: &'a ProviderKind, - ) -> futures::future::BoxFuture<'a, Result> { - Box::pin(async move { - self.provider_circuit_decision(provider_kind.as_str()) - .await - .map(|decision| match decision { - ProviderCircuitDecision::Allow => CoreCircuitDecision::Allow, - ProviderCircuitDecision::BlockedUntil(until) => { - CoreCircuitDecision::BlockedUntil(until.into()) - } - }) - .map_err(|_| ProviderCircuitError) - }) - } - - fn observe_failure<'a>( - &'a self, - provider_kind: &'a ProviderKind, - ) -> futures::future::BoxFuture<'a, Result<(), ProviderCircuitError>> { - Box::pin(async move { - self.observe_provider_failure(provider_kind.as_str()) - .await - .map(|_| ()) - .map_err(|_| ProviderCircuitError) - }) - } - - fn observe_success<'a>( - &'a self, - provider_kind: &'a ProviderKind, - ) -> futures::future::BoxFuture<'a, Result<(), ProviderCircuitError>> { - Box::pin(async move { - self.observe_provider_success(provider_kind.as_str()) - .await - .map_err(|_| ProviderCircuitError) - }) - } -} - -fn timestamp(value: &str) -> StoreResult> { - let value = value - .parse::() - .map_err(|_| invalid("Redis returned an invalid circuit timestamp"))?; - DateTime::from_timestamp_millis(value) - .ok_or_else(|| invalid("Redis returned an out-of-range circuit timestamp")) -} - -fn invalid(message: &str) -> StoreError { - StoreError::InvalidData { - entity: "provider circuit", - message: message.to_owned(), - } -} diff --git a/backend/crates/gateway-store/src/workers.rs b/backend/crates/gateway-store/src/workers.rs index 29dce232b..399450e2c 100644 --- a/backend/crates/gateway-store/src/workers.rs +++ b/backend/crates/gateway-store/src/workers.rs @@ -9,10 +9,9 @@ pub(crate) struct CommandStoreWriters { pub(crate) execution: postgres::ExecutionObservationWriter, pub(crate) client_key_usage: postgres::PgClientApiKeyUsageWriter, pub(crate) admission_release: redis::ClientAdmissionReleaseWriter, - pub(crate) circuit_feedback: redis::ProviderCircuitFeedbackWriter, } -/// 短生命周期 CLI 只运行数据面必需的四个写泵，不注册恢复、保留或维护 Worker。 +/// 短生命周期 CLI 只运行数据面必需的三个写泵，不注册恢复、保留或维护 Worker。 pub struct CommandStoreDrain { cancellation: gateway_core::lifecycle::CancellationToken, tasks: Vec>>, @@ -30,14 +29,12 @@ impl CommandStoreWriters { execution, client_key_usage, admission_release, - circuit_feedback, } = self; let execution_idle = execution.idle(); let tasks = vec![ spawn_command_writer(execution, cancellation.child_token()), spawn_command_writer(client_key_usage, cancellation.child_token()), spawn_command_writer(admission_release, cancellation.child_token()), - spawn_command_writer(circuit_feedback, cancellation.child_token()), ]; CommandStoreDrain { cancellation, @@ -102,7 +99,6 @@ pub(crate) fn store_worker_contributions( execution_writer: postgres::ExecutionObservationWriter, client_key_usage_writer: postgres::PgClientApiKeyUsageWriter, admission_release_writer: redis::ClientAdmissionReleaseWriter, - circuit_feedback_writer: redis::ProviderCircuitFeedbackWriter, retention: Arc, ) -> StoreResult> { let stale_id = WorkerId::try_new(WorkerKind::StaleModelRequestRecovery, "postgres") @@ -116,8 +112,6 @@ pub(crate) fn store_worker_contributions( .map_err(worker_definition_error)?; let admission_flush_id = WorkerId::try_new(WorkerKind::OpsFlush, "redis_admission") .map_err(worker_definition_error)?; - let circuit_flush_id = WorkerId::try_new(WorkerKind::OpsFlush, "redis_circuit") - .map_err(worker_definition_error)?; let ops_flush_restart = DaemonRestartPolicy::try_new(Duration::from_secs(1), Duration::from_secs(60)) .map_err(worker_definition_error)?; @@ -162,16 +156,6 @@ pub(crate) fn store_worker_contributions( ) .map_err(worker_definition_error)?, ), - WorkerContribution::Registration( - WorkerRegistration::try_new( - circuit_flush_id, - WorkerRunnable::Daemon { - restart: ops_flush_restart, - task: Box::new(circuit_feedback_writer), - }, - ) - .map_err(worker_definition_error)?, - ), ]) } diff --git a/backend/crates/gateway-store/tests/postgres/execution.rs b/backend/crates/gateway-store/tests/postgres/execution.rs index 40b67f086..250134edf 100644 --- a/backend/crates/gateway-store/tests/postgres/execution.rs +++ b/backend/crates/gateway-store/tests/postgres/execution.rs @@ -1995,3 +1995,36 @@ async fn zero_attempt_failure_does_not_enter_successful_usage_or_cost_aggregates assert_eq!(charges, 0); database.close().await; } + +#[tokio::test] +async fn entry_rejection_is_visible_without_a_fictitious_model_execution() { + let Some(database) = TestDatabase::create("entry_rejection").await else { + return; + }; + let repository = PgExecutionStore::new(database.pool.clone()); + repository + .record_entry_rejection(gateway_core::engine::EntryRejection { + request_id: ModelRequestId::new("req_entry_rejection").unwrap(), + client_key_id: ClientApiKeyId::new("key_entry").unwrap(), + error: GatewayError::new(GatewayErrorKind::NoAvailableProvider, "no route"), + latency: StdDuration::from_millis(12), + }) + .await + .unwrap(); + let row: (Option, String, String, i64) = sqlx::query_as( + "select model_request_id, failure_kind, message, latency_ms from ops_events where component = 'request_entry' and operation = 'reject'" + ).fetch_one(&database.pool).await.unwrap(); + assert!(row.0.is_none()); + assert_eq!(row.1, GatewayErrorKind::NoAvailableProvider.as_str()); + assert_eq!( + serde_json::from_str::(&row.2).unwrap()["requestId"], + "req_entry_rejection" + ); + assert_eq!(row.3, 12); + let count: i64 = sqlx::query_scalar("select count(*) from model_requests") + .fetch_one(&database.pool) + .await + .unwrap(); + assert_eq!(count, 0); + database.close().await; +} diff --git a/backend/crates/gateway-store/tests/postgres/plugins/artifacts.rs b/backend/crates/gateway-store/tests/postgres/plugins/artifacts.rs index c585b2422..64a82b0d1 100644 --- a/backend/crates/gateway-store/tests/postgres/plugins/artifacts.rs +++ b/backend/crates/gateway-store/tests/postgres/plugins/artifacts.rs @@ -176,7 +176,9 @@ async fn plugin_migration_preserves_main_settings_and_accepts_new_installations( .fetch_one(&database.pool) .await .unwrap(); - assert_eq!(after, before); + for (key, value) in before.as_object().unwrap() { + assert_eq!(after.get(key), Some(value), "existing setting {key}"); + } let store = PgPluginStore::new(database.pool.clone()); store .install_artifact( diff --git a/backend/crates/gateway-store/tests/redis/coordination_buffer.rs b/backend/crates/gateway-store/tests/redis/coordination_buffer.rs index 2a12e308b..f7260f57f 100644 --- a/backend/crates/gateway-store/tests/redis/coordination_buffer.rs +++ b/backend/crates/gateway-store/tests/redis/coordination_buffer.rs @@ -9,13 +9,9 @@ use gateway_core::engine::admission::{ ClientAdmissionDecision, ClientAdmissionError, ClientAdmissionPort, ClientAdmissionRecovery, ClientAdmissionRejection, ClientAdmissionRequest, ClientAdmissionRestoreResult, }; -use gateway_core::engine::execution::{ - ProviderCircuitDecision, ProviderCircuitError, ProviderCircuitPort, -}; use gateway_core::lifecycle::CancellationToken; use gateway_core::policy::{ClientApiKeyId, RateLimits}; -use gateway_core::routing::ProviderKind; -use gateway_store::redis::{BufferedClientAdmissionPort, BufferedProviderCircuitPort}; +use gateway_store::redis::BufferedClientAdmissionPort; #[derive(Default)] struct RecordingCoordination { @@ -73,35 +69,6 @@ impl ClientAdmissionPort for RecordingCoordination { } } -impl ProviderCircuitPort for RecordingCoordination { - fn decision<'a>( - &'a self, - _: &'a ProviderKind, - ) -> BoxFuture<'a, Result> { - Box::pin(async { Ok(ProviderCircuitDecision::Allow) }) - } - - fn observe_failure<'a>( - &'a self, - _: &'a ProviderKind, - ) -> BoxFuture<'a, Result<(), ProviderCircuitError>> { - Box::pin(async move { - self.record("circuit_failure"); - Ok(()) - }) - } - - fn observe_success<'a>( - &'a self, - _: &'a ProviderKind, - ) -> BoxFuture<'a, Result<(), ProviderCircuitError>> { - Box::pin(async move { - self.record("circuit_success"); - Ok(()) - }) - } -} - struct CapacityCoordination { active: Mutex>, maximum: usize, @@ -173,13 +140,8 @@ async fn full_recoverable_coordination_queues_should_drop_writes_without_waiting inner.clone(), NonZeroUsize::new(1).expect("capacity"), ); - let (circuits, _circuit_writer) = BufferedProviderCircuitPort::with_capacity( - inner.clone(), - NonZeroUsize::new(1).expect("capacity"), - ); let client = ClientApiKeyId::new("key_buffer_test").expect("client key"); let request = ModelRequestId::new("req_buffer_test").expect("request ID"); - let provider = ProviderKind::new("openai").expect("provider"); tokio::time::timeout(Duration::from_millis(50), async { admissions @@ -190,14 +152,6 @@ async fn full_recoverable_coordination_queues_should_drop_writes_without_waiting .release(&client, &request) .await .expect("full admission queue remains fail-open"); - circuits - .observe_success(&provider) - .await - .expect("first circuit enqueue"); - circuits - .observe_failure(&provider) - .await - .expect("full circuit queue remains fail-open"); }) .await .expect("coordination enqueue must never wait for Redis"); @@ -212,29 +166,20 @@ async fn redis_coordination_writers_should_flush_each_side_effect() { inner.clone(), NonZeroUsize::new(8).expect("capacity"), ); - let (circuits, circuit_writer) = BufferedProviderCircuitPort::with_capacity( - inner.clone(), - NonZeroUsize::new(8).expect("capacity"), - ); let client = ClientApiKeyId::new("key_writer_test").expect("client key"); let request = ModelRequestId::new("req_writer_test").expect("request ID"); - let provider = ProviderKind::new("openai").expect("provider"); admissions .release(&client, &request) .await .expect("enqueue admission release"); - circuits - .observe_success(&provider) - .await - .expect("enqueue circuit feedback"); let cancellation = CancellationToken::new(); - let tasks = [ - spawn_writer(Arc::new(admission_writer), cancellation.clone()), - spawn_writer(Arc::new(circuit_writer), cancellation.clone()), - ]; + let tasks = [spawn_writer( + Arc::new(admission_writer), + cancellation.clone(), + )]; tokio::time::timeout(Duration::from_secs(1), async { loop { - if inner.operations().len() == 2 { + if inner.operations().len() == 1 { break; } tokio::task::yield_now().await; @@ -251,7 +196,7 @@ async fn redis_coordination_writers_should_flush_each_side_effect() { let mut operations = inner.operations(); operations.sort_unstable(); - assert_eq!(operations, ["admission", "circuit_success"]); + assert_eq!(operations, ["admission"]); } #[tokio::test] diff --git a/backend/crates/gateway-store/tests/redis/mod.rs b/backend/crates/gateway-store/tests/redis/mod.rs index 55eed5254..94ca46141 100644 --- a/backend/crates/gateway-store/tests/redis/mod.rs +++ b/backend/crates/gateway-store/tests/redis/mod.rs @@ -7,7 +7,6 @@ mod credential_leases; mod credential_state; mod native_continuation; mod oauth_pending; -mod provider_circuit; mod provider_session_affinity; mod provider_session_exclusion; mod worker_lease; diff --git a/backend/crates/gateway-store/tests/redis/provider_circuit.rs b/backend/crates/gateway-store/tests/redis/provider_circuit.rs deleted file mode 100644 index 9ad4b285b..000000000 --- a/backend/crates/gateway-store/tests/redis/provider_circuit.rs +++ /dev/null @@ -1,191 +0,0 @@ -use std::{num::NonZeroU32, time::Duration}; - -use gateway_core::{ - engine::execution::{ProviderCircuitPolicy, provider_failure_affects_circuit}, - error::ProviderErrorKind, -}; -use gateway_store::redis::{ - ProviderCircuitDecision, ProviderCircuitRepository, RedisProviderCircuitRepository, -}; -use redis::aio::ConnectionManager; -use uuid::Uuid; - -#[test] -fn provider_circuit_default_has_positive_threshold() { - assert!(ProviderCircuitPolicy::default().failure_threshold.get() > 0); -} - -#[test] -fn provider_circuit_should_only_count_instance_attributable_failures() { - for error_kind in [ - ProviderErrorKind::Timeout, - ProviderErrorKind::Transport, - ProviderErrorKind::Protocol, - ProviderErrorKind::Unavailable, - ] { - assert!(provider_failure_affects_circuit(error_kind)); - } - for error_kind in [ - ProviderErrorKind::InvalidRequest, - ProviderErrorKind::Unsupported, - ProviderErrorKind::Unauthorized, - ProviderErrorKind::PermissionDenied, - ProviderErrorKind::RateLimited, - ProviderErrorKind::QuotaExhausted, - ProviderErrorKind::Cancelled, - ProviderErrorKind::ProcessTerminated, - ] { - assert!(!provider_failure_affects_circuit(error_kind)); - } -} - -#[tokio::test] -async fn provider_circuit_opens_at_threshold_and_success_resets_it() { - let Some((repository, mut connection, namespace)) = repository(2).await else { - return; - }; - let instance = "instance-circuit-primary"; - - assert_eq!( - repository - .provider_circuit_decision(instance) - .await - .expect("read empty provider circuit"), - ProviderCircuitDecision::Allow - ); - let first = repository - .observe_provider_failure(instance) - .await - .expect("record first provider failure"); - assert_eq!(first.failure_count, 1); - assert!(first.open_until.is_none()); - - let second = repository - .observe_provider_failure(instance) - .await - .expect("record threshold provider failure"); - assert_eq!(second.failure_count, 2); - assert!(second.open_until.is_some()); - assert!(matches!( - repository - .provider_circuit_decision(instance) - .await - .expect("read open provider circuit"), - ProviderCircuitDecision::BlockedUntil(_) - )); - - repository - .observe_provider_success(instance) - .await - .expect("reset provider circuit after success"); - assert_eq!( - repository - .provider_circuit_decision(instance) - .await - .expect("read reset provider circuit"), - ProviderCircuitDecision::Allow - ); - - delete_namespace_keys(&mut connection, &namespace).await; -} - -#[tokio::test] -async fn provider_circuit_keeps_instances_isolated() { - let Some((repository, mut connection, namespace)) = repository(1).await else { - return; - }; - - repository - .observe_provider_failure("instance-circuit-failed") - .await - .expect("open failed instance circuit"); - assert!(matches!( - repository - .provider_circuit_decision("instance-circuit-failed") - .await - .expect("read failed instance circuit"), - ProviderCircuitDecision::BlockedUntil(_) - )); - assert_eq!( - repository - .provider_circuit_decision("instance-circuit-healthy") - .await - .expect("read isolated healthy circuit"), - ProviderCircuitDecision::Allow - ); - - delete_namespace_keys(&mut connection, &namespace).await; -} - -#[tokio::test] -async fn provider_circuit_reopens_after_redis_time_deadline() { - let Some((repository, mut connection, namespace)) = - repository_with_policy(1, Duration::from_millis(40)).await - else { - return; - }; - let instance = "instance-circuit-deadline"; - - repository - .observe_provider_failure(instance) - .await - .expect("open short provider circuit"); - assert!(matches!( - repository - .provider_circuit_decision(instance) - .await - .expect("read short open circuit"), - ProviderCircuitDecision::BlockedUntil(_) - )); - tokio::time::sleep(Duration::from_millis(80)).await; - assert_eq!( - repository - .provider_circuit_decision(instance) - .await - .expect("read circuit after Redis deadline"), - ProviderCircuitDecision::Allow - ); - - delete_namespace_keys(&mut connection, &namespace).await; -} - -async fn repository( - threshold: u32, -) -> Option<(RedisProviderCircuitRepository, ConnectionManager, String)> { - repository_with_policy(threshold, Duration::from_secs(30)).await -} - -async fn repository_with_policy( - threshold: u32, - open_duration: Duration, -) -> Option<(RedisProviderCircuitRepository, ConnectionManager, String)> { - let redis_url = crate::support::test_env("CPR_TEST_REDIS_URL")?; - let client = redis::Client::open(redis_url).expect("valid CPR_TEST_REDIS_URL"); - let connection = client - .get_connection_manager() - .await - .expect("connect test Redis"); - let namespace = format!("gateway-store-circuit-test-{}", Uuid::new_v4()); - let policy = ProviderCircuitPolicy { - failure_threshold: NonZeroU32::new(threshold).expect("positive threshold"), - open_duration, - }; - let repository = RedisProviderCircuitRepository::new(connection.clone(), &namespace, policy) - .expect("valid provider circuit repository"); - Some((repository, connection, namespace)) -} - -async fn delete_namespace_keys(connection: &mut ConnectionManager, namespace: &str) { - let keys = redis::cmd("KEYS") - .arg(format!("{namespace}:*")) - .query_async::>(connection) - .await - .expect("list isolated provider circuit keys"); - if !keys.is_empty() { - redis::cmd("DEL") - .arg(keys) - .query_async::(connection) - .await - .expect("delete isolated provider circuit keys"); - } -} diff --git a/backend/crates/providers/openai/Cargo.toml b/backend/crates/providers/openai/Cargo.toml index 8742b2c55..2698f4ac4 100644 --- a/backend/crates/providers/openai/Cargo.toml +++ b/backend/crates/providers/openai/Cargo.toml @@ -52,6 +52,7 @@ tokio = { version = "=1.52.3", default-features = false, features = ["fs", "macr tokio-rustls = { version = "=0.26.4", default-features = false } tokio-util = { version = "0.7.18", features = ["rt"] } tower-service = "0.3.3" +tower-layer = "0.3.3" # 官方 Codex WebSocket 需要上游尚未发布的 proxy 与 permessage-deflate 实现，固定到已审计提交。 tokio-tungstenite = { git = "https://github.com/openai-oss-forks/tokio-tungstenite", rev = "0e5b2d73aa18dd9f0a50ee9ff199d5aef7594186", features = ["proxy", "rustls-tls-native-roots"] } tracing = "0.1.44" diff --git a/backend/crates/providers/openai/src/provider/execution.rs b/backend/crates/providers/openai/src/provider/execution.rs index 4310f8c37..25b17f34f 100644 --- a/backend/crates/providers/openai/src/provider/execution.rs +++ b/backend/crates/providers/openai/src/provider/execution.rs @@ -760,6 +760,7 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream { return; } }; + context.connection_budget().complete(); if !accepts_backend_transport(transport_policy, response.transport) { let failure = MappedProviderFailure::plain(provider_error( ProviderErrorKind::Protocol, diff --git a/backend/crates/providers/openai/src/provider/failure.rs b/backend/crates/providers/openai/src/provider/failure.rs index fe29feb93..b2ff52d40 100644 --- a/backend/crates/providers/openai/src/provider/failure.rs +++ b/backend/crates/providers/openai/src/provider/failure.rs @@ -810,7 +810,15 @@ pub(super) fn map_client_error( uncertain_state: UpstreamSendState, observe_transport: bool, ) -> MappedProviderFailure { - let diagnostic = client_diagnostic(&error); + let local_connection_capacity = crate::transport::connection::is_admission_failure(&error); + let diagnostic = if local_connection_capacity { + Some( + ProviderDiagnostic::new("OpenAI local connection capacity unavailable") + .with_classification("admission", "local_connection_capacity"), + ) + } else { + client_diagnostic(&error) + }; let raw_upstream_error = match &error { CodexClientError::WebSocket(error) => websocket_raw_error(error), _ => None, @@ -850,7 +858,13 @@ pub(super) fn map_client_error( if let Some(failure) = error.upstream_failure() { return map_upstream_failure(failure, observation, ReplayBoundary::BeforeSemanticOutput); } + let connect_retry = !local_connection_capacity + && matches!(&error, CodexClientError::Http(error) if transient_http_connect(error)); let mut failure = match error { + CodexClientError::ConnectionBudgetExhausted => MappedProviderFailure::plain( + provider_error(ProviderErrorKind::Timeout, UpstreamSendState::NotSent) + .with_connection_retry(gateway_core::engine::AttemptTransport::Fallback), + ), CodexClientError::Upstream { .. } => MappedProviderFailure::plain(provider_error( ProviderErrorKind::Protocol, UpstreamSendState::Sent, @@ -963,6 +977,17 @@ pub(super) fn map_client_error( failure } }; + if local_connection_capacity { + failure.error = provider_error( + ProviderErrorKind::ProviderInfrastructureUnavailable, + UpstreamSendState::NotSent, + ); + } + if connect_retry { + failure.error = failure + .error + .with_connection_retry(gateway_core::engine::AttemptTransport::Fallback); + } if let Some(continuation_failure) = continuation_failure { failure.error = failure .error @@ -989,6 +1014,11 @@ pub(super) fn map_client_error( /// 在消费 transport 错误前提取可持久化事实，不能使用可能携带 URL/凭据的 Display。 fn client_diagnostic(error: &CodexClientError) -> Option { let (stage, code, message) = match error { + CodexClientError::ConnectionBudgetExhausted => ( + "connect", + "connection_budget_exhausted", + "OpenAI connection recovery budget exhausted".to_owned(), + ), CodexClientError::WebSocket(error) => return Some(websocket_diagnostic(error)), CodexClientError::ErrorBodyRead { status, source, .. } => { return Some( @@ -1561,3 +1591,28 @@ pub(super) fn remaining(deadline: SystemTime) -> Option { .ok() .filter(|remaining| !remaining.is_zero()) } + +/// 只恢复明确未发送的 HTTP 建连失败；已知 TLS/配置错误不反复尝试同一路径。 +fn transient_http_connect(error: &reqwest::Error) -> bool { + if !error.is_connect() || error.is_builder() { + return false; + } + let mut source = std::error::Error::source(error); + while let Some(cause) = source { + if cause.is::() || cause.is::() { + return false; + } + if let Some(io) = cause.downcast_ref::() + && matches!( + io.kind(), + std::io::ErrorKind::PermissionDenied + | std::io::ErrorKind::InvalidInput + | std::io::ErrorKind::InvalidData + ) + { + return false; + } + source = cause.source(); + } + true +} diff --git a/backend/crates/providers/openai/src/provider/mod.rs b/backend/crates/providers/openai/src/provider/mod.rs index 505079abe..4bf006e06 100644 --- a/backend/crates/providers/openai/src/provider/mod.rs +++ b/backend/crates/providers/openai/src/provider/mod.rs @@ -482,23 +482,34 @@ impl Provider for CodexProvider { ) }); let selection_started_at = Instant::now(); - let lease = self - .selector - .select_with_cyber_policy( - &SelectCodexCredential { - upstream_model: upstream_model.as_str(), - request_url: &self.responses_url, - attempt: &context, - session_affinity_key: selection_session_affinity - .as_ref() - .map(|affinity| affinity.key()), - }, - selection_cyber_policy_key.as_ref(), - selection_session_affinity.as_ref(), - requires_websocket, - ) - .await - .map_err(map_selection_error)?; + let selection = async { + self.selector + .select_with_cyber_policy( + &SelectCodexCredential { + upstream_model: upstream_model.as_str(), + request_url: &self.responses_url, + attempt: &context, + session_affinity_key: selection_session_affinity + .as_ref() + .map(|affinity| affinity.key()), + }, + selection_cyber_policy_key.as_ref(), + selection_session_affinity.as_ref(), + requires_websocket, + ) + .await + .map_err(map_selection_error) + }; + // 恢复期间排队也消耗启动窗口；只包住选账号，不限制业务响应时长。 + let lease = if let Some(remaining) = context.connection_budget().startup_remaining() { + tokio::time::timeout(remaining, selection) + .await + .map_err(|_| { + provider_error(ProviderErrorKind::Timeout, UpstreamSendState::NotSent) + })?? + } else { + selection.await? + }; let account_selection_wait_ms = u64::try_from(selection_started_at.elapsed().as_millis()).unwrap_or(u64::MAX); let provider_kind = ProviderKind::new(PROVIDER_NAME) @@ -790,6 +801,7 @@ impl CodexProvider { provider_error(ProviderErrorKind::Unavailable, UpstreamSendState::NotSent) })? .with_authentication(lease.authentication()) + .with_connection_budget(context.connection_budget().clone()) .with_middleware_headers(middleware_headers), response_origin: self.responses_url.clone(), request: upstream_request, diff --git a/backend/crates/providers/openai/src/transport/client.rs b/backend/crates/providers/openai/src/transport/client.rs index ba646f8e3..f3b3e1cb7 100644 --- a/backend/crates/providers/openai/src/transport/client.rs +++ b/backend/crates/providers/openai/src/transport/client.rs @@ -46,7 +46,7 @@ const UPSTREAM_CONNECT_TIMEOUT: Duration = Duration::from_secs(15); pub(super) const UPSTREAM_STREAM_IDLE_TIMEOUT: Duration = Duration::from_secs(5 * 60); const X_CODEX_WS_STREAM_REQUEST_START_MS_CLIENT_METADATA_KEY: &str = "x-codex-ws-stream-request-start-ms"; -type ReqwestClientCacheKey = (Option, String); +type ReqwestClientCacheKey = (Option, String, Duration); type ReqwestClientCache = Mutex>; /// 构建带缓存、自动协商 HTTP/2 的 reqwest Client。 @@ -57,9 +57,21 @@ pub fn build_reqwest_client() -> Result { pub fn build_account_http_client( account_id: &str, proxy: Option<&gateway_core::account::OutboundProxy>, +) -> Result { + build_account_http_client_with_timeout(account_id, proxy, UPSTREAM_CONNECT_TIMEOUT) +} + +pub(super) fn build_account_http_client_with_timeout( + account_id: &str, + proxy: Option<&gateway_core::account::OutboundProxy>, + timeout: Duration, ) -> Result { super::tls::ensure_rustls_provider(); - let cache_key = (custom_ca_env_cache_key(), egress_key(account_id, proxy)); + let cache_key = ( + custom_ca_env_cache_key(), + egress_key(account_id, proxy), + timeout, + ); static CLIENTS: OnceLock = OnceLock::new(); let cache = CLIENTS.get_or_init(|| Mutex::new(HashMap::new())); if let Some(client) = cache @@ -75,7 +87,8 @@ pub fn build_account_http_client( .redirect(reqwest::redirect::Policy::none()) .pool_max_idle_per_host(4) .pool_idle_timeout(None::) - .connect_timeout(UPSTREAM_CONNECT_TIMEOUT) + .connect_timeout(timeout) + .connector_layer(super::connection::ConnectionLayer) .tcp_keepalive(Duration::from_secs(30)) .http2_keep_alive_interval(Duration::from_secs(30)) .http2_keep_alive_timeout(Duration::from_secs(5)) @@ -188,6 +201,8 @@ impl fmt::Debug for CodexClientVisibleUpstreamResponse { /// Codex 上游 HTTP 客户端错误。 #[derive(Error)] pub enum CodexClientError { + #[error("connection recovery budget exhausted")] + ConnectionBudgetExhausted, /// Reqwest 传输失败。 #[error("http transport error: {0}")] Http(#[from] reqwest::Error), @@ -269,6 +284,9 @@ pub enum CodexClientError { impl fmt::Debug for CodexClientError { fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { match self { + Self::ConnectionBudgetExhausted => { + formatter.write_str("CodexClientError::ConnectionBudgetExhausted") + } Self::Http(_) => formatter.write_str("CodexClientError::Http([REDACTED])"), Self::HttpJson(_) => formatter.write_str("CodexClientError::HttpJson([REDACTED])"), Self::ErrorBodyRead { @@ -338,7 +356,8 @@ impl CodexClientError { Self::Upstream { transport, .. } | Self::ErrorBodyRead { transport, .. } => { Some(*transport) } - Self::CustomCa(_) + Self::ConnectionBudgetExhausted + | Self::CustomCa(_) | Self::InvalidHeaderName(_) | Self::InvalidHeaderValue(_) | Self::MiddlewareHeaderConflict @@ -564,6 +583,7 @@ pub enum CodexTransportDecision { ExactWebSocket, RequiredWebSocket, Http2WebSocketBudgetExhausted, + Http2LocalConnectionCapacity, Http2BreakerOpen, Http2PoolUnavailable, } @@ -577,6 +597,7 @@ impl CodexTransportDecision { Self::ExactWebSocket => "ws_exact_required", Self::RequiredWebSocket => "ws_required", Self::Http2WebSocketBudgetExhausted => "http2_ws_budget_exhausted", + Self::Http2LocalConnectionCapacity => "http2_local_connection_capacity", Self::Http2BreakerOpen => "http2_breaker_open", Self::Http2PoolUnavailable => "http2_pool_unavailable", } @@ -668,6 +689,7 @@ impl OpenAiUpstreamProtocol { /// Codex HTTP/SSE 上游客户端。 #[derive(Clone)] pub struct CodexBackendClient { + pub(super) connection_budget: Option, pub(super) client: Client, pub(super) direct_client: Client, pub(super) base_url: String, @@ -683,6 +705,41 @@ pub struct CodexBackendClient { } impl CodexBackendClient { + pub(crate) fn with_connection_budget( + mut self, + budget: gateway_core::engine::connection::ConnectionBudget, + ) -> Self { + self.connection_budget = Some(budget); + self + } + + pub(super) fn connection_opening(&self) -> Result, CodexClientError> { + self.connection_budget.as_ref().map_or(Ok(None), |budget| { + budget + .begin() + .map_err(|_| CodexClientError::ConnectionBudgetExhausted) + }) + } + + pub(super) fn http_opening_client(&self) -> Result { + let Some(remaining) = self.connection_opening()? else { + return Ok(self.client.clone()); + }; + if remaining >= UPSTREAM_CONNECT_TIMEOUT { + return Ok(self.client.clone()); + } + // 按整秒向下取整限制缓存种类；不足一秒不缓存新的 client 配置。 + let timeout = Duration::from_secs(remaining.as_secs()); + if timeout.is_zero() { + return Err(CodexClientError::ConnectionBudgetExhausted); + } + Ok(build_account_http_client_with_timeout( + &self.egress_key, + self.outbound_proxy.as_ref(), + timeout, + )?) + } + /// 覆盖官方账号接口基址，用于隔离上游联调与协议测试。 #[must_use] pub fn with_official_base_url(mut self, official_base_url: impl Into) -> Self { @@ -1015,6 +1072,9 @@ pub(super) fn websocket_success_decision( pub(super) fn local_http_fallback_decision( error: &CodexWebSocketExchangeError, ) -> Option { + if crate::transport::connection::is_admission_failure(error) { + return Some(CodexTransportDecision::Http2LocalConnectionCapacity); + } match error.classified() { CodexWebSocketExchangeError::OriginCircuitOpen | CodexWebSocketExchangeError::OriginHalfOpenBusy => { diff --git a/backend/crates/providers/openai/src/transport/client_sse.rs b/backend/crates/providers/openai/src/transport/client_sse.rs index 1c24a763c..eff9cf786 100644 --- a/backend/crates/providers/openai/src/transport/client_sse.rs +++ b/backend/crates/providers/openai/src/transport/client_sse.rs @@ -64,6 +64,7 @@ impl CodexBackendClient { ) -> Self { let base_url = base_url.into().trim_end_matches('/').to_string(); Self { + connection_budget: None, direct_client: client.clone(), client, websocket_origin_key: websocket_origin_key(&base_url), @@ -134,7 +135,8 @@ impl CodexBackendClient { .map(|(name, value)| (name.as_str(), value.as_bytes())), ); trace.capture("upstream.request.body", &body); - let mut outbound = self.client.post(endpoint).headers(headers); + let client = self.http_opening_client()?; + let mut outbound = client.post(endpoint).headers(headers); let body = if self.protocol == OpenAiUpstreamProtocol::Codex { outbound = outbound.header(CONTENT_ENCODING, HeaderValue::from_static("zstd")); zstd::stream::encode_all(std::io::Cursor::new(body), 3) @@ -282,6 +284,7 @@ impl CodexBackendClient { ) .map_err(CodexClientError::WebSocketEncode)?; websocket_create.connection.outbound_proxy = self.outbound_proxy.clone(); + websocket_create.connection.connection_budget = self.connection_budget.clone(); context.trace.cloned().unwrap_or_default().headers( "upstream.request.headers", serde_json::json!({"transport": "websocket", "phase": "prepared_opening"}), diff --git a/backend/crates/providers/openai/src/transport/connection.rs b/backend/crates/providers/openai/src/transport/connection.rs new file mode 100644 index 000000000..846b06aea --- /dev/null +++ b/backend/crates/providers/openai/src/transport/connection.rs @@ -0,0 +1,98 @@ +//! HTTP 与 WS 共用的建连资源上限，不保留出口健康或冷却状态。 + +use std::future::Future; +use std::pin::Pin; +use std::sync::OnceLock; +use std::task::{Context, Poll}; +use tokio::sync::{Semaphore, SemaphorePermit}; +use tower_layer::Layer; +use tower_service::Service; + +// 为单副本的文件描述符和 TLS 握手保留硬上限；不限制已经建立的长流。 +const ACTIVE_CONNECTIONS: usize = 128; +const WAITING_CONNECTIONS: usize = 1024; +static ACTIVE: OnceLock = OnceLock::new(); +static WAITING: OnceLock = OnceLock::new(); + +#[derive(Debug, thiserror::Error)] +#[error("local connection capacity unavailable")] +struct ConnectionAdmissionRejected; + +/// 只根据自有类型识别本地容量拒绝，不将任意 IO 文案当作上游故障。 +pub(crate) fn is_admission_failure(error: &(dyn std::error::Error + 'static)) -> bool { + let mut source = Some(error); + while let Some(cause) = source { + if cause.is::() + || cause + .downcast_ref::() + .and_then(std::io::Error::get_ref) + .is_some_and(|inner| inner.is::()) + { + return true; + } + source = cause.source(); + } + false +} + +fn admission_rejected() -> std::io::Error { + std::io::Error::new(std::io::ErrorKind::WouldBlock, ConnectionAdmissionRejected) +} + +/// 可回退 HTTP 的 WS 快路径不排队，避免本地等待被当作上游 opening 变慢。 +pub(super) fn try_acquire() -> Result, std::io::Error> { + ACTIVE + .get_or_init(|| Semaphore::new(ACTIVE_CONNECTIONS)) + .try_acquire() + .map_err(|_| admission_rejected()) +} + +pub(super) async fn acquire() -> Result, std::io::Error> { + let active = ACTIVE.get_or_init(|| Semaphore::new(ACTIVE_CONNECTIONS)); + let waiting = WAITING.get_or_init(|| Semaphore::new(WAITING_CONNECTIONS)); + let queued = waiting.try_acquire().map_err(|_| admission_rejected())?; + let permit = active + .acquire() + .await + .map_err(|_| std::io::Error::other(ConnectionAdmissionRejected))?; + drop(queued); + Ok(permit) +} + +#[derive(Clone)] +pub(super) struct ConnectionLayer; + +impl Layer for ConnectionLayer { + type Service = ConnectionService; + fn layer(&self, inner: S) -> Self::Service { + ConnectionService(inner) + } +} + +#[derive(Clone)] +pub(super) struct ConnectionService(S); + +type BoxError = Box; + +impl Service for ConnectionService +where + S: Service, + S::Future: Send + 'static, + S::Response: 'static, +{ + type Response = S::Response; + type Error = BoxError; + type Future = Pin> + Send>>; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.0.poll_ready(cx) + } + + fn call(&mut self, request: R) -> Self::Future { + let connection = self.0.call(request); + Box::pin(async move { + let _permit = acquire().await?; + connection.await + }) + } +} diff --git a/backend/crates/providers/openai/src/transport/mod.rs b/backend/crates/providers/openai/src/transport/mod.rs index 510336f08..d7cff6e84 100644 --- a/backend/crates/providers/openai/src/transport/mod.rs +++ b/backend/crates/providers/openai/src/transport/mod.rs @@ -5,6 +5,7 @@ pub mod catalog; pub mod client; mod client_json; mod client_sse; +pub(crate) mod connection; pub mod diagnostics; mod downstream; pub mod endpoints; diff --git a/backend/crates/providers/openai/src/transport/websocket/coordinator.rs b/backend/crates/providers/openai/src/transport/websocket/coordinator.rs index e01665a4e..ff350df14 100644 --- a/backend/crates/providers/openai/src/transport/websocket/coordinator.rs +++ b/backend/crates/providers/openai/src/transport/websocket/coordinator.rs @@ -268,6 +268,7 @@ async fn prepare_pooled_websocket( pool.clone(), connect_lease, permit, + fast_path_budget.is_some(), ); match waiter.wait(fast_path_budget).await? { WebSocketFastPath::Ready(handoff) => Ok(WebSocketFastPath::Ready(PreparedWebSocket { @@ -323,6 +324,7 @@ fn start_pooled_websocket_connect( pool: CodexWebSocketPool, connect_lease: WebSocketPoolConnectLease, permit: Option, + fast_path: bool, ) -> PooledWebSocketConnectWaiter { let task_key = connect_lease.key().clone(); let started_at = connect_lease.started_at(); @@ -352,7 +354,7 @@ fn start_pooled_websocket_connect( ); return; } - result = connect_websocket_connection(&connection, keepalive, context) => result, + result = connect_websocket_connection(&connection, keepalive, context, fast_path) => result, }; match finish_breaker_attempt(permit, connected) { Ok((connection, connect_elapsed)) => { @@ -460,7 +462,7 @@ async fn connect_with_budget( match wait_for_fast_path( TokioInstant::now(), fast_path_budget, - connect_websocket_connection(connection, keepalive, context), + connect_websocket_connection(connection, keepalive, context, fast_path_budget.is_some()), ) .await { @@ -473,9 +475,11 @@ async fn connect_websocket_connection( connection: &CodexWebSocketConnection, keepalive: PumpKeepalive, context: PumpLogContext, + fast_path: bool, ) -> Result<(PooledWebSocketConnection, Duration), CodexWebSocketExchangeError> { let started_at = Instant::now(); - let (websocket, response) = connect_pumped_websocket(connection, keepalive, context).await?; + let (websocket, response) = + connect_pumped_websocket(connection, keepalive, context, fast_path).await?; Ok(( PooledWebSocketConnection { websocket, @@ -516,6 +520,10 @@ fn finish_breaker_attempt( permit.succeed(); Ok(connection) } + Err(error) if crate::transport::connection::is_admission_failure(&error) => { + permit.cancel(); + Err(error) + } Err(CodexWebSocketExchangeError::Upstream(upstream)) if upstream.status_code < 500 => { // 账号或请求级 opening 响应证明 origin 可达，不得污染 transport 熔断器。 permit.succeed(); diff --git a/backend/crates/providers/openai/src/transport/websocket/handshake.rs b/backend/crates/providers/openai/src/transport/websocket/handshake.rs index 7c3075259..043250464 100644 --- a/backend/crates/providers/openai/src/transport/websocket/handshake.rs +++ b/backend/crates/providers/openai/src/transport/websocket/handshake.rs @@ -66,6 +66,7 @@ impl CodexWebSocketConnection { endpoint, headers, outbound_proxy: None, + connection_budget: None, } } @@ -121,8 +122,9 @@ pub(super) async fn connect_pumped_websocket( connection: &CodexWebSocketConnection, keepalive: PumpKeepalive, context: PumpLogContext, + fast_path: bool, ) -> Result<(PumpedWebSocket, WsResponse>>), CodexWebSocketExchangeError> { - let (raw, response) = connect_websocket(connection).await?; + let (raw, response) = connect_websocket(connection, fast_path).await?; Ok((PumpedWebSocket::new(raw, keepalive, context), response)) } @@ -158,6 +160,7 @@ pub(super) fn websocket_connection_metadata( async fn connect_websocket( connection: &CodexWebSocketConnection, + fast_path: bool, ) -> Result<(RawWsStream, WsResponse>>), CodexWebSocketExchangeError> { let request = websocket_handshake_request(connection)?; let connector = tls::maybe_build_rustls_client_config_with_custom_ca() @@ -167,7 +170,27 @@ async fn connect_websocket( ))) })? .map(Connector::Rustls); - let result = timeout(WEBSOCKET_CONNECT_TIMEOUT, async { + let remaining = connection + .connection_budget + .as_ref() + .map(|budget| budget.begin()) + .transpose() + .map_err(|error| { + CodexWebSocketExchangeError::Connect(tungstenite::Error::Io(std::io::Error::other( + error, + ))) + })? + .flatten(); + let connect_timeout = remaining.map_or(WEBSOCKET_CONNECT_TIMEOUT, |remaining| { + remaining.min(WEBSOCKET_CONNECT_TIMEOUT) + }); + let result = timeout(connect_timeout, async { + let _permit = if fast_path { + crate::transport::connection::try_acquire() + } else { + crate::transport::connection::acquire().await + } + .map_err(tungstenite::Error::Io)?; // Preserve the native direct handshake; explicit egress never inherits a global proxy. if connection.outbound_proxy.is_none() && matches!( @@ -202,7 +225,7 @@ async fn connect_websocket( }) .await .map_err(|_| CodexWebSocketExchangeError::ConnectTimeout { - timeout: WEBSOCKET_CONNECT_TIMEOUT, + timeout: connect_timeout, })?; match result { Ok((websocket, response)) => Ok((websocket, response)), @@ -273,7 +296,6 @@ async fn dial_account( tokio_tungstenite::proxy::connect_via_proxy(BufWriter::new(stream), &config, &target, port) .await .map(BufWriter::into_inner) - .map_err(|_| invalid()) } async fn connect_tcp(host: &str, port: u16) -> Result { diff --git a/backend/crates/providers/openai/src/transport/websocket/model.rs b/backend/crates/providers/openai/src/transport/websocket/model.rs index ecbc0a3bf..23750c382 100644 --- a/backend/crates/providers/openai/src/transport/websocket/model.rs +++ b/backend/crates/providers/openai/src/transport/websocket/model.rs @@ -7,13 +7,24 @@ use crate::transport::protocol::responses::{ }; /// WebSocket opening 描述。 -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone)] pub struct CodexWebSocketConnection { + pub(crate) connection_budget: Option, pub(crate) outbound_proxy: Option, pub(super) endpoint: String, pub(super) headers: Vec<(String, String)>, } +// 请求预算不是连接身份；连接池复用不能由请求局部状态决定。 +impl PartialEq for CodexWebSocketConnection { + fn eq(&self, other: &Self) -> bool { + self.endpoint == other.endpoint + && self.headers == other.headers + && self.outbound_proxy == other.outbound_proxy + } +} +impl Eq for CodexWebSocketConnection {} + /// 已构造完成的 Responses WebSocket 请求描述。 #[derive(Debug, Clone, PartialEq, Eq)] pub struct CodexWebSocketRequest { @@ -115,6 +126,7 @@ impl CodexWebSocketConnection { endpoint: endpoint.into(), headers, outbound_proxy: None, + connection_budget: None, } } diff --git a/backend/crates/providers/openai/tests/provider/contract.rs b/backend/crates/providers/openai/tests/provider/contract.rs index e1878c807..626455857 100644 --- a/backend/crates/providers/openai/tests/provider/contract.rs +++ b/backend/crates/providers/openai/tests/provider/contract.rs @@ -11145,3 +11145,198 @@ async fn invalid_oauth_candidate_does_not_block_another_websocket_account() { "acct_provider_contract" ); } + +#[tokio::test] +async fn transient_socks_openings_offer_same_account_http_retry_and_send_one_payload() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let proxy = format!("socks5h://{}", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + for opening in 0..3 { + let (mut socket, _) = listener.accept().await.unwrap(); + let mut greeting = [0_u8; 2]; + socket.read_exact(&mut greeting).await.unwrap(); + assert_eq!(greeting[0], 5); + let mut methods = vec![0; usize::from(greeting[1])]; + socket.read_exact(&mut methods).await.unwrap(); + if opening < 2 { + continue; + } + socket.write_all(&[5, 0]).await.unwrap(); + let mut connect = [0_u8; 4]; + socket.read_exact(&mut connect).await.unwrap(); + assert_eq!(connect, [5, 1, 0, 3]); + let length = socket.read_u8().await.unwrap(); + let mut target = vec![0; usize::from(length) + 2]; + socket.read_exact(&mut target).await.unwrap(); + socket + .write_all(&[5, 0, 0, 1, 127, 0, 0, 1, 0, 80]) + .await + .unwrap(); + read_http_request(&mut socket).await; + let body = CAPTURE_COMPLETED_SSE; + socket.write_all(format!("HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).as_bytes()).await.unwrap(); + } + }); + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + store.set_egress( + "acct_provider_contract", + Some(gateway_core::account::OutboundProxy::parse(&proxy).unwrap()), + None, + ); + let provider = provider_with_base_url(&store, "http://upstream.invalid".to_owned()); + let context = fallback_transport_context("req_socks_recovery"); + for opening in 0..3 { + let mut stream = provider + .clone() + .execute( + planned_request("openai", generate_operation()), + context.clone(), + ) + .await + .unwrap(); + let mut failure = None; + while let Some(event) = stream.next().await { + if let Err(error) = event { + failure = Some(error); + break; + } + } + if opening < 2 { + let failure = failure.expect("SOCKS negotiation must fail before payload"); + assert_eq!(failure.send_state(), UpstreamSendState::NotSent); + assert!(matches!( + failure.pre_delivery_retry(), + Some(PreDeliveryRetry::SameAccountConnectionRetry { + transport: AttemptTransport::Fallback + }) + )); + } else { + assert!(failure.is_none()); + } + } + timeout(Duration::from_secs(5), server) + .await + .unwrap() + .unwrap(); +} + +#[tokio::test] +async fn lost_http_response_after_payload_does_not_request_connection_replay() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base = format!("http://{}", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.unwrap(); + read_http_request(&mut socket).await; + }); + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let mut stream = provider_with_base_url(&store, base) + .execute( + planned_request("openai", generate_operation()), + fallback_transport_context("req_response_lost"), + ) + .await + .unwrap(); + let mut failure = None; + while let Some(event) = stream.next().await { + if let Err(error) = event { + failure = Some(error); + break; + } + } + let failure = failure.unwrap(); + assert_ne!(failure.send_state(), UpstreamSendState::NotSent); + assert!(!matches!( + failure.pre_delivery_retry(), + Some(PreDeliveryRetry::SameAccountConnectionRetry { .. }) + )); + server.await.unwrap(); +} + +#[tokio::test] +async fn stalled_proxy_connections_share_thirty_seconds_instead_of_resetting_timeout() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let proxy = format!("socks5h://{}", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + let mut sockets = Vec::new(); + loop { + sockets.push(listener.accept().await.unwrap().0); + } + }); + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + store.set_egress( + "acct_provider_contract", + Some(gateway_core::account::OutboundProxy::parse(&proxy).unwrap()), + None, + ); + let provider = provider_with_base_url(&store, "http://upstream.invalid".to_owned()); + let context = fallback_transport_context("req_socks_timeout"); + let started = std::time::Instant::now(); + timeout(Duration::from_secs(32), async { + for _ in 0..4 { + let mut stream = provider + .clone() + .execute( + planned_request("openai", generate_operation()), + context.clone(), + ) + .await + .unwrap(); + let mut failure = None; + while let Some(event) = stream.next().await { + if let Err(error) = event { + failure = Some(error); + break; + } + } + let failure = failure.unwrap(); + assert_eq!(failure.kind(), ProviderErrorKind::Timeout); + assert_eq!(failure.send_state(), UpstreamSendState::NotSent); + } + }) + .await + .expect("shared recovery window must not become four 15-second timeouts"); + assert!(started.elapsed() >= Duration::from_secs(25)); + assert!(context.connection_budget().exhausted()); + server.abort(); + let _ = server.await; +} + +pub(crate) async fn assert_local_connection_capacity_is_not_an_upstream_failure() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + // 显式出口使 Provider 使用生产建连层，而不是测试注入的裸 reqwest client。 + store.set_egress( + "acct_provider_contract", + Some(gateway_core::account::OutboundProxy::parse("http://127.0.0.1:9").unwrap()), + None, + ); + let mut stream = provider_with_base_url(&store, "http://127.0.0.1:9".to_owned()) + .execute( + planned_request("openai", generate_operation()), + fallback_transport_context("req_local_capacity"), + ) + .await + .unwrap(); + let mut failure = None; + while let Some(event) = stream.next().await { + if let Err(error) = event { + failure = Some(error); + break; + } + } + let failure = failure.unwrap(); + assert_eq!( + failure.kind(), + ProviderErrorKind::ProviderInfrastructureUnavailable + ); + assert_eq!(failure.send_state(), UpstreamSendState::NotSent); + assert_eq!(failure.diagnostic().unwrap().stage(), Some("admission")); + assert_eq!( + failure.diagnostic().unwrap().code(), + Some("local_connection_capacity") + ); + assert!(failure.pre_delivery_retry().is_none()); +} diff --git a/backend/crates/providers/openai/tests/provider/mod.rs b/backend/crates/providers/openai/tests/provider/mod.rs index cbddfe24c..321fdeb5d 100644 --- a/backend/crates/providers/openai/tests/provider/mod.rs +++ b/backend/crates/providers/openai/tests/provider/mod.rs @@ -1,2 +1,4 @@ mod contract; mod failure; + +pub(crate) use contract::assert_local_connection_capacity_is_not_an_upstream_failure; diff --git a/backend/crates/providers/openai/tests/transport/connection.rs b/backend/crates/providers/openai/tests/transport/connection.rs new file mode 100644 index 000000000..5445ad0b9 --- /dev/null +++ b/backend/crates/providers/openai/tests/transport/connection.rs @@ -0,0 +1,152 @@ +//! 在独立测试进程中验证共享建连准入，避免其他网络测试占用全局名额。 + +use super::*; +use gateway_core::account::OutboundProxy; +use provider_openai::transport::client::build_account_http_client; + +#[tokio::test] +async fn cold_connection_admission_bounds_queue_and_releases_cancelled_work() { + const CHILD: &str = "CPR_CONNECTION_ADMISSION_TEST_CHILD"; + if std::env::var_os(CHILD).is_none() { + let status = Command::new(std::env::current_exe().unwrap()) + .args(["--exact", "transport::connection::cold_connection_admission_bounds_queue_and_releases_cancelled_work", "--nocapture"]) + .env(CHILD, "1") + .status().unwrap(); + assert!(status.success()); + return; + } + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let proxy = + OutboundProxy::parse(&format!("http://{}", listener.local_addr().unwrap())).unwrap(); + let client = build_account_http_client("acct_connection_admission", Some(&proxy)).unwrap(); + let (completed, mut results) = tokio::sync::mpsc::unbounded_channel(); + let mut tasks = Vec::new(); + let spawn = |client: reqwest::Client, completed: tokio::sync::mpsc::UnboundedSender| { + tokio::spawn(async move { + let result = client.get("https://upstream.invalid/").send().await; + let _ = completed.send(result.is_err_and(|error| error.is_connect())); + }) + }; + for _ in 0..128 { + tasks.push(spawn(client.clone(), completed.clone())); + } + let mut sockets = Vec::new(); + for _ in 0..128 { + sockets.push( + timeout(Duration::from_secs(10), listener.accept()) + .await + .unwrap() + .unwrap() + .0, + ); + } + // 128 条 CONNECT 握手保持挂起；1024 个等待者之外的一次请求应立即拒绝。 + for _ in 0..1025 { + tasks.push(spawn(client.clone(), completed.clone())); + } + assert_eq!( + timeout(Duration::from_secs(3), results.recv()) + .await + .unwrap(), + Some(true) + ); + assert!( + timeout(Duration::from_millis(100), results.recv()) + .await + .is_err() + ); + assert!( + timeout(Duration::from_millis(100), listener.accept()) + .await + .is_err() + ); + crate::provider::assert_local_connection_capacity_is_not_an_upstream_failure().await; + + let upstream = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base_url = format!("http://{}", upstream.local_addr().unwrap()); + let server = tokio::spawn(async move { + for _ in 0..3 { + let (mut http, _) = upstream.accept().await.unwrap(); + assert!( + read_http_request(&mut http) + .await + .starts_with("POST /codex/responses") + ); + write_completed_sse_response(&mut http).await; + } + let (stream, _) = upstream.accept().await.unwrap(); + let mut websocket = accept_codex_test_websocket(stream).await; + websocket.next().await.unwrap().unwrap(); + websocket + .send(tokio_tungstenite::tungstenite::Message::Text( + completed_websocket_response("resp_after_local_capacity", 2, 1).into(), + )) + .await + .unwrap(); + }); + let backend = CodexBackendClient::new( + reqwest::Client::builder().no_proxy().build().unwrap(), + base_url, + test_wire_profile(), + ); + for attempt in 0..3 { + let mut request = codex_request("gpt-5.5", "be brief", Vec::new()); + request.use_websocket = true; + request.local_conversation_id = Some(format!("capacity-{attempt}")); + let response = backend + .create_response( + &request, + request_context("req_local_ws_capacity", Some("chatgpt-account")), + ) + .await + .expect("local WS capacity should fall back to HTTP"); + assert_eq!(response.transport, CodexBackendTransport::HttpSse); + assert_eq!( + response.transport_metrics.decision, + Some(CodexTransportDecision::Http2LocalConnectionCapacity) + ); + } + + tasks[0].abort(); + sockets.push( + timeout(Duration::from_secs(3), listener.accept()) + .await + .unwrap() + .unwrap() + .0, + ); + for task in &tasks { + task.abort(); + } + for task in tasks { + let _ = task.await; + } + // 同时取消活动与排队请求后，新请求仍能取得名额。 + let fresh = spawn(client, completed); + let _socket = timeout(Duration::from_secs(3), listener.accept()) + .await + .unwrap() + .unwrap(); + fresh.abort(); + let _ = fresh.await; + drop(sockets); + + let mut request = codex_request("gpt-5.5", "be brief", Vec::new()); + request.use_websocket = true; + request.local_conversation_id = Some("capacity-recovered".to_owned()); + let response = timeout( + Duration::from_secs(3), + backend.create_response( + &request, + request_context("req_ws_after_capacity", Some("chatgpt-account")), + ), + ) + .await + .unwrap() + .expect("local capacity must not open the WS origin breaker"); + assert_eq!(response.transport, CodexBackendTransport::WebSocket); + timeout(Duration::from_secs(3), server) + .await + .unwrap() + .unwrap(); +} diff --git a/backend/crates/providers/openai/tests/transport/mod.rs b/backend/crates/providers/openai/tests/transport/mod.rs index 8b21d4049..842f3ee55 100644 --- a/backend/crates/providers/openai/tests/transport/mod.rs +++ b/backend/crates/providers/openai/tests/transport/mod.rs @@ -49,6 +49,7 @@ mod account_proxy; mod canonical; mod catalog; mod client; +mod connection; mod diagnostics; mod downstream; mod endpoints;