i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-10 21:20:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/pr/557'

Browse Source
This commit is contained in:
fawney19
2026-05-25 01:08:22 +08:00
parent d4ae6e0e64 480579a0d5
commit cc3ce8b5d7
21 changed files with 291 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -432,6 +432,7 @@ dependencies = [
"aether-contracts",
"aether-data-contracts",
"aether-wallet",
"chrono",
"regex",
"serde",
"serde_json",
apps/aether-gateway/src/ai_serving/planner/pool_scores.rs
+71 -2
View File
@@ -6,6 +6,7 @@ use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogKe
use aether_pool_core::{
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 crate::handlers::shared::{provider_key_health_summary, provider_key_status_snapshot_payload};
@@ -98,7 +99,8 @@ fn provider_key_score_input(
.as_object()
.and_then(|snapshot| snapshot.get("account"))
.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
.health_by_format
.as_ref()
@@ -125,7 +127,7 @@ fn provider_key_score_input(
.and_then(Value::as_bool)
.unwrap_or(false),
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(),
error_count: key.error_count.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
}
#[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);
}
}
apps/aether-gateway/src/handlers/admin/endpoint/health_builders/keys.rs
+15 -10
View File
@@ -1,6 +1,9 @@
use crate::handlers::admin::request::AdminAppState;
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 std::time::{SystemTime, UNIX_EPOCH};
@@ -18,6 +21,10 @@ pub(crate) async fn build_admin_key_health_payload(
.await
.ok()
.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
.read_provider_catalog_providers_by_ids(std::slice::from_ref(&key.provider_id))
.await
@@ -81,10 +88,10 @@ pub(crate) async fn build_admin_key_health_payload(
.and_then(|value| value.get("last_failure_at"))
.cloned()
.unwrap_or(serde_json::Value::Null);
payload["circuit_breaker_open"] = json!(circuit_data
.and_then(|value| value.get("open"))
.and_then(serde_json::Value::as_bool)
.unwrap_or(false));
payload["circuit_breaker_open"] =
json!(circuit_data.is_some_and(
|value| provider_key_circuit_payload_is_active_open_at(value, now_unix_secs)
));
payload["circuit_breaker_open_at"] = circuit_data
.and_then(|value| value.get("open_at"))
.cloned()
@@ -168,11 +175,9 @@ pub(crate) async fn build_admin_key_health_payload(
.reduce(f64::min)
.unwrap_or(1.0);
let any_circuit_open = formats_payload.values().any(|value| {
value
.get("circuit_breaker")
.and_then(|circuit| circuit.get("open"))
.and_then(serde_json::Value::as_bool)
.unwrap_or(false)
value.get("circuit_breaker").is_some_and(|circuit| {
provider_key_circuit_payload_is_active_open_at(circuit, now_unix_secs)
})
});
payload["key_health_score"] = json!(key_health_score);
apps/aether-gateway/src/handlers/admin/endpoint/health_builders/status.rs
+11 -16
View File
@@ -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::provider_key_auth::provider_key_effective_api_formats;
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 std::collections::{BTreeMap, BTreeSet};
use std::time::{SystemTime, UNIX_EPOCH};
@@ -99,7 +101,9 @@ pub(crate) async fn build_admin_endpoint_health_status_payload(
.entry(api_format.clone())
.or_default()
.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 =
provider_key_health_score(&key, &api_format).unwrap_or(1.0);
active_keys_by_format
@@ -229,6 +233,10 @@ pub(crate) async fn build_admin_health_summary_payload(
return None;
}
let now_unix_secs = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_secs())
.unwrap_or_default();
let providers = state
.list_provider_catalog_providers(false)
.await
@@ -286,20 +294,7 @@ pub(crate) async fn build_admin_health_summary_payload(
.count();
let circuit_open_keys = keys
.iter()
.filter(|key| {
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)
})
.filter(|key| any_provider_key_circuit_open_at(key, now_unix_secs))
.count();
Some(json!({
apps/aether-gateway/src/handlers/admin/model/routing.rs
+13 -5
View File
@@ -8,10 +8,12 @@ use aether_data_contracts::repository::provider_catalog::{
StoredProviderCatalogEndpoint, StoredProviderCatalogKey,
};
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 std::collections::BTreeMap;
use std::time::{SystemTime, UNIX_EPOCH};
use uuid::Uuid;
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()
.and_then(|value| value.as_bool())
.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
.config
@@ -163,9 +169,11 @@ pub(crate) async fn build_admin_global_model_routing_payload(
entries
.iter()
.filter_map(|(api_format, value)| {
value.get("open")
.and_then(serde_json::Value::as_bool)
.filter(|is_open| *is_open)
provider_key_circuit_payload_is_active_open_at(
value,
now_unix_secs,
)
.then_some(())
.map(|_| api_format.clone())
})
.collect::<Vec<_>>()
@@ -188,7 +196,7 @@ pub(crate) async fn build_admin_global_model_routing_payload(
"effective_rpm": effective_rpm,
"allowed_models": allowed_models,
"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,
"next_probe_at": next_probe_at,
});
apps/aether-gateway/src/handlers/admin/observability/monitoring/resilience/snapshot.rs
+8 -7
View File
@@ -1,6 +1,6 @@
use super::super::usage_helpers::admin_monitoring_usage_is_error;
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 aether_data_contracts::repository::{
provider_catalog::StoredProviderCatalogKey, usage::UsageMonitoringErrorListQuery,
@@ -99,7 +99,7 @@ pub(super) async fn build_admin_monitoring_resilience_snapshot(
last_failure_at,
circuit_breaker_open,
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 {
degraded_keys += 1;
}
@@ -110,11 +110,12 @@ pub(super) async fn build_admin_monitoring_resilience_snapshot(
let open_formats = circuit_by_format
.iter()
.filter_map(|(api_format, value)| {
value
.get("open")
.and_then(serde_json::Value::as_bool)
.filter(|open| *open)
.map(|_| api_format.clone())
aether_scheduler_core::provider_key_circuit_payload_is_active_open_at(
value,
now.timestamp().max(0) as u64,
)
.then_some(())
.map(|_| api_format.clone())
})
.collect::<Vec<_>>();
apps/aether-gateway/src/handlers/admin/observability/monitoring/tests/mod.rs
+2 -1
View File
@@ -1237,7 +1237,8 @@ async fn admin_monitoring_circuit_history_returns_local_payload() {
"openai:chat": {
"open": true,
"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": "错误率过高"
}
})),
apps/aether-gateway/src/handlers/admin/provider/pool_admin/payloads.rs
+4 -8
View File
@@ -13,6 +13,7 @@ use aether_data_contracts::repository::provider_catalog::{
StoredProviderCatalogEndpoint, StoredProviderCatalogKey,
};
use aether_data_contracts::repository::usage::StoredProviderApiKeyWindowUsageSummary;
use aether_scheduler_core::provider_key_circuit_payload_is_active_open_at;
use serde_json::json;
use std::collections::BTreeMap;
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
.as_ref()
.and_then(serde_json::Value::as_object)
.map(|formats| {
formats
.values()
.filter_map(serde_json::Value::as_object)
.any(|item| {
item.get("open")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false)
})
.any(|item| provider_key_circuit_payload_is_active_open_at(item, now_unix_secs))
})
.unwrap_or(false)
}
@@ -1032,7 +1028,7 @@ pub(super) fn build_admin_pool_key_payload(
.as_ref()
.and_then(|_| runtime.cooldown_ttl_by_key.get(&key.id).copied());
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 account_quota_exhausted = pool_config
.as_ref()
apps/aether-gateway/src/handlers/admin/provider/query/models/model_test.rs
+6 -4
View File
@@ -68,6 +68,7 @@ use aether_model_fetch::{
aggregate_models_for_cache, fetch_models_from_transports, json_string_list,
preset_models_for_provider, selected_models_fetch_endpoints,
};
use aether_scheduler_core::provider_key_circuit_payload_is_active_open_at;
use axum::{
body::{to_bytes, Body},
http::{self, HeaderMap, HeaderName, HeaderValue},
@@ -920,6 +921,7 @@ fn provider_query_test_key_sort_key(
provider_type: &str,
key: &StoredProviderCatalogKey,
endpoint_api_format: &str,
now_unix_secs: u64,
) -> (u8, u8, i32, u64, i32) {
let quota_exhausted =
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()
.and_then(Value::as_object)
.and_then(|value| value.get(endpoint_api_format))
.and_then(Value::as_object)
.and_then(|value| value.get("open"))
.and_then(Value::as_bool)
.unwrap_or(false);
.is_some_and(|value| provider_key_circuit_payload_is_active_open_at(value, now_unix_secs));
let health_score = key
.health_by_format
.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)
})
.collect::<Vec<_>>();
let now_unix_secs = current_unix_ms() / 1000;
let candidates = if test_mode.eq_ignore_ascii_case("pool") {
if let Some(pool_config) =
@@ -1411,6 +1411,7 @@ async fn provider_query_build_kiro_test_candidates(
provider.provider_type.as_str(),
key,
&endpoint.api_format,
now_unix_secs,
)
});
keys.into_iter()
@@ -1428,6 +1429,7 @@ async fn provider_query_build_kiro_test_candidates(
provider.provider_type.as_str(),
key,
&endpoint.api_format,
now_unix_secs,
)
});
keys.into_iter()
apps/aether-gateway/src/handlers/admin/shared/mod.rs
+4 -3
View File
@@ -13,7 +13,8 @@ pub(crate) use crate::handlers::shared::{
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,
parse_catalog_auth_config_json, provider_catalog_key_supports_format,
provider_key_health_summary, provider_key_status_snapshot_payload, query_param_bool,
query_param_optional_bool, query_param_value, take_secret_prefix, take_secret_suffix,
unix_secs_to_rfc3339, OFFICIAL_EXTERNAL_MODEL_PROVIDERS,
provider_key_health_summary, provider_key_health_summary_at,
provider_key_status_snapshot_payload, query_param_bool, query_param_optional_bool,
query_param_value, take_secret_prefix, take_secret_suffix, unix_secs_to_rfc3339,
OFFICIAL_EXTERNAL_MODEL_PROVIDERS,
};
apps/aether-gateway/src/handlers/public/system_modules_helpers/keys_grouped.rs
+5 -4
View File
@@ -2,6 +2,7 @@ use super::enabled_key_capability_short_names;
use crate::handlers::shared::unix_secs_to_rfc3339;
use crate::provider_key_auth::provider_key_effective_api_formats;
use crate::AppState;
use aether_scheduler_core::provider_key_circuit_payload_is_active_open_at;
use serde_json::json;
use std::collections::{BTreeMap, HashMap};
use std::time::{SystemTime, UNIX_EPOCH};
@@ -173,10 +174,10 @@ pub(crate) async fn build_admin_keys_grouped_by_format_payload(
.get("health_score")
.and_then(serde_json::Value::as_f64)
.unwrap_or(1.0),
"circuit_breaker_open": format_circuit
.get("open")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false),
"circuit_breaker_open": provider_key_circuit_payload_is_active_open_at(
&format_circuit,
now_unix_secs,
),
"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)),
"updated_at": unix_secs_to_rfc3339(key.updated_at_unix_secs.unwrap_or(now_unix_secs)),
apps/aether-gateway/src/handlers/shared/catalog.rs
+37 -13
View File
@@ -13,6 +13,7 @@ use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogKe
use aether_provider_pool::{
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 std::borrow::Cow;
use std::time::{SystemTime, UNIX_EPOCH};
@@ -1778,6 +1779,39 @@ pub(crate) fn provider_key_health_summary(
Option<String>,
bool,
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
.health_by_format
@@ -1820,12 +1854,7 @@ pub(crate) fn provider_key_health_summary(
}
}
let any_circuit_open = circuit_by_format.values().any(|value| {
value
.get("open")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false)
});
let any_circuit_open = circuit_by_format.values().any(circuit_is_open);
(
if health_by_format.is_empty() {
@@ -1974,15 +2003,10 @@ pub(crate) fn build_admin_provider_key_response(
last_failure_at,
circuit_breaker_open,
circuit_by_format,
) = provider_key_health_summary(key);
) = provider_key_health_summary_at(key, now_unix_secs);
let circuit_sample = circuit_by_format
.values()
.find(|value| {
value
.get("open")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false)
})
.find(|value| provider_key_circuit_payload_is_active_open_at(value, now_unix_secs))
.or_else(|| circuit_by_format.values().next());
let is_adaptive = key.rpm_limit.is_none();
let effective_limit = if is_adaptive {
apps/aether-gateway/src/handlers/shared/mod.rs
+3 -2
View File
@@ -26,8 +26,9 @@ pub(crate) use self::catalog::{
default_provider_key_status_snapshot, effective_catalog_encryption_key,
encrypt_catalog_secret_with_fallbacks, masked_catalog_api_key, parse_catalog_auth_config_json,
provider_catalog_key_supports_format, provider_key_health_summary,
provider_key_status_snapshot_payload, sync_provider_key_oauth_status_snapshot,
sync_provider_key_quota_status_snapshot, take_secret_prefix, take_secret_suffix,
provider_key_health_summary_at, provider_key_status_snapshot_payload,
sync_provider_key_oauth_status_snapshot, sync_provider_key_quota_status_snapshot,
take_secret_prefix, take_secret_suffix,
};
pub(crate) use self::email_templates::{
admin_email_template_definition, admin_email_template_html_key,
apps/aether-gateway/src/tests/control/admin/endpoints/keys.rs
+2 -2
View File
@@ -241,8 +241,8 @@ async fn gateway_provider_keys_expose_circuit_breaker_and_recover_clears_it() {
"open": true,
"open_at": "2026-03-26T12:00:00+00:00",
"reason": "consecutive_failures_8",
"next_probe_at": "2026-03-26T12:01:00+00:00",
"next_probe_at_unix_secs": 1774526460u64,
"next_probe_at": "2099-03-26T12:01:00+00:00",
"next_probe_at_unix_secs": 4078209660u64,
"probe_interval_minutes": 1,
"max_probe_interval_minutes": 32,
"half_open_until": null,
apps/aether-gateway/src/tests/control/admin/health_access.rs
+3 -3
View File
@@ -346,7 +346,7 @@ async fn gateway_handles_admin_key_health_locally_with_trusted_admin_principal()
Some(json!({"openai:chat": {
"open": true,
"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_successes": 1,
"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"],
"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_failures"], 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": {
"open": true,
"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_successes": 0,
"half_open_failures": 1
apps/aether-gateway/src/tests/control/admin/models/global.rs
+2 -2
View File
@@ -789,7 +789,7 @@ async fn gateway_handles_admin_global_model_routing_locally_with_trusted_admin_p
"openai:chat": {"health_score": 0.66}
}));
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(
@@ -907,7 +907,7 @@ async fn gateway_handles_admin_global_model_routing_locally_with_trusted_admin_p
openai_keys[0]["circuit_breaker_formats"],
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"]
.as_array()
apps/aether-gateway/src/tests/control/admin/monitoring.rs
+2 -1
View File
@@ -2040,7 +2040,8 @@ async fn gateway_handles_admin_monitoring_resilience_circuit_history_locally_wit
"openai:chat": {
"open": true,
"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": "错误率过高"
}
})),
crates/aether-admin/src/provider/pool.rs
+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 {
let now_unix_secs = Utc::now().timestamp().max(0) as u64;
key.circuit_breaker_by_format
.as_ref()
.and_then(Value::as_object)
.map(|formats| {
formats
.values()
.filter_map(Value::as_object)
.any(|item| item.get("open").and_then(Value::as_bool).unwrap_or(false))
.any(|item| admin_pool_circuit_payload_active_open_at(item, now_unix_secs))
})
.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> {
Utc.timestamp_opt(unix_secs as i64, 0)
.single()
crates/aether-scheduler-core/Cargo.toml
+1
View File
@@ -11,6 +11,7 @@ aether-ai-formats.workspace = true
aether-contracts.workspace = true
aether-data-contracts.workspace = true
aether-wallet.workspace = true
chrono.workspace = true
regex.workspace = true
serde.workspace = true
serde_json.workspace = true
crates/aether-scheduler-core/src/health.rs
+69 -6
View File
@@ -294,10 +294,33 @@ pub fn is_provider_key_circuit_open_at(
.as_ref()
.and_then(serde_json::Value::as_object)
.and_then(|values| values.get(api_format))
.and_then(serde_json::Value::as_object)
else {
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
.get("open")
.and_then(serde_json::Value::as_bool)
@@ -305,10 +328,26 @@ pub fn is_provider_key_circuit_open_at(
{
return false;
}
payload
if let Some(next_probe_at) = payload
.get("next_probe_at_unix_secs")
.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(
@@ -641,9 +680,9 @@ mod tests {
count_recent_rpm_requests_for_provider_key,
count_recent_rpm_requests_for_provider_key_since, effective_provider_key_health_score,
effective_provider_key_rpm_limit, is_candidate_in_recent_failure_cooldown,
is_provider_key_circuit_open, provider_key_health_bucket, provider_key_health_score,
provider_key_rpm_allows_request, provider_key_rpm_allows_request_since,
ProviderKeyHealthBucket,
is_provider_key_circuit_open, is_provider_key_circuit_open_at, provider_key_health_bucket,
provider_key_health_score, provider_key_rpm_allows_request,
provider_key_rpm_allows_request_since, ProviderKeyHealthBucket,
};
fn stored_candidate(
@@ -1334,6 +1373,30 @@ mod tests {
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]
fn aggregates_provider_key_health_score_with_lower_bound_strategy() {
let key = provider_catalog_key("key-a").with_health_fields(
crates/aether-scheduler-core/src/lib.rs
+9 -8
View File
@@ -27,14 +27,15 @@ pub use candidate::{
SchedulerPriorityMode,
};
pub use health::{
aggregate_provider_key_health_score, count_recent_active_requests_for_api_key,
count_recent_active_requests_for_provider, count_recent_active_requests_for_provider_key,
count_recent_rpm_requests_for_provider_key, count_recent_rpm_requests_for_provider_key_since,
effective_provider_key_health_score, effective_provider_key_rpm_limit,
is_candidate_in_recent_failure_cooldown, is_provider_key_circuit_open,
is_provider_key_circuit_open_at, provider_key_health_bucket, provider_key_health_score,
provider_key_rpm_allows_request, provider_key_rpm_allows_request_since,
ProviderKeyHealthBucket, PROVIDER_KEY_RPM_WINDOW_SECS,
aggregate_provider_key_health_score, any_provider_key_circuit_open_at,
count_recent_active_requests_for_api_key, count_recent_active_requests_for_provider,
count_recent_active_requests_for_provider_key, count_recent_rpm_requests_for_provider_key,
count_recent_rpm_requests_for_provider_key_since, effective_provider_key_health_score,
effective_provider_key_rpm_limit, is_candidate_in_recent_failure_cooldown,
is_provider_key_circuit_open, is_provider_key_circuit_open_at,
provider_key_circuit_payload_is_active_open_at, provider_key_health_bucket,
provider_key_health_score, provider_key_rpm_allows_request,
provider_key_rpm_allows_request_since, ProviderKeyHealthBucket, PROVIDER_KEY_RPM_WINDOW_SECS,
};
pub use model::{
candidate_model_names, extract_global_priority_for_format, matches_model_mapping,