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

Clarify API key concurrency skip reasons

Browse Source
This commit is contained in:
fawney19
2026-05-27 17:00:09 +08:00
parent d77a572dc7
commit 42e723ff7c
9 changed files with 96 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/aether-gateway/src/ai_serving/planner/candidate_materialization.rs
+2 -2
View File
@@ -42,7 +42,7 @@ use crate::clock::current_unix_ms;
use crate::dispatch::refs::dispatch_ref_for_local_candidate;
use crate::handlers::shared::provider_pool::admin_provider_pool_config_from_config_value;
use crate::orchestration::{local_attempt_slot_count, ExecutionAttemptIdentity};
use crate::scheduler::candidate::API_KEY_CONCURRENCY_LIMIT_SKIP_REASON;
use crate::scheduler::candidate::is_auth_api_key_concurrency_limit_skip_reason;
use crate::scheduler::config::SchedulerSchedulingMode;
use crate::{AppState, GatewayError};
@@ -988,7 +988,7 @@ fn page_is_exact_auth_api_key_concurrency_limited(
&& page
.skipped_candidates
.iter()
.all(|skipped| skipped.skip_reason == API_KEY_CONCURRENCY_LIMIT_SKIP_REASON)
.all(|skipped| is_auth_api_key_concurrency_limit_skip_reason(skipped.skip_reason))
}
async fn pop_attempt_from_items(
apps/aether-gateway/src/handlers/proxy/mod.rs
+21 -14
View File
@@ -55,6 +55,10 @@ use crate::headers::{
should_skip_request_header, RequestBodyNormalizationError,
};
use crate::router::RequestAdmissionError;
use crate::scheduler::candidate::{
is_auth_api_key_concurrency_limit_skip_reason, AUTH_API_KEY_CONCURRENCY_LIMIT_SKIP_REASON,
LEGACY_API_KEY_CONCURRENCY_LIMIT_SKIP_REASON,
};
use crate::scheduler::config::{read_scheduler_ordering_config, SchedulerSchedulingMode};
use crate::{
AppState, FrontdoorUserRpmOutcome, GatewayError, GatewayFallbackMetricKind,
@@ -92,7 +96,7 @@ const LOCAL_PROXY_PASSTHROUGH_REMOVED_DETAIL: &str =
const LOCAL_EXECUTION_LOOP_DETECTED_DETAIL: &str =
"Gateway detected an execution runtime request loop back into the local frontdoor";
const AUTH_API_KEY_CONCURRENCY_LIMIT_REACHED_DETAIL: &str =
"当前 API Key 并发请求数已达上限，请稍后重试";
"当前调用方 API Key 并发请求数已达上限，请稍后重试";
const REQUEST_BODY_READ_TIMEOUT_DETAIL: &str =
"Request body read timed out before the gateway could route the request";
const REQUEST_BODY_READ_FAILED_DETAIL: &str = "Failed to read request body";
@@ -1790,7 +1794,9 @@ pub(crate) async fn proxy_request(
let auth_api_key_concurrency_limited = diagnostic_is_auth_api_key_concurrency_limited(
local_execution_runtime_miss_diagnostic.as_ref(),
) || local_execution_runtime_miss_context
.all_candidates_skipped_for_reason("api_key_concurrency_limit_reached");
.all_candidates_skipped_for_reason(AUTH_API_KEY_CONCURRENCY_LIMIT_SKIP_REASON)
|| local_execution_runtime_miss_context
.all_candidates_skipped_for_reason(LEGACY_API_KEY_CONCURRENCY_LIMIT_SKIP_REASON);
let local_execution_runtime_miss_detail = local_execution_runtime_miss_detail(
control_decision,
local_execution_runtime_miss_diagnostic.as_ref(),
@@ -1917,7 +1923,7 @@ pub(crate) async fn proxy_request(
.map(ToOwned::to_owned)
.or_else(|| {
auth_api_key_concurrency_limited
.then_some("api_key_concurrency_limit_reached".to_string())
.then_some(AUTH_API_KEY_CONCURRENCY_LIMIT_SKIP_REASON.to_string())
});
if let Some(reason) = local_execution_runtime_miss_reason {
response.headers_mut().insert(
@@ -2142,7 +2148,9 @@ fn local_execution_runtime_miss_skip_reasons_summary(
fn local_execution_runtime_miss_skip_reason_label(reason: &str) -> &str {
match reason {
"api_key_concurrency_limit_reached" => "API Key 并发已达上限",
"auth_api_key_concurrency_limit_reached" | "api_key_concurrency_limit_reached" => {
"调用方 API Key 并发已达上限"
}
"auth_channel_mismatch" => "认证通道不匹配",
"auth_snapshot_missing" => "API Key 本地执行配置缺失",
"endpoint_api_format_changed" => "端点 API 格式已变更",
@@ -2157,7 +2165,9 @@ fn local_execution_runtime_miss_skip_reason_label(reason: &str) -> &str {
"pool_cost_limit_reached" => "池内账号成本额度已用尽",
"pool_group_exhausted" => "池化提供商没有可调度账号",
"pool_key_lease_busy" => "池内账号正被其他请求占用",
"provider_concurrency_limit_reached" => "上游提供商并发已达上限",
"provider_inactive" => "提供商未启用",
"provider_key_concurrency_limit_reached" => "上游账号并发已达上限",
"provider_request_body_missing" => "无法构建上游请求体",
"provider_request_body_build_failed" => "上游请求体转换失败",
"transport_api_format_mismatch" => "传输层 API 格式不匹配",
@@ -2217,15 +2227,12 @@ fn diagnostic_is_auth_api_key_concurrency_limited(
let Some(diagnostic) = diagnostic else {
return false;
};
diagnostic.reason == "api_key_concurrency_limit_reached"
is_auth_api_key_concurrency_limit_skip_reason(diagnostic.reason.as_str())
|| (diagnostic.reason == "all_candidates_skipped"
&& diagnostic.skip_reasons.len() == 1
&& diagnostic
.skip_reasons
.get("api_key_concurrency_limit_reached")
.copied()
.unwrap_or(0)
> 0)
&& diagnostic.skip_reasons.iter().any(|(reason, count)| {
is_auth_api_key_concurrency_limit_skip_reason(reason.as_str()) && *count > 0
}))
}
fn local_execution_runtime_miss_route_detail(
@@ -2514,7 +2521,7 @@ mod tests {
let diagnostic = LocalExecutionRuntimeMissDiagnostic {
reason: "all_candidates_skipped".to_string(),
skip_reasons: std::collections::BTreeMap::from([(
"api_key_concurrency_limit_reached".to_string(),
"auth_api_key_concurrency_limit_reached".to_string(),
1,
)]),
requested_model: Some("gpt-5.4".to_string()),
@@ -2526,7 +2533,7 @@ mod tests {
assert_eq!(
detail.as_deref(),
Some("当前 API Key 并发请求数已达上限，请稍后重试")
Some("当前调用方 API Key 并发请求数已达上限，请稍后重试")
);
assert!(diagnostic_is_auth_api_key_concurrency_limited(Some(
&diagnostic
@@ -2557,7 +2564,7 @@ mod tests {
assert_eq!(
detail.as_deref(),
Some("当前 API Key 并发请求数已达上限，请稍后重试")
Some("当前调用方 API Key 并发请求数已达上限，请稍后重试")
);
}
}
apps/aether-gateway/src/scheduler/candidate/mod.rs
+3 -1
View File
@@ -32,7 +32,9 @@ use sha2::{Digest, Sha256};
use std::collections::BTreeMap;
pub(crate) use self::selection::{
SchedulerSkippedCandidate, API_KEY_CONCURRENCY_LIMIT_SKIP_REASON,
is_auth_api_key_concurrency_limit_skip_reason, SchedulerSkippedCandidate,
API_KEY_CONCURRENCY_LIMIT_SKIP_REASON, AUTH_API_KEY_CONCURRENCY_LIMIT_SKIP_REASON,
LEGACY_API_KEY_CONCURRENCY_LIMIT_SKIP_REASON,
};
use crate::data::auth::GatewayAuthApiKeySnapshot;
apps/aether-gateway/src/scheduler/candidate/selection.rs
+14 -2
View File
@@ -20,7 +20,19 @@ pub(crate) struct SchedulerSkippedCandidate {
pub(crate) skip_reason: &'static str,
}
pub(crate) const API_KEY_CONCURRENCY_LIMIT_SKIP_REASON: &str = "api_key_concurrency_limit_reached";
pub(crate) const AUTH_API_KEY_CONCURRENCY_LIMIT_SKIP_REASON: &str =
"auth_api_key_concurrency_limit_reached";
pub(crate) const LEGACY_API_KEY_CONCURRENCY_LIMIT_SKIP_REASON: &str =
"api_key_concurrency_limit_reached";
pub(crate) const API_KEY_CONCURRENCY_LIMIT_SKIP_REASON: &str =
AUTH_API_KEY_CONCURRENCY_LIMIT_SKIP_REASON;
pub(crate) fn is_auth_api_key_concurrency_limit_skip_reason(reason: &str) -> bool {
matches!(
reason.trim(),
AUTH_API_KEY_CONCURRENCY_LIMIT_SKIP_REASON | LEGACY_API_KEY_CONCURRENCY_LIMIT_SKIP_REASON
)
}
pub(super) fn is_exact_all_skipped_by_auth_limit(
selected: &[SchedulerMinimalCandidateSelectionCandidate],
@@ -30,7 +42,7 @@ pub(super) fn is_exact_all_skipped_by_auth_limit(
&& !skipped.is_empty()
&& skipped
.iter()
.all(|candidate| candidate.skip_reason == API_KEY_CONCURRENCY_LIMIT_SKIP_REASON)
.all(|candidate| is_auth_api_key_concurrency_limit_skip_reason(candidate.skip_reason))
}
#[cfg_attr(not(test), allow(dead_code))]
apps/aether-gateway/src/tests/ai_execute/sync/cli.rs
+3 -3
View File
@@ -1249,12 +1249,12 @@ async fn gateway_returns_concurrency_limited_after_wait_budget_expires_for_opena
.headers()
.get(LOCAL_EXECUTION_RUNTIME_MISS_REASON_HEADER)
.and_then(|value| value.to_str().ok()),
Some("api_key_concurrency_limit_reached")
Some("auth_api_key_concurrency_limit_reached")
);
let payload: serde_json::Value = response.json().await.expect("body should parse");
assert_eq!(
payload["error"]["message"],
serde_json::Value::String("当前 API Key 并发请求数已达上限，请稍后重试".to_string())
serde_json::Value::String("当前调用方 API Key 并发请求数已达上限，请稍后重试".to_string())
);
let stored_candidates = request_candidate_repository
@@ -1265,7 +1265,7 @@ async fn gateway_returns_concurrency_limited_after_wait_budget_expires_for_opena
assert_eq!(stored_candidates[0].status, RequestCandidateStatus::Skipped);
assert_eq!(
stored_candidates[0].skip_reason.as_deref(),
Some("api_key_concurrency_limit_reached")
Some("auth_api_key_concurrency_limit_reached")
);
assert_eq!(
*execution_runtime_hits.lock().expect("mutex should lock"),
apps/aether-gateway/src/tests/architecture/runtime_and_security.rs
+3 -3
View File
@@ -371,7 +371,7 @@ fn scheduler_candidate_runtime_paths_depend_on_scheduler_core_and_state_trait()
"candidate/mod.rs should not own the core selectable-candidate collector anymore"
);
assert!(
!candidate_mod.contains("auth_api_key_concurrency_limit_reached"),
!candidate_mod.contains("auth_api_key_concurrency_limit_reached("),
"candidate/mod.rs should not own the core auth api key concurrency helper anymore"
);
assert!(
@@ -463,7 +463,7 @@ fn scheduler_candidate_runtime_paths_depend_on_scheduler_core_and_state_trait()
"read_provider_key_rpm_states(",
"candidate_is_selectable_with_runtime_state",
"collect_selectable_candidates_from_keys",
"auth_api_key_concurrency_limit_reached",
"auth_api_key_concurrency_limit_reached(",
"build_provider_concurrent_limit_map(",
"reorder_candidates_by_scheduler_health",
] {
@@ -487,7 +487,7 @@ fn scheduler_candidate_runtime_paths_depend_on_scheduler_core_and_state_trait()
"candidate/runtime.rs should keep affinity out of runtime eligibility checks"
);
assert!(
runtime.contains("auth_api_key_concurrency_limit_reached"),
runtime.contains("auth_api_key_concurrency_limit_reached("),
"candidate/runtime.rs should depend on core auth api key concurrency helper"
);
assert!(