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 <noreply@openai.com>
This commit is contained in:
zyycn
2026-07-31 23:23:23 +08:00
co-authored by Codex
parent c9b613659c
commit 4cd98889a7
11 changed files with 69 additions and 105 deletions
@@ -11,14 +11,8 @@ use super::Revision;
/// 客户端模型到上游模型的全局精确映射。
pub type ModelMappings = BTreeMap<PublicModelId, UpstreamModelId>;
/// 账号调度策略。
#[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)]
@@ -2180,19 +2180,6 @@ fn relative_time(value: Option<DateTime<Utc>>, now: DateTime<Utc>) -> 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(),
}
}
@@ -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<RuntimeSettings> 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<RotationStrategy> {
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
@@ -698,6 +698,17 @@ impl RotationStrategy {
Self::Sticky => "sticky",
}
}
#[must_use]
pub fn parse(value: &str) -> Option<Self> {
match value {
"smart" => Some(Self::Smart),
"quota_reset_priority" => Some(Self::QuotaResetPriority),
"round_robin" => Some(Self::RoundRobin),
"sticky" => Some(Self::Sticky),
_ => None,
}
}
}
/// 从 `runtime_settings` 冻结到一次请求计划的账号调度策略。
@@ -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<RoutingContext, GatewayError> {
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 => {
@@ -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)
@@ -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();
+7 -21
View File
@@ -330,7 +330,7 @@ pub async fn initialize(mut config: StoreConfig) -> StoreResult<StoreBundle> {
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<dyn gateway_core::engine::continuation::NativeContinuationPort> =
@@ -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<AdminRuntimeSettings> {
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<S
.collect()
}
const fn admin_rotation_name(strategy: AdminRotationStrategy) -> &'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> {
Revision::new(revision.get()).map_err(|error| admin_store_error("config revision", error))
}
@@ -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",
@@ -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,
@@ -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;