//! 供应商连通性检查服务(reachability) //! //! 仅探测供应商 `base_url` 是否可达,**不发送真实大模型请求**: //! - 收到任意 HTTP 响应(200/4xx/5xx)即判定"可达"(端口通、网关存活); //! - 仅 DNS / 连接被拒 / TLS / 超时等网络级错误判定"不可达"; //! - 延迟 = 收到响应头的耗时(TTFB,真实往返)。 //! //! ## 设计取舍:可达 ≠ 配置正确 //! //! 本检查刻意不验证鉴权或模型,因此不会被第三方供应商的鉴权拦截 / 模型校验 //! 误判为"不可用"。代价是它无法告诉你鉴权对不对、模型存不存在。 //! //! ## 与故障转移的关系(重要不变量) //! //! 连通性检查 **绝不** 触碰故障转移熔断器:一个返回 403/401 的供应商在本检查里 //! 算"可达",但它对真实流量是坏的。熔断器只由 `proxy/forwarder.rs` 转发真实流量 //! 的成败驱动(被动)。两者职责分离——可达性回答"能不能到",真实流量回答"能不能用"。 use reqwest::header::HeaderValue; use reqwest::Client; use serde::{Deserialize, Serialize}; use std::time::Instant; use crate::app_config::AppType; use crate::error::AppError; use crate::provider::Provider; use crate::proxy::providers::{get_adapter, ClaudeAdapter, ProviderAdapter}; /// 健康状态枚举 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] #[serde(rename_all = "lowercase")] pub enum HealthStatus { Operational, Degraded, Failed, } /// 连通性检查配置 #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct StreamCheckConfig { /// 单次探测超时(秒) pub timeout_secs: u64, /// 超时类失败的最大重试次数 pub max_retries: u32, /// 降级阈值(毫秒):可达但 TTFB 超过该值判定为"较慢" pub degraded_threshold_ms: u64, } impl Default for StreamCheckConfig { fn default() -> Self { // 可达性探测打的是 base_url 的小请求(仅读响应头),不等待模型生成,故超时远小于 // 旧的真实请求检查(45s → 8s);降级阈值沿用旧尺度 6000ms——探测 TTFB 一般远低于 // 此,仅在确实很慢时才标"较慢",避免把 1 秒多的正常延迟误判为降级。 Self { timeout_secs: 8, max_retries: 1, degraded_threshold_ms: 6000, } } } /// 连通性检查结果 #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct StreamCheckResult { pub status: HealthStatus, pub success: bool, pub message: String, pub response_time_ms: Option, pub http_status: Option, /// 保留字段以兼容 `stream_check_logs` 表结构;连通性检查恒为空串。 pub model_used: String, pub tested_at: i64, pub retry_count: u32, /// 细粒度错误分类;连通性检查不再细分,恒为 None。 #[serde(skip_serializing_if = "Option::is_none")] pub error_category: Option, } /// 连通性检查服务 pub struct StreamCheckService; impl StreamCheckService { /// 执行连通性检查(仅对超时类失败重试)。 /// /// `base_url_override`:用于 Copilot 等需要从 OAuth 管理器动态解析端点的供应商, /// 由命令层预先解析后传入;其余供应商传 `None`,由本服务从 `settings_config` 提取。 pub async fn check_with_retry( app_type: &AppType, provider: &Provider, config: &StreamCheckConfig, base_url_override: Option, ) -> Result { let effective = Self::merge_provider_config(provider, config); let mut last_result: Option = None; for attempt in 0..=effective.max_retries { let start = Instant::now(); let result = Self::check_once( app_type, provider, &effective, base_url_override.clone(), start, ) .await?; if result.success { return Ok(StreamCheckResult { retry_count: attempt, ..result }); } // 仅超时 / abort 类网络抖动值得重试;连接被拒、DNS 失败等立即返回。 if Self::should_retry(&result.message) && attempt < effective.max_retries { last_result = Some(result); continue; } return Ok(StreamCheckResult { retry_count: attempt, ..result }); } Ok(last_result.unwrap_or_else(|| StreamCheckResult { status: HealthStatus::Failed, success: false, message: "Check failed".to_string(), response_time_ms: None, http_status: None, model_used: String::new(), tested_at: chrono::Utc::now().timestamp(), retry_count: effective.max_retries, error_category: None, })) } /// 合并供应商单独配置(`meta.testConfig`,仅当 `enabled`)与全局配置。 fn merge_provider_config(provider: &Provider, global: &StreamCheckConfig) -> StreamCheckConfig { let tc = provider .meta .as_ref() .and_then(|m| m.test_config.as_ref()) .filter(|tc| tc.enabled); match tc { Some(tc) => StreamCheckConfig { timeout_secs: tc.timeout_secs.unwrap_or(global.timeout_secs), max_retries: tc.max_retries.unwrap_or(global.max_retries), degraded_threshold_ms: tc .degraded_threshold_ms .unwrap_or(global.degraded_threshold_ms), }, None => global.clone(), } } /// 单次连通性探测。 async fn check_once( app_type: &AppType, provider: &Provider, config: &StreamCheckConfig, base_url_override: Option, start: Instant, ) -> Result { let base_url = match base_url_override { Some(b) => b, None => Self::resolve_base_url(app_type, provider)?, }; let client = crate::proxy::http_client::get(); let timeout = std::time::Duration::from_secs(config.timeout_secs); let ua = Self::custom_user_agent(provider); let result = Self::probe_reachability(&client, &base_url, timeout, ua).await; let response_time = start.elapsed().as_millis() as u64; Ok(Self::build_result( result, response_time, config.degraded_threshold_ms, )) } /// 解析供应商 `base_url`。 /// /// 连通性探测只需打到 base(origin 或用户配置的 base 路径)即可——任何 HTTP /// 响应都证明端口可达,因此无需像旧的真实请求检查那样解析具体 API 路径 /// (`/v1/messages` vs `/chat/completions` vs `:streamGenerateContent`)。 /// /// 官方供应商(`category == "official"`)base_url 故意留空(走客户端默认/OAuth 端点), /// 没有 cc-switch 能可靠探测的目标——这类供应商的连通检测按钮在前端已隐藏 /// (见 `ProviderCard.tsx`),故此处对其提取失败直接报错即可,不做官方端点回退。 fn resolve_base_url(app_type: &AppType, provider: &Provider) -> Result { match app_type { // 累加模式应用的 settings_config 结构与 Claude/Codex/Gemini 不同, // 不走 adapter,直接按各自约定提取 base_url。 AppType::OpenCode => { let npm = Self::extract_opencode_npm(provider); Self::resolve_opencode_base_url(provider, npm.as_deref()) } AppType::OpenClaw => Self::extract_openclaw_base_url(provider), AppType::Hermes => Self::extract_hermes_base_url(provider), AppType::ClaudeDesktop => ClaudeAdapter::new() .extract_base_url(provider) .map_err(|e| AppError::Message(format!("Failed to extract base_url: {e}"))), _ => get_adapter(app_type) .extract_base_url(provider) .map_err(|e| AppError::Message(format!("Failed to extract base_url: {e}"))), } } /// 轻量可达性探测:GET `base_url`,收到任意 HTTP 响应即可达。 /// /// - `send()` 在收到响应头时即返回,故计时天然是 TTFB;不读 body。 /// - reqwest 对任何 HTTP 状态码都返回 `Ok`,只有网络级错误进 `Err`—— /// 这正是"任何响应都算可达、只有连不上才算失败"的语义。 async fn probe_reachability( client: &Client, base_url: &str, timeout: std::time::Duration, custom_ua: Option, ) -> Result { let url = base_url.trim(); if url.is_empty() { return Err(AppError::Message("base_url 为空".to_string())); } let mut req = client .get(url) .timeout(timeout) .header("accept", "*/*") .header("accept-encoding", "identity"); // 复用供应商自定义 UA(部分网关按 UA 白名单放行),与转发路径口径一致。 if let Some(ua) = custom_ua { req = req.header("user-agent", ua); } match req.send().await { Ok(resp) => Ok(resp.status().as_u16()), Err(e) => Err(Self::map_request_error(e)), } } /// 将探测原始结果包装成 `StreamCheckResult`。 fn build_result( result: Result, response_time: u64, degraded_threshold_ms: u64, ) -> StreamCheckResult { let tested_at = chrono::Utc::now().timestamp(); match result { Ok(status) => StreamCheckResult { status: Self::determine_status(response_time, degraded_threshold_ms), success: true, message: "Reachable".to_string(), response_time_ms: Some(response_time), http_status: Some(status), model_used: String::new(), tested_at, retry_count: 0, error_category: None, }, Err(e) => StreamCheckResult { status: HealthStatus::Failed, success: false, message: e.to_string(), response_time_ms: Some(response_time), http_status: None, model_used: String::new(), tested_at, retry_count: 0, error_category: None, }, } } fn determine_status(latency_ms: u64, threshold: u64) -> HealthStatus { if latency_ms <= threshold { HealthStatus::Operational } else { HealthStatus::Degraded } } fn should_retry(msg: &str) -> bool { let lower = msg.to_lowercase(); lower.contains("timeout") || lower.contains("abort") || lower.contains("timed out") } fn map_request_error(e: reqwest::Error) -> AppError { if e.is_timeout() { AppError::Message("Request timeout".to_string()) } else if e.is_connect() { AppError::Message(format!("Connection failed: {e}")) } else { AppError::Message(e.to_string()) } } /// Provider 级自定义 User-Agent(`meta.customUserAgent`),与转发路径共用单一口径: /// trim、空串视为未设置、非法值静默忽略(返回 `None`)。 fn custom_user_agent(provider: &Provider) -> Option { provider .meta .as_ref() .and_then(|meta| meta.custom_user_agent_header().ok().flatten()) } // ===== 各应用 base_url 提取(settings_config 结构互不相同)===== /// OpenClaw: `{ baseUrl, apiKey, api, ... }`(camelCase) fn extract_openclaw_base_url(provider: &Provider) -> Result { provider .settings_config .get("baseUrl") .and_then(|v| v.as_str()) .map(|s| s.trim().to_string()) .filter(|s| !s.is_empty()) .ok_or_else(|| { AppError::localized( "openclaw_base_url_missing", "OpenClaw 供应商缺少 baseUrl", "OpenClaw provider is missing `baseUrl`", ) }) } /// Hermes: `{ base_url, api_key, api_mode }`(snake_case) fn extract_hermes_base_url(provider: &Provider) -> Result { provider .settings_config .get("base_url") .and_then(|v| v.as_str()) .map(|s| s.trim().to_string()) .filter(|s| !s.is_empty()) .ok_or_else(|| { AppError::localized( "hermes_base_url_missing", "Hermes 供应商缺少 base_url", "Hermes provider is missing `base_url`", ) }) } /// OpenCode: `{ npm, options: { baseURL, apiKey }, ... }` /// /// 用户未显式填 `options.baseURL` 时,按 `npm`(AI SDK 包)回退到包自带默认端点。 /// `@ai-sdk/openai-compatible` 无默认端点,必须显式填。 fn resolve_opencode_base_url( provider: &Provider, npm: Option<&str>, ) -> Result { if let Some(explicit) = Self::extract_opencode_base_url(provider) { return Ok(explicit); } let fallback = match npm { Some("@ai-sdk/openai") => Some("https://api.openai.com/v1"), Some("@ai-sdk/anthropic") => Some("https://api.anthropic.com"), Some("@ai-sdk/google") => Some("https://generativelanguage.googleapis.com"), _ => None, }; fallback.map(|s| s.to_string()).ok_or_else(|| { AppError::localized( "opencode_base_url_missing", "OpenCode 供应商缺少 options.baseURL,且当前 SDK 包没有默认端点", "OpenCode provider is missing `options.baseURL` and the SDK package has no default endpoint", ) }) } fn extract_opencode_base_url(provider: &Provider) -> Option { provider .settings_config .get("options") .and_then(|v| v.get("baseURL")) .and_then(|v| v.as_str()) .map(|s| s.trim().to_string()) .filter(|s| !s.is_empty()) } fn extract_opencode_npm(provider: &Provider) -> Option { provider .settings_config .get("npm") .and_then(|v| v.as_str()) .map(|s| s.trim().to_string()) .filter(|s| !s.is_empty()) } } #[cfg(test)] mod tests { use super::*; fn make_provider(settings_config: serde_json::Value) -> Provider { Provider::with_id( "test".to_string(), "Test".to_string(), settings_config, None, ) } #[test] fn test_default_config_uses_reachability_friendly_values() { let config = StreamCheckConfig::default(); assert_eq!(config.timeout_secs, 8); assert_eq!(config.max_retries, 1); // 降级阈值沿用旧尺度,避免把 1 秒多的正常延迟误判为"较慢" assert_eq!(config.degraded_threshold_ms, 6000); } #[test] fn test_determine_status() { assert_eq!( StreamCheckService::determine_status(1000, 1500), HealthStatus::Operational ); assert_eq!( StreamCheckService::determine_status(1500, 1500), HealthStatus::Operational ); assert_eq!( StreamCheckService::determine_status(1501, 1500), HealthStatus::Degraded ); } #[test] fn test_should_retry_only_on_timeout_like_errors() { assert!(StreamCheckService::should_retry("Request timeout")); assert!(StreamCheckService::should_retry("request timed out")); assert!(StreamCheckService::should_retry("connection abort")); // 连接被拒 / DNS 失败不重试 assert!(!StreamCheckService::should_retry( "Connection failed: dns error" )); assert!(!StreamCheckService::should_retry("Reachable")); } #[test] fn test_build_result_any_http_status_is_reachable() { // 任何 HTTP 状态码都算可达(success=true) for status in [200u16, 401, 403, 404, 429, 500, 503] { let r = StreamCheckService::build_result(Ok(status), 100, 1500); assert!(r.success, "status {status} should be reachable"); assert_eq!(r.status, HealthStatus::Operational); assert_eq!(r.http_status, Some(status)); assert!(r.model_used.is_empty()); assert!(r.error_category.is_none()); } } #[test] fn test_build_result_network_error_is_unreachable() { let r = StreamCheckService::build_result( Err(AppError::Message("Connection failed: refused".to_string())), 5, 1500, ); assert!(!r.success); assert_eq!(r.status, HealthStatus::Failed); assert!(r.http_status.is_none()); } #[test] fn test_build_result_slow_response_is_degraded() { let r = StreamCheckService::build_result(Ok(200), 3000, 1500); assert!(r.success); assert_eq!(r.status, HealthStatus::Degraded); } #[test] fn test_merge_provider_config_override_and_default() { use crate::provider::{ProviderMeta, ProviderTestConfig}; let global = StreamCheckConfig::default(); // 无 testConfig → 用全局 let p = make_provider(serde_json::json!({})); let merged = StreamCheckService::merge_provider_config(&p, &global); assert_eq!(merged.timeout_secs, global.timeout_secs); // testConfig 启用并覆盖部分字段 let mut p2 = make_provider(serde_json::json!({})); p2.meta = Some(ProviderMeta { test_config: Some(ProviderTestConfig { enabled: true, timeout_secs: Some(20), degraded_threshold_ms: Some(3000), max_retries: None, }), ..Default::default() }); let merged2 = StreamCheckService::merge_provider_config(&p2, &global); assert_eq!(merged2.timeout_secs, 20); assert_eq!(merged2.degraded_threshold_ms, 3000); assert_eq!(merged2.max_retries, global.max_retries); // 未覆盖 → 全局 // testConfig 存在但未启用 → 忽略,用全局 let mut p3 = make_provider(serde_json::json!({})); p3.meta = Some(ProviderMeta { test_config: Some(ProviderTestConfig { enabled: false, timeout_secs: Some(99), degraded_threshold_ms: None, max_retries: None, }), ..Default::default() }); let merged3 = StreamCheckService::merge_provider_config(&p3, &global); assert_eq!(merged3.timeout_secs, global.timeout_secs); } #[test] fn test_resolve_opencode_base_url_explicit_wins() { let p = make_provider(serde_json::json!({ "npm": "@ai-sdk/openai", "options": { "baseURL": "https://proxy.local/v1", "apiKey": "k" }, "models": {}, })); let resolved = StreamCheckService::resolve_opencode_base_url(&p, Some("@ai-sdk/openai")).unwrap(); assert_eq!(resolved, "https://proxy.local/v1"); } #[test] fn test_resolve_opencode_base_url_falls_back_for_known_npm() { let p = make_provider(serde_json::json!({ "npm": "@ai-sdk/anthropic", "options": { "apiKey": "k" }, "models": {}, })); let resolved = StreamCheckService::resolve_opencode_base_url(&p, Some("@ai-sdk/anthropic")).unwrap(); assert_eq!(resolved, "https://api.anthropic.com"); } #[test] fn test_resolve_opencode_base_url_errors_for_openai_compatible_without_url() { let p = make_provider(serde_json::json!({ "npm": "@ai-sdk/openai-compatible", "options": { "apiKey": "k" }, "models": {}, })); let result = StreamCheckService::resolve_opencode_base_url(&p, Some("@ai-sdk/openai-compatible")); assert!(result.is_err()); } #[test] fn test_extract_openclaw_base_url_missing_errors() { let p = make_provider(serde_json::json!({ "apiKey": "k", "api": "openai-completions" })); assert!(StreamCheckService::extract_openclaw_base_url(&p).is_err()); let p2 = make_provider(serde_json::json!({ "baseUrl": "https://api.deepseek.com/v1" })); assert_eq!( StreamCheckService::extract_openclaw_base_url(&p2).unwrap(), "https://api.deepseek.com/v1" ); } #[test] fn test_resolve_base_url_uses_explicit_url_or_errors_when_missing() { // 有显式 base_url → 直接用 let p = make_provider( serde_json::json!({ "env": { "ANTHROPIC_BASE_URL": "https://relay.example/v1" } }), ); assert_eq!( StreamCheckService::resolve_base_url(&AppType::Claude, &p).unwrap(), "https://relay.example/v1" ); // 缺 base_url(官方留空 / 用户忘填)→ 报错。官方供应商的检测按钮在前端已隐藏, // 不会走到这里;不做官方端点回退(避免给忘填地址的第三方误显绿灯)。 let empty = make_provider(serde_json::json!({ "env": {} })); assert!(StreamCheckService::resolve_base_url(&AppType::Claude, &empty).is_err()); } }