diff --git a/src-tauri/src/commands/failover.rs b/src-tauri/src/commands/failover.rs index e4e48ddc1..a0c40ce66 100644 --- a/src-tauri/src/commands/failover.rs +++ b/src-tauri/src/commands/failover.rs @@ -86,11 +86,19 @@ pub async fn set_auto_failover_enabled( "[Failover] Setting auto_failover_enabled: app_type='{app_type}', enabled={enabled}" ); - // 强一致语义：开启故障转移后立即切到队列 P1（并确保队列非空） - // - // 说明： - // - 仅在 enabled=true 时执行“切到 P1” - // - 若队列为空，则尝试把“当前供应商”自动加入队列作为 P1，避免用户在 UI 上陷入死锁（无法先加队列再开启） + // 读取当前配置 + let mut config = state + .db + .get_proxy_config_for_app(&app_type) + .await + .map_err(|e| e.to_string())?; + + if enabled && !config.enabled { + return Err("需要先启用该应用的代理接管，再开启故障转移".to_string()); + } + + // 队列为空时把当前供应商自动加入作为 P1，避免用户陷入"必须先加队列才能开启"的死锁 + let mut auto_added_provider_id: Option = None; let p1_provider_id = if enabled { let mut queue = state .db @@ -112,6 +120,7 @@ pub async fn set_auto_failover_enabled( .db .add_to_failover_queue(&app_type, ¤t_id) .map_err(|e| e.to_string())?; + auto_added_provider_id = Some(current_id); queue = state .db @@ -127,12 +136,20 @@ pub async fn set_auto_failover_enabled( String::new() }; - // 读取当前配置 - let mut config = state - .db - .get_proxy_config_for_app(&app_type) - .await - .map_err(|e| e.to_string())?; + // 开启前先切到 P1。只有切换成功后才写入 auto_failover_enabled=true， + // 避免 P1 不可切换（例如 official provider）时留下“开关已开但目标未切”的脏状态。 + if enabled { + if let Err(e) = state + .proxy_service + .switch_proxy_target(&app_type, &p1_provider_id) + .await + { + if let Some(provider_id) = auto_added_provider_id { + let _ = state.db.remove_from_failover_queue(&app_type, &provider_id); + } + return Err(e); + } + } // 更新 auto_failover_enabled 字段 config.auto_failover_enabled = enabled; @@ -144,13 +161,7 @@ pub async fn set_auto_failover_enabled( .await .map_err(|e| e.to_string())?; - // 开启后立即切到 P1：更新 is_current + 本地 settings + Live 备份（接管模式下） if enabled { - state - .proxy_service - .switch_proxy_target(&app_type, &p1_provider_id) - .await?; - // 发射 provider-switched 事件（让前端刷新当前供应商） let event_data = serde_json::json!({ "appType": app_type, diff --git a/src-tauri/src/commands/proxy.rs b/src-tauri/src/commands/proxy.rs index f6feaf94d..93f62b403 100644 --- a/src-tauri/src/commands/proxy.rs +++ b/src-tauri/src/commands/proxy.rs @@ -133,9 +133,17 @@ pub async fn update_proxy_config_for_app( config: AppProxyConfig, ) -> Result<(), String> { let db = &state.db; + let app_type = config.app_type.clone(); + let circuit_config = CircuitBreakerConfig::from(&config); + db.update_proxy_config_for_app(config) .await - .map_err(|e| e.to_string()) + .map_err(|e| e.to_string())?; + + state + .proxy_service + .update_circuit_breaker_config_for_app(&app_type, circuit_config) + .await } async fn get_default_cost_multiplier_internal( diff --git a/src-tauri/src/proxy/circuit_breaker.rs b/src-tauri/src/proxy/circuit_breaker.rs index 890a5138f..03ef8aead 100644 --- a/src-tauri/src/proxy/circuit_breaker.rs +++ b/src-tauri/src/proxy/circuit_breaker.rs @@ -3,6 +3,7 @@ //! 实现熔断器模式，用于防止向不健康的供应商发送请求 use super::log_codes::cb as log_cb; +use super::types::AppProxyConfig; use serde::{Deserialize, Serialize}; use std::sync::atomic::{AtomicU32, Ordering}; use std::sync::Arc; @@ -47,6 +48,18 @@ pub struct CircuitBreakerConfig { pub min_requests: u32, } +impl From<&AppProxyConfig> for CircuitBreakerConfig { + fn from(config: &AppProxyConfig) -> Self { + Self { + failure_threshold: config.circuit_failure_threshold, + success_threshold: config.circuit_success_threshold, + timeout_seconds: config.circuit_timeout_seconds as u64, + error_rate_threshold: config.circuit_error_rate_threshold, + min_requests: config.circuit_min_requests, + } + } +} + impl Default for CircuitBreakerConfig { fn default() -> Self { Self { diff --git a/src-tauri/src/proxy/forwarder.rs b/src-tauri/src/proxy/forwarder.rs index f0ba93066..74996b429 100644 --- a/src-tauri/src/proxy/forwarder.rs +++ b/src-tauri/src/proxy/forwarder.rs @@ -25,6 +25,7 @@ use crate::commands::{CodexOAuthState, CopilotAuthState}; use crate::proxy::providers::codex_oauth_auth::CodexOAuthManager; use crate::proxy::providers::copilot_auth::CopilotAuthManager; use crate::{app_config::AppType, provider::Provider}; +use futures::StreamExt; use http::Extensions; use serde_json::Value; use std::sync::Arc; @@ -1164,8 +1165,9 @@ impl RequestForwarder { &filtered_body, self.session_client_provided, ); - let force_identity_encoding = needs_transform - || should_force_identity_encoding(&effective_endpoint, &filtered_body, headers); + let request_is_streaming = + is_streaming_request(&effective_endpoint, &filtered_body, headers); + let force_identity_encoding = needs_transform || request_is_streaming; // Codex OAuth 需要注入的 ChatGPT-Account-Id（在动态 token 获取期间填充） let mut codex_oauth_account_id: Option = None; @@ -1609,8 +1611,6 @@ impl RequestForwarder { ); let client = super::http_client::get(); let mut request = client.request(method.clone(), &url); - let request_is_streaming = - is_streaming_request(&effective_endpoint, &filtered_body, headers); if request_is_streaming { // reqwest 的 timeout 是整请求超时；流式请求交给 response_processor // 的首包/静默期超时控制，避免长流被总时长误杀。 @@ -1663,6 +1663,9 @@ impl RequestForwarder { let status = response.status(); if status.is_success() { + let response = self + .prepare_success_response_for_failover(response, request_is_streaming) + .await?; Ok((response, resolved_claude_api_format)) } else { let status_code = status.as_u16(); @@ -1675,6 +1678,73 @@ impl RequestForwarder { } } + /// 故障转移开启时，成功不能只看上游响应头。 + /// + /// - 非流式：先把完整 body 读到内存，读超时/连接中断会回到 retry loop 尝试下一家。 + /// - 流式：至少等首个 chunk 到达，避免上游返回 200 后一直不吐 SSE 时被误记成功。 + async fn prepare_success_response_for_failover( + &self, + response: ProxyResponse, + request_is_streaming: bool, + ) -> Result { + if request_is_streaming { + return self.prime_streaming_response(response).await; + } + + if self.non_streaming_timeout.is_zero() { + return Ok(response); + } + + let status = response.status(); + let headers = response.headers().clone(); + let body_timeout = self.non_streaming_timeout; + let body = tokio::time::timeout(body_timeout, response.bytes()) + .await + .map_err(|_| { + ProxyError::Timeout(format!( + "响应体读取超时: {}s（上游发完响应头后 body 未到达）", + body_timeout.as_secs() + )) + })??; + + Ok(ProxyResponse::buffered(status, headers, body)) + } + + async fn prime_streaming_response( + &self, + response: ProxyResponse, + ) -> Result { + if self.streaming_first_byte_timeout.is_zero() { + return Ok(response); + } + + let status = response.status(); + let headers = response.headers().clone(); + let timeout = self.streaming_first_byte_timeout; + let mut stream = Box::pin(response.bytes_stream()); + + let first = tokio::time::timeout(timeout, stream.next()) + .await + .map_err(|_| { + ProxyError::Timeout(format!( + "流式响应首包超时: {}s（上游已返回响应头但未返回数据）", + timeout.as_secs() + )) + })?; + + let Some(first) = first else { + return Err(ProxyError::ForwardFailed( + "流式响应在首包到达前结束".to_string(), + )); + }; + + let first = + first.map_err(|e| ProxyError::ForwardFailed(format!("读取流式响应首包失败: {e}")))?; + + let replay = futures::stream::once(async move { Ok(first) }).chain(stream); + Ok(ProxyResponse::streamed(status, headers, replay)) + } + async fn resolve_claude_api_format( &self, provider: &Provider, @@ -2213,9 +2283,14 @@ fn value_for_log(value: &Value) -> String { #[cfg(test)] mod tests { use super::*; + use crate::database::Database; use axum::http::header::{HeaderValue, ACCEPT}; use axum::http::HeaderMap; + use bytes::Bytes; + use http::StatusCode; use serde_json::json; + use std::collections::HashMap; + use std::time::Duration; fn test_provider_with_type(provider_type: Option<&str>) -> Provider { Provider { @@ -2237,6 +2312,31 @@ mod tests { } } + fn test_forwarder( + non_streaming_timeout: Duration, + streaming_first_byte_timeout: Duration, + ) -> RequestForwarder { + let db = Arc::new(Database::memory().expect("memory db")); + + RequestForwarder { + router: Arc::new(ProviderRouter::new(db.clone())), + status: Arc::new(RwLock::new(ProxyStatus::default())), + current_providers: Arc::new(RwLock::new(HashMap::new())), + gemini_shadow: Arc::new(GeminiShadowStore::new()), + failover_manager: Arc::new(FailoverSwitchManager::new(db)), + app_handle: None, + current_provider_id_at_start: String::new(), + session_id: String::new(), + session_client_provided: false, + rectifier_config: RectifierConfig::default(), + optimizer_config: OptimizerConfig::default(), + copilot_optimizer_config: CopilotOptimizerConfig::default(), + non_streaming_timeout, + streaming_first_byte_timeout, + max_attempts: 1, + } + } + #[test] fn single_provider_retryable_log_uses_single_provider_code() { let error = ProxyError::UpstreamError { @@ -2382,6 +2482,96 @@ mod tests { ); } + #[tokio::test] + async fn non_streaming_success_is_buffered_before_marking_provider_successful() { + let forwarder = test_forwarder(Duration::from_secs(1), Duration::from_secs(1)); + let response = ProxyResponse::streamed( + StatusCode::OK, + HeaderMap::new(), + futures::stream::once(async { + tokio::time::sleep(Duration::from_millis(10)).await; + Ok::(Bytes::from_static(b"{\"ok\":true}")) + }), + ); + + let prepared = forwarder + .prepare_success_response_for_failover(response, false) + .await + .expect("response should be buffered"); + + assert_eq!( + prepared.bytes().await.unwrap(), + Bytes::from_static(b"{\"ok\":true}") + ); + } + + #[tokio::test] + async fn non_streaming_body_read_error_is_retryable_before_success_record() { + let forwarder = test_forwarder(Duration::from_secs(1), Duration::from_secs(1)); + let response = ProxyResponse::streamed( + StatusCode::OK, + HeaderMap::new(), + futures::stream::once(async { + Err::(std::io::Error::other("body boom")) + }), + ); + + let err = match forwarder + .prepare_success_response_for_failover(response, false) + .await + { + Ok(_) => panic!("body read errors should fail the attempt"), + Err(err) => err, + }; + + assert!(matches!(err, ProxyError::ForwardFailed(_))); + } + + #[tokio::test] + async fn streaming_success_primes_first_chunk_and_replays_it() { + let forwarder = test_forwarder(Duration::from_secs(1), Duration::from_secs(1)); + let response = ProxyResponse::streamed( + StatusCode::OK, + HeaderMap::new(), + futures::stream::iter(vec![ + Ok::(Bytes::from_static(b"first")), + Ok::(Bytes::from_static(b"second")), + ]), + ); + + let prepared = forwarder + .prepare_success_response_for_failover(response, true) + .await + .expect("stream should be primed"); + + assert_eq!( + prepared.bytes().await.unwrap(), + Bytes::from_static(b"firstsecond") + ); + } + + #[tokio::test] + async fn streaming_first_chunk_error_is_retryable_before_success_record() { + let forwarder = test_forwarder(Duration::from_secs(1), Duration::from_secs(1)); + let response = ProxyResponse::streamed( + StatusCode::OK, + HeaderMap::new(), + futures::stream::once(async { + Err::(std::io::Error::other("first chunk boom")) + }), + ); + + let err = match forwarder + .prepare_success_response_for_failover(response, true) + .await + { + Ok(_) => panic!("first chunk errors should fail the attempt"), + Err(err) => err, + }; + + assert!(matches!(err, ProxyError::ForwardFailed(_))); + } + #[test] fn codex_oauth_session_headers_match_codex_cache_identity() { let headers = build_codex_oauth_session_headers("session-123"); diff --git a/src-tauri/src/proxy/hyper_client.rs b/src-tauri/src/proxy/hyper_client.rs index f4533759b..9dad42b67 100644 --- a/src-tauri/src/proxy/hyper_client.rs +++ b/src-tauri/src/proxy/hyper_client.rs @@ -6,7 +6,7 @@ use super::ProxyError; use bytes::Bytes; -use futures::stream::Stream; +use futures::{stream::Stream, StreamExt}; use http_body_util::BodyExt; use hyper_rustls::HttpsConnectorBuilder; use hyper_util::{client::legacy::Client, rt::TokioExecutor}; @@ -79,13 +79,44 @@ fn global_hyper_client() -> &'static HyperClient { pub enum ProxyResponse { Hyper(hyper::Response), Reqwest(reqwest::Response), + Buffered { + status: http::StatusCode, + headers: http::HeaderMap, + body: Bytes, + }, + Streamed { + status: http::StatusCode, + headers: http::HeaderMap, + stream: std::pin::Pin> + Send>>, + }, } impl ProxyResponse { + pub fn buffered(status: http::StatusCode, headers: http::HeaderMap, body: Bytes) -> Self { + Self::Buffered { + status, + headers, + body, + } + } + + pub fn streamed( + status: http::StatusCode, + headers: http::HeaderMap, + stream: impl Stream> + Send + 'static, + ) -> Self { + Self::Streamed { + status, + headers, + stream: Box::pin(stream), + } + } + pub fn status(&self) -> http::StatusCode { match self { Self::Hyper(r) => r.status(), Self::Reqwest(r) => r.status(), + Self::Buffered { status, .. } | Self::Streamed { status, .. } => *status, } } @@ -93,6 +124,7 @@ impl ProxyResponse { match self { Self::Hyper(r) => r.headers(), Self::Reqwest(r) => r.headers(), + Self::Buffered { headers, .. } | Self::Streamed { headers, .. } => headers, } } @@ -122,6 +154,17 @@ impl ProxyResponse { Self::Reqwest(r) => r.bytes().await.map_err(|e| { ProxyError::ForwardFailed(format!("Failed to read response body: {e}")) }), + Self::Buffered { body, .. } => Ok(body), + Self::Streamed { mut stream, .. } => { + let mut body = bytes::BytesMut::new(); + while let Some(chunk) = stream.next().await { + let chunk = chunk.map_err(|e| { + ProxyError::ForwardFailed(format!("Failed to read response body: {e}")) + })?; + body.extend_from_slice(&chunk); + } + Ok(body.freeze()) + } } } @@ -161,6 +204,9 @@ impl ProxyResponse { .map(|r| r.map_err(|e| std::io::Error::other(e.to_string()))); Box::pin(stream) } + Self::Buffered { body, .. } => Box::pin(futures::stream::once(async move { Ok(body) })) + as std::pin::Pin> + Send>>, + Self::Streamed { stream, .. } => stream, } } } diff --git a/src-tauri/src/proxy/provider_router.rs b/src-tauri/src/proxy/provider_router.rs index 28e23f87e..28d2b8a2d 100644 --- a/src-tauri/src/proxy/provider_router.rs +++ b/src-tauri/src/proxy/provider_router.rs @@ -201,6 +201,17 @@ impl ProviderRouter { } } + /// 更新指定应用已创建熔断器的配置（热更新） + pub async fn update_app_configs(&self, app_type: &str, config: CircuitBreakerConfig) { + let prefix = format!("{app_type}:"); + let breakers = self.circuit_breakers.read().await; + for (key, breaker) in breakers.iter() { + if key.starts_with(&prefix) { + breaker.update_config(config.clone()).await; + } + } + } + /// 获取熔断器状态 #[allow(dead_code)] pub async fn get_circuit_breaker_stats( diff --git a/src-tauri/src/proxy/server.rs b/src-tauri/src/proxy/server.rs index b0595b83e..2bc9fe2b9 100644 --- a/src-tauri/src/proxy/server.rs +++ b/src-tauri/src/proxy/server.rs @@ -360,6 +360,17 @@ impl ProxyServer { self.state.provider_router.update_all_configs(config).await; } + pub async fn update_circuit_breaker_config_for_app( + &self, + app_type: &str, + config: super::circuit_breaker::CircuitBreakerConfig, + ) { + self.state + .provider_router + .update_app_configs(app_type, config) + .await; + } + /// 重置指定 Provider 的熔断器 pub async fn reset_provider_circuit_breaker(&self, provider_id: &str, app_type: &str) { self.state diff --git a/src-tauri/src/services/proxy.rs b/src-tauri/src/services/proxy.rs index f5a18fdf7..bcb446588 100644 --- a/src-tauri/src/services/proxy.rs +++ b/src-tauri/src/services/proxy.rs @@ -1993,6 +1993,23 @@ impl ProxyService { Ok(()) } + /// 热更新指定应用的熔断器配置 + pub async fn update_circuit_breaker_config_for_app( + &self, + app_type: &str, + config: crate::proxy::CircuitBreakerConfig, + ) -> Result<(), String> { + if let Some(server) = self.server.read().await.as_ref() { + server + .update_circuit_breaker_config_for_app(app_type, config) + .await; + log::info!("已热更新 {app_type} 运行中的熔断器配置"); + } else { + log::debug!("{app_type} 熔断器配置将在下次代理启动时生效"); + } + Ok(()) + } + /// 重置指定 Provider 的熔断器 /// /// 如果代理服务器正在运行，立即重置内存中的熔断器状态 diff --git a/src/components/providers/ProviderCard.tsx b/src/components/providers/ProviderCard.tsx index 859393dcd..cfa46231d 100644 --- a/src/components/providers/ProviderCard.tsx +++ b/src/components/providers/ProviderCard.tsx @@ -364,6 +364,7 @@ export function ProviderCard({ {isProxyRunning && isInFailoverQueue && health && ( )} diff --git a/src/components/providers/ProviderHealthBadge.tsx b/src/components/providers/ProviderHealthBadge.tsx index aa6b55ad6..3e03dbc88 100644 --- a/src/components/providers/ProviderHealthBadge.tsx +++ b/src/components/providers/ProviderHealthBadge.tsx @@ -4,6 +4,7 @@ import { useTranslation } from "react-i18next"; interface ProviderHealthBadgeProps { consecutiveFailures: number; + isHealthy?: boolean; className?: string; } @@ -13,6 +14,7 @@ interface ProviderHealthBadgeProps { */ export function ProviderHealthBadge({ consecutiveFailures, + isHealthy, className, }: ProviderHealthBadgeProps) { const { t } = useTranslation(); @@ -29,7 +31,7 @@ export function ProviderHealthBadge({ bgColor: "bg-green-500/10", textColor: "text-green-600 dark:text-green-400", }; - } else if (consecutiveFailures < 5) { + } else if (isHealthy !== false) { return { labelKey: "health.degraded", labelFallback: "降级", diff --git a/src/components/proxy/FailoverToggle.tsx b/src/components/proxy/FailoverToggle.tsx index 9e6e23121..757a73018 100644 --- a/src/components/proxy/FailoverToggle.tsx +++ b/src/components/proxy/FailoverToggle.tsx @@ -10,6 +10,7 @@ import { useAutoFailoverEnabled, useSetAutoFailoverEnabled, } from "@/lib/query/failover"; +import { useProxyStatus } from "@/hooks/useProxyStatus"; import { cn } from "@/lib/utils"; import { useTranslation } from "react-i18next"; import type { AppId } from "@/lib/api"; @@ -24,8 +25,11 @@ export function FailoverToggle({ className, activeApp }: FailoverToggleProps) { const { data: isEnabled = false, isLoading } = useAutoFailoverEnabled(activeApp); const setEnabled = useSetAutoFailoverEnabled(); + const { takeoverStatus } = useProxyStatus(); + const takeoverEnabled = takeoverStatus?.[activeApp] ?? false; const handleToggle = (checked: boolean) => { + if (checked && !takeoverEnabled) return; setEnabled.mutate({ appType: activeApp, enabled: checked }); }; @@ -36,15 +40,20 @@ export function FailoverToggle({ className, activeApp }: FailoverToggleProps) { ? "Codex" : "Gemini"; - const tooltipText = isEnabled - ? t("failover.tooltip.enabled", { + const tooltipText = !takeoverEnabled + ? t("failover.tooltip.takeoverRequired", { app: appLabel, - defaultValue: `${appLabel} 故障转移已启用\n按队列优先级（P1→P2→...）选择供应商`, + defaultValue: `请先接管 ${appLabel}，再启用故障转移`, }) - : t("failover.tooltip.disabled", { - app: appLabel, - defaultValue: `启用 ${appLabel} 故障转移\n将立即切换到队列 P1，并在失败时自动切换到下一个`, - }); + : isEnabled + ? t("failover.tooltip.enabled", { + app: appLabel, + defaultValue: `${appLabel} 故障转移已启用\n按队列优先级（P1→P2→...）选择供应商`, + }) + : t("failover.tooltip.disabled", { + app: appLabel, + defaultValue: `启用 ${appLabel} 故障转移\n将立即切换到队列 P1，并在失败时自动切换到下一个`, + }); return (
); diff --git a/src/components/proxy/ProxyPanel.tsx b/src/components/proxy/ProxyPanel.tsx index fd6b9cb02..a736c6b8c 100644 --- a/src/components/proxy/ProxyPanel.tsx +++ b/src/components/proxy/ProxyPanel.tsx @@ -725,6 +725,7 @@ function ProviderQueueItem({ {/* 健康徽章 */} ); diff --git a/src/components/settings/ProxyTabContent.tsx b/src/components/settings/ProxyTabContent.tsx index 08ba6543b..a89a7a535 100644 --- a/src/components/settings/ProxyTabContent.tsx +++ b/src/components/settings/ProxyTabContent.tsx @@ -35,6 +35,7 @@ export function ProxyTabContent({ const { isRunning, + takeoverStatus, startProxyServer, stopWithRestore, isPending: isProxyPending, @@ -176,72 +177,38 @@ export function ProxyTabContent({ Codex Gemini - -
-
-

- {t("proxy.failoverQueue.title")} -

-

- {t("proxy.failoverQueue.description")} -

-
- -
-
- -
- - -
-
-

- {t("proxy.failoverQueue.title")} -

-

- {t("proxy.failoverQueue.description")} -

-
- -
-
- -
- - -
-
-

- {t("proxy.failoverQueue.title")} -

-

- {t("proxy.failoverQueue.description")} -

-
- -
-
- -
- + {(["claude", "codex", "gemini"] as const).map((appType) => { + const failoverDisabled = + !isRunning || !(takeoverStatus?.[appType] ?? false); + return ( + +
+
+

+ {t("proxy.failoverQueue.title")} +

+

+ {t("proxy.failoverQueue.description")} +

+
+ +
+
+ +
+ + ); + })} diff --git a/src/lib/query/proxy.ts b/src/lib/query/proxy.ts index 38e162488..451d39769 100644 --- a/src/lib/query/proxy.ts +++ b/src/lib/query/proxy.ts @@ -227,8 +227,12 @@ export function useUpdateAppProxyConfig() { queryClient.invalidateQueries({ queryKey: ["appProxyConfig", variables.appType], }); + queryClient.invalidateQueries({ + queryKey: ["autoFailoverEnabled", variables.appType], + }); queryClient.invalidateQueries({ queryKey: ["proxyConfig"] }); queryClient.invalidateQueries({ queryKey: ["circuitBreakerConfig"] }); + queryClient.invalidateQueries({ queryKey: ["proxyStatus"] }); }, onError: (error: Error) => { toast.error(