Feat/auto failover switch (#440)

* feat(failover): add auto-failover master switch with proxy integration - Add persistent auto_failover_enabled setting in database - Add get/set_auto_failover_enabled commands - Provider router respects master switch state - Proxy shutdown automatically disables failover - Enabling failover auto-starts proxy server - Optimistic updates for failover queue toggle * feat(proxy): persist proxy takeover state across app restarts - Add proxy_takeover_{app_type} settings for per-app state tracking - Restore proxy takeover state automatically on app startup - Preserve state on normal exit, clear on manual stop - Add stop_with_restore_keep_state method for graceful shutdown * fix(proxy): set takeover state for all apps in start_with_takeover * fix(windows): hide console window when checking CLI versions Add CREATE_NO_WINDOW flag to prevent command prompt from flashing when detecting claude/codex/gemini CLI versions on Windows. * refactor(failover): make auto-failover toggle per-app independent - Change setting key from 'auto_failover_enabled' to 'auto_failover_enabled_{app_type}' - Update provider_router to check per-app failover setting - When failover disabled, use current provider only; when enabled, use queue order - Add unit tests for failover enabled/disabled behavior * feat(failover): auto-switch to higher priority provider on recovery - After circuit breaker reset, check if recovered provider has higher priority - Automatically switch back if queue_order is lower (higher priority) - Stream health check now resets circuit breaker on success/degraded * chore: remove unused start_proxy_with_takeover command - Remove command registration from lib.rs - Add comment clarifying failover queue is preserved on proxy stop * feat(ui): integrate failover controls into provider cards - Add failover toggle button to provider card actions - Show priority badge (P1, P2, ...) for queued providers - Highlight active provider with green border in failover mode - Sync drag-drop order with failover queue - Move per-app failover toggle to FailoverQueueManager - Simplify SettingsPage failover section * test(providers): add mocks for failover hooks in ProviderList tests * refactor(failover): merge failover_queue table into providers - Add in_failover_queue field to providers table - Remove standalone failover_queue table and related indexes - Simplify queue ordering by reusing sort_index field - Remove reorder_failover_queue and set_failover_item_enabled commands - Update frontend to use simplified FailoverQueueItem type * fix(database): ensure in_failover_queue column exists for v2 databases Add column check in create_tables to handle existing v2 databases that were created before the failover queue refactor. * fix(ui): differentiate active provider border color by proxy mode - Use green border/gradient when proxy takeover is active - Use blue border/gradient in normal mode (no proxy) - Improves visual distinction between proxy and non-proxy states * fix(database): clear provider health record when removing from failover queue When a provider is removed from the failover queue, its health monitoring is no longer needed. This change ensures the health record is also deleted from the database to prevent stale data. * fix(failover): improve cache cleanup for provider health and circuit breaker - Use removeQueries instead of invalidateQueries when stopping proxy to completely clear health and circuit breaker caches - Clear provider health and circuit breaker caches when removing from failover queue - Refresh failover queue after drag-sort since queue order depends on sort_index - Only show health badge when provider is in failover queue * style: apply prettier formatting to App.tsx and ProviderList.tsx * fix(proxy): handle missing health records and clear health on proxy stop - Return default healthy state when provider health record not found - Add clear_provider_health_for_app to clear health for specific app - Clear app health records when stopping proxy takeover * fix(proxy): track actual provider used in forwarding for accurate logging Introduce ForwardResult and ForwardError structs to return the actual provider that handled the request. This ensures usage statistics and error logs reflect the correct provider after failover.
2026-07-29 09:37:37 +08:00 · 2025-12-23 12:37:36 +08:00
parent bf570b6d2a
commit 1586451862
38 changed files with 972 additions and 810 deletions
@@ -17,6 +17,16 @@ use std::sync::Arc;
 use std::time::{Duration, Instant};
 use tokio::sync::RwLock;

+pub struct ForwardResult {
+    pub response: Response,
+    pub provider: Provider,
+}
+
+pub struct ForwardError {
+    pub error: ProxyError,
+    pub provider: Option<Provider>,
+}
+
 pub struct RequestForwarder {
    client: Client,
    /// 共享的 ProviderRouter（持有熔断器状态）
@@ -134,13 +144,16 @@ impl RequestForwarder {
        body: Value,
        headers: axum::http::HeaderMap,
        providers: Vec<Provider>,
-    ) -> Result<Response, ProxyError> {
+    ) -> Result<ForwardResult, ForwardError> {
        // 获取适配器
        let adapter = get_adapter(app_type);
        let app_type_str = app_type.as_str();

        if providers.is_empty() {
-            return Err(ProxyError::NoAvailableProvider);
+            return Err(ForwardError {
+                error: ProxyError::NoAvailableProvider,
+                provider: None,
+            });
        }

        log::info!(
@@ -150,6 +163,7 @@ impl RequestForwarder {
        );

        let mut last_error = None;
+        let mut last_provider = None;
        let mut attempted_providers = 0usize;

        // 依次尝试每个供应商
@@ -269,7 +283,10 @@ impl RequestForwarder {
                        latency
                    );

-                    return Ok(response);
+                    return Ok(ForwardResult {
+                        response,
+                        provider: provider.clone(),
+                    });
                }
                Err(e) => {
                    let latency = start.elapsed().as_millis() as u64;
@@ -310,6 +327,7 @@ impl RequestForwarder {
                            );

                            last_error = Some(e);
+                            last_provider = Some(provider.clone());
                            // 继续尝试下一个供应商
                            continue;
                        }
@@ -331,7 +349,10 @@ impl RequestForwarder {
                                provider.name,
                                e
                            );
-                            return Err(e);
+                            return Err(ForwardError {
+                                error: e,
+                                provider: Some(provider.clone()),
+                            });
                        }
                    }
                }
@@ -349,7 +370,10 @@ impl RequestForwarder {
                        (status.success_requests as f32 / status.total_requests as f32) * 100.0;
                }
            }
-            return Err(ProxyError::NoAvailableProvider);
+            return Err(ForwardError {
+                error: ProxyError::NoAvailableProvider,
+                provider: None,
+            });
        }

        // 所有供应商都失败了
@@ -369,7 +393,10 @@ impl RequestForwarder {
            providers.len()
        );

-        Err(last_error.unwrap_or(ProxyError::MaxRetriesExceeded))
+        Err(ForwardError {
+            error: last_error.unwrap_or(ProxyError::MaxRetriesExceeded),
+            provider: last_provider,
+        })
    }

    /// 转发单个请求（使用适配器）
@@ -61,27 +61,16 @@ pub async fn handle_messages(
    headers: axum::http::HeaderMap,
    Json(body): Json<Value>,
 ) -> Result<axum::response::Response, ProxyError> {
-    let ctx = RequestContext::new(&state, &body, AppType::Claude, "Claude", "claude").await?;
-
-    // 检查是否需要格式转换（OpenRouter 等中转服务）
-    let adapter = get_adapter(&AppType::Claude);
-    let needs_transform = adapter.needs_transform(&ctx.provider);
+    let mut ctx = RequestContext::new(&state, &body, AppType::Claude, "Claude", "claude").await?;

    let is_stream = body
        .get("stream")
        .and_then(|s| s.as_bool())
        .unwrap_or(false);

-    log::info!(
-        "[Claude] Provider: {}, needs_transform: {}, is_stream: {}",
-        ctx.provider.name,
-        needs_transform,
-        is_stream
-    );
-
    // 转发请求
    let forwarder = ctx.create_forwarder(&state);
-    let response = match forwarder
+    let result = match forwarder
        .forward_with_retry(
            &AppType::Claude,
            "/v1/messages",
@@ -91,13 +80,30 @@ pub async fn handle_messages(
        )
        .await
    {
-        Ok(resp) => resp,
-        Err(e) => {
-            log_forward_error(&state, &ctx, is_stream, &e);
-            return Err(e);
+        Ok(result) => result,
+        Err(mut err) => {
+            if let Some(provider) = err.provider.take() {
+                ctx.provider = provider;
+            }
+            log_forward_error(&state, &ctx, is_stream, &err.error);
+            return Err(err.error);
        }
    };

+    ctx.provider = result.provider;
+    let response = result.response;
+
+    // 检查是否需要格式转换（OpenRouter 等中转服务）
+    let adapter = get_adapter(&AppType::Claude);
+    let needs_transform = adapter.needs_transform(&ctx.provider);
+
+    log::info!(
+        "[Claude] Provider: {}, needs_transform: {}, is_stream: {}",
+        ctx.provider.name,
+        needs_transform,
+        is_stream
+    );
+
    let status = response.status();
    log::info!("[Claude] 上游响应状态: {status}");

@@ -295,7 +301,7 @@ pub async fn handle_chat_completions(
 ) -> Result<axum::response::Response, ProxyError> {
    log::info!("[Codex] ====== /v1/chat/completions 请求开始 ======");

-    let ctx = RequestContext::new(&state, &body, AppType::Codex, "Codex", "codex").await?;
+    let mut ctx = RequestContext::new(&state, &body, AppType::Codex, "Codex", "codex").await?;

    let is_stream = body
        .get("stream")
@@ -309,7 +315,7 @@ pub async fn handle_chat_completions(
    );

    let forwarder = ctx.create_forwarder(&state);
-    let response = match forwarder
+    let result = match forwarder
        .forward_with_retry(
            &AppType::Codex,
            "/v1/chat/completions",
@@ -319,13 +325,19 @@ pub async fn handle_chat_completions(
        )
        .await
    {
-        Ok(resp) => resp,
-        Err(e) => {
-            log_forward_error(&state, &ctx, is_stream, &e);
-            return Err(e);
+        Ok(result) => result,
+        Err(mut err) => {
+            if let Some(provider) = err.provider.take() {
+                ctx.provider = provider;
+            }
+            log_forward_error(&state, &ctx, is_stream, &err.error);
+            return Err(err.error);
        }
    };

+    ctx.provider = result.provider;
+    let response = result.response;
+
    log::info!("[Codex] 上游响应状态: {}", response.status());

    process_response(response, &ctx, &state, &OPENAI_PARSER_CONFIG).await
@@ -337,7 +349,7 @@ pub async fn handle_responses(
    headers: axum::http::HeaderMap,
    Json(body): Json<Value>,
 ) -> Result<axum::response::Response, ProxyError> {
-    let ctx = RequestContext::new(&state, &body, AppType::Codex, "Codex", "codex").await?;
+    let mut ctx = RequestContext::new(&state, &body, AppType::Codex, "Codex", "codex").await?;

    let is_stream = body
        .get("stream")
@@ -345,7 +357,7 @@ pub async fn handle_responses(
        .unwrap_or(false);

    let forwarder = ctx.create_forwarder(&state);
-    let response = match forwarder
+    let result = match forwarder
        .forward_with_retry(
            &AppType::Codex,
            "/v1/responses",
@@ -355,13 +367,19 @@ pub async fn handle_responses(
        )
        .await
    {
-        Ok(resp) => resp,
-        Err(e) => {
-            log_forward_error(&state, &ctx, is_stream, &e);
-            return Err(e);
+        Ok(result) => result,
+        Err(mut err) => {
+            if let Some(provider) = err.provider.take() {
+                ctx.provider = provider;
+            }
+            log_forward_error(&state, &ctx, is_stream, &err.error);
+            return Err(err.error);
        }
    };

+    ctx.provider = result.provider;
+    let response = result.response;
+
    log::info!("[Codex] 上游响应状态: {}", response.status());

    process_response(response, &ctx, &state, &CODEX_PARSER_CONFIG).await
@@ -379,7 +397,7 @@ pub async fn handle_gemini(
    Json(body): Json<Value>,
 ) -> Result<axum::response::Response, ProxyError> {
    // Gemini 的模型名称在 URI 中
-    let ctx = RequestContext::new(&state, &body, AppType::Gemini, "Gemini", "gemini")
+    let mut ctx = RequestContext::new(&state, &body, AppType::Gemini, "Gemini", "gemini")
        .await?
        .with_model_from_uri(&uri);

@@ -397,7 +415,7 @@ pub async fn handle_gemini(
        .unwrap_or(false);

    let forwarder = ctx.create_forwarder(&state);
-    let response = match forwarder
+    let result = match forwarder
        .forward_with_retry(
            &AppType::Gemini,
            endpoint,
@@ -407,13 +425,19 @@ pub async fn handle_gemini(
        )
        .await
    {
-        Ok(resp) => resp,
-        Err(e) => {
-            log_forward_error(&state, &ctx, is_stream, &e);
-            return Err(e);
+        Ok(result) => result,
+        Err(mut err) => {
+            if let Some(provider) = err.provider.take() {
+                ctx.provider = provider;
+            }
+            log_forward_error(&state, &ctx, is_stream, &err.error);
+            return Err(err.error);
        }
    };

+    ctx.provider = result.provider;
+    let response = result.response;
+
    log::info!("[Gemini] 上游响应状态: {}", response.status());

    process_response(response, &ctx, &state, &GEMINI_PARSER_CONFIG).await
@@ -30,85 +30,92 @@ impl ProviderRouter {
    /// 选择可用的供应商（支持故障转移）
    ///
    /// 返回按优先级排序的可用供应商列表：
-    /// 1. 当前供应商（is_current=true）始终第一位
-    /// 2. 故障转移队列中的其他供应商（按 queue_order 排序）- 仅当自动故障转移开关开启时
-    /// 3. 只返回熔断器未打开的供应商
+    /// - 故障转移关闭时：仅返回当前供应商
+    /// - 故障转移开启时：完全按照故障转移队列顺序返回，忽略当前供应商设置
    pub async fn select_providers(&self, app_type: &str) -> Result<Vec<Provider>, AppError> {
        let mut result = Vec::new();
-        let all_providers = self.db.get_all_providers(app_type)?;

-        // 检查自动故障转移总开关是否开启
-        let auto_failover_enabled = self
-            .db
-            .get_setting("auto_failover_enabled")
-            .map(|v| v.map(|s| s == "true").unwrap_or(false)) // 默认关闭
-            .unwrap_or(false);
+        // 检查该应用的自动故障转移开关是否开启
+        let failover_key = format!("auto_failover_enabled_{app_type}");
+        let auto_failover_enabled = match self.db.get_setting(&failover_key) {
+            Ok(Some(value)) => {
+                let enabled = value == "true";
+                log::info!(
+                    "[{app_type}] Failover setting '{failover_key}' = '{value}', enabled: {enabled}"
+                );
+                enabled
+            }
+            Ok(None) => {
+                log::warn!(
+                    "[{app_type}] Failover setting '{failover_key}' not found in database, defaulting to disabled"
+                );
+                false
+            }
+            Err(e) => {
+                log::error!(
+                    "[{app_type}] Failed to read failover setting '{failover_key}': {e}, defaulting to disabled"
+                );
+                false
+            }
+        };

-        // 1. 当前供应商始终第一位
-        if let Some(current_id) = self.db.get_current_provider(app_type)? {
-            if let Some(current) = all_providers.get(&current_id) {
-                let circuit_key = format!("{}:{}", app_type, current.id);
+        if auto_failover_enabled {
+            // 故障转移开启：使用 in_failover_queue 标记的供应商，按 sort_index 排序
+            let failover_providers = self.db.get_failover_providers(app_type)?;
+            log::info!(
+                "[{}] Failover enabled, using queue order ({} items)",
+                app_type,
+                failover_providers.len()
+            );
+
+            for provider in failover_providers {
+                // 检查熔断器状态
+                let circuit_key = format!("{}:{}", app_type, provider.id);
                let breaker = self.get_or_create_circuit_breaker(&circuit_key).await;

                if breaker.is_available().await {
                    log::info!(
-                        "[{}] Current provider available: {} ({})",
+                        "[{}] Queue provider available: {} ({}) at sort_index {:?}",
                        app_type,
-                        current.name,
-                        current.id
+                        provider.name,
+                        provider.id,
+                        provider.sort_index
                    );
-                    result.push(current.clone());
+                    result.push(provider);
                } else {
-                    log::warn!(
-                        "[{}] Current provider {} circuit breaker open, checking failover queue",
+                    log::debug!(
+                        "[{}] Queue provider {} circuit breaker open, skipping",
                        app_type,
-                        current.name
+                        provider.name
                    );
                }
            }
-        }
+        } else {
+            // 故障转移关闭：仅使用当前供应商
+            log::info!("[{app_type}] Failover disabled, using current provider only");

-        // 2. 获取故障转移队列中的供应商（仅当自动故障转移开关开启时）
-        if auto_failover_enabled {
-            let queue = self.db.get_failover_queue(app_type)?;
-
-            for item in queue {
-                // 跳过已添加的当前供应商
-                if result.iter().any(|p| p.id == item.provider_id) {
-                    continue;
-                }
-
-                // 跳过禁用的队列项
-                if !item.enabled {
-                    continue;
-                }
-
-                // 获取供应商信息
-                if let Some(provider) = all_providers.get(&item.provider_id) {
-                    // 检查熔断器状态
-                    let circuit_key = format!("{}:{}", app_type, provider.id);
+            if let Some(current_id) = self.db.get_current_provider(app_type)? {
+                if let Some(current) = self.db.get_provider_by_id(&current_id, app_type)? {
+                    let circuit_key = format!("{}:{}", app_type, current.id);
                    let breaker = self.get_or_create_circuit_breaker(&circuit_key).await;

                    if breaker.is_available().await {
                        log::info!(
-                            "[{}] Failover provider available: {} ({}) at queue position {}",
+                            "[{}] Current provider available: {} ({})",
                            app_type,
-                            provider.name,
-                            provider.id,
-                            item.queue_order
+                            current.name,
+                            current.id
                        );
-                        result.push(provider.clone());
+                        result.push(current);
                    } else {
-                        log::debug!(
-                            "[{}] Failover provider {} circuit breaker open, skipping",
+                        log::warn!(
+                            "[{}] Current provider {} circuit breaker open",
                            app_type,
-                            provider.name
+                            current.name
                        );
                    }
                }
            }
-        } else {
-            log::info!("[{app_type}] Auto-failover disabled, using only current provider");
        }

        if result.is_empty() {
@@ -118,7 +125,7 @@ impl ProviderRouter {
        }

        log::info!(
-            "[{}] Failover chain: {} provider(s) available",
+            "[{}] Provider chain: {} provider(s) available",
            app_type,
            result.len()
        );
@@ -275,25 +282,14 @@ mod tests {
        let db = Arc::new(Database::memory().unwrap());
        let router = ProviderRouter::new(db);

-        // 测试创建熔断器
        let breaker = router.get_or_create_circuit_breaker("claude:test").await;
        assert!(breaker.allow_request().await.allowed);
    }

    #[tokio::test]
-    async fn select_providers_does_not_consume_half_open_permit() {
+    async fn test_failover_disabled_uses_current_provider() {
        let db = Arc::new(Database::memory().unwrap());

-        // 配置：让熔断器 Open 后立刻进入 HalfOpen（timeout_seconds=0），并用 1 次失败就打开熔断器
-        db.update_circuit_breaker_config(&CircuitBreakerConfig {
-            failure_threshold: 1,
-            timeout_seconds: 0,
-            ..Default::default()
-        })
-        .await
-        .unwrap();
-
-        // 准备 2 个 Provider：A（当前）+ B（队列）
        let provider_a =
            Provider::with_id("a".to_string(), "Provider A".to_string(), json!({}), None);
        let provider_b =
@@ -304,19 +300,78 @@ mod tests {
        db.set_current_provider("claude", "a").unwrap();
        db.add_to_failover_queue("claude", "b").unwrap();

+        let router = ProviderRouter::new(db.clone());
+        let providers = router.select_providers("claude").await.unwrap();
+
+        assert_eq!(providers.len(), 1);
+        assert_eq!(providers[0].id, "a");
+    }
+
+    #[tokio::test]
+    async fn test_failover_enabled_uses_queue_order() {
+        let db = Arc::new(Database::memory().unwrap());
+
+        // 设置 sort_index 来控制顺序：b=1, a=2
+        let mut provider_a =
+            Provider::with_id("a".to_string(), "Provider A".to_string(), json!({}), None);
+        provider_a.sort_index = Some(2);
+        let mut provider_b =
+            Provider::with_id("b".to_string(), "Provider B".to_string(), json!({}), None);
+        provider_b.sort_index = Some(1);
+
+        db.save_provider("claude", &provider_a).unwrap();
+        db.save_provider("claude", &provider_b).unwrap();
+        db.set_current_provider("claude", "a").unwrap();
+
+        db.add_to_failover_queue("claude", "b").unwrap();
+        db.add_to_failover_queue("claude", "a").unwrap();
+        db.set_setting("auto_failover_enabled_claude", "true")
+            .unwrap();
+
+        let router = ProviderRouter::new(db.clone());
+        let providers = router.select_providers("claude").await.unwrap();
+
+        assert_eq!(providers.len(), 2);
+        // 按 sort_index 排序：b(1) 在前，a(2) 在后
+        assert_eq!(providers[0].id, "b");
+        assert_eq!(providers[1].id, "a");
+    }
+
+    #[tokio::test]
+    async fn test_select_providers_does_not_consume_half_open_permit() {
+        let db = Arc::new(Database::memory().unwrap());
+
+        db.update_circuit_breaker_config(&CircuitBreakerConfig {
+            failure_threshold: 1,
+            timeout_seconds: 0,
+            ..Default::default()
+        })
+        .await
+        .unwrap();
+
+        let provider_a =
+            Provider::with_id("a".to_string(), "Provider A".to_string(), json!({}), None);
+        let provider_b =
+            Provider::with_id("b".to_string(), "Provider B".to_string(), json!({}), None);
+
+        db.save_provider("claude", &provider_a).unwrap();
+        db.save_provider("claude", &provider_b).unwrap();
+
+        db.add_to_failover_queue("claude", "a").unwrap();
+        db.add_to_failover_queue("claude", "b").unwrap();
+        db.set_setting("auto_failover_enabled_claude", "true")
+            .unwrap();
+
        let router = ProviderRouter::new(db.clone());

-        // 让 B 进入 Open 状态（failure_threshold=1）
        router
            .record_result("b", "claude", false, false, Some("fail".to_string()))
            .await
            .unwrap();

-        // select_providers 只做“可用性判断”，不应占用 HalfOpen 探测名额
        let providers = router.select_providers("claude").await.unwrap();
        assert_eq!(providers.len(), 2);

-        // 如果 select_providers 错误地消耗了 HalfOpen 名额，这里会返回 false（被限流拒绝）
        assert!(router.allow_provider_request("b", "claude").await.allowed);
    }
 }
@@ -270,6 +270,7 @@ mod tests {
            meta: None,
            icon: None,
            icon_color: None,
+            in_failover_queue: false,
        }
    }

@@ -174,6 +174,7 @@ mod tests {
            meta: None,
            icon: None,
            icon_color: None,
+            in_failover_queue: false,
        }
    }

@@ -250,6 +250,7 @@ mod tests {
            meta: None,
            icon: None,
            icon_color: None,
+            in_failover_queue: false,
        }
    }

@@ -209,6 +209,7 @@ mod tests {
            meta: None,
            icon: None,
            icon_color: None,
+            in_failover_queue: false,
        }
    }

@@ -394,6 +394,7 @@ mod tests {
            meta: None,
            icon: None,
            icon_color: None,
+            in_failover_queue: false,
        }
    }