From 4cd98889a7e2b3ff741de6443d3067c9bae91a0c Mon Sep 17 00:00:00 2001 From: zyycn Date: Fri, 31 Jul 2026 23:23:23 +0800 Subject: [PATCH] refactor(core): centralize rotation strategy and circuit policy Own RotationStrategy wire mapping in gateway-core and re-export it from gateway-admin, move ProviderCircuitPolicy into the execution port contract, and rename the Core coordination timeout without a Redis implementation leak. Co-authored-by: Codex --- .../gateway-admin/src/model/settings.rs | 10 ++----- .../gateway-api/src/admin/observability.rs | 15 +--------- .../crates/gateway-api/src/admin/settings.rs | 28 ++----------------- .../gateway-core/src/engine/credential.rs | 11 ++++++++ .../gateway-core/src/engine/execution.rs | 25 ++++++++++++++--- .../gateway-core/src/routing/snapshot.rs | 9 ++---- .../gateway-core/tests/engine/credential.rs | 13 +++++++++ backend/crates/gateway-store/src/lib.rs | 28 +++++-------------- .../src/postgres/runtime_settings.rs | 6 ++-- .../src/redis/provider_circuit.rs | 21 ++------------ .../tests/redis/provider_circuit.rs | 8 ++++-- 11 files changed, 69 insertions(+), 105 deletions(-) diff --git a/backend/crates/gateway-admin/src/model/settings.rs b/backend/crates/gateway-admin/src/model/settings.rs index da5687f6c..3ac8b5741 100644 --- a/backend/crates/gateway-admin/src/model/settings.rs +++ b/backend/crates/gateway-admin/src/model/settings.rs @@ -11,14 +11,8 @@ use super::Revision; /// 客户端模型到上游模型的全局精确映射。 pub type ModelMappings = BTreeMap; -/// 账号调度策略。 -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum RotationStrategy { - Smart, - QuotaResetPriority, - RoundRobin, - Sticky, -} +/// 账号调度策略；由 Core 拥有稳定值与 wire 映射。 +pub use gateway_core::engine::credential::RotationStrategy; /// 完整运行设置事实。 #[derive(Debug, Clone, PartialEq, Eq)] diff --git a/backend/crates/gateway-api/src/admin/observability.rs b/backend/crates/gateway-api/src/admin/observability.rs index 786cd1b05..4424ea202 100644 --- a/backend/crates/gateway-api/src/admin/observability.rs +++ b/backend/crates/gateway-api/src/admin/observability.rs @@ -2180,19 +2180,6 @@ fn relative_time(value: Option>, now: DateTime) -> String { format!("{} 天前", elapsed.num_days()) } -fn rotation_strategy_name( - strategy: gateway_admin::model::settings::RotationStrategy, -) -> &'static str { - match strategy { - gateway_admin::model::settings::RotationStrategy::Smart => "smart", - gateway_admin::model::settings::RotationStrategy::QuotaResetPriority => { - "quota_reset_priority" - } - gateway_admin::model::settings::RotationStrategy::RoundRobin => "round_robin", - gateway_admin::model::settings::RotationStrategy::Sticky => "sticky", - } -} - fn dashboard_view(result: domain::DashboardResult, kind: TrendKind) -> DashboardDataView { let domain::DashboardResult { observation, @@ -2296,7 +2283,7 @@ fn dashboard_view(result: domain::DashboardResult, kind: TrendKind) -> Dashboard used_slots: capacity.used_slots, available_slots: capacity.available_slots, }, - rotation_strategy: rotation_strategy_name(rotation_strategy).to_owned(), + rotation_strategy: rotation_strategy.as_str().to_owned(), } } diff --git a/backend/crates/gateway-api/src/admin/settings.rs b/backend/crates/gateway-api/src/admin/settings.rs index 9859fa9d0..de3de60e3 100644 --- a/backend/crates/gateway-api/src/admin/settings.rs +++ b/backend/crates/gateway-api/src/admin/settings.rs @@ -72,10 +72,7 @@ impl UpdateRuntimeSettingsRequest { if i64::try_from(self.request_interval_ms).is_err() { return Err(WireValidationError::new("requestIntervalMs")); } - if !matches!( - self.rotation_strategy.as_str(), - "smart" | "quota_reset_priority" | "round_robin" | "sticky" - ) { + if RotationStrategy::parse(&self.rotation_strategy).is_none() { return Err(WireValidationError::new("rotationStrategy")); } Ok(()) @@ -91,7 +88,7 @@ impl UpdateRuntimeSettingsRequest { max_concurrent_per_account: u32::try_from(self.max_concurrent_per_account) .map_err(|_| WireValidationError::new("settingsMaxConcurrencyOverflow"))?, request_interval_ms: self.request_interval_ms, - rotation_strategy: parse_rotation_strategy(&self.rotation_strategy) + rotation_strategy: RotationStrategy::parse(&self.rotation_strategy) .ok_or_else(|| WireValidationError::new("rotationStrategy"))?, usage_retention_days: u32::try_from(self.usage_retention_days) .map_err(|_| WireValidationError::new("settingsUsageRetentionOverflow"))?, @@ -111,7 +108,7 @@ impl From for RuntimeSettingsView { refresh_concurrency: u64::from(settings.refresh_concurrency), max_concurrent_per_account: u64::from(settings.max_concurrent_per_account), request_interval_ms: settings.request_interval_ms, - rotation_strategy: rotation_strategy_name(settings.rotation_strategy).to_owned(), + rotation_strategy: settings.rotation_strategy.as_str().to_owned(), usage_retention_days: u64::from(settings.usage_retention_days), ops_event_retention_days: u64::from(settings.ops_event_retention_days), audit_retention_days: u64::from(settings.audit_retention_days), @@ -319,25 +316,6 @@ fn wire_model_mappings(mappings: DomainModelMappings) -> ModelMappings { .collect() } -fn parse_rotation_strategy(value: &str) -> Option { - match value { - "smart" => Some(RotationStrategy::Smart), - "quota_reset_priority" => Some(RotationStrategy::QuotaResetPriority), - "round_robin" => Some(RotationStrategy::RoundRobin), - "sticky" => Some(RotationStrategy::Sticky), - _ => None, - } -} - -const fn rotation_strategy_name(strategy: RotationStrategy) -> &'static str { - match strategy { - RotationStrategy::Smart => "smart", - RotationStrategy::QuotaResetPriority => "quota_reset_priority", - RotationStrategy::RoundRobin => "round_robin", - RotationStrategy::Sticky => "sticky", - } -} - fn valid_model_name(value: &str, max_len: usize) -> bool { !value.is_empty() && value.len() <= max_len diff --git a/backend/crates/gateway-core/src/engine/credential.rs b/backend/crates/gateway-core/src/engine/credential.rs index a89dc9900..41448fdb0 100644 --- a/backend/crates/gateway-core/src/engine/credential.rs +++ b/backend/crates/gateway-core/src/engine/credential.rs @@ -698,6 +698,17 @@ impl RotationStrategy { Self::Sticky => "sticky", } } + + #[must_use] + pub fn parse(value: &str) -> Option { + match value { + "smart" => Some(Self::Smart), + "quota_reset_priority" => Some(Self::QuotaResetPriority), + "round_robin" => Some(Self::RoundRobin), + "sticky" => Some(Self::Sticky), + _ => None, + } + } } /// 从 `runtime_settings` 冻结到一次请求计划的账号调度策略。 diff --git a/backend/crates/gateway-core/src/engine/execution.rs b/backend/crates/gateway-core/src/engine/execution.rs index 4b049e550..a3773c4a4 100644 --- a/backend/crates/gateway-core/src/engine/execution.rs +++ b/backend/crates/gateway-core/src/engine/execution.rs @@ -3,6 +3,7 @@ use std::collections::BTreeSet; use std::fmt; use std::net::IpAddr; +use std::num::NonZeroU32; use std::sync::Arc; use std::time::{Duration, SystemTime}; @@ -39,7 +40,7 @@ use crate::routing::{ }; const MODEL_REQUEST_DEADLINE: Duration = Duration::from_secs(10 * 60); -const REDIS_COORDINATION_TIMEOUT: Duration = Duration::from_millis(100); +const COORDINATION_TIMEOUT: Duration = Duration::from_millis(100); #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ClientTransport { @@ -173,6 +174,22 @@ pub enum ProviderCircuitDecision { #[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, @@ -254,7 +271,7 @@ impl DefaultExecutionService { let continuation = match request.metadata.previous_response_id.as_ref() { Some(previous) => { let resolve = self.continuation.resolve(previous).fuse(); - let timeout = Delay::new(REDIS_COORDINATION_TIMEOUT).fuse(); + let timeout = Delay::new(COORDINATION_TIMEOUT).fuse(); pin_mut!(resolve, timeout); let pin = select_biased! { result = resolve => result.ok().flatten(), @@ -393,7 +410,7 @@ impl DefaultExecutionService { provider_kind: &ProviderKind, ) -> Result { let decision = self.circuits.decision(provider_kind).fuse(); - let timeout = Delay::new(REDIS_COORDINATION_TIMEOUT).fuse(); + let timeout = Delay::new(COORDINATION_TIMEOUT).fuse(); pin_mut!(decision, timeout); let decision = select_biased! { result = decision => Some(result), @@ -934,7 +951,7 @@ async fn record_native_continuation( pin: NativeContinuationPin, ) { let record = continuation.record(pin).fuse(); - let timeout = Delay::new(REDIS_COORDINATION_TIMEOUT).fuse(); + let timeout = Delay::new(COORDINATION_TIMEOUT).fuse(); pin_mut!(record, timeout); select_biased! { result = record => { diff --git a/backend/crates/gateway-core/src/routing/snapshot.rs b/backend/crates/gateway-core/src/routing/snapshot.rs index 1e2ad8bc1..e74c1626c 100644 --- a/backend/crates/gateway-core/src/routing/snapshot.rs +++ b/backend/crates/gateway-core/src/routing/snapshot.rs @@ -244,13 +244,8 @@ async fn compile_runtime_snapshot( } let model_mappings = facts.settings.model_mappings; - let rotation_strategy = match facts.settings.rotation_strategy.as_str() { - "smart" => RotationStrategy::Smart, - "quota_reset_priority" => RotationStrategy::QuotaResetPriority, - "round_robin" => RotationStrategy::RoundRobin, - "sticky" => RotationStrategy::Sticky, - _ => return Err(RuntimeSnapshotCompileError::InvalidData), - }; + let rotation_strategy = RotationStrategy::parse(facts.settings.rotation_strategy.as_str()) + .ok_or(RuntimeSnapshotCompileError::InvalidData)?; let selection_policy = AccountSelectionPolicy::new( rotation_strategy, NonZeroU32::new(facts.settings.max_concurrent_per_account) diff --git a/backend/crates/gateway-core/tests/engine/credential.rs b/backend/crates/gateway-core/tests/engine/credential.rs index be44ff393..0f0f9626a 100644 --- a/backend/crates/gateway-core/tests/engine/credential.rs +++ b/backend/crates/gateway-core/tests/engine/credential.rs @@ -56,6 +56,19 @@ fn context(strategy: RotationStrategy) -> AccountSelectionContext { } } +#[test] +fn rotation_strategy_parse_should_round_trip_stable_wire_values() { + for strategy in [ + RotationStrategy::Smart, + RotationStrategy::QuotaResetPriority, + RotationStrategy::RoundRobin, + RotationStrategy::Sticky, + ] { + assert_eq!(RotationStrategy::parse(strategy.as_str()), Some(strategy)); + } + assert_eq!(RotationStrategy::parse("unknown"), None); +} + #[test] fn plaintext_credential_debug_should_redact_values() { let mut object = Map::new(); diff --git a/backend/crates/gateway-store/src/lib.rs b/backend/crates/gateway-store/src/lib.rs index 858259945..3fea0ff94 100644 --- a/backend/crates/gateway-store/src/lib.rs +++ b/backend/crates/gateway-store/src/lib.rs @@ -330,7 +330,7 @@ pub async fn initialize(mut config: StoreConfig) -> StoreResult { Arc::new(redis::RedisProviderCircuitRepository::new( redis_connection.clone(), REDIS_NAMESPACE, - redis::ProviderCircuitPolicy::default(), + gateway_core::engine::execution::ProviderCircuitPolicy::default(), )?); // Continuation affinity 是下一轮请求的路由事实，Core 必须直接等待 Redis 确认。 let continuation: Arc = @@ -613,7 +613,7 @@ impl SettingsStore for AdminSettingsStoreAdapter { refresh_concurrency: command.refresh_concurrency, max_concurrent_per_account: command.max_concurrent_per_account, request_interval_ms: command.request_interval_ms, - rotation_strategy: admin_rotation_name(command.rotation_strategy).to_owned(), + rotation_strategy: command.rotation_strategy.as_str().to_owned(), model_mappings: store_model_mappings(command.model_mappings), usage_retention_days: command.usage_retention_days, ops_event_retention_days: command.ops_event_retention_days, @@ -712,19 +712,14 @@ impl AdminSettingsStoreAdapter { fn admin_runtime_settings( settings: postgres::RuntimeSettings, ) -> AdminStoreResult { - let rotation_strategy = match settings.rotation_strategy.as_str() { - "smart" => AdminRotationStrategy::Smart, - "quota_reset_priority" => AdminRotationStrategy::QuotaResetPriority, - "round_robin" => AdminRotationStrategy::RoundRobin, - "sticky" => AdminRotationStrategy::Sticky, - _ => { - return Err(AdminStoreError::new( + let rotation_strategy = AdminRotationStrategy::parse(settings.rotation_strategy.as_str()) + .ok_or_else(|| { + AdminStoreError::new( AdminStoreErrorKind::Invalid, "runtime settings", "rotation strategy is invalid", - )); - } - }; + ) + })?; let model_mappings = settings .model_mappings .into_iter() @@ -768,15 +763,6 @@ fn store_model_mappings(mappings: ModelMappings) -> std::collections::BTreeMap &'static str { - match strategy { - AdminRotationStrategy::Smart => "smart", - AdminRotationStrategy::QuotaResetPriority => "quota_reset_priority", - AdminRotationStrategy::RoundRobin => "round_robin", - AdminRotationStrategy::Sticky => "sticky", - } -} - pub(crate) fn store_revision(revision: AdminRevision) -> AdminStoreResult { Revision::new(revision.get()).map_err(|error| admin_store_error("config revision", error)) } diff --git a/backend/crates/gateway-store/src/postgres/runtime_settings.rs b/backend/crates/gateway-store/src/postgres/runtime_settings.rs index 358af7254..2adee0e73 100644 --- a/backend/crates/gateway-store/src/postgres/runtime_settings.rs +++ b/backend/crates/gateway-store/src/postgres/runtime_settings.rs @@ -8,6 +8,7 @@ use async_trait::async_trait; use chrono::{DateTime, Utc}; use sqlx::{PgPool, Postgres, Transaction}; +use gateway_core::engine::credential::RotationStrategy; use gateway_core::provider_ports::{ ProviderRefreshPolicy, ProviderRuntimePolicyPort, ProviderStoreError, ProviderStoreErrorKind, }; @@ -93,10 +94,7 @@ impl RuntimeSettingsUpdate { || self.ops_event_retention_days == 0 || self.audit_retention_days == 0 || !valid_model_mappings(&self.model_mappings) - || !matches!( - self.rotation_strategy.as_str(), - "smart" | "quota_reset_priority" | "round_robin" | "sticky" - ) + || RotationStrategy::parse(&self.rotation_strategy).is_none() { return Err(StoreError::InvalidData { entity: "runtime settings", diff --git a/backend/crates/gateway-store/src/redis/provider_circuit.rs b/backend/crates/gateway-store/src/redis/provider_circuit.rs index 6552ace20..1e6006254 100644 --- a/backend/crates/gateway-store/src/redis/provider_circuit.rs +++ b/backend/crates/gateway-store/src/redis/provider_circuit.rs @@ -1,12 +1,10 @@ //! Provider 的可重建 Redis circuit。 -use std::num::NonZeroU32; -use std::time::Duration; - use async_trait::async_trait; use chrono::{DateTime, Utc}; use gateway_core::engine::execution::{ - ProviderCircuitDecision as CoreCircuitDecision, ProviderCircuitError, ProviderCircuitPort, + ProviderCircuitDecision as CoreCircuitDecision, ProviderCircuitError, ProviderCircuitPolicy, + ProviderCircuitPort, }; use gateway_core::routing::ProviderKind; use redis::{Script, aio::ConnectionManager}; @@ -37,21 +35,6 @@ if open_until > now_ms then return {0, tostring(open_until)} end return {1, '0'} "#; -#[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), - } - } -} - #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ProviderCircuitDecision { Allow, diff --git a/backend/crates/gateway-store/tests/redis/provider_circuit.rs b/backend/crates/gateway-store/tests/redis/provider_circuit.rs index 9804b9784..9ad4b285b 100644 --- a/backend/crates/gateway-store/tests/redis/provider_circuit.rs +++ b/backend/crates/gateway-store/tests/redis/provider_circuit.rs @@ -1,9 +1,11 @@ use std::{num::NonZeroU32, time::Duration}; -use gateway_core::{engine::execution::provider_failure_affects_circuit, error::ProviderErrorKind}; +use gateway_core::{ + engine::execution::{ProviderCircuitPolicy, provider_failure_affects_circuit}, + error::ProviderErrorKind, +}; use gateway_store::redis::{ - ProviderCircuitDecision, ProviderCircuitPolicy, ProviderCircuitRepository, - RedisProviderCircuitRepository, + ProviderCircuitDecision, ProviderCircuitRepository, RedisProviderCircuitRepository, }; use redis::aio::ConnectionManager; use uuid::Uuid;