fix(gateway): expire provider key circuit cooldowns

This commit is contained in:
MMEXA
2026-05-25 00:40:38 +08:00
parent 40b4e52508
commit 480579a0d5
21 changed files with 291 additions and 99 deletions
Generated
+1
View File
@@ -432,6 +432,7 @@ dependencies = [
"aether-contracts", "aether-contracts",
"aether-data-contracts", "aether-data-contracts",
"aether-wallet", "aether-wallet",
"chrono",
"regex", "regex",
"serde", "serde",
"serde_json", "serde_json",
@@ -6,6 +6,7 @@ use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogKe
use aether_pool_core::{ use aether_pool_core::{
score_pool_member_with_rules, PoolMemberScoreInput, PoolMemberScoreRules, POOL_SCORE_VERSION, score_pool_member_with_rules, PoolMemberScoreInput, PoolMemberScoreRules, POOL_SCORE_VERSION,
}; };
use aether_scheduler_core::any_provider_key_circuit_open_at;
use serde_json::Value; use serde_json::Value;
use crate::handlers::shared::{provider_key_health_summary, provider_key_status_snapshot_payload}; use crate::handlers::shared::{provider_key_health_summary, provider_key_status_snapshot_payload};
@@ -98,7 +99,8 @@ fn provider_key_score_input(
.as_object() .as_object()
.and_then(|snapshot| snapshot.get("account")) .and_then(|snapshot| snapshot.get("account"))
.and_then(Value::as_object); .and_then(Value::as_object);
let (health_score, _, _, any_circuit_open, _) = provider_key_health_summary(key); let (health_score, _, _, _, _) = provider_key_health_summary(key);
let active_circuit_open = any_provider_key_circuit_open_at(key, now_unix_secs);
let health_score = key let health_score = key
.health_by_format .health_by_format
.as_ref() .as_ref()
@@ -125,7 +127,7 @@ fn provider_key_score_input(
.and_then(Value::as_bool) .and_then(Value::as_bool)
.unwrap_or(false), .unwrap_or(false),
oauth_invalid_reason: key.oauth_invalid_reason.clone(), oauth_invalid_reason: key.oauth_invalid_reason.clone(),
circuit_open: any_circuit_open, circuit_open: active_circuit_open,
success_count: key.success_count.unwrap_or(0).into(), success_count: key.success_count.unwrap_or(0).into(),
error_count: key.error_count.unwrap_or(0).into(), error_count: key.error_count.unwrap_or(0).into(),
total_response_time_ms: key.total_response_time_ms.unwrap_or(0).into(), total_response_time_ms: key.total_response_time_ms.unwrap_or(0).into(),
@@ -159,3 +161,70 @@ fn stable_hash(bytes: &[u8]) -> u64 {
} }
hash hash
} }
#[cfg(test)]
mod tests {
use super::*;
use aether_data_contracts::repository::pool_scores::PoolMemberHardState;
use serde_json::json;
fn sample_key_with_circuit_next_probe(
next_probe_at_unix_secs: u64,
) -> StoredProviderCatalogKey {
let mut key = StoredProviderCatalogKey::new(
"key-gemini-5".to_string(),
"provider-google-api".to_string(),
"5".to_string(),
"api_key".to_string(),
None,
true,
)
.expect("sample key should be valid");
key.health_by_format = Some(json!({
"gemini:generate_content": {
"health_score": 0.2,
"consecutive_failures": 8
}
}));
key.circuit_breaker_by_format = Some(json!({
"gemini:generate_content": {
"open": true,
"reason": "consecutive_failures_8",
"next_probe_at_unix_secs": next_probe_at_unix_secs
}
}));
key
}
#[test]
fn expired_circuit_probe_deadline_does_not_leave_pool_score_in_cooldown() {
let now_unix_secs = 1_000;
let key = sample_key_with_circuit_next_probe(900);
let score = build_provider_key_pool_score_upsert(
&key,
"custom",
None,
now_unix_secs,
PoolMemberScoreRules::default(),
);
assert_eq!(score.hard_state, PoolMemberHardState::Available);
}
#[test]
fn future_circuit_probe_deadline_keeps_pool_score_in_cooldown() {
let now_unix_secs = 1_000;
let key = sample_key_with_circuit_next_probe(1_100);
let score = build_provider_key_pool_score_upsert(
&key,
"custom",
None,
now_unix_secs,
PoolMemberScoreRules::default(),
);
assert_eq!(score.hard_state, PoolMemberHardState::Cooldown);
}
}
@@ -1,6 +1,9 @@
use crate::handlers::admin::request::AdminAppState; use crate::handlers::admin::request::AdminAppState;
use crate::provider_key_auth::provider_key_effective_api_formats; use crate::provider_key_auth::provider_key_effective_api_formats;
use aether_scheduler_core::count_recent_rpm_requests_for_provider_key_since; use aether_scheduler_core::{
count_recent_rpm_requests_for_provider_key_since,
provider_key_circuit_payload_is_active_open_at,
};
use serde_json::json; use serde_json::json;
use std::time::{SystemTime, UNIX_EPOCH}; use std::time::{SystemTime, UNIX_EPOCH};
@@ -18,6 +21,10 @@ pub(crate) async fn build_admin_key_health_payload(
.await .await
.ok() .ok()
.and_then(|mut keys| keys.drain(..).next())?; .and_then(|mut keys| keys.drain(..).next())?;
let now_unix_secs = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_secs())
.unwrap_or_default();
let provider = state let provider = state
.read_provider_catalog_providers_by_ids(std::slice::from_ref(&key.provider_id)) .read_provider_catalog_providers_by_ids(std::slice::from_ref(&key.provider_id))
.await .await
@@ -81,10 +88,10 @@ pub(crate) async fn build_admin_key_health_payload(
.and_then(|value| value.get("last_failure_at")) .and_then(|value| value.get("last_failure_at"))
.cloned() .cloned()
.unwrap_or(serde_json::Value::Null); .unwrap_or(serde_json::Value::Null);
payload["circuit_breaker_open"] = json!(circuit_data payload["circuit_breaker_open"] =
.and_then(|value| value.get("open")) json!(circuit_data.is_some_and(
.and_then(serde_json::Value::as_bool) |value| provider_key_circuit_payload_is_active_open_at(value, now_unix_secs)
.unwrap_or(false)); ));
payload["circuit_breaker_open_at"] = circuit_data payload["circuit_breaker_open_at"] = circuit_data
.and_then(|value| value.get("open_at")) .and_then(|value| value.get("open_at"))
.cloned() .cloned()
@@ -168,11 +175,9 @@ pub(crate) async fn build_admin_key_health_payload(
.reduce(f64::min) .reduce(f64::min)
.unwrap_or(1.0); .unwrap_or(1.0);
let any_circuit_open = formats_payload.values().any(|value| { let any_circuit_open = formats_payload.values().any(|value| {
value value.get("circuit_breaker").is_some_and(|circuit| {
.get("circuit_breaker") provider_key_circuit_payload_is_active_open_at(circuit, now_unix_secs)
.and_then(|circuit| circuit.get("open")) })
.and_then(serde_json::Value::as_bool)
.unwrap_or(false)
}); });
payload["key_health_score"] = json!(key_health_score); payload["key_health_score"] = json!(key_health_score);
@@ -4,7 +4,9 @@ use crate::handlers::public::{api_format_display_name, build_public_health_timel
use crate::handlers::shared::unix_ms_to_rfc3339; use crate::handlers::shared::unix_ms_to_rfc3339;
use crate::provider_key_auth::provider_key_effective_api_formats; use crate::provider_key_auth::provider_key_effective_api_formats;
use aether_data_contracts::repository::candidates::PublicHealthTimelineBucket; use aether_data_contracts::repository::candidates::PublicHealthTimelineBucket;
use aether_scheduler_core::{is_provider_key_circuit_open, provider_key_health_score}; use aether_scheduler_core::{
any_provider_key_circuit_open_at, is_provider_key_circuit_open_at, provider_key_health_score,
};
use serde_json::json; use serde_json::json;
use std::collections::{BTreeMap, BTreeSet}; use std::collections::{BTreeMap, BTreeSet};
use std::time::{SystemTime, UNIX_EPOCH}; use std::time::{SystemTime, UNIX_EPOCH};
@@ -99,7 +101,9 @@ pub(crate) async fn build_admin_endpoint_health_status_payload(
.entry(api_format.clone()) .entry(api_format.clone())
.or_default() .or_default()
.insert(key.id.clone()); .insert(key.id.clone());
if key.is_active && !is_provider_key_circuit_open(&key, &api_format) { if key.is_active
&& !is_provider_key_circuit_open_at(&key, &api_format, now_unix_secs)
{
let key_health_score = let key_health_score =
provider_key_health_score(&key, &api_format).unwrap_or(1.0); provider_key_health_score(&key, &api_format).unwrap_or(1.0);
active_keys_by_format active_keys_by_format
@@ -229,6 +233,10 @@ pub(crate) async fn build_admin_health_summary_payload(
return None; return None;
} }
let now_unix_secs = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_secs())
.unwrap_or_default();
let providers = state let providers = state
.list_provider_catalog_providers(false) .list_provider_catalog_providers(false)
.await .await
@@ -286,20 +294,7 @@ pub(crate) async fn build_admin_health_summary_payload(
.count(); .count();
let circuit_open_keys = keys let circuit_open_keys = keys
.iter() .iter()
.filter(|key| { .filter(|key| any_provider_key_circuit_open_at(key, now_unix_secs))
key.circuit_breaker_by_format
.as_ref()
.and_then(serde_json::Value::as_object)
.map(|formats| {
formats.values().any(|circuit| {
circuit
.get("open")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false)
})
})
.unwrap_or(false)
})
.count(); .count();
Some(json!({ Some(json!({
@@ -8,10 +8,12 @@ use aether_data_contracts::repository::provider_catalog::{
StoredProviderCatalogEndpoint, StoredProviderCatalogKey, StoredProviderCatalogEndpoint, StoredProviderCatalogKey,
}; };
use aether_scheduler_core::{ use aether_scheduler_core::{
is_provider_key_circuit_open, matches_model_mapping, provider_key_health_score, is_provider_key_circuit_open_at, matches_model_mapping,
provider_key_circuit_payload_is_active_open_at, provider_key_health_score,
}; };
use serde_json::json; use serde_json::json;
use std::collections::BTreeMap; use std::collections::BTreeMap;
use std::time::{SystemTime, UNIX_EPOCH};
use uuid::Uuid; use uuid::Uuid;
pub(crate) async fn build_admin_global_model_routing_payload( pub(crate) async fn build_admin_global_model_routing_payload(
@@ -86,6 +88,10 @@ pub(crate) async fn build_admin_global_model_routing_payload(
.flatten() .flatten()
.and_then(|value| value.as_bool()) .and_then(|value| value.as_bool())
.unwrap_or(false); .unwrap_or(false);
let now_unix_secs = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_secs())
.unwrap_or_default();
let global_model_mappings = global_model let global_model_mappings = global_model
.config .config
@@ -163,9 +169,11 @@ pub(crate) async fn build_admin_global_model_routing_payload(
entries entries
.iter() .iter()
.filter_map(|(api_format, value)| { .filter_map(|(api_format, value)| {
value.get("open") provider_key_circuit_payload_is_active_open_at(
.and_then(serde_json::Value::as_bool) value,
.filter(|is_open| *is_open) now_unix_secs,
)
.then_some(())
.map(|_| api_format.clone()) .map(|_| api_format.clone())
}) })
.collect::<Vec<_>>() .collect::<Vec<_>>()
@@ -188,7 +196,7 @@ pub(crate) async fn build_admin_global_model_routing_payload(
"effective_rpm": effective_rpm, "effective_rpm": effective_rpm,
"allowed_models": allowed_models, "allowed_models": allowed_models,
"health_score": provider_key_health_score(key, &endpoint.api_format), "health_score": provider_key_health_score(key, &endpoint.api_format),
"circuit_breaker_open": is_provider_key_circuit_open(key, &endpoint.api_format), "circuit_breaker_open": is_provider_key_circuit_open_at(key, &endpoint.api_format, now_unix_secs),
"circuit_breaker_formats": circuit_breaker_formats, "circuit_breaker_formats": circuit_breaker_formats,
"next_probe_at": next_probe_at, "next_probe_at": next_probe_at,
}); });
@@ -1,6 +1,6 @@
use super::super::usage_helpers::admin_monitoring_usage_is_error; use super::super::usage_helpers::admin_monitoring_usage_is_error;
use crate::handlers::admin::request::AdminAppState; use crate::handlers::admin::request::AdminAppState;
use crate::handlers::admin::shared::{provider_key_health_summary, unix_secs_to_rfc3339}; use crate::handlers::admin::shared::{provider_key_health_summary_at, unix_secs_to_rfc3339};
use crate::GatewayError; use crate::GatewayError;
use aether_data_contracts::repository::{ use aether_data_contracts::repository::{
provider_catalog::StoredProviderCatalogKey, usage::UsageMonitoringErrorListQuery, provider_catalog::StoredProviderCatalogKey, usage::UsageMonitoringErrorListQuery,
@@ -99,7 +99,7 @@ pub(super) async fn build_admin_monitoring_resilience_snapshot(
last_failure_at, last_failure_at,
circuit_breaker_open, circuit_breaker_open,
circuit_by_format, circuit_by_format,
) = provider_key_health_summary(key); ) = provider_key_health_summary_at(key, now.timestamp().max(0) as u64);
if health_score < 0.8 { if health_score < 0.8 {
degraded_keys += 1; degraded_keys += 1;
} }
@@ -110,11 +110,12 @@ pub(super) async fn build_admin_monitoring_resilience_snapshot(
let open_formats = circuit_by_format let open_formats = circuit_by_format
.iter() .iter()
.filter_map(|(api_format, value)| { .filter_map(|(api_format, value)| {
value aether_scheduler_core::provider_key_circuit_payload_is_active_open_at(
.get("open") value,
.and_then(serde_json::Value::as_bool) now.timestamp().max(0) as u64,
.filter(|open| *open) )
.map(|_| api_format.clone()) .then_some(())
.map(|_| api_format.clone())
}) })
.collect::<Vec<_>>(); .collect::<Vec<_>>();
@@ -1237,7 +1237,8 @@ async fn admin_monitoring_circuit_history_returns_local_payload() {
"openai:chat": { "openai:chat": {
"open": true, "open": true,
"open_at": "2026-03-30T12:00:00+00:00", "open_at": "2026-03-30T12:00:00+00:00",
"next_probe_at": "2026-03-30T12:05:00+00:00", "next_probe_at": "2099-03-30T12:05:00+00:00",
"recovery_seconds": 300,
"reason": "错误率过高" "reason": "错误率过高"
} }
})), })),
@@ -13,6 +13,7 @@ use aether_data_contracts::repository::provider_catalog::{
StoredProviderCatalogEndpoint, StoredProviderCatalogKey, StoredProviderCatalogEndpoint, StoredProviderCatalogKey,
}; };
use aether_data_contracts::repository::usage::StoredProviderApiKeyWindowUsageSummary; use aether_data_contracts::repository::usage::StoredProviderApiKeyWindowUsageSummary;
use aether_scheduler_core::provider_key_circuit_payload_is_active_open_at;
use serde_json::json; use serde_json::json;
use std::collections::BTreeMap; use std::collections::BTreeMap;
use std::time::{SystemTime, UNIX_EPOCH}; use std::time::{SystemTime, UNIX_EPOCH};
@@ -921,19 +922,14 @@ fn admin_pool_health_score(key: &StoredProviderCatalogKey) -> f64 {
} }
} }
fn admin_pool_circuit_breaker_open(key: &StoredProviderCatalogKey) -> bool { fn admin_pool_circuit_breaker_open(key: &StoredProviderCatalogKey, now_unix_secs: u64) -> bool {
key.circuit_breaker_by_format key.circuit_breaker_by_format
.as_ref() .as_ref()
.and_then(serde_json::Value::as_object) .and_then(serde_json::Value::as_object)
.map(|formats| { .map(|formats| {
formats formats
.values() .values()
.filter_map(serde_json::Value::as_object) .any(|item| provider_key_circuit_payload_is_active_open_at(item, now_unix_secs))
.any(|item| {
item.get("open")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false)
})
}) })
.unwrap_or(false) .unwrap_or(false)
} }
@@ -1032,7 +1028,7 @@ pub(super) fn build_admin_pool_key_payload(
.as_ref() .as_ref()
.and_then(|_| runtime.cooldown_ttl_by_key.get(&key.id).copied()); .and_then(|_| runtime.cooldown_ttl_by_key.get(&key.id).copied());
let health_score = admin_pool_health_score(key); let health_score = admin_pool_health_score(key);
let circuit_breaker_open = admin_pool_circuit_breaker_open(key); let circuit_breaker_open = admin_pool_circuit_breaker_open(key, now_unix_secs);
let auth_semantics = provider_key_auth_semantics(key, provider_type); let auth_semantics = provider_key_auth_semantics(key, provider_type);
let account_quota_exhausted = pool_config let account_quota_exhausted = pool_config
.as_ref() .as_ref()
@@ -68,6 +68,7 @@ use aether_model_fetch::{
aggregate_models_for_cache, fetch_models_from_transports, json_string_list, aggregate_models_for_cache, fetch_models_from_transports, json_string_list,
preset_models_for_provider, selected_models_fetch_endpoints, preset_models_for_provider, selected_models_fetch_endpoints,
}; };
use aether_scheduler_core::provider_key_circuit_payload_is_active_open_at;
use axum::{ use axum::{
body::{to_bytes, Body}, body::{to_bytes, Body},
http::{self, HeaderMap, HeaderName, HeaderValue}, http::{self, HeaderMap, HeaderName, HeaderValue},
@@ -920,6 +921,7 @@ fn provider_query_test_key_sort_key(
provider_type: &str, provider_type: &str,
key: &StoredProviderCatalogKey, key: &StoredProviderCatalogKey,
endpoint_api_format: &str, endpoint_api_format: &str,
now_unix_secs: u64,
) -> (u8, u8, i32, u64, i32) { ) -> (u8, u8, i32, u64, i32) {
let quota_exhausted = let quota_exhausted =
admin_provider_pool_pure::admin_pool_key_account_quota_exhausted(key, provider_type); admin_provider_pool_pure::admin_pool_key_account_quota_exhausted(key, provider_type);
@@ -928,10 +930,7 @@ fn provider_query_test_key_sort_key(
.as_ref() .as_ref()
.and_then(Value::as_object) .and_then(Value::as_object)
.and_then(|value| value.get(endpoint_api_format)) .and_then(|value| value.get(endpoint_api_format))
.and_then(Value::as_object) .is_some_and(|value| provider_key_circuit_payload_is_active_open_at(value, now_unix_secs));
.and_then(|value| value.get("open"))
.and_then(Value::as_bool)
.unwrap_or(false);
let health_score = key let health_score = key
.health_by_format .health_by_format
.as_ref() .as_ref()
@@ -1390,6 +1389,7 @@ async fn provider_query_build_kiro_test_candidates(
provider_query_key_supports_endpoint(key, &provider.provider_type, &endpoint.api_format) provider_query_key_supports_endpoint(key, &provider.provider_type, &endpoint.api_format)
}) })
.collect::<Vec<_>>(); .collect::<Vec<_>>();
let now_unix_secs = current_unix_ms() / 1000;
let candidates = if test_mode.eq_ignore_ascii_case("pool") { let candidates = if test_mode.eq_ignore_ascii_case("pool") {
if let Some(pool_config) = if let Some(pool_config) =
@@ -1411,6 +1411,7 @@ async fn provider_query_build_kiro_test_candidates(
provider.provider_type.as_str(), provider.provider_type.as_str(),
key, key,
&endpoint.api_format, &endpoint.api_format,
now_unix_secs,
) )
}); });
keys.into_iter() keys.into_iter()
@@ -1428,6 +1429,7 @@ async fn provider_query_build_kiro_test_candidates(
provider.provider_type.as_str(), provider.provider_type.as_str(),
key, key,
&endpoint.api_format, &endpoint.api_format,
now_unix_secs,
) )
}); });
keys.into_iter() keys.into_iter()
@@ -13,7 +13,8 @@ pub(crate) use crate::handlers::shared::{
effective_catalog_encryption_key, encrypt_catalog_secret_with_fallbacks, json_string_list, effective_catalog_encryption_key, encrypt_catalog_secret_with_fallbacks, json_string_list,
masked_catalog_api_key, normalize_json_array, normalize_json_object, normalize_string_list, masked_catalog_api_key, normalize_json_array, normalize_json_object, normalize_string_list,
parse_catalog_auth_config_json, provider_catalog_key_supports_format, parse_catalog_auth_config_json, provider_catalog_key_supports_format,
provider_key_health_summary, provider_key_status_snapshot_payload, query_param_bool, provider_key_health_summary, provider_key_health_summary_at,
query_param_optional_bool, query_param_value, take_secret_prefix, take_secret_suffix, provider_key_status_snapshot_payload, query_param_bool, query_param_optional_bool,
unix_secs_to_rfc3339, OFFICIAL_EXTERNAL_MODEL_PROVIDERS, query_param_value, take_secret_prefix, take_secret_suffix, unix_secs_to_rfc3339,
OFFICIAL_EXTERNAL_MODEL_PROVIDERS,
}; };
@@ -2,6 +2,7 @@ use super::enabled_key_capability_short_names;
use crate::handlers::shared::unix_secs_to_rfc3339; use crate::handlers::shared::unix_secs_to_rfc3339;
use crate::provider_key_auth::provider_key_effective_api_formats; use crate::provider_key_auth::provider_key_effective_api_formats;
use crate::AppState; use crate::AppState;
use aether_scheduler_core::provider_key_circuit_payload_is_active_open_at;
use serde_json::json; use serde_json::json;
use std::collections::{BTreeMap, HashMap}; use std::collections::{BTreeMap, HashMap};
use std::time::{SystemTime, UNIX_EPOCH}; use std::time::{SystemTime, UNIX_EPOCH};
@@ -173,10 +174,10 @@ pub(crate) async fn build_admin_keys_grouped_by_format_payload(
.get("health_score") .get("health_score")
.and_then(serde_json::Value::as_f64) .and_then(serde_json::Value::as_f64)
.unwrap_or(1.0), .unwrap_or(1.0),
"circuit_breaker_open": format_circuit "circuit_breaker_open": provider_key_circuit_payload_is_active_open_at(
.get("open") &format_circuit,
.and_then(serde_json::Value::as_bool) now_unix_secs,
.unwrap_or(false), ),
"last_used_at": key.last_used_at_unix_secs.and_then(unix_secs_to_rfc3339), "last_used_at": key.last_used_at_unix_secs.and_then(unix_secs_to_rfc3339),
"created_at": unix_secs_to_rfc3339(key.created_at_unix_ms.unwrap_or(now_unix_secs)), "created_at": unix_secs_to_rfc3339(key.created_at_unix_ms.unwrap_or(now_unix_secs)),
"updated_at": unix_secs_to_rfc3339(key.updated_at_unix_secs.unwrap_or(now_unix_secs)), "updated_at": unix_secs_to_rfc3339(key.updated_at_unix_secs.unwrap_or(now_unix_secs)),
@@ -13,6 +13,7 @@ use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogKe
use aether_provider_pool::{ use aether_provider_pool::{
grok_pool_tier_from_quota_bucket, grok_supported_quota_windows_for_tier, grok_pool_tier_from_quota_bucket, grok_supported_quota_windows_for_tier,
}; };
use aether_scheduler_core::provider_key_circuit_payload_is_active_open_at;
use serde_json::{json, Map, Value}; use serde_json::{json, Map, Value};
use std::borrow::Cow; use std::borrow::Cow;
use std::time::{SystemTime, UNIX_EPOCH}; use std::time::{SystemTime, UNIX_EPOCH};
@@ -1778,6 +1779,39 @@ pub(crate) fn provider_key_health_summary(
Option<String>, Option<String>,
bool, bool,
serde_json::Map<String, serde_json::Value>, serde_json::Map<String, serde_json::Value>,
) {
provider_key_health_summary_with_circuit_predicate(key, |value| {
value
.get("open")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false)
})
}
pub(crate) fn provider_key_health_summary_at(
key: &StoredProviderCatalogKey,
now_unix_secs: u64,
) -> (
f64,
i64,
Option<String>,
bool,
serde_json::Map<String, serde_json::Value>,
) {
provider_key_health_summary_with_circuit_predicate(key, |value| {
provider_key_circuit_payload_is_active_open_at(value, now_unix_secs)
})
}
fn provider_key_health_summary_with_circuit_predicate(
key: &StoredProviderCatalogKey,
circuit_is_open: impl Fn(&serde_json::Value) -> bool,
) -> (
f64,
i64,
Option<String>,
bool,
serde_json::Map<String, serde_json::Value>,
) { ) {
let health_by_format = key let health_by_format = key
.health_by_format .health_by_format
@@ -1820,12 +1854,7 @@ pub(crate) fn provider_key_health_summary(
} }
} }
let any_circuit_open = circuit_by_format.values().any(|value| { let any_circuit_open = circuit_by_format.values().any(circuit_is_open);
value
.get("open")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false)
});
( (
if health_by_format.is_empty() { if health_by_format.is_empty() {
@@ -1974,15 +2003,10 @@ pub(crate) fn build_admin_provider_key_response(
last_failure_at, last_failure_at,
circuit_breaker_open, circuit_breaker_open,
circuit_by_format, circuit_by_format,
) = provider_key_health_summary(key); ) = provider_key_health_summary_at(key, now_unix_secs);
let circuit_sample = circuit_by_format let circuit_sample = circuit_by_format
.values() .values()
.find(|value| { .find(|value| provider_key_circuit_payload_is_active_open_at(value, now_unix_secs))
value
.get("open")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false)
})
.or_else(|| circuit_by_format.values().next()); .or_else(|| circuit_by_format.values().next());
let is_adaptive = key.rpm_limit.is_none(); let is_adaptive = key.rpm_limit.is_none();
let effective_limit = if is_adaptive { let effective_limit = if is_adaptive {
@@ -26,8 +26,9 @@ pub(crate) use self::catalog::{
default_provider_key_status_snapshot, effective_catalog_encryption_key, default_provider_key_status_snapshot, effective_catalog_encryption_key,
encrypt_catalog_secret_with_fallbacks, masked_catalog_api_key, parse_catalog_auth_config_json, encrypt_catalog_secret_with_fallbacks, masked_catalog_api_key, parse_catalog_auth_config_json,
provider_catalog_key_supports_format, provider_key_health_summary, provider_catalog_key_supports_format, provider_key_health_summary,
provider_key_status_snapshot_payload, sync_provider_key_oauth_status_snapshot, provider_key_health_summary_at, provider_key_status_snapshot_payload,
sync_provider_key_quota_status_snapshot, take_secret_prefix, take_secret_suffix, sync_provider_key_oauth_status_snapshot, sync_provider_key_quota_status_snapshot,
take_secret_prefix, take_secret_suffix,
}; };
pub(crate) use self::email_templates::{ pub(crate) use self::email_templates::{
admin_email_template_definition, admin_email_template_html_key, admin_email_template_definition, admin_email_template_html_key,
@@ -241,8 +241,8 @@ async fn gateway_provider_keys_expose_circuit_breaker_and_recover_clears_it() {
"open": true, "open": true,
"open_at": "2026-03-26T12:00:00+00:00", "open_at": "2026-03-26T12:00:00+00:00",
"reason": "consecutive_failures_8", "reason": "consecutive_failures_8",
"next_probe_at": "2026-03-26T12:01:00+00:00", "next_probe_at": "2099-03-26T12:01:00+00:00",
"next_probe_at_unix_secs": 1774526460u64, "next_probe_at_unix_secs": 4078209660u64,
"probe_interval_minutes": 1, "probe_interval_minutes": 1,
"max_probe_interval_minutes": 32, "max_probe_interval_minutes": 32,
"half_open_until": null, "half_open_until": null,
@@ -346,7 +346,7 @@ async fn gateway_handles_admin_key_health_locally_with_trusted_admin_principal()
Some(json!({"openai:chat": { Some(json!({"openai:chat": {
"open": true, "open": true,
"open_at": "2026-03-26T12:01:00+00:00", "open_at": "2026-03-26T12:01:00+00:00",
"next_probe_at": "2026-03-26T12:05:00+00:00", "next_probe_at": "2099-03-26T12:05:00+00:00",
"half_open_until": null, "half_open_until": null,
"half_open_successes": 1, "half_open_successes": 1,
"half_open_failures": 0 "half_open_failures": 0
@@ -399,7 +399,7 @@ async fn gateway_handles_admin_key_health_locally_with_trusted_admin_principal()
payload["circuit_breaker_open_at"], payload["circuit_breaker_open_at"],
"2026-03-26T12:01:00+00:00" "2026-03-26T12:01:00+00:00"
); );
assert_eq!(payload["next_probe_at"], "2026-03-26T12:05:00+00:00"); assert_eq!(payload["next_probe_at"], "2099-03-26T12:05:00+00:00");
assert_eq!(payload["half_open_successes"], 1); assert_eq!(payload["half_open_successes"], 1);
assert_eq!(payload["half_open_failures"], 0); assert_eq!(payload["half_open_failures"], 0);
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0); assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
@@ -442,7 +442,7 @@ async fn gateway_recovers_admin_key_health_locally_with_trusted_admin_principal(
Some(json!({"openai:chat": { Some(json!({"openai:chat": {
"open": true, "open": true,
"open_at": "2026-03-26T12:01:00+00:00", "open_at": "2026-03-26T12:01:00+00:00",
"next_probe_at": "2026-03-26T12:05:00+00:00", "next_probe_at": "2099-03-26T12:05:00+00:00",
"half_open_until": null, "half_open_until": null,
"half_open_successes": 0, "half_open_successes": 0,
"half_open_failures": 1 "half_open_failures": 1
@@ -789,7 +789,7 @@ async fn gateway_handles_admin_global_model_routing_locally_with_trusted_admin_p
"openai:chat": {"health_score": 0.66} "openai:chat": {"health_score": 0.66}
})); }));
primary_key.circuit_breaker_by_format = Some(json!({ primary_key.circuit_breaker_by_format = Some(json!({
"openai:chat": {"open": true, "next_probe_at": "2026-03-27T15:00:00Z"} "openai:chat": {"open": true, "next_probe_at": "2099-03-27T15:00:00Z"}
})); }));
let mut mapped_key = sample_key( let mut mapped_key = sample_key(
@@ -907,7 +907,7 @@ async fn gateway_handles_admin_global_model_routing_locally_with_trusted_admin_p
openai_keys[0]["circuit_breaker_formats"], openai_keys[0]["circuit_breaker_formats"],
json!(["openai:chat"]) json!(["openai:chat"])
); );
assert_eq!(openai_keys[0]["next_probe_at"], "2026-03-27T15:00:00Z"); assert_eq!(openai_keys[0]["next_probe_at"], "2099-03-27T15:00:00Z");
let alt_endpoints = providers[1]["endpoints"] let alt_endpoints = providers[1]["endpoints"]
.as_array() .as_array()
@@ -2040,7 +2040,8 @@ async fn gateway_handles_admin_monitoring_resilience_circuit_history_locally_wit
"openai:chat": { "openai:chat": {
"open": true, "open": true,
"open_at": "2026-03-30T12:00:00+00:00", "open_at": "2026-03-30T12:00:00+00:00",
"next_probe_at": "2026-03-30T12:05:00+00:00", "next_probe_at": "2099-03-30T12:05:00+00:00",
"recovery_seconds": 300,
"reason": "错误率过高" "reason": "错误率过高"
} }
})), })),
+23 -2
View File
@@ -157,18 +157,39 @@ fn admin_pool_health_score(key: &StoredProviderCatalogKey) -> f64 {
} }
fn admin_pool_circuit_breaker_open(key: &StoredProviderCatalogKey) -> bool { fn admin_pool_circuit_breaker_open(key: &StoredProviderCatalogKey) -> bool {
let now_unix_secs = Utc::now().timestamp().max(0) as u64;
key.circuit_breaker_by_format key.circuit_breaker_by_format
.as_ref() .as_ref()
.and_then(Value::as_object) .and_then(Value::as_object)
.map(|formats| { .map(|formats| {
formats formats
.values() .values()
.filter_map(Value::as_object) .any(|item| admin_pool_circuit_payload_active_open_at(item, now_unix_secs))
.any(|item| item.get("open").and_then(Value::as_bool).unwrap_or(false))
}) })
.unwrap_or(false) .unwrap_or(false)
} }
fn admin_pool_circuit_payload_active_open_at(value: &Value, now_unix_secs: u64) -> bool {
let Some(item) = value.as_object() else {
return false;
};
if !item.get("open").and_then(Value::as_bool).unwrap_or(false) {
return false;
}
if let Some(next_probe_at) = item.get("next_probe_at_unix_secs").and_then(Value::as_u64) {
return now_unix_secs < next_probe_at;
}
if let Some(next_probe_at) = item
.get("next_probe_at")
.and_then(Value::as_str)
.and_then(|value| chrono::DateTime::parse_from_rfc3339(value).ok())
.and_then(|value| u64::try_from(value.timestamp()).ok())
{
return now_unix_secs < next_probe_at;
}
true
}
fn unix_secs_to_rfc3339(unix_secs: u64) -> Option<String> { fn unix_secs_to_rfc3339(unix_secs: u64) -> Option<String> {
Utc.timestamp_opt(unix_secs as i64, 0) Utc.timestamp_opt(unix_secs as i64, 0)
.single() .single()
+1
View File
@@ -11,6 +11,7 @@ aether-ai-formats.workspace = true
aether-contracts.workspace = true aether-contracts.workspace = true
aether-data-contracts.workspace = true aether-data-contracts.workspace = true
aether-wallet.workspace = true aether-wallet.workspace = true
chrono.workspace = true
regex.workspace = true regex.workspace = true
serde.workspace = true serde.workspace = true
serde_json.workspace = true serde_json.workspace = true
+69 -6
View File
@@ -294,10 +294,33 @@ pub fn is_provider_key_circuit_open_at(
.as_ref() .as_ref()
.and_then(serde_json::Value::as_object) .and_then(serde_json::Value::as_object)
.and_then(|values| values.get(api_format)) .and_then(|values| values.get(api_format))
.and_then(serde_json::Value::as_object)
else { else {
return false; return false;
}; };
provider_key_circuit_payload_is_active_open_at(payload, now_unix_secs)
}
pub fn any_provider_key_circuit_open_at(
key: &StoredProviderCatalogKey,
now_unix_secs: u64,
) -> bool {
key.circuit_breaker_by_format
.as_ref()
.and_then(serde_json::Value::as_object)
.is_some_and(|values| {
values.values().any(|payload| {
provider_key_circuit_payload_is_active_open_at(payload, now_unix_secs)
})
})
}
pub fn provider_key_circuit_payload_is_active_open_at(
payload: &serde_json::Value,
now_unix_secs: u64,
) -> bool {
let Some(payload) = payload.as_object() else {
return false;
};
if !payload if !payload
.get("open") .get("open")
.and_then(serde_json::Value::as_bool) .and_then(serde_json::Value::as_bool)
@@ -305,10 +328,26 @@ pub fn is_provider_key_circuit_open_at(
{ {
return false; return false;
} }
payload if let Some(next_probe_at) = payload
.get("next_probe_at_unix_secs") .get("next_probe_at_unix_secs")
.and_then(serde_json::Value::as_u64) .and_then(serde_json::Value::as_u64)
.is_none_or(|next_probe_at| now_unix_secs < next_probe_at) {
return now_unix_secs < next_probe_at;
}
if let Some(next_probe_at) = payload
.get("next_probe_at")
.and_then(serde_json::Value::as_str)
.and_then(rfc3339_to_unix_secs)
{
return now_unix_secs < next_probe_at;
}
true
}
fn rfc3339_to_unix_secs(value: &str) -> Option<u64> {
chrono::DateTime::parse_from_rfc3339(value)
.ok()
.and_then(|value| u64::try_from(value.timestamp()).ok())
} }
fn available_provider_key_rpm_slots_for_new_user( fn available_provider_key_rpm_slots_for_new_user(
@@ -641,9 +680,9 @@ mod tests {
count_recent_rpm_requests_for_provider_key, count_recent_rpm_requests_for_provider_key,
count_recent_rpm_requests_for_provider_key_since, effective_provider_key_health_score, count_recent_rpm_requests_for_provider_key_since, effective_provider_key_health_score,
effective_provider_key_rpm_limit, is_candidate_in_recent_failure_cooldown, effective_provider_key_rpm_limit, is_candidate_in_recent_failure_cooldown,
is_provider_key_circuit_open, provider_key_health_bucket, provider_key_health_score, is_provider_key_circuit_open, is_provider_key_circuit_open_at, provider_key_health_bucket,
provider_key_rpm_allows_request, provider_key_rpm_allows_request_since, provider_key_health_score, provider_key_rpm_allows_request,
ProviderKeyHealthBucket, provider_key_rpm_allows_request_since, ProviderKeyHealthBucket,
}; };
fn stored_candidate( fn stored_candidate(
@@ -1334,6 +1373,30 @@ mod tests {
assert!(!is_provider_key_circuit_open(&key, "openai:responses")); assert!(!is_provider_key_circuit_open(&key, "openai:responses"));
} }
#[test]
fn provider_key_circuit_open_at_allows_probe_after_rfc3339_deadline() {
let key = provider_catalog_key("key-a").with_health_fields(
None,
Some(serde_json::json!({
"openai:chat": {
"open": true,
"next_probe_at": "2026-05-24T14:45:27Z"
}
})),
);
assert!(is_provider_key_circuit_open_at(
&key,
"openai:chat",
1_779_633_926
));
assert!(!is_provider_key_circuit_open_at(
&key,
"openai:chat",
1_779_633_927
));
}
#[test] #[test]
fn aggregates_provider_key_health_score_with_lower_bound_strategy() { fn aggregates_provider_key_health_score_with_lower_bound_strategy() {
let key = provider_catalog_key("key-a").with_health_fields( let key = provider_catalog_key("key-a").with_health_fields(
+9 -8
View File
@@ -27,14 +27,15 @@ pub use candidate::{
SchedulerPriorityMode, SchedulerPriorityMode,
}; };
pub use health::{ pub use health::{
aggregate_provider_key_health_score, count_recent_active_requests_for_api_key, aggregate_provider_key_health_score, any_provider_key_circuit_open_at,
count_recent_active_requests_for_provider, count_recent_active_requests_for_provider_key, count_recent_active_requests_for_api_key, count_recent_active_requests_for_provider,
count_recent_rpm_requests_for_provider_key, count_recent_rpm_requests_for_provider_key_since, count_recent_active_requests_for_provider_key, count_recent_rpm_requests_for_provider_key,
effective_provider_key_health_score, effective_provider_key_rpm_limit, count_recent_rpm_requests_for_provider_key_since, effective_provider_key_health_score,
is_candidate_in_recent_failure_cooldown, is_provider_key_circuit_open, effective_provider_key_rpm_limit, is_candidate_in_recent_failure_cooldown,
is_provider_key_circuit_open_at, provider_key_health_bucket, provider_key_health_score, is_provider_key_circuit_open, is_provider_key_circuit_open_at,
provider_key_rpm_allows_request, provider_key_rpm_allows_request_since, provider_key_circuit_payload_is_active_open_at, provider_key_health_bucket,
ProviderKeyHealthBucket, PROVIDER_KEY_RPM_WINDOW_SECS, provider_key_health_score, provider_key_rpm_allows_request,
provider_key_rpm_allows_request_since, ProviderKeyHealthBucket, PROVIDER_KEY_RPM_WINDOW_SECS,
}; };
pub use model::{ pub use model::{
candidate_model_names, extract_global_priority_for_format, matches_model_mapping, candidate_model_names, extract_global_priority_for_format, matches_model_mapping,