refactor(architecture): consolidate usage and provider contracts

Move Dashboard quota projection to the backend, share pagination request state, and fix independent usage loads while keeping trend switches free of loading overlays.

Stream billing facts within the query budget, share pending authorization and usage validation, and separate Core value ownership and response observations. Remove the unused theme helper and strengthen dependency checks.

Validated with workspace integration tests against isolated PostgreSQL and Redis, strict Clippy, Rust formatting, and frontend typecheck, lint and build.

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-09-05 05:29:12 +08:00
co-authored by Codex
parent 93a34b0f56
commit 2585d4cd7d
72 changed files with 1745 additions and 1261 deletions
+1
View File
@@ -1533,6 +1533,7 @@ dependencies = [
name = "gateway-store"
version = "0.1.0"
dependencies = [
"async-stream",
"async-trait",
"aws-credential-types",
"aws-sdk-s3",
+1 -1
View File
@@ -26,7 +26,7 @@ tokio = { version = "1.52.3", default-features = false, features = ["rt-multi-th
[dev-dependencies]
config = { version = "0.15.23", default-features = false, features = ["yaml"] }
serde_json = "1"
syn = { version = "3.0.3", features = ["full"] }
syn = { version = "3.0.3", features = ["full", "visit"] }
tempfile = "3.27.0"
[lints]
+119 -20
View File
@@ -6,7 +6,7 @@ use std::{
path::{Path, PathBuf},
};
use syn::Item;
use syn::{Item, visit::Visit};
/// workspace 成员冻结清单;新增 crate 必须同步扩展本文件的依赖规则。
pub(super) const WORKSPACE_MEMBERS: &[&str] = &[
@@ -61,6 +61,82 @@ fn gateway_core_depends_on_no_http_db_redis_or_provider_crate() {
);
}
#[test]
fn core_value_owners_do_not_depend_on_execution_or_routing() {
let root = backend_root().join("crates/gateway-core/src");
for relative in super::rust_files(&root) {
let allowed: Option<&[&str]> = match relative.to_str().expect("source path") {
"validation.rs" => Some(&[]),
"identity.rs" => Some(&["validation"]),
"upstream.rs" => Some(&["validation"]),
"event.rs" => Some(&["metering", "operation", "upstream", "validation"]),
"account/store.rs" => Some(&["account", "error", "identity", "validation"]),
path if path.starts_with("policy/") => Some(&["account", "policy", "validation"]),
path if path.starts_with("account/") => Some(&["account", "identity", "validation"]),
_ => None,
};
let Some(allowed) = allowed else { continue };
let source = fs::read_to_string(root.join(&relative)).expect("read core source");
let syntax = syn::parse_file(&source).expect("parse core source");
let mut references = CrateReferences::default();
references.visit_file(&syntax);
for dependency in references.0 {
assert!(
allowed.contains(&dependency.as_str()),
"{} must not depend on crate::{dependency}",
relative.display()
);
}
}
}
#[derive(Default)]
struct CrateReferences(BTreeSet<String>);
impl<'ast> Visit<'ast> for CrateReferences {
fn visit_path(&mut self, path: &'ast syn::Path) {
let mut segments = path.segments.iter();
if segments
.next()
.is_some_and(|segment| segment.ident == "crate")
&& let Some(owner) = segments.next()
{
self.0.insert(owner.ident.to_string());
}
syn::visit::visit_path(self, path);
}
fn visit_item_use(&mut self, item: &'ast syn::ItemUse) {
self.use_tree(&item.tree, false);
}
}
impl CrateReferences {
fn use_tree(&mut self, tree: &syn::UseTree, crate_root: bool) {
match tree {
syn::UseTree::Path(path) if crate_root => {
self.0.insert(path.ident.to_string());
}
syn::UseTree::Name(name) if crate_root => {
self.0.insert(name.ident.to_string());
}
syn::UseTree::Rename(rename) if crate_root => {
self.0.insert(rename.ident.to_string());
}
syn::UseTree::Path(path) => self.use_tree(&path.tree, path.ident == "crate"),
syn::UseTree::Group(group) => {
for item in &group.items {
self.use_tree(item, crate_root);
}
}
syn::UseTree::Glob(_) if crate_root => {
self.0.insert("*".to_owned());
}
_ => {}
}
}
}
#[test]
fn gateway_protocol_has_no_workspace_dependencies() {
for name in dependency_names("crates/gateway-protocol") {
@@ -156,6 +232,46 @@ const ALLOWED_INTERNAL_EDGES: &[(&str, &str)] = &[
#[test]
fn workspace_internal_dependency_edges_match_frozen_dag() {
let metadata = cargo_metadata_json();
let actual = workspace_runtime_dependency_edges(&metadata);
let mut expected: Vec<(String, String)> = ALLOWED_INTERNAL_EDGES
.iter()
.map(|(from, to)| ((*from).to_owned(), (*to).to_owned()))
.collect();
expected.sort();
assert_eq!(
actual, expected,
"workspace internal dependency edges diverged from the frozen DAG"
);
}
#[test]
fn runtime_dependency_edges_exclude_dev_build_and_external_dependencies() {
let metadata = serde_json::json!({
"packages": [
{
"name": "gateway-api",
"dependencies": [
{ "name": "gateway-core", "kind": null },
{ "name": "gateway-store", "kind": "dev" },
{ "name": "gateway-host", "kind": "build" },
{ "name": "serde", "kind": null }
]
},
{
"name": "external-package",
"dependencies": [{ "name": "gateway-core", "kind": null }]
}
]
});
assert_eq!(
workspace_runtime_dependency_edges(&metadata),
vec![("gateway-api".to_owned(), "gateway-core".to_owned())]
);
}
fn workspace_runtime_dependency_edges(metadata: &serde_json::Value) -> Vec<(String, String)> {
let packages = metadata["packages"]
.as_array()
.expect("cargo metadata packages");
@@ -173,31 +289,14 @@ fn workspace_internal_dependency_edges_match_frozen_dag() {
if !internal_names.contains(dep_name) {
continue;
}
let is_runtime_edge = dependency["dep_kinds"]
.as_array()
.map(|kinds| {
kinds
.iter()
.any(|kind| kind["kind"].as_str() == Some("normal"))
})
.unwrap_or(true);
if is_runtime_edge {
if dependency.get("kind").expect("dependency kind").is_null() {
actual.push((name.to_owned(), dep_name.to_owned()));
}
}
}
let mut expected: Vec<(String, String)> = ALLOWED_INTERNAL_EDGES
.iter()
.map(|(from, to)| ((*from).to_owned(), (*to).to_owned()))
.collect();
actual.sort();
expected.sort();
assert_eq!(
actual, expected,
"workspace internal dependency edges diverged from the frozen DAG"
);
actual
}
#[test]
+1 -1
View File
@@ -22,6 +22,7 @@ gateway-core = { path = "../gateway-core" }
hex = "0.4.3"
rand_core = { version = "0.6.4", features = ["getrandom"] }
serde = { version = "1.0.228", features = ["derive"] }
serde_json = { version = "1.0.150", features = ["arbitrary_precision", "preserve_order"] }
secrecy = { version = "0.10.3", features = ["serde"] }
subtle = "2.6.1"
thiserror = "2.0.18"
@@ -32,7 +33,6 @@ chrono-tz = "0.10"
tokio = { version = "1.52", features = ["rt", "time", "macros", "sync"] }
[dev-dependencies]
serde_json = { version = "1.0.150", features = ["arbitrary_precision", "preserve_order"] }
sha2 = "0.11"
tempfile = "3"
tokio = { version = "1.52.3", default-features = false, features = ["macros", "rt"] }
+1
View File
@@ -304,6 +304,7 @@ pub async fn initialize(
client_distribution: Arc::new(DefaultClientDistributionService::new(client_distribution)),
observability: Arc::new(DefaultObservabilityService::new(
store.observability(),
store.accounts(),
store.settings(),
registry,
)),
@@ -491,6 +491,8 @@ pub struct DashboardAccountUsage {
///
/// `None` 表示上游未提供可比较的百分比,不应伪造为零。
pub quota_used_percent: Option<f64>,
/// Provider 返回的代表窗口;缺失时保留未知语义。
pub quota_window: Option<super::provider_credentials::ProviderQuotaWindow>,
pub models: Vec<DashboardAccountModelUsage>,
}
@@ -10,6 +10,8 @@ use gateway_core::{
account::{OpaqueProviderData, ProviderAccountId, ProviderAccountIdentity},
routing::{ProviderKind, UpstreamModelId},
};
use serde::Deserialize;
use serde_json::{Map, Value};
use uuid::Uuid;
use super::{
@@ -279,6 +281,101 @@ impl PendingAuthorizationMutation {
}
}
impl PendingAuthorizationMutation {
/// 编码 v1 中立事务字段;版本号的位置和 Provider 私有外层文档由 adapter 保持。
#[must_use]
pub fn to_storage_v1(&self) -> Map<String, Value> {
let target = match &self.target {
AuthorizationMutationTarget::Create { name } => {
serde_json::json!({"kind": "create", "name": name})
}
AuthorizationMutationTarget::Reauthorize { account_id } => {
serde_json::json!({"kind": "reauthorize", "account_id": account_id.to_string()})
}
};
let owner = match self.owner_binding.owner() {
AuthorizationOwner::AdminSession { admin_user_id } => {
serde_json::json!({"kind": "admin_session", "admin_user_id": admin_user_id})
}
AuthorizationOwner::AdminApiKey => serde_json::json!({"kind": "admin_api_key"}),
AuthorizationOwner::System => serde_json::json!({"kind": "system"}),
};
Map::from_iter([
(
"provider_kind".to_owned(),
Value::String(self.provider_kind.as_str().to_owned()),
),
("target".to_owned(), target),
("owner".to_owned(), owner),
(
"started_request_id".to_owned(),
Value::String(self.owner_binding.started_request_id().to_owned()),
),
])
}
/// 恢复 adapter 已确认版本为 v1 的中立事务字段。
///
/// # Errors
///
/// 文档字段、Provider kind 或账号身份非法时返回错误,不携带原始 pending 内容。
pub fn from_storage_v1(value: Value) -> Result<Self, AdminError> {
let invalid = || AdminError::invalid("invalid pending authorization mutation");
let document: StoredAuthorizationMutationV1 =
serde_json::from_value(value).map_err(|_| invalid())?;
let provider_kind = ProviderKind::new(document.provider_kind).map_err(|_| invalid())?;
let target = match document.target {
StoredAuthorizationTargetV1::Create { name } => {
AuthorizationMutationTarget::Create { name }
}
StoredAuthorizationTargetV1::Reauthorize { account_id } => {
AuthorizationMutationTarget::Reauthorize {
account_id: ProviderAccountId::new(account_id).map_err(|_| invalid())?,
}
}
};
let owner = match document.owner {
StoredAuthorizationOwnerV1::AdminSession { admin_user_id } => {
AuthorizationOwner::AdminSession { admin_user_id }
}
StoredAuthorizationOwnerV1::AdminApiKey => AuthorizationOwner::AdminApiKey,
StoredAuthorizationOwnerV1::System => AuthorizationOwner::System,
};
Ok(Self::new(
provider_kind,
target,
AuthorizationOwnerBinding {
owner,
started_request_id: document.started_request_id,
},
))
}
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct StoredAuthorizationMutationV1 {
provider_kind: String,
target: StoredAuthorizationTargetV1,
owner: StoredAuthorizationOwnerV1,
started_request_id: String,
}
#[derive(Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
enum StoredAuthorizationTargetV1 {
Create { name: String },
Reauthorize { account_id: String },
}
#[derive(Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
enum StoredAuthorizationOwnerV1 {
AdminSession { admin_user_id: String },
AdminApiKey,
System,
}
/// 启动 Provider OAuth Authorization Code 流程。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StartAuthorization {
@@ -776,6 +873,20 @@ impl ProviderQuota {
.map(|(_, used_percent)| used_percent)
}
/// 选择 Dashboard 展示的真实窗口;没有百分比时仍保留 Provider 的滚动窗口语义。
#[must_use]
pub fn representative_window(&self) -> Option<&ProviderQuotaWindow> {
self.representative_used_window()
.map(|(index, _)| &self.windows[index])
.or_else(|| {
self.windows
.iter()
.enumerate()
.min_by_key(|(index, window)| (quota_usage_priority(window), *index))
.map(|(_, window)| window)
})
}
/// 将已确认的账号级额度耗尽事实投影到展示窗口。
///
/// Provider 已指出具体触顶窗口时只归一化这些窗口;否则归一化 Dashboard
@@ -6,6 +6,7 @@ use std::sync::Arc;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use futures::stream::BoxStream;
use super::backup::BackupStorePorts;
use crate::model::{
@@ -289,6 +290,10 @@ pub trait AccountGroupStore: Send + Sync {
) -> AdminStoreResult<AccountGroupMutation>;
}
/// 逐条读取的已计算费用事实;消费结束或丢弃时释放查询资源。
pub type UsageCalculatedBillingStream<'a> =
BoxStream<'a, AdminStoreResult<UsageCalculatedBillingFact>>;
/// 用量、趋势、诊断与运维错误的只读能力。
#[async_trait]
pub trait ObservabilityStore: Send + Sync {
@@ -318,12 +323,13 @@ pub trait ObservabilityStore: Send + Sync {
filter: UsageFilter,
) -> AdminStoreResult<Vec<RequestMetricPoint>>;
/// 返回可由 Provider 重新校验的已计算费用事实,用于恢复标准费用趋势。
async fn usage_calculated_billing_facts(
/// 流式返回可由 Provider 重新校验的已计算费用事实,不保证顺序。
/// 查询及解码错误由流返回;调用方应逐条聚合,避免收集整个区间。
fn usage_calculated_billing_facts(
&self,
range: TimeRange,
filter: UsageFilter,
) -> AdminStoreResult<Vec<UsageCalculatedBillingFact>>;
) -> UsageCalculatedBillingStream<'_>;
async fn list_usage_records(&self, query: UsageQuery) -> AdminStoreResult<UsagePage>;
@@ -9,6 +9,7 @@ use std::{
use async_trait::async_trait;
use chrono::{DateTime, Duration, Timelike as _, Utc};
use futures::TryStreamExt;
use gateway_core::{account::ProviderAccountId, routing::ProviderKind};
use crate::{
@@ -29,7 +30,7 @@ use crate::{
},
ports::{
provider::{ProviderAdminErrorKind, ProviderAdminRegistry},
store::{ObservabilityStore, SettingsStore},
store::{AccountStore, ObservabilityStore, SettingsStore, UsageCalculatedBillingStream},
},
};
@@ -126,6 +127,7 @@ pub trait ObservabilityService: Send + Sync {
pub(crate) struct DefaultObservabilityService {
store: Arc<dyn ObservabilityStore>,
accounts: Arc<dyn AccountStore>,
settings: Arc<dyn SettingsStore>,
providers: ProviderAdminRegistry,
dashboard_queries: DashboardQueryCache,
@@ -135,11 +137,13 @@ impl DefaultObservabilityService {
#[must_use]
pub(crate) fn new(
store: Arc<dyn ObservabilityStore>,
accounts: Arc<dyn AccountStore>,
settings: Arc<dyn SettingsStore>,
providers: ProviderAdminRegistry,
) -> Self {
Self {
store,
accounts,
settings,
providers,
dashboard_queries: DashboardQueryCache::default(),
@@ -301,13 +305,21 @@ impl ObservabilityService for DefaultObservabilityService {
range: TimeRange,
filter: UsageFilter,
) -> Result<UsageInsights, AdminError> {
let (overview, trend, billing_facts) = futures::try_join!(
self.store.usage_summary(range, filter.clone()),
self.store.usage_trend(range, filter.clone()),
self.store.usage_calculated_billing_facts(range, filter),
)
.map_err(|error| map_store_error(error, "usage insights"))?;
let standard_costs = recover_standard_costs(&self.providers, billing_facts)?;
let aggregates = async {
futures::try_join!(
self.store.usage_summary(range, filter.clone()),
self.store.usage_trend(range, filter.clone()),
)
.map_err(|error| map_store_error(error, "usage insights"))
};
let ((overview, trend), standard_costs) = futures::try_join!(
aggregates,
recover_standard_costs(
&self.providers,
self.store
.usage_calculated_billing_facts(range, filter.clone()),
),
)?;
build_usage_insights(overview, trend, standard_costs)
}
@@ -590,12 +602,16 @@ struct UsageCostScenarios {
no_cache: ScenarioCosts,
}
fn recover_standard_costs(
async fn recover_standard_costs(
providers: &ProviderAdminRegistry,
facts: Vec<UsageCalculatedBillingFact>,
mut facts: UsageCalculatedBillingStream<'_>,
) -> Result<UsageCostScenarios, AdminError> {
let mut scenarios = UsageCostScenarios::default();
for fact in facts {
while let Some(fact) = facts
.try_next()
.await
.map_err(|error| map_store_error(error, "usage billing facts"))?
{
let Ok(provider_kind) = ProviderKind::new(fact.provider_kind.clone()) else {
continue;
};
@@ -708,8 +724,31 @@ fn usd_cost(costs: &[CurrencyCost]) -> Option<DecimalAmount> {
impl DefaultObservabilityService {
async fn enrich_dashboard_quotas(&self, accounts: &mut [DashboardAccountUsage]) {
if accounts.is_empty() {
return;
}
let now = Utc::now();
let ids = accounts
.iter()
.map(|account| account.account_id.clone())
.collect::<Vec<_>>();
let mut rolling_usage = self
.accounts
.load_account_usage(
TimeRange {
start: now - Duration::hours(24),
end: now,
},
&ids,
)
.await
.unwrap_or_default()
.into_iter()
.map(|usage| (usage.account_id.clone(), usage))
.collect::<HashMap<_, _>>();
let providers = self.providers.clone();
let quota_reads = accounts.iter().map(|account| {
let rolling_usage = rolling_usage.remove(&account.account_id);
let account_id = ProviderAccountId::new(account.account_id.clone()).ok();
let provider = ProviderKind::new(account.provider_kind.clone())
.ok()
@@ -722,16 +761,19 @@ impl DefaultObservabilityService {
.quota(ProviderQuotaRequest {
account_id,
refresh: false,
rolling_usage: None,
rolling_usage,
})
.await
.ok()
.and_then(|quota| quota.representative_used_percent())
}
});
let quota_used = futures::future::join_all(quota_reads).await;
for (account, used_percent) in accounts.iter_mut().zip(quota_used) {
account.quota_used_percent = used_percent;
let quotas = futures::future::join_all(quota_reads).await;
for (account, quota) in accounts.iter_mut().zip(quotas) {
if let Some(mut quota) = quota {
quota.apply_limit_reached_display();
account.quota_used_percent = quota.representative_used_percent();
account.quota_window = quota.representative_window().cloned();
}
}
}
@@ -1,3 +1,4 @@
//! 模型模块测试:镜像 src/model/ 中带逻辑的模块。
mod backup;
mod provider_credentials;
@@ -0,0 +1,134 @@
use gateway_admin::model::provider_credentials::{
AuthorizationMutationTarget, AuthorizationOwnerBinding, PendingAuthorizationMutation,
ProviderQuota, ProviderQuotaWindow, ProviderQuotaWindowRole, QuotaLocalUsageAttribution,
};
use gateway_admin::model::{MutationActor, MutationContext};
use gateway_core::account::ProviderAccountId;
use gateway_core::routing::ProviderKind;
use serde_json::{Value, json};
#[test]
fn dashboard_quota_should_preserve_unknown_and_actual_window_facts() {
let mut quota = ProviderQuota {
observed_at: None,
refresh_token_expires_at: None,
windows: Vec::new(),
limit_reached: false,
provider_data: None,
};
assert!(quota.representative_window().is_none());
assert!(quota.representative_used_percent().is_none());
let rolling = ProviderQuotaWindow {
key: "provider_rolling".to_owned(),
group: "requests".to_owned(),
label: "Provider rolling window".to_owned(),
limit_id: None,
limit_name: None,
role: Some(ProviderQuotaWindowRole::Primary),
local_usage_attribution: QuotaLocalUsageAttribution::AccountWide,
window_seconds: Some(86_400),
used_percent: None,
reset_at: None,
limit_reached: false,
local_usage: None,
provider_data: None,
};
quota.windows.push(rolling.clone());
assert_eq!(quota.representative_window(), Some(&rolling));
assert!(quota.representative_used_percent().is_none());
quota.windows[0].used_percent = Some(99.6);
quota.apply_limit_reached_display();
let actual = quota.representative_window().expect("actual window");
assert_eq!(actual.used_percent, Some(99.6));
assert!(
!actual.limit_reached,
"display rounding cannot exhaust quota"
);
assert_eq!(actual.window_seconds, Some(86_400));
assert_eq!(actual.reset_at, None);
quota.windows.insert(
0,
ProviderQuotaWindow {
key: "model_specific".to_owned(),
used_percent: Some(100.0),
local_usage_attribution: QuotaLocalUsageAttribution::Unavailable,
..rolling
},
);
assert_eq!(
quota
.representative_window()
.map(|window| window.key.as_str()),
Some("provider_rolling")
);
}
#[test]
fn pending_mutation_v1_should_round_trip_all_targets_and_owners() {
for provider in ["openai", "xai"] {
for target in [
AuthorizationMutationTarget::Create {
name: "account".to_owned(),
},
AuthorizationMutationTarget::Reauthorize {
account_id: ProviderAccountId::new("acct_1").expect("account id"),
},
] {
for actor in [
MutationActor::AdminSession {
admin_user_id: "admin-1".to_owned(),
},
MutationActor::AdminApiKey,
MutationActor::System,
] {
let expected = PendingAuthorizationMutation::new(
ProviderKind::new(provider).expect("provider"),
target.clone(),
AuthorizationOwnerBinding::from_context(&MutationContext {
actor,
request_id: "request-1".to_owned(),
}),
);
assert_eq!(
PendingAuthorizationMutation::from_storage_v1(Value::Object(
expected.to_storage_v1()
)),
Ok(expected),
);
}
}
}
}
#[test]
fn pending_mutation_v1_should_keep_existing_storage_fields_and_reject_invalid_documents() {
let legacy = json!({
"provider_kind": "openai",
"target": { "kind": "reauthorize", "account_id": "acct_1" },
"owner": { "kind": "admin_session", "admin_user_id": "admin-1" },
"started_request_id": "request-1"
});
let restored =
PendingAuthorizationMutation::from_storage_v1(legacy.clone()).expect("v1 document");
assert_eq!(Value::Object(restored.to_storage_v1()), legacy);
for (pointer, invalid) in [
("/provider_kind", json!("")),
("/target/account_id", json!("")),
("/target/kind", json!("other")),
("/owner/kind", json!("other")),
("/started_request_id", json!(null)),
] {
let mut document = legacy.clone();
*document.pointer_mut(pointer).expect("field") = invalid;
assert!(
PendingAuthorizationMutation::from_storage_v1(document).is_err(),
"{pointer}"
);
}
let mut document = legacy;
document["schema_version"] = json!(2);
assert!(PendingAuthorizationMutation::from_storage_v1(document).is_err());
}
@@ -536,13 +536,14 @@ impl ObservabilityStore for UnavailableStore {
Err(unavailable("usage trend"))
}
async fn usage_calculated_billing_facts(
fn usage_calculated_billing_facts(
&self,
_: TimeRange,
_: UsageFilter,
) -> AdminStoreResult<Vec<gateway_admin::model::observability::UsageCalculatedBillingFact>>
{
Err(unavailable("usage billing facts"))
) -> gateway_admin::ports::store::UsageCalculatedBillingStream<'_> {
Box::pin(futures::stream::once(async {
Err(unavailable("usage billing facts"))
}))
}
async fn list_usage_records(&self, _: UsageQuery) -> AdminStoreResult<UsagePage> {
@@ -2,7 +2,7 @@ use std::{
str::FromStr as _,
sync::{
Arc, Mutex,
atomic::{AtomicUsize, Ordering},
atomic::{AtomicBool, AtomicUsize, Ordering},
},
time::Duration as StdDuration,
};
@@ -537,10 +537,41 @@ async fn usage_records_should_enrich_provider_reported_totals_when_pricing_match
));
}
#[tokio::test]
async fn usage_insights_should_reject_partial_costs_when_billing_stream_fails() {
let now = Utc::now();
let range = observation_range(now);
let store = Arc::new(FixtureObservabilityStore::new(range));
store.replace_calculated_billing_facts(vec![UsageCalculatedBillingFact {
bucket_start: quarter_hour_start(now),
provider_kind: "openai".to_owned(),
upstream_model_id: "gpt-5.5".to_owned(),
service_tier: None,
input_tokens: Some(800),
output_tokens: Some(200),
cached_tokens: Some(0),
cache_write_tokens: Some(0),
total: CurrencyCost {
currency: "USD".to_owned(),
amount: "1.25".parse().expect("cost"),
},
}]);
store.billing_stream_fails.store(true, Ordering::SeqCst);
let services = observability_services_with_calculated_billing(store).await;
assert!(
services
.observability()
.usage_insights(range, UsageFilter::default())
.await
.is_err()
);
}
struct FixtureObservabilityStore {
trend: Mutex<Vec<RequestMetricPoint>>,
overview: Mutex<UsageOverview>,
calculated_billing_facts: Mutex<Vec<UsageCalculatedBillingFact>>,
billing_stream_fails: AtomicBool,
diagnostics: Mutex<Vec<DiagnosticObservation>>,
runtime_slots: Mutex<Option<DashboardRuntimeSlots>>,
summary_observed_at: Mutex<Option<DateTime<Utc>>>,
@@ -561,6 +592,7 @@ impl FixtureObservabilityStore {
providers: Vec::new(),
}),
calculated_billing_facts: Mutex::new(Vec::new()),
billing_stream_fails: AtomicBool::new(false),
diagnostics: Mutex::new(Vec::new()),
runtime_slots: Mutex::new(None),
summary_observed_at: Mutex::new(None),
@@ -663,16 +695,23 @@ impl ObservabilityStore for FixtureObservabilityStore {
Ok(self.trend.lock().expect("trend").clone())
}
async fn usage_calculated_billing_facts(
fn usage_calculated_billing_facts(
&self,
_: TimeRange,
_: UsageFilter,
) -> AdminStoreResult<Vec<UsageCalculatedBillingFact>> {
Ok(self
) -> gateway_admin::ports::store::UsageCalculatedBillingStream<'_> {
let facts = self
.calculated_billing_facts
.lock()
.expect("calculated billing facts")
.clone())
.clone();
let error = self
.billing_stream_fails
.load(Ordering::SeqCst)
.then(|| Err(super::unavailable("billing stream failed")));
Box::pin(futures::stream::iter(
facts.into_iter().map(Ok).chain(error),
))
}
async fn list_usage_records(&self, query: UsageQuery) -> AdminStoreResult<UsagePage> {
@@ -70,6 +70,7 @@ mod wire;
pub use credentials::*;
pub use handlers::{profile_avatar_response, router};
pub(super) use presenter::quota_window_view;
pub use wire::*;
use credentials::{
@@ -140,7 +140,7 @@ pub(super) fn account_quota_view(
)
}
pub(super) fn quota_window_view(window: ProviderQuotaWindow) -> AccountQuotaWindowView {
pub(crate) fn quota_window_view(window: ProviderQuotaWindow) -> AccountQuotaWindowView {
let ProviderQuotaWindow {
key,
group,
@@ -864,6 +864,28 @@ pub(crate) fn dashboard_view(
} = observation;
let mut account_usage_views = Vec::with_capacity(account_usage.len());
for credential in account_usage {
let local_requests = credential
.quota_window
.as_ref()
.filter(|window| window.used_percent.is_none())
.and_then(|window| window.local_usage.as_ref())
.map(|usage| usage.request_count);
let (metric_label, metric_value) = match local_requests {
Some(count) => (
"次数".to_owned(),
if count > 0 {
format_compact_number(count)
} else {
"—".to_owned()
},
),
None => (
"今日 Token".to_owned(),
credential
.total_tokens
.map_or_else(|| "—".to_owned(), format_compact_number),
),
};
account_usage_views.push(DashboardAccountUsageView {
id: credential.account_id.clone(),
provider: credential.provider_kind.clone(),
@@ -886,6 +908,11 @@ pub(crate) fn dashboard_view(
})
.collect(),
quota_used_percent: credential.quota_used_percent,
usage_window: credential
.quota_window
.map(crate::admin::accounts::quota_window_view),
metric_label,
metric_value,
last_used: relative_time(credential.last_used_at, range.end),
});
}
@@ -375,6 +375,9 @@ pub struct DashboardAccountUsageView {
pub request_count: u64,
pub request_buckets: Vec<DashboardAccountRequestBucketView>,
pub quota_used_percent: Option<f64>,
pub usage_window: Option<crate::admin::accounts::AccountQuotaWindowView>,
pub metric_label: String,
pub metric_value: String,
pub last_used: String,
}
@@ -875,13 +875,14 @@ impl ObservabilityStore for UnusedStore {
Err(unavailable("usage trend"))
}
async fn usage_calculated_billing_facts(
fn usage_calculated_billing_facts(
&self,
_: TimeRange,
_: UsageFilter,
) -> AdminStoreResult<Vec<gateway_admin::model::observability::UsageCalculatedBillingFact>>
{
Err(unavailable("usage billing facts"))
) -> gateway_admin::ports::store::UsageCalculatedBillingStream<'_> {
Box::pin(futures::stream::once(async {
Err(unavailable("usage billing facts"))
}))
}
async fn list_usage_records(&self, query: UsageQuery) -> AdminStoreResult<UsagePage> {
@@ -71,6 +71,9 @@ fn dashboard_account_usage_should_keep_daily_request_timeline() {
request_count: 3,
}],
quota_used_percent: None,
usage_window: None,
metric_label: "次数".to_owned(),
metric_value: "3".to_owned(),
last_used: "刚刚".to_owned(),
})
.expect("dashboard account usage");
@@ -2,6 +2,7 @@
mod error;
mod model;
pub mod scope;
mod selection;
mod store;
@@ -6,8 +6,8 @@ use std::time::{Duration, SystemTime};
use serde_json::{Map, Value};
use crate::error::{IdentifierError, validate_text};
use crate::routing::ProviderKind;
use crate::identity::ProviderKind;
use crate::validation::{IdentifierError, validate_text};
use super::CredentialError;
@@ -0,0 +1,291 @@
//! 请求认证时冻结的账号范围与目录,不依赖路由选择器。
use super::ProviderAccountId;
use crate::identity::ProviderKind;
use crate::validation::{IdentifierError, RoutingError};
use std::{
collections::{BTreeMap, BTreeSet},
fmt,
sync::Arc,
};
/// `account_groups.id` 的核心值对象。
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct AccountGroupId(String);
impl AccountGroupId {
/// 校验并创建账号分组 ID。
pub fn new(value: impl Into<String>) -> Result<Self, IdentifierError> {
let value = value.into();
let Some(suffix) = value.strip_prefix("grp_") else {
return Err(IdentifierError::MissingPrefix { expected: "grp_" });
};
if suffix.len() != 32
|| !suffix
.bytes()
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
{
return Err(IdentifierError::InvalidFormat);
}
Ok(Self(value))
}
#[must_use]
pub fn as_str(&self) -> &str {
&self.0
}
}
impl fmt::Display for AccountGroupId {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(&self.0)
}
}
/// 快照中一个账号的 Provider 与分组归属。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RuntimeAccount {
provider_kind: ProviderKind,
group_ids: Arc<BTreeSet<AccountGroupId>>,
}
impl RuntimeAccount {
#[must_use]
pub fn new(provider_kind: ProviderKind, group_ids: BTreeSet<AccountGroupId>) -> Self {
Self {
provider_kind,
group_ids: Arc::new(group_ids),
}
}
#[must_use]
pub const fn provider_kind(&self) -> &ProviderKind {
&self.provider_kind
}
#[must_use]
pub fn group_ids(&self) -> &BTreeSet<AccountGroupId> {
&self.group_ids
}
}
/// 全快照共享的账号、Provider 与分组反向索引。
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct RuntimeAccountDirectory {
accounts: BTreeMap<ProviderAccountId, RuntimeAccount>,
providers_with_accounts: BTreeSet<ProviderKind>,
providers_by_group: BTreeMap<AccountGroupId, BTreeSet<ProviderKind>>,
}
impl RuntimeAccountDirectory {
#[must_use]
pub fn new(accounts: BTreeMap<ProviderAccountId, RuntimeAccount>) -> Self {
let mut providers_with_accounts = BTreeSet::new();
let mut providers_by_group = BTreeMap::<AccountGroupId, BTreeSet<ProviderKind>>::new();
for account in accounts.values() {
providers_with_accounts.insert(account.provider_kind.clone());
for group_id in account.group_ids.iter() {
providers_by_group
.entry(group_id.clone())
.or_default()
.insert(account.provider_kind.clone());
}
}
Self {
accounts,
providers_with_accounts,
providers_by_group,
}
}
#[must_use]
pub fn account(&self, account_id: &ProviderAccountId) -> Option<&RuntimeAccount> {
self.accounts.get(account_id)
}
#[must_use]
pub fn providers_with_accounts(&self) -> &BTreeSet<ProviderKind> {
&self.providers_with_accounts
}
#[must_use]
pub fn providers_for_groups<'a>(
&self,
group_ids: impl IntoIterator<Item = &'a AccountGroupId>,
) -> BTreeSet<ProviderKind> {
group_ids
.into_iter()
.filter_map(|group_id| self.providers_by_group.get(group_id))
.flatten()
.cloned()
.collect()
}
}
/// 历史请求保存的账号范围种类。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AccountRoutingScopeKind {
All,
Groups,
}
impl AccountRoutingScopeKind {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::All => "all",
Self::Groups => "groups",
}
}
}
/// 请求开始时冻结的分组 ID 与名称。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RoutingGroupSnapshot {
id: AccountGroupId,
name: String,
}
impl RoutingGroupSnapshot {
#[must_use]
pub fn new(id: AccountGroupId, name: String) -> Self {
Self { id, name }
}
#[must_use]
pub const fn id(&self) -> &AccountGroupId {
&self.id
}
#[must_use]
pub fn name(&self) -> &str {
&self.name
}
}
/// 请求历史所需的完整、稳定账号范围快照。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AccountRoutingSnapshot {
kind: AccountRoutingScopeKind,
groups: Arc<[RoutingGroupSnapshot]>,
}
impl AccountRoutingSnapshot {
#[must_use]
pub fn all() -> Self {
Self {
kind: AccountRoutingScopeKind::All,
groups: Arc::from([]),
}
}
#[must_use]
pub fn groups(groups: Vec<RoutingGroupSnapshot>) -> Self {
Self {
kind: AccountRoutingScopeKind::Groups,
groups: Arc::from(groups),
}
}
#[must_use]
pub const fn kind(&self) -> AccountRoutingScopeKind {
self.kind
}
#[must_use]
pub fn groups_snapshot(&self) -> &[RoutingGroupSnapshot] {
&self.groups
}
}
/// Key 持久 binding 编译出的账号权限。
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ClientRoutingScope {
AllAccounts,
Restricted {
bound_groups: Arc<[RoutingGroupSnapshot]>,
enabled_group_ids: Arc<BTreeSet<AccountGroupId>>,
provider_kinds: Arc<BTreeSet<ProviderKind>>,
},
}
impl ClientRoutingScope {
#[must_use]
pub fn all_accounts() -> Self {
Self::AllAccounts
}
pub fn restricted(
bound_groups: Vec<RoutingGroupSnapshot>,
enabled_group_ids: BTreeSet<AccountGroupId>,
provider_kinds: BTreeSet<ProviderKind>,
) -> Result<Self, RoutingError> {
if bound_groups.is_empty() {
return Err(RoutingError::InvalidAccountScope);
}
Ok(Self::Restricted {
bound_groups: Arc::from(bound_groups),
enabled_group_ids: Arc::new(enabled_group_ids),
provider_kinds: Arc::new(provider_kinds),
})
}
}
/// 一次认证随 RuntimeSnapshot 冻结的账号目录与 Key 权限。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FrozenAccountScope {
directory: Arc<RuntimeAccountDirectory>,
client_scope: ClientRoutingScope,
}
impl FrozenAccountScope {
#[must_use]
pub const fn new(
directory: Arc<RuntimeAccountDirectory>,
client_scope: ClientRoutingScope,
) -> Self {
Self {
directory,
client_scope,
}
}
#[must_use]
pub fn allows(&self, account_id: &ProviderAccountId) -> bool {
let Some(account) = self.directory.account(account_id) else {
return false;
};
match &self.client_scope {
ClientRoutingScope::AllAccounts => true,
ClientRoutingScope::Restricted {
enabled_group_ids, ..
} => account
.group_ids()
.iter()
.any(|group_id| enabled_group_ids.contains(group_id)),
}
}
#[must_use]
pub fn provider_kinds(&self) -> &BTreeSet<ProviderKind> {
match &self.client_scope {
ClientRoutingScope::AllAccounts => self.directory.providers_with_accounts(),
ClientRoutingScope::Restricted { provider_kinds, .. } => provider_kinds,
}
}
#[must_use]
pub fn routing_snapshot(&self) -> AccountRoutingSnapshot {
match &self.client_scope {
ClientRoutingScope::AllAccounts => AccountRoutingSnapshot::all(),
ClientRoutingScope::Restricted { bound_groups, .. } => {
AccountRoutingSnapshot::groups(bound_groups.to_vec())
}
}
}
#[must_use]
pub const fn directory(&self) -> &Arc<RuntimeAccountDirectory> {
&self.directory
}
}
@@ -7,7 +7,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Mutex, RwLock};
use std::time::{Duration, Instant, SystemTime};
use crate::routing::ProviderKind;
use crate::identity::ProviderKind;
use super::{AccountStatus, ProviderAccount, ProviderAccountId};
@@ -388,7 +388,7 @@ pub struct AccountSelectionContext {
pub preferred_account: Option<ProviderAccountId>,
pub round_robin_cursor: u64,
pub eligibility: AccountEligibilityPolicy,
pub account_scope: Option<std::sync::Arc<crate::routing::FrozenAccountScope>>,
pub account_scope: Option<std::sync::Arc<crate::account::scope::FrozenAccountScope>>,
}
/// 选择账号时是否执行本地调度资格投影。
@@ -3,7 +3,7 @@
use async_trait::async_trait;
use crate::error::StoreError;
use crate::routing::ProviderKind;
use crate::identity::ProviderKind;
use super::{
AccountStateChange, CredentialCasOutcome, CredentialCasUpdate, CredentialRevision,
@@ -8,9 +8,9 @@ use std::fmt;
use futures::future::BoxFuture;
use crate::account::ProviderAccountId;
use crate::identity::ProviderKind;
use crate::operation::ProviderSessionState;
use crate::policy::ClientApiKeyId;
use crate::routing::ProviderKind;
/// 客户端或 Provider 传递的 opaque response handle。
///
@@ -6,16 +6,18 @@ use std::num::NonZeroU32;
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime};
use super::observation::ResponseObservation;
use crate::engine::continuation::{
ContinuationBinding, NativeContinuationPin, NativeContinuationScope, PreviousResponseId,
};
use crate::engine::provider::{Provider, ProviderCallMetadata, ProviderRequest, ProviderStream};
use crate::engine::{
AccountAttemptContext, AttemptContext, AttemptRecord, AttemptTransport, AttemptTrigger,
CommitRequirement, ContinuationAttempt, CoordinatedEvent, EngineError, ExecutionOutcome,
ExecutionStore, GatewayEngine, IntermediateFailure, ModelRequestFailureObservation,
ModelRequestFinalization, ModelRequestId, ModelRequestTimings, NewModelRequest,
ProviderAccountStateOwner, ProviderAttemptOutcome, RequestAttemptContext, UpstreamSendState,
ModelRequestFinalization, ModelRequestId, NewModelRequest, ProviderAccountStateOwner,
ProviderAttemptOutcome, RequestAttemptContext, UpstreamSendState,
};
use crate::error::{
ContinuationRecoveryDisposition, GatewayError, GatewayErrorKind, ProviderError,
@@ -25,7 +27,6 @@ use crate::event::{
GatewayEvent, ProviderEvent, ProviderResponseHeader, ProviderResponseObservation,
};
use crate::lifecycle::CancellationToken;
use crate::metering::{CostEstimate, CostSource, Usage};
use crate::operation::{Operation, ProviderSessionState};
use crate::routing::RoutingPlan;
use futures::future::Fuse;
@@ -148,7 +149,7 @@ where
engine: Arc::clone(&self.engine),
request_id,
client_api_key_ref,
timing_started_at,
observation: ResponseObservation::new(timing_started_at),
deadline,
deadline_timer: Delay::new(
deadline
@@ -179,12 +180,7 @@ where
delivery_pending: false,
upstream_complete: false,
finalized: false,
usage: Usage::new(),
image_generation_requested,
cost: CostEstimate::unavailable(),
timings: ModelRequestTimings::default(),
client_response_id: None,
upstream_response_id: None,
last_retryable_failure: None,
last_retryable_failure_events: Vec::new(),
provider_attempt_outcomes: Vec::new(),
@@ -241,7 +237,7 @@ pub struct ResponseExecutionSession<S: ?Sized> {
engine: Arc<GatewayEngine<S>>,
request_id: ModelRequestId,
client_api_key_ref: crate::policy::ClientApiKeyId,
timing_started_at: Instant,
observation: ResponseObservation,
deadline: SystemTime,
/// 会话级 deadline 计时器;deadline 固定,帧循环内复用而非逐事件新建。
deadline_timer: Fuse<Delay>,
@@ -276,12 +272,7 @@ pub struct ResponseExecutionSession<S: ?Sized> {
delivery_pending: bool,
upstream_complete: bool,
finalized: bool,
usage: Usage,
image_generation_requested: bool,
cost: CostEstimate,
timings: ModelRequestTimings,
client_response_id: Option<String>,
upstream_response_id: Option<String>,
/// 最近一次为无感恢复而被丢弃的原始上游错误;只在后续空选路时成为终态。
last_retryable_failure: Option<ProviderError>,
/// 与 `last_retryable_failure` 同属一个 attempt 的原始失败批次。
@@ -500,7 +491,7 @@ where
return None;
}
let account = current.metadata.provider_account_id().clone();
let response_id = self.upstream_response_id.as_deref()?;
let response_id = self.observation.upstream_response_id.as_deref()?;
let previous_response_id = PreviousResponseId::new(response_id.to_owned());
let upstream_response_id = PreviousResponseId::new(response_id.to_owned());
Some(
@@ -579,7 +570,7 @@ where
self.observe_response(observation);
}
for fact in event.canonical_facts() {
self.observe_response_identity(fact);
self.observation.observe_identity(fact);
}
for fact in event.canonical_facts() {
self.observe_event(fact).await;
@@ -691,7 +682,7 @@ where
}));
let context = AttemptContext::new(
RequestAttemptContext::new(self.request_id.clone(), self.client_api_key_ref.clone())
.with_timing_started_at(self.timing_started_at),
.with_timing_started_at(self.observation.timing_started_at),
next_attempt,
self.deadline,
self.plan.account_selection_policy(),
@@ -954,26 +945,10 @@ where
}
async fn observe_event(&mut self, event: &GatewayEvent) {
self.observe_event_facts(event);
self.observation.observe_event(event);
self.mark_send_observed().await;
}
fn observe_event_facts(&mut self, event: &GatewayEvent) {
let elapsed = elapsed_ms(self.timing_started_at);
observe_event_timing(&mut self.timings, event, elapsed);
if let GatewayEvent::Usage(observed) = event {
self.usage.merge(observed);
}
if let GatewayEvent::CalculatedCost(observed) = event
&& self.cost.source() != CostSource::ProviderReported
{
self.cost = observed.into_estimate();
}
if let GatewayEvent::ProviderCost(observed) = event {
self.cost = observed.into_estimate();
}
}
async fn observe_wire_event(&mut self) {
self.mark_send_observed().await;
}
@@ -1020,44 +995,10 @@ where
{
return;
}
let observed = observation.timings();
if let Some(value) = observed.transport_decision_wait_ms {
self.timings.transport_decision_wait_ms = Some(value);
}
if let Some(value) = observed.connect_ms {
self.timings.connect_ms = Some(value);
}
if let Some(value) = observed.headers_ms {
self.timings.headers_ms = Some(value);
}
if let Some(value) = observed.first_event_ms {
self.timings.first_event_ms = Some(value);
}
if let Some(value) = observed.first_reasoning_ms {
self.timings.first_reasoning_ms = Some(value);
}
if let Some(value) = observed.first_text_ms {
self.timings.first_text_ms = Some(value);
}
if let Some(value) = observed.first_token_ms {
self.timings.first_token_ms = Some(value);
}
if let Some(value) = observed.provider_processing_ms {
self.timings.provider_processing_ms = Some(value);
}
self.observation.observe_response(&observation);
current.response_observation = Some(observation);
}
fn observe_response_identity(&mut self, event: &GatewayEvent) {
let metadata = match event {
GatewayEvent::Started(metadata) | GatewayEvent::Completed(metadata) => metadata,
_ => return,
};
let response_id = metadata.response_id().to_owned();
self.client_response_id = Some(response_id.clone());
self.upstream_response_id = Some(response_id);
}
/// 失败可重试时要求丢弃本 attempt;预算耗尽时可返回最后一批原始失败事件。
async fn handle_stream_error(
&mut self,
@@ -1227,8 +1168,8 @@ where
self.observe_response(observation);
}
for fact in event.canonical_facts() {
self.observe_response_identity(fact);
self.observe_event_facts(fact);
self.observation.observe_identity(fact);
self.observation.observe_event(fact);
}
}
}
@@ -1327,18 +1268,7 @@ where
}
fn reset_uncommitted_observations(&mut self) {
self.usage = Usage::new();
self.cost = CostEstimate::unavailable();
self.client_response_id = None;
self.upstream_response_id = None;
self.timings.transport_decision_wait_ms = None;
self.timings.connect_ms = None;
self.timings.headers_ms = None;
self.timings.first_event_ms = None;
self.timings.first_reasoning_ms = None;
self.timings.first_text_ms = None;
self.timings.first_token_ms = None;
self.timings.provider_processing_ms = None;
self.observation.reset_for_attempt();
self.upstream_complete = false;
}
@@ -1347,7 +1277,7 @@ where
return Ok(());
}
let completed_at = SystemTime::now();
self.timings.latency_ms = Some(elapsed_ms(self.timing_started_at));
self.observation.finish();
let upstream_request_id = self.current.as_ref().and_then(|current| {
current
.response_observation
@@ -1387,9 +1317,9 @@ where
downstream_committed_at: self.downstream_committed_at,
client_status_code: self.client_status_code,
upstream_status_code,
client_response_id: self.client_response_id.clone(),
client_response_id: self.observation.client_response_id.clone(),
upstream_request_id,
upstream_response_id: self.upstream_response_id.clone(),
upstream_response_id: self.observation.upstream_response_id.clone(),
upstream_transport,
http_version,
websocket_pool,
@@ -1400,10 +1330,10 @@ where
raw_upstream_error: None,
failure_observation: ModelRequestFailureObservation::default(),
retry_after_ms: None,
usage: self.usage.clone(),
usage: self.observation.usage.clone(),
image_generation_succeeded: self.image_generation_succeeded(),
cost: self.cost.clone(),
timings: self.timings.clone(),
cost: self.observation.cost.clone(),
timings: self.observation.timings.clone(),
completed_at,
}),
)
@@ -1518,7 +1448,7 @@ where
return Ok(());
}
let completed_at = SystemTime::now();
self.timings.latency_ms = Some(elapsed_ms(self.timing_started_at));
self.observation.finish();
let upstream_request_id = self.current.as_ref().and_then(|current| {
current
.response_observation
@@ -1559,9 +1489,9 @@ where
upstream_status_code: finalization
.upstream_status_code
.or(observed_status_code),
client_response_id: self.client_response_id.clone(),
client_response_id: self.observation.client_response_id.clone(),
upstream_request_id,
upstream_response_id: self.upstream_response_id.clone(),
upstream_response_id: self.observation.upstream_response_id.clone(),
upstream_transport,
http_version,
websocket_pool,
@@ -1572,10 +1502,10 @@ where
raw_upstream_error: finalization.raw_upstream_error,
failure_observation: finalization.observation,
retry_after_ms: finalization.retry_after_ms,
usage: self.usage.clone(),
usage: self.observation.usage.clone(),
image_generation_succeeded: self.image_generation_succeeded(),
cost: self.cost.clone(),
timings: self.timings.clone(),
cost: self.observation.cost.clone(),
timings: self.observation.timings.clone(),
completed_at,
}),
)
@@ -1623,7 +1553,11 @@ where
if matches!(self.operation, Operation::GenerateImage(_)) {
self.upstream_complete
} else {
self.usage.image_output_tokens.unwrap_or_default() > 0
self.observation
.usage
.image_output_tokens
.unwrap_or_default()
> 0
}
})
}
@@ -1661,7 +1595,7 @@ where
fn record_provider_failure(
&mut self,
provider_kind: crate::routing::ProviderKind,
provider_kind: crate::identity::ProviderKind,
error_kind: ProviderErrorKind,
) {
self.provider_attempt_outcomes
@@ -1850,31 +1784,6 @@ const fn escalate_send_state(a: UpstreamSendState, b: UpstreamSendState) -> Upst
}
}
fn observe_event_timing(timings: &mut ModelRequestTimings, event: &GatewayEvent, elapsed_ms: u64) {
timings.first_event_ms.get_or_insert(elapsed_ms);
match event {
GatewayEvent::ReasoningDelta(_) => {
timings.first_reasoning_ms.get_or_insert(elapsed_ms);
timings.first_token_ms.get_or_insert(elapsed_ms);
}
GatewayEvent::TextDelta(_) => {
timings.first_text_ms.get_or_insert(elapsed_ms);
timings.first_token_ms.get_or_insert(elapsed_ms);
}
// `response.output_item.added` 会先投影一个空参数的 tool delta;它只是结构帧,
// 不能抢在真实工具参数之前成为首个可消费 token。
GatewayEvent::ToolCallDelta(delta) if !delta.arguments_delta.is_empty() => {
timings.first_token_ms.get_or_insert(elapsed_ms);
}
GatewayEvent::CalculatedCost(_) | GatewayEvent::ProviderCost(_) => {}
_ => {}
}
}
fn elapsed_ms(started_at: Instant) -> u64 {
duration_ms(started_at.elapsed())
}
fn duration_ms(duration: Duration) -> u64 {
u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
}
@@ -31,12 +31,12 @@ use crate::engine::{
};
use crate::error::{GatewayError, GatewayErrorKind, ProviderErrorKind, StoreError};
use crate::event::{GatewayEvent, ProviderEvent, ProviderResponseHeader};
use crate::identity::ProviderKind;
use crate::lifecycle::CancellationToken;
use crate::operation::{Operation, ProviderSessionState};
use crate::policy::{ClientApiKeyId, ClientPolicy};
use crate::routing::{
ProviderKind, PublicModelId, PublicModelProfile, RoutingContext, RuntimeSnapshot,
UpstreamModelId,
PublicModelId, PublicModelProfile, RoutingContext, RuntimeSnapshot, UpstreamModelId,
};
use crate::runtime::RuntimeSnapshotHandle;
@@ -1202,11 +1202,11 @@ fn new_request_id() -> Result<ModelRequestId, GatewayError> {
.map_err(|_| GatewayError::new(GatewayErrorKind::Internal, "failed to allocate request ID"))
}
fn map_routing_error(error: crate::error::RoutingError) -> GatewayError {
fn map_routing_error(error: crate::validation::RoutingError) -> GatewayError {
match error {
crate::error::RoutingError::NoCapableProvider { .. }
| crate::error::RoutingError::NoCapableProviderEndpoint { .. }
| crate::error::RoutingError::EmptyAccountScope => GatewayError::new(
crate::validation::RoutingError::NoCapableProvider { .. }
| crate::validation::RoutingError::NoCapableProviderEndpoint { .. }
| crate::validation::RoutingError::EmptyAccountScope => GatewayError::new(
GatewayErrorKind::NoAvailableProvider,
"no provider can execute this request",
),
+12 -7
View File
@@ -4,6 +4,7 @@ pub mod admission;
pub mod continuation;
pub mod coordinator;
pub mod execution;
mod observation;
pub mod probe;
pub mod provider;
@@ -22,17 +23,18 @@ use thiserror::Error;
use crate::account::{AccountSelectionPolicy, ProviderAccountId};
use crate::engine::continuation::{ContinuationBinding, NativeContinuationPin};
use crate::error::{
GatewayError, IdentifierError, ProviderConnectionObservation, ProviderError, ProviderErrorKind,
StoreError, validate_text,
GatewayError, ProviderConnectionObservation, ProviderError, ProviderErrorKind, StoreError,
};
use crate::event::ProviderEvent;
use crate::identity::ProviderKind;
use crate::lifecycle::CancellationToken;
use crate::metering::{CostEstimate, Usage};
use crate::operation::OperationKind;
use crate::operation::ProviderSessionState;
use crate::policy::ClientApiKeyId;
use crate::routing::{ConfigRevision, ProviderKind, PublicModelId, UpstreamModelId};
use crate::routing::{ConfigRevision, PublicModelId, UpstreamModelId};
use crate::upstream::UpstreamSendState;
use crate::validation::{IdentifierError, validate_text};
/// `model_requests.id`。
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
@@ -153,7 +155,7 @@ pub struct AccountAttemptContext {
state_owner: Option<ProviderAccountStateOwner>,
credential_recovery_attempted: bool,
diagnostic_required_account: bool,
account_scope: Option<Arc<crate::routing::FrozenAccountScope>>,
account_scope: Option<Arc<crate::account::scope::FrozenAccountScope>>,
}
impl AccountAttemptContext {
@@ -194,7 +196,10 @@ impl AccountAttemptContext {
/// 附着普通请求认证时冻结的账号范围。
#[must_use]
pub fn with_account_scope(mut self, scope: Arc<crate::routing::FrozenAccountScope>) -> Self {
pub fn with_account_scope(
mut self,
scope: Arc<crate::account::scope::FrozenAccountScope>,
) -> Self {
self.account_scope = Some(scope);
self
}
@@ -233,7 +238,7 @@ impl AccountAttemptContext {
}
#[must_use]
pub const fn account_scope(&self) -> Option<&Arc<crate::routing::FrozenAccountScope>> {
pub const fn account_scope(&self) -> Option<&Arc<crate::account::scope::FrozenAccountScope>> {
self.account_scope.as_ref()
}
}
@@ -469,7 +474,7 @@ impl AttemptContext {
/// 普通请求认证时冻结的账号范围;管理端诊断为 `None`。
#[must_use]
pub const fn account_scope(&self) -> Option<&Arc<crate::routing::FrozenAccountScope>> {
pub const fn account_scope(&self) -> Option<&Arc<crate::account::scope::FrozenAccountScope>> {
self.account.account_scope()
}
@@ -0,0 +1,127 @@
//! 单次响应的计量、时间与响应身份事实;重试丢弃时统一清理。
use super::ModelRequestTimings;
use crate::event::GatewayEvent;
use crate::event::ProviderResponseObservation;
use crate::metering::{CostEstimate, CostSource, Usage};
use std::time::Instant;
pub(super) struct ResponseObservation {
pub(super) timing_started_at: Instant,
pub(super) usage: Usage,
pub(super) cost: CostEstimate,
pub(super) timings: ModelRequestTimings,
pub(super) client_response_id: Option<String>,
pub(super) upstream_response_id: Option<String>,
}
impl ResponseObservation {
pub(super) fn new(timing_started_at: Instant) -> Self {
Self {
timing_started_at,
usage: Usage::new(),
cost: CostEstimate::unavailable(),
timings: ModelRequestTimings::default(),
client_response_id: None,
upstream_response_id: None,
}
}
fn elapsed_ms(&self) -> u64 {
u64::try_from(self.timing_started_at.elapsed().as_millis()).unwrap_or(u64::MAX)
}
pub(super) fn finish(&mut self) {
self.timings.latency_ms = Some(self.elapsed_ms());
}
pub(super) fn observe_event(&mut self, event: &GatewayEvent) {
let elapsed = self.elapsed_ms();
observe_event_timing(&mut self.timings, event, elapsed);
if let GatewayEvent::Usage(observed) = event {
self.usage.merge(observed);
}
if let GatewayEvent::CalculatedCost(observed) = event
&& self.cost.source() != CostSource::ProviderReported
{
self.cost = observed.into_estimate();
}
if let GatewayEvent::ProviderCost(observed) = event {
self.cost = observed.into_estimate();
}
}
pub(super) fn observe_identity(&mut self, event: &GatewayEvent) {
let metadata = match event {
GatewayEvent::Started(metadata) | GatewayEvent::Completed(metadata) => metadata,
_ => return,
};
let response_id = metadata.response_id().to_owned();
self.client_response_id = Some(response_id.clone());
self.upstream_response_id = Some(response_id);
}
pub(super) fn reset_for_attempt(&mut self) {
self.usage = Usage::new();
self.cost = CostEstimate::unavailable();
self.client_response_id = None;
self.upstream_response_id = None;
self.timings.transport_decision_wait_ms = None;
self.timings.connect_ms = None;
self.timings.headers_ms = None;
self.timings.first_event_ms = None;
self.timings.first_reasoning_ms = None;
self.timings.first_text_ms = None;
self.timings.first_token_ms = None;
self.timings.provider_processing_ms = None;
}
pub(super) fn observe_response(&mut self, observation: &ProviderResponseObservation) {
let observed = observation.timings();
if let Some(value) = observed.transport_decision_wait_ms {
self.timings.transport_decision_wait_ms = Some(value);
}
if let Some(value) = observed.connect_ms {
self.timings.connect_ms = Some(value);
}
if let Some(value) = observed.headers_ms {
self.timings.headers_ms = Some(value);
}
if let Some(value) = observed.first_event_ms {
self.timings.first_event_ms = Some(value);
}
if let Some(value) = observed.first_reasoning_ms {
self.timings.first_reasoning_ms = Some(value);
}
if let Some(value) = observed.first_text_ms {
self.timings.first_text_ms = Some(value);
}
if let Some(value) = observed.first_token_ms {
self.timings.first_token_ms = Some(value);
}
if let Some(value) = observed.provider_processing_ms {
self.timings.provider_processing_ms = Some(value);
}
}
}
fn observe_event_timing(timings: &mut ModelRequestTimings, event: &GatewayEvent, elapsed_ms: u64) {
timings.first_event_ms.get_or_insert(elapsed_ms);
match event {
GatewayEvent::ReasoningDelta(_) => {
timings.first_reasoning_ms.get_or_insert(elapsed_ms);
timings.first_token_ms.get_or_insert(elapsed_ms);
}
GatewayEvent::TextDelta(_) => {
timings.first_text_ms.get_or_insert(elapsed_ms);
timings.first_token_ms.get_or_insert(elapsed_ms);
}
// `response.output_item.added` 会先投影一个空参数的 tool delta;它只是结构帧,
// 不能抢在真实工具参数之前成为首个可消费 token。
GatewayEvent::ToolCallDelta(delta) if !delta.arguments_delta.is_empty() => {
timings.first_token_ms.get_or_insert(elapsed_ms);
}
GatewayEvent::CalculatedCost(_) | GatewayEvent::ProviderCost(_) => {}
_ => {}
}
}
@@ -6,7 +6,8 @@ use bytes::Bytes;
use futures::future::BoxFuture;
use crate::error::{ClientVisibleUpstreamResponse, GatewayError, GatewayErrorKind};
use crate::routing::{ProviderKind, UpstreamModelId};
use crate::identity::ProviderKind;
use crate::routing::UpstreamModelId;
use crate::{account::ProviderAccountId, operation::Operation, upstream::UpstreamSendState};
#[derive(Debug, Clone, PartialEq)]
@@ -15,13 +15,13 @@ use crate::account::{
AccountAttemptFeedback, AccountCapacitySnapshot, AccountFeedbackStats, ProviderAccountId,
};
use crate::engine::AttemptContext;
use crate::error::{OpaqueUpstreamValue, PreDeliveryRetry, ProviderError, ProviderErrorKind};
use crate::error::{PreDeliveryRetry, ProviderError, ProviderErrorKind};
use crate::event::{EventSequenceValidator, ProviderEvent};
use crate::identity::ProviderKind;
use crate::operation::Operation;
use crate::policy::ClientApiKeyId;
use crate::routing::{
ModelCapabilities, ModelPresentation, ProviderCandidate, ProviderKind, UpstreamModelId,
};
use crate::routing::{ModelCapabilities, ModelPresentation, ProviderCandidate, UpstreamModelId};
use crate::upstream::OpaqueUpstreamValue;
use crate::upstream::{UpstreamSendState, UpstreamTransport};
/// Box 只出现在 Provider Registry 的统一 event envelope 边界。
+4 -172
View File
@@ -8,126 +8,11 @@ use bytes::Bytes;
use thiserror::Error;
use crate::event::{ProviderEvent, ProviderResponseHeader};
pub use crate::upstream::OpaqueUpstreamValue;
use crate::upstream::UpstreamSendState;
/// 应用层标识不满足约束。
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum IdentifierError {
/// 标识为空。
#[error("identifier must not be empty")]
Empty,
/// 标识超过核心允许的字节数。
#[error("identifier exceeds {max_bytes} bytes")]
TooLong {
/// 最大字节数。
max_bytes: usize,
},
/// 标识使用了保留的系统前缀。
#[error("identifier uses the reserved system prefix")]
ReservedPrefix,
/// 标识包含控制字符。
#[error("identifier contains control characters")]
ControlCharacter,
/// 标识缺少规定的语义前缀。
#[error("identifier must start with `{expected}`")]
MissingPrefix {
/// 规定前缀。
expected: &'static str,
},
/// 标识不满足该领域值对象的固定格式。
#[error("identifier has an invalid format")]
InvalidFormat,
}
/// Operation 构造或校验失败。
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum OperationError {
/// 必填文本为空。
#[error("`{field}` must not be empty")]
EmptyField {
/// 字段名。
field: &'static str,
},
/// 数量字段为零。
#[error("`{field}` must be greater than zero")]
ZeroValue {
/// 字段名。
field: &'static str,
},
/// JSON 字段必须是 object。
#[error("`{field}` must be a JSON object")]
JsonObjectRequired {
/// 字段名。
field: &'static str,
},
}
/// 路由快照或 Route Plan 不满足不变量。
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum RoutingError {
/// 动态 Provider/model 标识无法构造。
#[error("routing identifier is invalid")]
InvalidIdentifier,
/// 配置 revision 必须为正数。
#[error("config revision must be greater than zero")]
InvalidRevision,
/// Restricted scope 必须保留至少一个持久化分组 binding。
#[error("account routing scope is invalid")]
InvalidAccountScope,
/// Key 的账号范围中当前没有任何可参与路由的账号。
#[error("account routing scope is empty")]
EmptyAccountScope,
/// 快照中存在重复实体。
#[error("duplicate {entity} `{id}`")]
DuplicateEntity {
/// 实体类型。
entity: &'static str,
/// 实体 ID。
id: String,
},
/// 实体引用不存在。
#[error("{entity} `{id}` was not found")]
NotFound {
/// 实体类型。
entity: &'static str,
/// 实体 ID。
id: String,
},
/// 固定平台内没有可执行本次请求的 Provider。
#[error("no provider can execute model `{model}`")]
NoCapableProvider {
/// 客户端提交的模型名称。
model: String,
},
/// 固定 Provider 的原生端点当前不可执行。
#[error("provider endpoint `{provider}` is unavailable")]
NoCapableProviderEndpoint {
/// adapter 已绑定的 Provider。
provider: String,
},
}
/// 调用方策略不满足约束或拒绝请求。
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum PolicyError {
/// 请求超过调用方策略。
#[error("request was denied by caller policy: {reason}")]
Denied {
/// 稳定拒绝原因。
reason: &'static str,
},
}
/// 用量或价格估算不满足事实约束。
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum MeteringError {
/// 十进制定点值格式无效或超过 `numeric(20, 10)`。
#[error("decimal value must fit unsigned numeric(20, 10)")]
InvalidDecimal,
/// 货币代码无效。
#[error("currency must be a three-letter uppercase ASCII code")]
InvalidCurrency,
}
pub use crate::validation::{
IdentifierError, MeteringError, OperationError, PolicyError, RoutingError,
};
/// 跨 Provider 稳定的上游失败分类。
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
@@ -231,33 +116,6 @@ pub enum PreDeliveryRetry {
SameAccountTransportFallback,
}
/// Adapter 已明确分类为非 bearer 的不透明上游值。
///
/// Core 不解释或校验内容,只通过自定义 [`Debug`] 避免它意外进入日志。协议 adapter
/// 可在原客户端响应中读取原值。
#[derive(Clone, PartialEq, Eq, Hash)]
pub struct OpaqueUpstreamValue(String);
impl OpaqueUpstreamValue {
/// 原样保存上游值。
#[must_use]
pub fn new(value: impl Into<String>) -> Self {
Self(value.into())
}
/// 返回已经由 adapter 安全分类的原值。
#[must_use]
pub fn as_str(&self) -> &str {
&self.0
}
}
impl fmt::Debug for OpaqueUpstreamValue {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("OpaqueUpstreamValue(<redacted-from-Debug>)")
}
}
/// 已由 Provider 从结构化上游错误中提取、仅供原客户端协议展示的错误详情。
///
/// 它不是诊断日志或持久化错误信息:[`Debug`] 会隐藏全部值,调用方应只在
@@ -1242,29 +1100,3 @@ impl StoreError {
self.kind
}
}
pub(crate) fn validate_text(
value: &str,
max_bytes: usize,
reject_reserved_prefix: bool,
required_prefix: Option<&'static str>,
) -> Result<(), IdentifierError> {
if value.is_empty() {
return Err(IdentifierError::Empty);
}
if value.len() > max_bytes {
return Err(IdentifierError::TooLong { max_bytes });
}
if value.chars().any(char::is_control) {
return Err(IdentifierError::ControlCharacter);
}
if reject_reserved_prefix && value.starts_with("__") {
return Err(IdentifierError::ReservedPrefix);
}
if let Some(prefix) = required_prefix
&& !value.starts_with(prefix)
{
return Err(IdentifierError::MissingPrefix { expected: prefix });
}
Ok(())
}
+2 -1
View File
@@ -7,10 +7,11 @@ use bytes::Bytes;
use serde_json::Value;
use thiserror::Error;
use crate::error::{IdentifierError, OpaqueUpstreamValue, validate_text};
use crate::metering::{CalculatedCost, ProviderReportedCost, Usage};
use crate::operation::ProviderSessionState;
use crate::upstream::OpaqueUpstreamValue;
use crate::upstream::UpstreamTransport;
use crate::validation::{IdentifierError, validate_text};
/// 一次响应的稳定元数据。
#[derive(Debug, Clone, PartialEq, Eq)]
@@ -0,0 +1,28 @@
//! 编译期 Provider 身份;账号与路由共同使用,不依赖路由计划。
use crate::validation::{IdentifierError, validate_text};
use std::fmt;
/// 编译进二进制的 Provider adapter slug。
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct ProviderKind(String);
impl ProviderKind {
/// 校验 Provider slug。
pub fn new(value: impl Into<String>) -> Result<Self, IdentifierError> {
let value = value.into();
validate_text(&value, 64, true, None)?;
Ok(Self(value))
}
#[must_use]
pub fn as_str(&self) -> &str {
&self.0
}
}
impl fmt::Display for ProviderKind {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(&self.0)
}
}
+2
View File
@@ -8,6 +8,7 @@ pub mod engine;
pub mod error;
pub mod event;
pub mod health;
pub mod identity;
pub mod lifecycle;
pub mod metering;
pub mod operation;
@@ -17,6 +18,7 @@ pub mod routing;
pub mod runtime;
pub mod task;
pub mod upstream;
pub mod validation;
use std::sync::Arc;
use std::time::SystemTime;
@@ -3,7 +3,7 @@
use std::fmt;
use std::str::FromStr;
use crate::error::MeteringError;
use crate::validation::MeteringError;
const DECIMAL_SCALE: u128 = 10_000_000_000;
const MAX_SCALED_DECIMAL: u128 = 99_999_999_999_999_999_999;
@@ -12,7 +12,7 @@ use bytes::Bytes;
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value};
use crate::error::{OperationError, validate_text};
use crate::validation::{OperationError, validate_text};
/// 网关支持的稳定 operation 分类。
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
@@ -11,8 +11,8 @@ pub use client_version::{
use std::fmt;
use std::sync::Arc;
use crate::error::{IdentifierError, PolicyError, validate_text};
use crate::routing::FrozenAccountScope;
use crate::account::scope::FrozenAccountScope;
use crate::validation::{IdentifierError, PolicyError, validate_text};
/// `client_api_keys.id` 的核心值对象。
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
@@ -12,9 +12,10 @@ use crate::account::{
AccountFeedbackStats, AccountRuntimeSignals, CredentialRevision, CredentialState,
OpaqueProviderData, ProviderAccountId, ProviderAccountStore,
};
use crate::error::{IdentifierError, validate_text};
use crate::identity::ProviderKind;
use crate::policy::ClientApiKeyId;
use crate::routing::{ProviderKind, UpstreamModelId};
use crate::routing::UpstreamModelId;
use crate::validation::{IdentifierError, validate_text};
const MAX_PENDING_FLOW_TTL: Duration = Duration::from_secs(30 * 60);
+7 -307
View File
@@ -2,6 +2,11 @@
pub mod snapshot;
pub use crate::account::scope::{
AccountGroupId, AccountRoutingScopeKind, AccountRoutingSnapshot, ClientRoutingScope,
FrozenAccountScope, RoutingGroupSnapshot, RuntimeAccount, RuntimeAccountDirectory,
};
pub use crate::identity::ProviderKind;
pub use snapshot::RuntimeSnapshot;
use std::collections::{BTreeMap, BTreeSet};
@@ -9,317 +14,12 @@ use std::fmt;
use std::num::{NonZeroU32, NonZeroU64};
use std::sync::Arc;
use crate::account::{AccountSelectionPolicy, ProviderAccountId};
use crate::error::{IdentifierError, RoutingError, validate_text};
use crate::account::AccountSelectionPolicy;
use crate::operation::{CapabilityRequirements, Feature, OperationKind};
use crate::validation::{IdentifierError, RoutingError, validate_text};
const MAX_REQUEST_ATTEMPTS: u32 = 32;
/// `account_groups.id` 的核心值对象。
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct AccountGroupId(String);
impl AccountGroupId {
/// 校验并创建账号分组 ID。
pub fn new(value: impl Into<String>) -> Result<Self, IdentifierError> {
let value = value.into();
let Some(suffix) = value.strip_prefix("grp_") else {
return Err(IdentifierError::MissingPrefix { expected: "grp_" });
};
if suffix.len() != 32
|| !suffix
.bytes()
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
{
return Err(IdentifierError::InvalidFormat);
}
Ok(Self(value))
}
#[must_use]
pub fn as_str(&self) -> &str {
&self.0
}
}
impl fmt::Display for AccountGroupId {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(&self.0)
}
}
/// 快照中一个账号的 Provider 与分组归属。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RuntimeAccount {
provider_kind: ProviderKind,
group_ids: Arc<BTreeSet<AccountGroupId>>,
}
impl RuntimeAccount {
#[must_use]
pub fn new(provider_kind: ProviderKind, group_ids: BTreeSet<AccountGroupId>) -> Self {
Self {
provider_kind,
group_ids: Arc::new(group_ids),
}
}
#[must_use]
pub const fn provider_kind(&self) -> &ProviderKind {
&self.provider_kind
}
#[must_use]
pub fn group_ids(&self) -> &BTreeSet<AccountGroupId> {
&self.group_ids
}
}
/// 全快照共享的账号、Provider 与分组反向索引。
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct RuntimeAccountDirectory {
accounts: BTreeMap<ProviderAccountId, RuntimeAccount>,
providers_with_accounts: BTreeSet<ProviderKind>,
providers_by_group: BTreeMap<AccountGroupId, BTreeSet<ProviderKind>>,
}
impl RuntimeAccountDirectory {
#[must_use]
pub fn new(accounts: BTreeMap<ProviderAccountId, RuntimeAccount>) -> Self {
let mut providers_with_accounts = BTreeSet::new();
let mut providers_by_group = BTreeMap::<AccountGroupId, BTreeSet<ProviderKind>>::new();
for account in accounts.values() {
providers_with_accounts.insert(account.provider_kind.clone());
for group_id in account.group_ids.iter() {
providers_by_group
.entry(group_id.clone())
.or_default()
.insert(account.provider_kind.clone());
}
}
Self {
accounts,
providers_with_accounts,
providers_by_group,
}
}
#[must_use]
pub fn account(&self, account_id: &ProviderAccountId) -> Option<&RuntimeAccount> {
self.accounts.get(account_id)
}
#[must_use]
pub fn providers_with_accounts(&self) -> &BTreeSet<ProviderKind> {
&self.providers_with_accounts
}
#[must_use]
pub fn providers_for_groups<'a>(
&self,
group_ids: impl IntoIterator<Item = &'a AccountGroupId>,
) -> BTreeSet<ProviderKind> {
group_ids
.into_iter()
.filter_map(|group_id| self.providers_by_group.get(group_id))
.flatten()
.cloned()
.collect()
}
}
/// 历史请求保存的账号范围种类。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AccountRoutingScopeKind {
All,
Groups,
}
impl AccountRoutingScopeKind {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::All => "all",
Self::Groups => "groups",
}
}
}
/// 请求开始时冻结的分组 ID 与名称。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RoutingGroupSnapshot {
id: AccountGroupId,
name: String,
}
impl RoutingGroupSnapshot {
#[must_use]
pub fn new(id: AccountGroupId, name: String) -> Self {
Self { id, name }
}
#[must_use]
pub const fn id(&self) -> &AccountGroupId {
&self.id
}
#[must_use]
pub fn name(&self) -> &str {
&self.name
}
}
/// 请求历史所需的完整、稳定账号范围快照。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AccountRoutingSnapshot {
kind: AccountRoutingScopeKind,
groups: Arc<[RoutingGroupSnapshot]>,
}
impl AccountRoutingSnapshot {
#[must_use]
pub fn all() -> Self {
Self {
kind: AccountRoutingScopeKind::All,
groups: Arc::from([]),
}
}
#[must_use]
pub fn groups(groups: Vec<RoutingGroupSnapshot>) -> Self {
Self {
kind: AccountRoutingScopeKind::Groups,
groups: Arc::from(groups),
}
}
#[must_use]
pub const fn kind(&self) -> AccountRoutingScopeKind {
self.kind
}
#[must_use]
pub fn groups_snapshot(&self) -> &[RoutingGroupSnapshot] {
&self.groups
}
}
/// Key 持久 binding 编译出的账号权限。
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ClientRoutingScope {
AllAccounts,
Restricted {
bound_groups: Arc<[RoutingGroupSnapshot]>,
enabled_group_ids: Arc<BTreeSet<AccountGroupId>>,
provider_kinds: Arc<BTreeSet<ProviderKind>>,
},
}
impl ClientRoutingScope {
#[must_use]
pub fn all_accounts() -> Self {
Self::AllAccounts
}
pub fn restricted(
bound_groups: Vec<RoutingGroupSnapshot>,
enabled_group_ids: BTreeSet<AccountGroupId>,
provider_kinds: BTreeSet<ProviderKind>,
) -> Result<Self, RoutingError> {
if bound_groups.is_empty() {
return Err(RoutingError::InvalidAccountScope);
}
Ok(Self::Restricted {
bound_groups: Arc::from(bound_groups),
enabled_group_ids: Arc::new(enabled_group_ids),
provider_kinds: Arc::new(provider_kinds),
})
}
}
/// 一次认证随 RuntimeSnapshot 冻结的账号目录与 Key 权限。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FrozenAccountScope {
directory: Arc<RuntimeAccountDirectory>,
client_scope: ClientRoutingScope,
}
impl FrozenAccountScope {
#[must_use]
pub const fn new(
directory: Arc<RuntimeAccountDirectory>,
client_scope: ClientRoutingScope,
) -> Self {
Self {
directory,
client_scope,
}
}
#[must_use]
pub fn allows(&self, account_id: &ProviderAccountId) -> bool {
let Some(account) = self.directory.account(account_id) else {
return false;
};
match &self.client_scope {
ClientRoutingScope::AllAccounts => true,
ClientRoutingScope::Restricted {
enabled_group_ids, ..
} => account
.group_ids()
.iter()
.any(|group_id| enabled_group_ids.contains(group_id)),
}
}
#[must_use]
pub fn provider_kinds(&self) -> &BTreeSet<ProviderKind> {
match &self.client_scope {
ClientRoutingScope::AllAccounts => self.directory.providers_with_accounts(),
ClientRoutingScope::Restricted { provider_kinds, .. } => provider_kinds,
}
}
#[must_use]
pub fn routing_snapshot(&self) -> AccountRoutingSnapshot {
match &self.client_scope {
ClientRoutingScope::AllAccounts => AccountRoutingSnapshot::all(),
ClientRoutingScope::Restricted { bound_groups, .. } => {
AccountRoutingSnapshot::groups(bound_groups.to_vec())
}
}
}
#[must_use]
pub const fn directory(&self) -> &Arc<RuntimeAccountDirectory> {
&self.directory
}
}
/// 编译进二进制的 Provider adapter slug。
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct ProviderKind(String);
impl ProviderKind {
/// 校验 Provider slug。
pub fn new(value: impl Into<String>) -> Result<Self, IdentifierError> {
let value = value.into();
validate_text(&value, 64, true, None)?;
Ok(Self(value))
}
#[must_use]
pub fn as_str(&self) -> &str {
&self.0
}
}
impl fmt::Display for ProviderKind {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(&self.0)
}
}
/// 客户端请求中的模型名称。
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct PublicModelId(String);
@@ -9,12 +9,12 @@ use futures::future::BoxFuture;
use crate::account::{AccountSelectionPolicy, ProviderAccountId, RotationStrategy};
use crate::engine::provider::{ProviderCatalogGeneration, ProviderRegistry};
use crate::error::RoutingError;
use crate::operation::Operation;
use crate::policy::{
ClientApiKeyId, ClientPolicy, CodexClientMinVersions, CodexClientVersion,
PlaintextClientApiKey, RateLimits,
};
use crate::validation::RoutingError;
use super::{
AccountGroupId, ClientRoutingScope, ConfigRevision, FrozenAccountScope, ModelCapabilities,
@@ -11,11 +11,12 @@ use futures_timer::Delay;
use crate::engine::provider::ProviderCatalogGeneration;
use crate::health::{HealthProbe, HealthState};
use crate::identity::ProviderKind;
use crate::lifecycle::CancellationToken;
use crate::routing::ConfigRevision;
use crate::routing::snapshot::{
RuntimeSnapshot, RuntimeSnapshotCompileError, RuntimeSnapshotCompiler, SnapshotStorePort,
};
use crate::routing::{ConfigRevision, ProviderKind};
use crate::task::{
DaemonRestartPolicy, DaemonTask, ScheduledTask, WorkerContribution, WorkerCycleContext,
WorkerDefinitionError, WorkerId, WorkerKind, WorkerRegistration, WorkerRunnable,
+30 -1
View File
@@ -1,6 +1,8 @@
//! 上游调用跨 Engine、Event、Error 与 Provider 共享的中立边界事实。
use crate::error::{IdentifierError, validate_text};
use std::fmt;
use crate::validation::{IdentifierError, validate_text};
/// 上游是否可能已经收到业务 payload。
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
@@ -43,3 +45,30 @@ impl UpstreamTransport {
&self.0
}
}
/// Adapter 已明确分类为非 bearer 的不透明上游值。
///
/// Core 不解释或校验内容,只通过自定义 [`Debug`] 避免它意外进入日志。协议 adapter
/// 可在原客户端响应中读取原值。
#[derive(Clone, PartialEq, Eq, Hash)]
pub struct OpaqueUpstreamValue(String);
impl OpaqueUpstreamValue {
/// 原样保存上游值。
#[must_use]
pub fn new(value: impl Into<String>) -> Self {
Self(value.into())
}
/// 返回已经由 adapter 安全分类的原值。
#[must_use]
pub fn as_str(&self) -> &str {
&self.0
}
}
impl fmt::Debug for OpaqueUpstreamValue {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("OpaqueUpstreamValue(<redacted-from-Debug>)")
}
}
@@ -0,0 +1,148 @@
//! 领域值对象的纯校验错误与文本约束,不依赖执行事件或路由状态。
use thiserror::Error;
/// 应用层标识不满足约束。
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum IdentifierError {
/// 标识为空。
#[error("identifier must not be empty")]
Empty,
/// 标识超过核心允许的字节数。
#[error("identifier exceeds {max_bytes} bytes")]
TooLong {
/// 最大字节数。
max_bytes: usize,
},
/// 标识使用了保留的系统前缀。
#[error("identifier uses the reserved system prefix")]
ReservedPrefix,
/// 标识包含控制字符。
#[error("identifier contains control characters")]
ControlCharacter,
/// 标识缺少规定的语义前缀。
#[error("identifier must start with `{expected}`")]
MissingPrefix {
/// 规定前缀。
expected: &'static str,
},
/// 标识不满足该领域值对象的固定格式。
#[error("identifier has an invalid format")]
InvalidFormat,
}
/// Operation 构造或校验失败。
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum OperationError {
/// 必填文本为空。
#[error("`{field}` must not be empty")]
EmptyField {
/// 字段名。
field: &'static str,
},
/// 数量字段为零。
#[error("`{field}` must be greater than zero")]
ZeroValue {
/// 字段名。
field: &'static str,
},
/// JSON 字段必须是 object。
#[error("`{field}` must be a JSON object")]
JsonObjectRequired {
/// 字段名。
field: &'static str,
},
}
/// 路由快照或 Route Plan 不满足不变量。
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum RoutingError {
/// 动态 Provider/model 标识无法构造。
#[error("routing identifier is invalid")]
InvalidIdentifier,
/// 配置 revision 必须为正数。
#[error("config revision must be greater than zero")]
InvalidRevision,
/// Restricted scope 必须保留至少一个持久化分组 binding。
#[error("account routing scope is invalid")]
InvalidAccountScope,
/// Key 的账号范围中当前没有任何可参与路由的账号。
#[error("account routing scope is empty")]
EmptyAccountScope,
/// 快照中存在重复实体。
#[error("duplicate {entity} `{id}`")]
DuplicateEntity {
/// 实体类型。
entity: &'static str,
/// 实体 ID。
id: String,
},
/// 实体引用不存在。
#[error("{entity} `{id}` was not found")]
NotFound {
/// 实体类型。
entity: &'static str,
/// 实体 ID。
id: String,
},
/// 固定平台内没有可执行本次请求的 Provider。
#[error("no provider can execute model `{model}`")]
NoCapableProvider {
/// 客户端提交的模型名称。
model: String,
},
/// 固定 Provider 的原生端点当前不可执行。
#[error("provider endpoint `{provider}` is unavailable")]
NoCapableProviderEndpoint {
/// adapter 已绑定的 Provider。
provider: String,
},
}
/// 调用方策略不满足约束或拒绝请求。
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum PolicyError {
/// 请求超过调用方策略。
#[error("request was denied by caller policy: {reason}")]
Denied {
/// 稳定拒绝原因。
reason: &'static str,
},
}
/// 用量或价格估算不满足事实约束。
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum MeteringError {
/// 十进制定点值格式无效或超过 `numeric(20, 10)`。
#[error("decimal value must fit unsigned numeric(20, 10)")]
InvalidDecimal,
/// 货币代码无效。
#[error("currency must be a three-letter uppercase ASCII code")]
InvalidCurrency,
}
pub(crate) fn validate_text(
value: &str,
max_bytes: usize,
reject_reserved_prefix: bool,
required_prefix: Option<&'static str>,
) -> Result<(), IdentifierError> {
if value.is_empty() {
return Err(IdentifierError::Empty);
}
if value.len() > max_bytes {
return Err(IdentifierError::TooLong { max_bytes });
}
if value.chars().any(char::is_control) {
return Err(IdentifierError::ControlCharacter);
}
if reject_reserved_prefix && value.starts_with("__") {
return Err(IdentifierError::ReservedPrefix);
}
if let Some(prefix) = required_prefix
&& !value.starts_with(prefix)
{
return Err(IdentifierError::MissingPrefix { expected: prefix });
}
Ok(())
}
@@ -98,6 +98,37 @@ pub fn extract_usage(body: &Value) -> Option<TokenUsage> {
})
}
/// 校验计价所需的输入、输出和缓存读取字段,拒绝缺失、非法或不一致的数值。
///
/// 用量提取允许缺失字段补零;计价必须额外确认这些零值有明确的上游事实。
#[must_use]
pub fn billable_usage_is_complete(response: &Value, usage: TokenUsage) -> bool {
let Some(raw) = response.get("usage").filter(|value| value.is_object()) else {
return false;
};
let input = raw
.get("input_tokens")
.or_else(|| raw.get("prompt_tokens"))
.and_then(Value::as_u64);
let output = raw
.get("output_tokens")
.or_else(|| raw.get("completion_tokens"))
.and_then(Value::as_u64);
let cached = raw
.pointer("/input_tokens_details/cached_tokens")
.or_else(|| raw.pointer("/prompt_tokens_details/cached_tokens"))
.or_else(|| raw.get("cached_tokens"));
let cached = match cached {
Some(value) => value.as_u64(),
None => Some(0),
};
input == Some(usage.input_tokens)
&& output == Some(usage.output_tokens)
&& cached == Some(usage.cached_tokens)
&& usage.cached_tokens <= usage.input_tokens
}
/// 从完整 SSE 文本中提取最终可见用量。
///
/// 如果存在 `response.completed` 的 usage,则优先返回它;否则回退到最后一条
@@ -1,10 +1,55 @@
use gateway_protocol::openai::events::{
RateLimitKeySource, RateLimitWindow, TokenUsage, extract_sse_usage, extract_usage,
is_codex_quota_header_name, is_rate_limit_header_name, parse_rate_limit_headers,
RateLimitKeySource, RateLimitWindow, TokenUsage, billable_usage_is_complete, extract_sse_usage,
extract_usage, is_codex_quota_header_name, is_rate_limit_header_name, parse_rate_limit_headers,
parse_rate_limits_event, rate_limits_to_header_pairs, retry_after_seconds_from_body,
};
use serde_json::json;
#[test]
fn billable_usage_should_require_valid_input_output_and_cache_facts() {
for (raw, expected) in [
(json!({}), false),
(json!({"input_tokens": 0}), false),
(json!({"input_tokens": 0, "output_tokens": 0}), true),
(
json!({"prompt_tokens": 8, "completion_tokens": 4, "cached_tokens": 2}),
true,
),
(json!({"input_tokens": -1, "output_tokens": 1}), false),
(
json!({"input_tokens": "8", "prompt_tokens": 8, "output_tokens": 1}),
false,
),
(json!({"input_tokens": 8, "output_tokens": 1.5}), false),
(
json!({"input_tokens": 8, "output_tokens": 1, "cached_tokens": 9}),
false,
),
(
json!({"input_tokens": 8, "output_tokens": 1, "cached_tokens": null}),
false,
),
(
json!({"input_tokens": 8, "output_tokens": 1, "input_tokens_details": {"cached_tokens": "2"}, "cached_tokens": 2}),
false,
),
(
json!({"input_tokens": 8, "prompt_tokens": 9, "output_tokens": 1}),
true,
),
] {
let body = json!({"usage": raw});
let usage = extract_usage(&body).unwrap_or_default();
assert_eq!(billable_usage_is_complete(&body, usage), expected, "{body}");
}
let body = json!({"usage": {"input_tokens": 8, "output_tokens": 4}});
assert!(!billable_usage_is_complete(&body, TokenUsage::default()));
assert!(!billable_usage_is_complete(
&json!({}),
TokenUsage::default()
));
}
#[test]
fn extract_usage_should_read_codex_usage_shape() {
let body = json!({
+1
View File
@@ -12,6 +12,7 @@ name = "main"
path = "tests/main.rs"
[dependencies]
async-stream = "0.3.6"
async-trait = "0.1.89"
aws-credential-types = "1.2"
aws-sdk-s3 = { version = "1.140", default-features = false, features = ["rt-tokio", "default-https-client"] }
@@ -1,5 +1,7 @@
//! Pg 观测 adapter:实现 `ObservabilityRepository` 与 `AdminObservabilityStore`。
use futures::{StreamExt, TryStreamExt, stream::BoxStream};
use gateway_admin::ports::store::UsageCalculatedBillingStream;
use gateway_core::provider_ports::ProviderCooldownPort;
use std::sync::Arc;
@@ -185,17 +187,15 @@ impl ObservabilityRepository for PgObservabilityRepository {
.await
}
async fn usage_calculated_billing_facts(
fn usage_calculated_billing_facts(
&self,
range: ObservabilityRange,
filter: UsageRecordFilter,
) -> StoreResult<Vec<CalculatedUsageBillingFact>> {
self.query_budget
.run(
"load calculated usage billing facts",
calculated_usage_billing_facts(&self.pool, range, &filter),
)
.await
) -> BoxStream<'_, StoreResult<CalculatedUsageBillingFact>> {
self.query_budget.run_stream(
"load calculated usage billing facts",
calculated_usage_billing_facts(&self.pool, range, filter),
)
}
async fn provider_account_usage(
@@ -387,18 +387,20 @@ impl AdminObservabilityStore for PgAdminObservabilityStore {
.collect()
}
async fn usage_calculated_billing_facts(
fn usage_calculated_billing_facts(
&self,
range: admin_observability::TimeRange,
filter: admin_observability::UsageFilter,
) -> AdminStoreResult<Vec<admin_observability::UsageCalculatedBillingFact>> {
) -> UsageCalculatedBillingStream<'_> {
let range = match store_range(range) {
Ok(range) => range,
Err(error) => return Box::pin(futures::stream::once(async { Err(error) })),
};
self.repository
.usage_calculated_billing_facts(store_range(range)?, store_usage_filter(filter))
.await
.map_err(observability_error)?
.into_iter()
.map(admin_calculated_usage_billing_fact)
.collect()
.usage_calculated_billing_facts(range, store_usage_filter(filter))
.map_err(observability_error)
.map(|fact| fact.and_then(admin_calculated_usage_billing_fact))
.boxed()
}
async fn list_usage_records(
@@ -319,6 +319,7 @@ pub(crate) fn admin_dashboard_account_usage(
)
.collect(),
quota_used_percent: None,
quota_window: None,
models: usage
.models
.into_iter()
@@ -1,6 +1,7 @@
//! 查询模型、校验与观测端口契约。
use super::*;
use futures::stream::BoxStream;
pub(crate) const MAX_PAGE_SIZE: u16 = 100;
pub(crate) const MAX_FILTER_BYTES: usize = 256;
@@ -762,11 +763,11 @@ pub trait ObservabilityRepository: Send + Sync {
range: ObservabilityRange,
filter: UsageRecordFilter,
) -> StoreResult<Vec<RequestMetricPoint>>;
async fn usage_calculated_billing_facts(
fn usage_calculated_billing_facts(
&self,
range: ObservabilityRange,
filter: UsageRecordFilter,
) -> StoreResult<Vec<CalculatedUsageBillingFact>>;
) -> BoxStream<'_, StoreResult<CalculatedUsageBillingFact>>;
async fn provider_account_usage(
&self,
query: ProviderAccountUsageQuery,
@@ -283,42 +283,44 @@ async fn request_metric_series_inner(
fill_metric_gaps(range, granularity, points)
}
pub(crate) async fn calculated_usage_billing_facts(
pub(crate) fn calculated_usage_billing_facts(
pool: &PgPool,
range: ObservabilityRange,
filter: &UsageRecordFilter,
) -> StoreResult<Vec<CalculatedUsageBillingFact>> {
filter.validate()?;
let granularity = granularity_for(range);
let mut query = QueryBuilder::<Postgres>::new("select date_bin(");
query.push_bind(granularity.sql_interval());
query.push(
"::interval, mr.started_at, timestamptz '1970-01-01 00:00:00+00') as bucket_start,
mr.provider_kind, mr.upstream_model_id, mr.service_tier,
mr.input_tokens, mr.output_tokens, mr.cached_tokens, mr.cache_write_tokens,
mr.cost_currency, mr.cost_amount::text as amount
from model_requests mr where mr.started_at >= ",
);
query.push_bind(range.start);
query.push(" and mr.started_at < ");
query.push_bind(range.end);
query.push(
" and mr.cost_source = 'calculated'
and mr.cost_amount is not null and mr.cost_currency is not null
and nullif(mr.provider_kind, '') is not null
and nullif(mr.upstream_model_id, '') is not null",
);
push_completed_usage_fact_filter(&mut query, "mr");
push_usage_filter(&mut query, filter, "mr");
query.push(" order by bucket_start, mr.id");
query
.build()
.fetch_all(pool)
.await
.map_err(|_| postgres_unavailable("load calculated usage billing facts"))?
.iter()
.map(calculated_usage_billing_fact_from_row)
.collect()
filter: UsageRecordFilter,
) -> futures::stream::BoxStream<'_, StoreResult<CalculatedUsageBillingFact>> {
use futures::TryStreamExt;
Box::pin(async_stream::try_stream! {
filter.validate()?;
let granularity = granularity_for(range);
let mut query = QueryBuilder::<Postgres>::new("select date_bin(");
query.push_bind(granularity.sql_interval());
query.push(
"::interval, mr.started_at, timestamptz '1970-01-01 00:00:00+00') as bucket_start,
mr.provider_kind, mr.upstream_model_id, mr.service_tier,
mr.input_tokens, mr.output_tokens, mr.cached_tokens, mr.cache_write_tokens,
mr.cost_currency, mr.cost_amount::text as amount
from model_requests mr where mr.started_at >= ",
);
query.push_bind(range.start);
query.push(" and mr.started_at < ");
query.push_bind(range.end);
query.push(
" and mr.cost_source = 'calculated'
and mr.cost_amount is not null and mr.cost_currency is not null
and nullif(mr.provider_kind, '') is not null
and nullif(mr.upstream_model_id, '') is not null",
);
push_completed_usage_fact_filter(&mut query, "mr");
push_usage_filter(&mut query, &filter, "mr");
// 费用逐条累加与行顺序无关,避免全量物化和不必要的数据库排序。
let mut rows = query.build().fetch(pool);
while let Some(row) = rows.try_next().await
.map_err(|_| postgres_unavailable("load calculated usage billing facts"))?
{
yield calculated_usage_billing_fact_from_row(&row)?;
}
})
}
pub(crate) async fn request_costs_by_bucket(
@@ -6,6 +6,7 @@ use std::{
time::{Duration, Instant},
};
use futures::{Stream, TryStreamExt, stream::BoxStream};
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
use crate::{StoreBackend, StoreError, StoreResult};
@@ -49,6 +50,43 @@ impl ObservabilityQueryBudget {
query.await
}
/// 将槽位持有到查询流消费结束、报错或被丢弃,而不是仅持有到流创建完成。
pub fn run_stream<'a, T, S>(
&'a self,
operation: &'static str,
query: S,
) -> BoxStream<'a, StoreResult<T>>
where
T: Send + 'a,
S: Stream<Item = StoreResult<T>> + Send + 'a,
{
Box::pin(async_stream::stream! {
let mut query = Box::pin(query);
let slot = match self.acquire(operation).await {
Ok(slot) => slot,
Err(error) => {
drop(query);
yield Err(error);
return;
}
};
loop {
match query.try_next().await {
Ok(Some(item)) => yield Ok(item),
Ok(None) => break,
Err(error) => {
// 返回错误后调用方可能不再 poll,但仍保留流对象。
// 必须在交付错误之前释放数据库流和预算。
drop(query);
drop(slot);
yield Err(error);
return;
}
}
}
})
}
async fn acquire(&self, operation: &'static str) -> StoreResult<OwnedSemaphorePermit> {
let started_at = Instant::now();
let acquisition = Arc::clone(&self.slots).acquire_owned();
@@ -1,3 +1,4 @@
use futures::TryStreamExt;
use std::{
collections::BTreeMap,
sync::Arc,
@@ -595,6 +596,7 @@ async fn calculated_usage_billing_facts_keep_only_completed_calculated_costs() {
let facts = repository
.usage_calculated_billing_facts(range, UsageRecordFilter::default())
.try_collect::<Vec<_>>()
.await
.expect("calculated usage billing facts");
assert_eq!(facts.len(), 1);
@@ -612,6 +614,7 @@ async fn calculated_usage_billing_facts_keep_only_completed_calculated_costs() {
.expect("admin observability range"),
admin_observability::UsageFilter::default(),
)
.try_collect::<Vec<_>>()
.await
.expect("admin calculated usage billing facts");
assert_eq!(facts.len(), 1);
@@ -1,5 +1,6 @@
use std::time::Duration;
use futures::{StreamExt, TryStreamExt, stream};
use gateway_store::{StoreError, postgres::ObservabilityQueryBudget};
use tokio::sync::{mpsc, oneshot};
@@ -8,6 +9,44 @@ fn query_budget_rejects_an_empty_budget() {
assert!(ObservabilityQueryBudget::try_new(0, Duration::from_secs(1)).is_err());
}
#[tokio::test]
async fn query_stream_holds_budget_until_completion_error_or_drop() {
let budget =
ObservabilityQueryBudget::try_new(1, Duration::from_millis(10)).expect("query budget");
for terminal in [
None,
Some(Ok(2)),
Some(Err(StoreError::InvalidData {
entity: "test stream",
message: "decode failed".to_owned(),
})),
] {
let pending = terminal.is_none();
let query = stream::iter([Ok(1)]).chain(stream::iter(terminal));
let mut facts = budget.run_stream("stream test", query);
assert_eq!(facts.try_next().await.expect("first row"), Some(1));
assert!(budget.run("blocked query", async { Ok(()) }).await.is_err());
if pending {
drop(facts);
} else {
match facts.try_next().await {
Ok(Some(2)) => assert!(facts.try_next().await.expect("end of stream").is_none()),
Err(StoreError::InvalidData { .. }) => {}
other => panic!("unexpected stream result: {other:?}"),
}
// The finished stream may remain alive; its connection slot must already be free.
budget
.run("after terminal", async { Ok(()) })
.await
.expect("released slot");
}
budget
.run("after stream", async { Ok(()) })
.await
.expect("released slot");
}
}
#[tokio::test]
async fn query_budget_times_out_without_exceeding_its_connection_limit() {
let budget = ObservabilityQueryBudget::try_new(4, Duration::from_millis(10))
+13 -113
View File
@@ -6,6 +6,7 @@ use std::time::SystemTime;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use futures::TryStreamExt as _;
use gateway_admin::model::Revision;
use gateway_admin::model::accounts::AccountRecord;
use gateway_admin::model::observability::{
CalculatedBillingBreakdown, CurrencyCost, DashboardDesktopRelease, DashboardWireAttribute,
@@ -13,8 +14,8 @@ use gateway_admin::model::observability::{
ProviderBillingInput,
};
use gateway_admin::model::provider_credentials::{
AuthorizationMutationTarget, AuthorizationOwner, AuthorizationOwnerBinding,
AuthorizationStarted, CompleteAuthorization, ConsumeProviderResetCredit, CredentialCommitGuard,
AuthorizationMutationTarget, AuthorizationOwnerBinding, AuthorizationStarted,
CompleteAuthorization, ConsumeProviderResetCredit, CredentialCommitGuard,
PendingAuthorizationMutation, PrepareCredentialImport, PrepareCredentialRefresh,
PrepareCredentialRotation, PreparedAuthorizationCommit, PreparedAuthorizationCredential,
PreparedCredentialCreate, PreparedCredentialImport, PreparedCredentialRotation,
@@ -26,7 +27,6 @@ use gateway_admin::model::provider_credentials::{
ProviderResetCredit, ProviderResetCreditResult, ProviderResetCredits,
QuotaLocalUsageAttribution,
};
use gateway_admin::model::{MutationActor, MutationContext, Revision};
use gateway_admin::ports::provider::{ProviderAdmin, ProviderAdminError, ProviderAdminErrorKind};
use gateway_core::account::{
CredentialCasUpdateParts, CredentialRevision, LoadedCredential, NewProviderAccount,
@@ -1129,32 +1129,7 @@ struct PendingDocument {
code_verifier: String,
installation_id: String,
reauthorization_account_id: Option<String>,
mutation: PendingMutationDocument,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct PendingMutationDocument {
schema_version: u64,
provider_kind: String,
target: PendingTargetDocument,
owner: PendingOwnerDocument,
started_request_id: String,
}
#[derive(Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
enum PendingTargetDocument {
Create { name: String },
Reauthorize { account_id: String },
}
#[derive(Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
enum PendingOwnerDocument {
AdminSession { admin_user_id: String },
AdminApiKey,
System,
mutation: Map<String, Value>,
}
fn encode_pending(pending: &CodexPendingAuthorization) -> Map<String, Value> {
@@ -1206,65 +1181,11 @@ fn encode_pending(pending: &CodexPendingAuthorization) -> Map<String, Value> {
}
fn encode_mutation(mutation: &PendingAuthorizationMutation) -> Map<String, Value> {
let mut document = Map::new();
let mut document = mutation.to_storage_v1();
document.insert(
"schema_version".to_owned(),
Value::Number(Number::from(PENDING_DOCUMENT_SCHEMA_VERSION)),
);
document.insert(
"provider_kind".to_owned(),
Value::String(mutation.provider_kind().as_str().to_owned()),
);
document.insert(
"target".to_owned(),
Value::Object(encode_target(mutation.target())),
);
document.insert(
"owner".to_owned(),
Value::Object(encode_owner(mutation.owner_binding().owner())),
);
document.insert(
"started_request_id".to_owned(),
Value::String(mutation.owner_binding().started_request_id().to_owned()),
);
document
}
fn encode_target(target: &AuthorizationMutationTarget) -> Map<String, Value> {
let mut document = Map::new();
match target {
AuthorizationMutationTarget::Create { name } => {
document.insert("kind".to_owned(), Value::String("create".to_owned()));
document.insert("name".to_owned(), Value::String(name.clone()));
}
AuthorizationMutationTarget::Reauthorize { account_id } => {
document.insert("kind".to_owned(), Value::String("reauthorize".to_owned()));
document.insert(
"account_id".to_owned(),
Value::String(account_id.to_string()),
);
}
}
document
}
fn encode_owner(owner: &AuthorizationOwner) -> Map<String, Value> {
let mut document = Map::new();
match owner {
AuthorizationOwner::AdminSession { admin_user_id } => {
document.insert("kind".to_owned(), Value::String("admin_session".to_owned()));
document.insert(
"admin_user_id".to_owned(),
Value::String(admin_user_id.clone()),
);
}
AuthorizationOwner::AdminApiKey => {
document.insert("kind".to_owned(), Value::String("admin_api_key".to_owned()));
}
AuthorizationOwner::System => {
document.insert("kind".to_owned(), Value::String("system".to_owned()));
}
}
document
}
@@ -1289,38 +1210,17 @@ fn decode_pending(
}
fn decode_mutation(
document: PendingMutationDocument,
mut document: Map<String, Value>,
) -> Result<PendingAuthorizationMutation, CodexOAuthPendingStoreError> {
if document.schema_version != PENDING_DOCUMENT_SCHEMA_VERSION {
if document
.remove("schema_version")
.and_then(|value| value.as_u64())
!= Some(PENDING_DOCUMENT_SCHEMA_VERSION)
{
return Err(CodexOAuthPendingStoreError::InvalidValue);
}
let provider_kind = ProviderKind::new(document.provider_kind)
.map_err(|_| CodexOAuthPendingStoreError::InvalidValue)?;
let target = match document.target {
PendingTargetDocument::Create { name } => AuthorizationMutationTarget::Create { name },
PendingTargetDocument::Reauthorize { account_id } => {
AuthorizationMutationTarget::Reauthorize {
account_id: ProviderAccountId::new(account_id)
.map_err(|_| CodexOAuthPendingStoreError::InvalidValue)?,
}
}
};
let actor = match document.owner {
PendingOwnerDocument::AdminSession { admin_user_id } => {
MutationActor::AdminSession { admin_user_id }
}
PendingOwnerDocument::AdminApiKey => MutationActor::AdminApiKey,
PendingOwnerDocument::System => MutationActor::System,
};
let context = MutationContext {
actor,
request_id: document.started_request_id,
};
Ok(PendingAuthorizationMutation::new(
provider_kind,
target,
AuthorizationOwnerBinding::from_context(&context),
))
PendingAuthorizationMutation::from_storage_v1(Value::Object(document))
.map_err(|_| CodexOAuthPendingStoreError::InvalidValue)
}
fn binding(value: &str) -> Result<OAuthPendingBinding, CodexOAuthPendingStoreError> {
@@ -13,7 +13,7 @@ use gateway_core::event::{
};
use gateway_core::metering::Usage;
use gateway_core::upstream::UpstreamSendState;
use gateway_protocol::openai::events::{TokenUsage, extract_usage};
use gateway_protocol::openai::events::{TokenUsage, billable_usage_is_complete, extract_usage};
use gateway_protocol::openai::sse::{SseEvent, SseEventDecoder, SseFrame, sse_frame_is_done};
use serde_json::Value;
use thiserror::Error;
@@ -1022,33 +1022,6 @@ fn web_search_call_count(response: &Value) -> Option<u64> {
Some(calls)
}
fn billable_usage_is_complete(response: &Value, usage: TokenUsage) -> bool {
let Some(raw) = response.get("usage").filter(|value| value.is_object()) else {
return false;
};
let input = raw
.get("input_tokens")
.or_else(|| raw.get("prompt_tokens"))
.and_then(Value::as_u64);
let output = raw
.get("output_tokens")
.or_else(|| raw.get("completion_tokens"))
.and_then(Value::as_u64);
let cached = raw
.pointer("/input_tokens_details/cached_tokens")
.or_else(|| raw.pointer("/prompt_tokens_details/cached_tokens"))
.or_else(|| raw.get("cached_tokens"));
let cached = match cached {
Some(value) => value.as_u64(),
None => Some(0),
};
input == Some(usage.input_tokens)
&& output == Some(usage.output_tokens)
&& cached == Some(usage.cached_tokens)
&& usage.cached_tokens <= usage.input_tokens
}
fn incomplete_finish_reason(response: &Value) -> FinishReason {
match response
.pointer("/incomplete_details/reason")
+5 -119
View File
@@ -23,7 +23,7 @@ use gateway_admin::model::provider_credentials::{
ProviderExportCredentialInput, ProviderModel, ProviderModels, ProviderQuota,
ProviderQuotaRequest, ProviderQuotaWindow, QuotaLocalUsageAttribution,
};
use gateway_admin::model::{AdminError, MutationActor, MutationContext, Revision};
use gateway_admin::model::{AdminError, MutationContext, Revision};
use gateway_admin::ports::provider::{ProviderAdmin, ProviderAdminError, ProviderAdminErrorKind};
use gateway_core::account::{
CredentialCasUpdateParts, CredentialRevision, LoadedCredential, NewProviderAccount,
@@ -1170,31 +1170,7 @@ struct PendingDocument {
owner_ref: String,
expires_at: DateTime<Utc>,
server_state: String,
mutation: PendingMutationDocument,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct PendingMutationDocument {
provider_kind: String,
target: PendingTargetDocument,
owner: PendingOwnerDocument,
started_request_id: String,
}
#[derive(Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
enum PendingTargetDocument {
Create { name: String },
Reauthorize { account_id: String },
}
#[derive(Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
enum PendingOwnerDocument {
AdminSession { admin_user_id: String },
AdminApiKey,
System,
mutation: Map<String, Value>,
}
fn encode_pending(pending: &StoredXaiAuthorization) -> Map<String, Value> {
@@ -1218,70 +1194,11 @@ fn encode_pending(pending: &StoredXaiAuthorization) -> Map<String, Value> {
);
document.insert(
"mutation".to_owned(),
Value::Object(encode_mutation(&pending.mutation)),
Value::Object(pending.mutation.to_storage_v1()),
);
document
}
fn encode_mutation(mutation: &PendingAuthorizationMutation) -> Map<String, Value> {
let mut document = Map::new();
document.insert(
"provider_kind".to_owned(),
Value::String(mutation.provider_kind().as_str().to_owned()),
);
document.insert(
"target".to_owned(),
Value::Object(encode_target(mutation.target())),
);
document.insert(
"owner".to_owned(),
Value::Object(encode_owner(mutation.owner_binding().owner())),
);
document.insert(
"started_request_id".to_owned(),
Value::String(mutation.owner_binding().started_request_id().to_owned()),
);
document
}
fn encode_target(target: &AuthorizationMutationTarget) -> Map<String, Value> {
let mut document = Map::new();
match target {
AuthorizationMutationTarget::Create { name } => {
document.insert("kind".to_owned(), Value::String("create".to_owned()));
document.insert("name".to_owned(), Value::String(name.clone()));
}
AuthorizationMutationTarget::Reauthorize { account_id } => {
document.insert("kind".to_owned(), Value::String("reauthorize".to_owned()));
document.insert(
"account_id".to_owned(),
Value::String(account_id.to_string()),
);
}
}
document
}
fn encode_owner(owner: &AuthorizationOwner) -> Map<String, Value> {
let mut document = Map::new();
match owner {
AuthorizationOwner::AdminSession { admin_user_id } => {
document.insert("kind".to_owned(), Value::String("admin_session".to_owned()));
document.insert(
"admin_user_id".to_owned(),
Value::String(admin_user_id.clone()),
);
}
AuthorizationOwner::AdminApiKey => {
document.insert("kind".to_owned(), Value::String("admin_api_key".to_owned()));
}
AuthorizationOwner::System => {
document.insert("kind".to_owned(), Value::String("system".to_owned()));
}
}
document
}
fn decode_pending(
payload: OpaqueProviderData,
oauth_config: &GrokOAuthConfig,
@@ -1295,7 +1212,8 @@ fn decode_pending(
{
return Err(provider_error(ProviderAdminErrorKind::Invalid));
}
let mutation = decode_mutation(document.mutation)?;
let mutation = PendingAuthorizationMutation::from_storage_v1(Value::Object(document.mutation))
.map_err(|_| provider_error(ProviderAdminErrorKind::Invalid))?;
if mutation.provider_kind().as_str() != XAI_PROVIDER_NAME
|| owner_ref(mutation.owner_binding().owner()) != document.owner_ref
{
@@ -1315,38 +1233,6 @@ fn decode_pending(
})
}
fn decode_mutation(
document: PendingMutationDocument,
) -> Result<PendingAuthorizationMutation, ProviderAdminError> {
let provider_kind = ProviderKind::new(document.provider_kind)
.map_err(|_| provider_error(ProviderAdminErrorKind::Invalid))?;
let target = match document.target {
PendingTargetDocument::Create { name } => AuthorizationMutationTarget::Create { name },
PendingTargetDocument::Reauthorize { account_id } => {
AuthorizationMutationTarget::Reauthorize {
account_id: ProviderAccountId::new(account_id)
.map_err(|_| provider_error(ProviderAdminErrorKind::Invalid))?,
}
}
};
let actor = match document.owner {
PendingOwnerDocument::AdminSession { admin_user_id } => {
MutationActor::AdminSession { admin_user_id }
}
PendingOwnerDocument::AdminApiKey => MutationActor::AdminApiKey,
PendingOwnerDocument::System => MutationActor::System,
};
let context = MutationContext {
actor,
request_id: document.started_request_id,
};
Ok(PendingAuthorizationMutation::new(
provider_kind,
target,
AuthorizationOwnerBinding::from_context(&context),
))
}
fn random_flow_id() -> Result<String, ProviderAdminError> {
let mut random = [0_u8; 32];
getrandom::fill(&mut random)
@@ -14,7 +14,7 @@ use gateway_core::metering::{
CurrencyCode, Decimal, Money, ProviderReportedCost, Usage,
};
use gateway_core::upstream::UpstreamSendState;
use gateway_protocol::openai::events::{TokenUsage, extract_usage};
use gateway_protocol::openai::events::{TokenUsage, billable_usage_is_complete, extract_usage};
use gateway_protocol::openai::sse::{SseEvent, SseEventDecoder};
use serde_json::Value;
@@ -750,33 +750,6 @@ fn calculated_cost(response: &Value, model: &str, usage: TokenUsage) -> Option<C
Some(breakdown.calculated_cost())
}
fn billable_usage_is_complete(response: &Value, usage: TokenUsage) -> bool {
let Some(raw) = response.get("usage").filter(|value| value.is_object()) else {
return false;
};
let input = raw
.get("input_tokens")
.or_else(|| raw.get("prompt_tokens"))
.and_then(Value::as_u64);
let output = raw
.get("output_tokens")
.or_else(|| raw.get("completion_tokens"))
.and_then(Value::as_u64);
let cached = raw
.pointer("/input_tokens_details/cached_tokens")
.or_else(|| raw.pointer("/prompt_tokens_details/cached_tokens"))
.or_else(|| raw.get("cached_tokens"));
let cached = match cached {
Some(value) => value.as_u64(),
None => Some(0),
};
input == Some(usage.input_tokens)
&& output == Some(usage.output_tokens)
&& cached == Some(usage.cached_tokens)
&& usage.cached_tokens <= usage.input_tokens
}
fn model_pricing(model: &str) -> Option<ModelPricing> {
match model {
"grok-4.5" | "grok-4.5-latest" | "grok-4.5-build-free" | "grok-build-latest" => {
+5
View File
@@ -550,6 +550,11 @@ request/response/upstream ID、outcome 与搜索文本。诊断 `dimension` 可
汇总与洞察中的请求数与 outcome 分布覆盖筛选范围内全部请求;token、缓存、延迟与成本聚合仅统计
已完整交付客户端的成功响应。
Dashboard 的 `accountUsage[]` 由后端提供 `usageWindow`、`metricLabel`、`metricValue`。
`usageWindow` 复用账号额度窗口合同,缺失额度事实时为 `null`;窗口标签、百分比、触顶状态、重置时间
和本地用量由 Provider/Admin 投影。前端不得从套餐缺失推断免费套餐,也不得从显示时舍入的百分比推断
触顶。滚动窗口使用相应时间范围的本地用量,独立于 Dashboard 的今日统计范围。
OpenAI 的 `serviceTier` 只接受上游响应生命周期事件确认的实际 `response.service_tier`;请求里的
期望档位只保留在 request summary,不能冒充响应事实。计费展示把 `priority`/`fast` 映射为 `Fast`,
`flex` 映射为 `Flex`,缺失或 `default` 映射为 `Default`;未知非空值原样展示。Fast 优先使用模型的
+11 -2
View File
@@ -103,14 +103,23 @@ flowchart LR
| 模块 | 唯一责任 |
| --- | --- |
| `account` | Provider 账号/credential/quota 值对象、持久化端口与请求级账号选择;门面下分 `model`、`store`、`selection`、`error` |
| `validation` | 纯值对象校验错误和文本约束,不依赖事件、执行错误或路由 |
| `identity` | Provider 身份值 `ProviderKind`,只依赖纯校验 |
| `account` | Provider 账号/credential/quota 值对象、持久化端口与请求级账号选择;`scope` 持有分组、账号目录和冻结账号范围 |
| `metering` | 标准化 Usage、金额、费用估算与费用明细;不表示账号或开票系统 |
| `upstream` | 跨 Engine、Event、Error 与 Provider 共用的 transport 名称和发送状态 |
| `upstream` | 跨 Engine、Event、Error 与 Provider 共用的 transport 名称、发送状态和不透明上游值 |
| `lifecycle` | 取消信号、连接注册与 drain 合同 |
| `engine` | attempt、发送/提交屏障、执行编排和持久化调用时序 |
| `routing` | 冻结路由事实、请求计划以及运行时快照的表示与编译 |
| `runtime` | 当前快照的发布、读取、revision 订阅与周期对账任务 |
`event` 通过 `validation` / `upstream` 使用基础值,不再依赖承载原始事件的执行错误;账号值对象和
选择策略通过 `identity` / `account::scope` 使用身份与范围,不依赖路由计划。`routing` 和 `error` 的
既有公开类型路径保留 re-export,定义与 Core 内部使用均归属上述 owner。架构测试约束这些叶子依赖。
`engine::observation` 统一维护单次响应的用量、费用、时间和响应 ID,并负责重试前清理;协调器继续
独占发送、提交、重试和终结顺序。Provider 上报费用优先于本地估算,丢弃的 attempt 不得污染最终计量。
## 4. 数据面请求生命周期
```mermaid
+4
View File
@@ -1,3 +1,4 @@
import type { AccountQuotaWindow } from './accounts'
import type { UsageListRecord } from './usage'
import request from '../request'
@@ -145,6 +146,9 @@ export interface DashboardAccountUsage {
requestCount: number
requestBuckets: DashboardAccountRequestBucket[]
quotaUsedPercent: number | null
usageWindow: AccountQuotaWindow | null
metricLabel: string
metricValue: string
lastUsed: string
}
+8 -26
View File
@@ -1,7 +1,7 @@
import { clamp } from 'es-toolkit'
import { onScopeDispose, shallowRef } from 'vue'
import { shallowRef } from 'vue'
import { errorMessage } from '@/utils/async'
import { useRequestState } from './useRequestState'
interface PageResult {
items: unknown[]
@@ -23,23 +23,18 @@ export function usePagedQuery<Result extends PageResult>(options: {
const pageSize = shallowRef(options.initialPageSize)
const total = shallowRef(0)
const items = shallowRef<Result['items'][number][]>([])
const loading = shallowRef(false)
const error = shallowRef('')
let requestSequence = 0
const request = useRequestState(options.onError)
const { loading, error, invalidate } = request
async function execute(execution: { silent?: boolean } = {}) {
const requestId = ++requestSequence
if (!execution.silent) {
loading.value = true
error.value = ''
}
const requestId = request.start(execution.silent)
try {
const result = await options.load({
page: page.value,
pageSize: pageSize.value,
})
if (requestId !== requestSequence)
if (!request.isCurrent(requestId))
return false
if (result.items.length === 0 && result.page.total > 0 && result.page.page > 1) {
@@ -55,27 +50,14 @@ export function usePagedQuery<Result extends PageResult>(options: {
return true
}
catch (cause: unknown) {
if (requestId !== requestSequence)
return false
if (!execution.silent) {
error.value = errorMessage(cause, '加载失败')
options.onError?.(cause)
}
request.fail(requestId, cause, execution.silent)
return false
}
finally {
if (requestId === requestSequence)
loading.value = false
request.finish(requestId)
}
}
function invalidate() {
requestSequence += 1
loading.value = false
}
onScopeDispose(invalidate)
return {
page,
pageSize,
@@ -0,0 +1,43 @@
import { onScopeDispose, shallowRef } from 'vue'
import { errorMessage } from '@/utils/async'
export function useRequestState(onError?: (error: unknown) => void) {
const loading = shallowRef(false)
const error = shallowRef('')
let sequence = 0
function start(silent = false) {
const requestId = ++sequence
if (!silent) {
loading.value = true
error.value = ''
}
return requestId
}
function isCurrent(requestId: number) {
return requestId === sequence
}
function fail(requestId: number, cause: unknown, silent = false) {
if (!isCurrent(requestId) || silent)
return
error.value = errorMessage(cause, '加载失败')
onError?.(cause)
}
function finish(requestId: number) {
if (isCurrent(requestId))
loading.value = false
}
function invalidate() {
sequence += 1
loading.value = false
}
onScopeDispose(invalidate)
return { loading, error, start, isCurrent, fail, finish, invalidate }
}
@@ -1,6 +1,6 @@
import { onScopeDispose, shallowRef } from 'vue'
import { shallowRef } from 'vue'
import { errorMessage } from '@/utils/async'
import { useRequestState } from './useRequestState'
interface PageResult {
items: unknown[]
@@ -14,7 +14,6 @@ interface PageRequest {
pageSize: number
}
/** Element Plus 风格的服务端页码分页;一次 execute 只发起一次列表请求。 */
export function useStablePagedQuery<Result extends PageResult>(options: {
initialPageSize: number
load: (pagination: PageRequest) => Promise<Result>
@@ -25,23 +24,18 @@ export function useStablePagedQuery<Result extends PageResult>(options: {
const pageSize = shallowRef(options.initialPageSize)
const total = shallowRef(0)
const items = shallowRef<Result['items'][number][]>([])
const loading = shallowRef(false)
const error = shallowRef('')
let requestSequence = 0
const request = useRequestState(options.onError)
const { loading, error, invalidate } = request
async function execute(targetPage = currentPage.value, execution: { silent?: boolean } = {}) {
const requestId = ++requestSequence
if (!execution.silent) {
loading.value = true
error.value = ''
}
const requestId = request.start(execution.silent)
try {
const result = await options.load({
currentPage: Math.max(1, targetPage),
pageSize: pageSize.value,
})
if (requestId !== requestSequence)
if (!request.isCurrent(requestId))
return false
items.value = result.items
@@ -52,17 +46,11 @@ export function useStablePagedQuery<Result extends PageResult>(options: {
return true
}
catch (cause: unknown) {
if (requestId !== requestSequence)
return false
if (!execution.silent) {
error.value = errorMessage(cause, '加载失败')
options.onError?.(cause)
}
request.fail(requestId, cause, execution.silent)
return false
}
finally {
if (requestId === requestSequence)
loading.value = false
request.finish(requestId)
}
}
@@ -72,13 +60,6 @@ export function useStablePagedQuery<Result extends PageResult>(options: {
return execute(1, execution)
}
function invalidate() {
requestSequence += 1
loading.value = false
}
onScopeDispose(invalidate)
return {
currentPage,
pageSize,
-11
View File
@@ -120,17 +120,6 @@ function mixHue(first: number, second: number, secondWeight: number): number {
return (first + difference * secondWeight + 360) % 360
}
/** 浅色 Surface 限制色度,并随层级加深继续收敛。 */
export function mixSurfaceTone(
background: string,
foreground: string,
foregroundWeight: number,
): string {
return mixColorTone(background, foreground, foregroundWeight, {
saturationLimit: Math.max(0, 0.42 - foregroundWeight),
})
}
export function withAlpha(color: string, alpha: number): string {
const normalized = normalizeHexColor(color) ?? DEFAULT_CUSTOM_THEME_COLOR
const alphaHex = Math.round(Math.min(1, Math.max(0, alpha)) * 255)
@@ -263,7 +263,7 @@ const statusBars = computed(() => {
</span>
<AccountUsageWindow
:window="account.usageWindow"
:window="account.usageWindow ?? undefined"
:show-local-value="false"
variant="metric"
/>
@@ -25,6 +25,8 @@ withDefaults(
metrics: DashboardSnapshotView['metrics']
trendPoints: DashboardTrendView['points']
trendSummary: DashboardTrendView['summary']
trendLoading?: boolean
trendError?: string
healthTimeline: DashboardSnapshotView['healthTimeline']
accountUsage: DashboardSnapshotView['accountUsage']
wireProfiles: DashboardSnapshotView['wireProfiles']
@@ -37,6 +39,8 @@ withDefaults(
loading: false,
refreshing: false,
lastRefreshedAt: '',
trendLoading: false,
trendError: '',
},
)
@@ -86,6 +90,8 @@ const trendKind = defineModel<DashboardTrendKind>('trendKind', { required: true
v-model:kind="trendKind"
:points="trendPoints"
:summary="trendSummary"
:loading="trendLoading"
:error="trendError"
@trend-change="emit('trendChange', $event)"
/>
<WireProfileCard :profiles="wireProfiles" />
@@ -11,10 +11,15 @@ import { useRequestTrendChart } from '../composables/useRequestTrendChart'
type TrendKind = ReturnType<typeof normalizeDashboardTrendKind>
type TrendView = ReturnType<typeof dashboardTrendView>
const props = defineProps<{
const props = withDefaults(defineProps<{
points: TrendView['points']
summary: TrendView['summary']
}>()
loading?: boolean
error?: string
}>(), {
loading: false,
error: '',
})
const emit = defineEmits<{
trendChange: [kind: TrendKind]
@@ -92,7 +97,12 @@ const {
<div class="relative h-55 w-full overflow-hidden">
<BaseChart v-if="hasSamples" :option="chartOption" :height="220" />
<BaseEmpty v-if="!hasSamples" surface="none" title="暂无趋势数据" class="h-full place-content-center" />
<BaseEmpty
v-else-if="!loading"
surface="none"
:title="error || '暂无趋势数据'"
class="h-full place-content-center"
/>
</div>
</div>
</template>
@@ -1,22 +1,24 @@
import type { Component } from 'vue'
import { Activity, FileText, Timer, Users } from '@lucide/vue'
import { useIntervalFn } from '@vueuse/core'
import { clamp } from 'es-toolkit'
import { computed, onMounted, shallowRef } from 'vue'
import { computed, onMounted, onScopeDispose, shallowRef } from 'vue'
import { getDashboardSummary, getDashboardTrend } from '@/api'
import { withMinimumDuration } from '@/utils/async'
import { errorMessage, withMinimumDuration } from '@/utils/async'
import { formatDateTime } from '@/utils/date'
import { formatCompactNumber, formatInteger } from '@/utils/number'
export function useDashboard() {
const activeTrendKind = shallowRef(normalizeDashboardTrendKind('usage'))
const snapshot = shallowRef(dashboardSnapshotView(null))
const trend = shallowRef(dashboardTrendView(null))
const trend = shallowRef<DashboardTrend | null>(null)
const trendLoading = shallowRef(false)
const trendError = shallowRef('')
const loading = shallowRef(false)
const refreshing = shallowRef(false)
const lastRefreshedAt = shallowRef('')
let trendRequestId = 0
let disposed = false
const metrics = computed(() => snapshot.value.metrics)
const healthTimeline = computed(() => snapshot.value.healthTimeline)
@@ -26,8 +28,11 @@ export function useDashboard() {
const poolSummary = computed(() => snapshot.value.poolSummary)
const capacityInfo = computed(() => snapshot.value.capacityInfo)
const rotationStrategy = computed(() => snapshot.value.rotationStrategy)
const trendPoints = computed(() => trend.value.points)
const trendSummary = computed(() => trend.value.summary)
const trendView = computed(() => dashboardTrendView(
trend.value?.kind === activeTrendKind.value ? trend.value : null,
))
const trendPoints = computed(() => trendView.value.points)
const trendSummary = computed(() => trendView.value.summary)
const { resume: startAutoRefresh } = useIntervalFn(
() => {
@@ -71,24 +76,46 @@ export function useDashboard() {
const trendKind = normalizeDashboardTrendKind(kind)
activeTrendKind.value = trendKind
const requestId = ++trendRequestId
trendLoading.value = true
trendError.value = ''
try {
const result = await getDashboardTrend({ kind: trendKind })
if (isCurrentTrendRequest(requestId, trendKind))
trend.value = dashboardTrendView(result)
trend.value = result
}
catch {
// 趋势请求失败时保留当前趋势,下一次刷新继续尝试。
catch (error: unknown) {
if (isCurrentTrendRequest(requestId, trendKind))
trendError.value = errorMessage(error, '趋势加载失败')
}
finally {
if (isCurrentTrendRequest(requestId, trendKind))
trendLoading.value = false
}
}
async function loadDashboardSnapshot() {
const trendKind = activeTrendKind.value
const requestId = ++trendRequestId
const summary = await getDashboardSummary({ kind: trendKind })
snapshot.value = dashboardSnapshotView(summary)
lastRefreshedAt.value = formatDateTime()
if (isCurrentTrendRequest(requestId, trendKind)) {
trend.value = dashboardTrendView(summary.trend)
trendLoading.value = true
trendError.value = ''
try {
const summary = await getDashboardSummary({ kind: trendKind })
if (disposed)
return
snapshot.value = dashboardSnapshotView(summary)
lastRefreshedAt.value = formatDateTime()
if (isCurrentTrendRequest(requestId, trendKind)) {
trend.value = summary.trend
}
}
catch (error: unknown) {
if (isCurrentTrendRequest(requestId, trendKind))
trendError.value = errorMessage(error, '趋势加载失败')
throw error
}
finally {
if (isCurrentTrendRequest(requestId, trendKind))
trendLoading.value = false
}
}
@@ -96,7 +123,7 @@ export function useDashboard() {
requestId: number,
kind: ReturnType<typeof normalizeDashboardTrendKind>,
) {
return requestId === trendRequestId && activeTrendKind.value === kind
return !disposed && requestId === trendRequestId && activeTrendKind.value === kind
}
onMounted(() => {
@@ -104,6 +131,11 @@ export function useDashboard() {
startAutoRefresh()
})
onScopeDispose(() => {
disposed = true
trendRequestId += 1
})
return {
loading,
refreshing,
@@ -112,6 +144,8 @@ export function useDashboard() {
metrics,
trendPoints,
trendSummary,
trendLoading,
trendError,
healthTimeline,
accountUsage,
wireProfiles,
@@ -212,7 +246,7 @@ export function dashboardSnapshotView(summary: DashboardSummary | null) {
return {
metrics: metricCards(summary?.cards ?? emptyCards, trendPoints),
healthTimeline: summary?.healthTimeline ?? emptyHealthTimeline,
accountUsage: (summary?.accountUsage ?? []).map(accountUsageItem),
accountUsage: (summary?.accountUsage ?? []),
wireProfiles: summary?.wireProfiles ?? [],
usageRecords: summary?.usageRecords ?? [],
poolSummary: summary?.poolSummary ?? null,
@@ -437,84 +471,6 @@ function usageTrendSummary(points: ReturnType<typeof aggregateUsageTrend>) {
]
}
function accountUsageItem(item: DashboardSummary['accountUsage'][number]) {
const usageWindow = dashboardUsageWindow(item)
const requestCount = usageWindow.localUsage?.requestCount
const usesDailyRequests = typeof requestCount === 'number'
return {
id: item.id,
provider: item.provider,
authenticationKind: item.authenticationKind,
email: item.email,
planType: item.planType,
tokens: item.tokens,
lastUsed: item.lastUsed,
metricLabel: usesDailyRequests ? '次数' : '今日 Token',
metricValue: usesDailyRequests
? requestCount > 0 ? usageWindow.localUsage?.requestCountDisplay : '—'
: item.tokens,
usageWindow,
}
}
function dashboardUsageWindow(item: DashboardSummary['accountUsage'][number]) {
const provider = item.provider.trim().toLowerCase()
const planType = item.planType?.trim().toLowerCase() || 'free'
if (provider !== 'xai' || planType !== 'free') {
const usedPercent = quotaUsedPercent(item.quotaUsedPercent)
return {
key: 'quota',
group: 'other',
limitId: null,
limitName: null,
role: null,
labelDisplay: '额度',
windowLabelDisplay: '额度',
usedPercent,
usedPercentDisplay: usedPercent === null ? '—' : `${usedPercent}%`,
limitReached: (usedPercent ?? 0) >= 100,
windowSeconds: null,
resetAtDisplay: '—',
}
}
const requestCount = typeof item.requestCount === 'number' && Number.isFinite(item.requestCount)
? Math.max(0, item.requestCount)
: 0
const requestBuckets = Array.isArray(item.requestBuckets)
? item.requestBuckets.map((bucket: { bucketStart: string, requestCount: number }) => ({
bucketStart: bucket.bucketStart,
requestCount: bucket.requestCount,
}))
: []
return {
key: 'dailyRequests',
group: 'other',
limitId: null,
limitName: null,
role: null,
labelDisplay: '日请求',
windowLabelDisplay: '日请求',
usedPercent: null,
usedPercentDisplay: '—',
limitReached: false,
windowSeconds: 86_400,
resetAtDisplay: '—',
localUsage: {
requestCount,
requestCountDisplay: formatCompactNumber(requestCount),
requestBuckets,
},
}
}
function quotaUsedPercent(value: number | null) {
if (typeof value !== 'number' || !Number.isFinite(value))
return null
return clamp(Math.round(value), 0, 100)
}
function formatDashboardUsd(value: string) {
const normalized = value.trim()
if (!normalized.startsWith('$'))
+4
View File
@@ -9,6 +9,8 @@ const {
metrics,
trendPoints,
trendSummary,
trendLoading,
trendError,
healthTimeline,
accountUsage,
wireProfiles,
@@ -31,6 +33,8 @@ const {
:metrics="metrics"
:trend-points="trendPoints"
:trend-summary="trendSummary"
:trend-loading="trendLoading"
:trend-error="trendError"
:health-timeline="healthTimeline"
:account-usage="accountUsage"
:wire-profiles="wireProfiles"
@@ -301,6 +301,22 @@ export const themeDashboardSummary: DashboardSummaryResponse = {
requestCount: 1_284,
requestBuckets: [],
quotaUsedPercent: 32,
metricLabel: '今日 Token',
metricValue: '926K',
usageWindow: {
key: 'short',
group: 'shortTerm',
limitId: null,
limitName: null,
role: 'primary',
labelDisplay: '5 小时',
windowLabelDisplay: '5 小时',
windowSeconds: 18_000,
usedPercent: 32,
usedPercentDisplay: '32%',
limitReached: false,
resetAtDisplay: '—',
},
lastUsed: '刚刚',
},
{
@@ -316,6 +332,30 @@ export const themeDashboardSummary: DashboardSummaryResponse = {
requestCount: Math.round(requestCount / 3),
})),
quotaUsedPercent: null,
metricLabel: '次数',
metricValue: '894',
usageWindow: {
key: 'free-rolling-24h',
group: 'shortTerm',
limitId: null,
limitName: null,
role: null,
labelDisplay: '日限额',
windowLabelDisplay: '日限额',
windowSeconds: 86_400,
usedPercent: null,
usedPercentDisplay: '—',
limitReached: false,
resetAtDisplay: '—',
localUsage: {
requestCount: 894,
requestCountDisplay: '894',
requestBuckets: requestSeries.slice(-12).map((requestCount, index) => ({
bucketStart: `2026-08-23T${String(index + 12).padStart(2, '0')}:00:00+08:00`,
requestCount: Math.round(requestCount / 3),
})),
},
},
lastUsed: '2 分钟前',
},
{
@@ -328,6 +368,22 @@ export const themeDashboardSummary: DashboardSummaryResponse = {
requestCount: 742,
requestBuckets: [],
quotaUsedPercent: 68,
metricLabel: '今日 Token',
metricValue: '512K',
usageWindow: {
key: 'short',
group: 'shortTerm',
limitId: null,
limitName: null,
role: 'primary',
labelDisplay: '5 小时',
windowLabelDisplay: '5 小时',
windowSeconds: 18_000,
usedPercent: 68,
usedPercentDisplay: '68%',
limitReached: false,
resetAtDisplay: '—',
},
lastUsed: '5 分钟前',
},
],
@@ -44,7 +44,8 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) {
})
const refreshingList = shallowRef(false)
const diagnosticDimension = shallowRef('model')
let loadRequestId = 0
let tableRequestId = 0
let analyticsRequestId = 0
let diagnosticRequestId = 0
const scopedParams = () => ({
...options.timeRangeParams.value,
@@ -66,91 +67,83 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) {
async function loadUsageRecords(loadOptions: UsageLoadOptions = {}) {
const { scope = 'all', background = false } = loadOptions
const requestId = ++loadRequestId
try {
if (background) {
loading.value = false
if (scope === 'all') {
analyticsLoading.value = false
}
}
else {
loading.value = true
if (scope === 'all') {
analyticsLoading.value = true
}
}
if (scope === 'all') {
diagnosticRequestId += 1
}
const globalParams = scopedParams()
if (scope === 'all') {
resetPagination()
tableTimeRangeParams.value = { ...globalParams }
}
const globalParams = scopedParams()
if (scope === 'all') {
resetPagination()
tableTimeRangeParams.value = { ...globalParams }
}
const tableParams = {
...tableTimeRangeParams.value,
...filterParams(),
}
const resultPromise = getUsageRecords({
await Promise.all([
loadUsagePage(background),
...(scope === 'all' ? [loadUsageAnalytics(globalParams, background)] : []),
])
}
async function loadUsagePage(background: boolean) {
const requestId = ++tableRequestId
loading.value = !background
try {
const result = await getUsageRecords({
currentPage: currentPage.value,
pageSize: pageSize.value,
...tableParams,
...tableTimeRangeParams.value,
...filterParams(),
})
const analyticsPromise
= scope === 'all'
? loadUsageAnalytics(globalParams)
: Promise.resolve({
summary: summary.value,
insights: insights.value,
})
const [result, nextAnalytics] = await Promise.all([resultPromise, analyticsPromise])
if (requestId !== loadRequestId)
if (requestId !== tableRequestId)
return
records.value = result.items
summary.value = nextAnalytics.summary
insights.value = {
...nextAnalytics.insights,
diagnostics:
nextAnalytics.insights.diagnostics.dimension === diagnosticDimension.value
? nextAnalytics.insights.diagnostics
: insights.value.diagnostics,
}
pageSize.value = result.pageSize
totalRecords.value = result.total
currentPage.value = result.currentPage
}
catch (error: unknown) {
if (requestId !== loadRequestId)
if (requestId !== tableRequestId)
return
toast.error(errorMessage(error, '加载失败'))
}
finally {
if (requestId === loadRequestId && !background) {
if (requestId === tableRequestId) {
loading.value = false
if (scope === 'all') {
analyticsLoading.value = false
}
}
}
}
async function loadUsageAnalytics(globalParams = scopedParams()) {
async function loadUsageAnalytics(globalParams: ReturnType<typeof scopedParams>, background: boolean) {
const requestId = ++analyticsRequestId
const diagnosticsId = ++diagnosticRequestId
const dimension = diagnosticDimension.value
const [nextSummary, overview, diagnostics] = await Promise.all([
getUsageRecordSummary(globalParams),
getUsageRecordInsightsOverview(globalParams),
getUsageRecordInsightsDiagnostics({
...globalParams,
dimension,
}),
])
analyticsLoading.value = !background
try {
const [nextSummary, overview, diagnostics] = await Promise.all([
getUsageRecordSummary(globalParams),
getUsageRecordInsightsOverview(globalParams),
getUsageRecordInsightsDiagnostics({
...globalParams,
dimension,
}),
])
if (requestId !== analyticsRequestId)
return
return {
summary: nextSummary,
insights: { overview, diagnostics },
summary.value = nextSummary
insights.value = {
overview,
diagnostics:
diagnosticsId === diagnosticRequestId && dimension === diagnosticDimension.value
? diagnostics
: insights.value.diagnostics,
}
}
catch (error: unknown) {
if (requestId !== analyticsRequestId)
return
toast.error(errorMessage(error, '加载失败'))
}
finally {
if (requestId === analyticsRequestId) {
analyticsLoading.value = false
}
}
}
@@ -171,6 +164,8 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) {
}
}
catch (error: unknown) {
if (requestId !== diagnosticRequestId || dimension !== diagnosticDimension.value)
return
toast.error(errorMessage(error, '加载失败'))
}
}
@@ -222,7 +217,8 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) {
)
onScopeDispose(() => {
loadRequestId += 1
tableRequestId += 1
analyticsRequestId += 1
diagnosticRequestId += 1
})