diff --git a/backend/scripts/check_architecture.sh b/backend/scripts/check_architecture.sh index 5fdf124c5..cbcfcdc1d 100755 --- a/backend/scripts/check_architecture.sh +++ b/backend/scripts/check_architecture.sh @@ -20,8 +20,8 @@ domain_paths=( src/accounts src/models src/dispatch ) -deny "domain -> api/bootstrap" 'crate::(api|bootstrap)\b' "${domain_paths[@]}" -deny "upstream -> domain" 'crate::(accounts|dispatch|telemetry|keys|auth|settings|update|models)\b' src/upstream +deny "domain -> api/bootstrap" '(^|[^[:alnum:]_])(api|bootstrap)::' "${domain_paths[@]}" +deny "upstream -> domain" '(^|[^[:alnum:]_])(accounts|dispatch|telemetry|keys|auth|settings|update|models)::' src/upstream deny "axum outside api/bootstrap" '\baxum(::|\b)' "${domain_paths[@]}" deny "sqlx in api" '\bsqlx(::|\b)' src/api deny "redis in api" '\bredis::' src/api @@ -33,6 +33,38 @@ deny "persistence repository naming" '\b[A-Za-z0-9_]*repository[A-Za-z0-9_]*\b' src/upstream/openai/fingerprint deny "Arc alias" '\bArc[[:space:]]+as[[:space:]]+' src +storage_command_check() { + local kind=$1 + local pattern=$2 + local file + while IFS= read -r file; do + case "$file" in + */store.rs | */store/*) + continue + ;; + esac + if [ "$kind" = sql ]; then + case "$file" in + src/infra/database.rs | src/bootstrap/import_sqlite.rs | src/bootstrap/import_sqlite/* | src/telemetry/rebuild.rs | */query.rs) + continue + ;; + esac + else + case "$file" in + src/infra/redis.rs | src/accounts/refresh/lease.rs) + continue + ;; + esac + fi + echo "$file" + echo "ARCH VIOLATION: $kind command outside an allowed storage adapter" >&2 + fail=1 + done < <(grep -rl --include='*.rs' -E "$pattern" src || true) +} + +storage_command_check sql 'sqlx::(query|query_as|query_scalar|raw_sql)' +storage_command_check redis 'redis::(cmd|pipe)|[.]query_async[[:space:]]*[(]|use redis::.*AsyncCommands' + banned=$(find src \( -name 'importing.rs' -o -name 'exporting.rs' -o -name 'testing.rs' \ -o -name 'util.rs' -o -name 'helper.rs' -o -name 'common.rs' -o -name 'misc.rs' \)) if [ -n "$banned" ]; then diff --git a/backend/src/api/admin/dashboard_routes.rs b/backend/src/api/admin/dashboard_routes.rs index e6807fcf9..09e7a6c95 100644 --- a/backend/src/api/admin/dashboard_routes.rs +++ b/backend/src/api/admin/dashboard_routes.rs @@ -36,13 +36,13 @@ use crate::{ }, }, telemetry::{ - account_usage::query::AccountUsageTimeBucket, + account_usage::query::{AccountUsageSummary, AccountUsageTimeBucket}, diagnostics::{ fingerprint_diagnostics, AccountCapacityDiagnostics, AccountPoolDiagnostics, }, usage::{ query::UsageQueryFilter, - store::{UsageRecordAccountUsage, UsageRecordSummary}, + store::UsageRecordAccountUsage, types::{metadata_string, UsageRecord}, }, }, @@ -292,12 +292,7 @@ pub(crate) async fn dashboard_summary( let capacity = state.services.account_pool.capacity_summary_now().await; let now = Utc::now(); let today_filter = today_usage_record_filter(now); - let usage_summary = state - .services - .usage_records - .summary(UsageQueryFilter::default()) - .await - .unwrap_or_else(|_| UsageRecordSummary::default()); + let lifetime_usage = state.services.usage.summary().await.unwrap_or_default(); let account_usage_records = state .services .usage_records @@ -327,7 +322,7 @@ pub(crate) async fn dashboard_summary( let trend = dashboard_trend_data(&time_buckets, query.kind.unwrap_or_default()); let quota_used_by_account = account_quota_used_percent_by_id(&state, &account_usage_records).await; - let settings = state.services.settings.current(); + let settings = state.services.settings.current().await; let recent_usage_records = recent_events; let account_emails = state .services @@ -341,7 +336,7 @@ pub(crate) async fn dashboard_summary( Ok(AdminResponse::new( StatusCode::OK, AdminEnvelope::ok(DashboardSummaryData { - cards: dashboard_cards(&accounts, &time_buckets, &usage_summary), + cards: dashboard_cards(&accounts, &time_buckets, &lifetime_usage), trend, health_timeline: dashboard_health_timeline_data(&time_buckets), account_usage: account_usage_data( diff --git a/backend/src/api/admin/dashboard_routes/data.rs b/backend/src/api/admin/dashboard_routes/data.rs index a3e5d87e3..02e898b1d 100644 --- a/backend/src/api/admin/dashboard_routes/data.rs +++ b/backend/src/api/admin/dashboard_routes/data.rs @@ -3,15 +3,15 @@ use super::*; pub(super) fn dashboard_cards( accounts: &[Account], buckets: &[AccountUsageTimeBucket], - usage_summary: &UsageRecordSummary, + lifetime_usage: &AccountUsageSummary, ) -> DashboardCardsData { - dashboard_cards_at(accounts, buckets, usage_summary, Utc::now()) + dashboard_cards_at(accounts, buckets, lifetime_usage, Utc::now()) } pub(super) fn dashboard_cards_at( accounts: &[Account], buckets: &[AccountUsageTimeBucket], - usage_summary: &UsageRecordSummary, + lifetime_usage: &AccountUsageSummary, now: DateTime, ) -> DashboardCardsData { let today_start = china_day_start(now); @@ -19,10 +19,10 @@ pub(super) fn dashboard_cards_at( let today = usage_window(buckets, today_start, now); let yesterday = usage_window(buckets, yesterday_start, today_start); let today_cost = cost_window(buckets, today_start, now).unwrap_or(0.0); - let total_requests = usage_summary.total_requests; - let total_tokens = usage_summary.total_tokens; - let total_cached_tokens = usage_summary.cached_tokens; - let total_hit_rate = summary_cache_hit_rate(usage_summary); + let total_requests = nonnegative_i64_to_u64(lifetime_usage.request_count); + let total_tokens = nonnegative_i64_to_u64(lifetime_usage.total_tokens); + let total_cached_tokens = nonnegative_i64_to_u64(lifetime_usage.cached_tokens); + let total_hit_rate = summary_cache_hit_rate(lifetime_usage); let total_accounts = accounts.len() as u64; let enabled_accounts = accounts @@ -77,10 +77,10 @@ pub(super) fn dashboard_cards_at( } } -pub(super) fn summary_cache_hit_rate(summary: &UsageRecordSummary) -> Option { - let input_tokens = summary.input_tokens; +pub(super) fn summary_cache_hit_rate(summary: &AccountUsageSummary) -> Option { + let input_tokens = nonnegative_i64_to_u64(summary.input_tokens); if input_tokens > 0 { - Some(summary.cached_tokens as f64 / input_tokens as f64) + Some(nonnegative_i64_to_u64(summary.cached_tokens) as f64 / input_tokens as f64) } else { None } diff --git a/backend/src/api/admin/settings_routes.rs b/backend/src/api/admin/settings_routes.rs index 6b37d233e..eb653f020 100644 --- a/backend/src/api/admin/settings_routes.rs +++ b/backend/src/api/admin/settings_routes.rs @@ -52,7 +52,7 @@ pub(crate) async fn settings( State(state): State, _auth: AdminAuth, ) -> Result { - let config = state.services.settings.current(); + let config = state.services.settings.current().await; Ok(AdminResponse::new( StatusCode::OK, AdminEnvelope::ok(AdminSettingsData::from_settings(&config)), @@ -68,11 +68,9 @@ pub(crate) async fn update_settings( let patch = parse_settings_patch(&body)?; let config = state .services - .settings - .update(patch) + .update_settings(patch) .await .map_err(settings_service_error)?; - state.services.apply_settings(&config).await; Ok(AdminResponse::new( StatusCode::OK, diff --git a/backend/src/api/admin/usage_routes.rs b/backend/src/api/admin/usage_routes.rs index 19842f0c5..0123c47fd 100644 --- a/backend/src/api/admin/usage_routes.rs +++ b/backend/src/api/admin/usage_routes.rs @@ -447,11 +447,9 @@ fn optional_datetime( fn log_error(error: &UsageQueryError) -> AdminError { match error { - UsageQueryError::List - | UsageQueryError::Get - | UsageQueryError::Clear - | UsageQueryError::Append - | UsageQueryError::Retention => AdminError::internal(error.to_string()), + UsageQueryError::List | UsageQueryError::Get | UsageQueryError::Clear => { + AdminError::internal(error.to_string()) + } UsageQueryError::Accounts => AdminError::usage_record_accounts_failed(error.to_string()), } } diff --git a/backend/src/api/client/errors.rs b/backend/src/api/client/errors.rs index 054e08a2f..311aa0ada 100644 --- a/backend/src/api/client/errors.rs +++ b/backend/src/api/client/errors.rs @@ -7,9 +7,10 @@ use axum::{ }; use serde_json::{json, Value}; -use crate::dispatch::{errors::DispatchFailureClass, responses::errors::ResponseDispatchError}; - -use super::responses::response_failed_sse_event as encode_response_failed_sse_event; +use crate::{ + dispatch::{errors::DispatchFailureClass, responses::errors::ResponseDispatchError}, + upstream::openai::protocol::sse::response_failed_sse_event as encode_response_failed_sse_event, +}; const NO_ACTIVE_UPSTREAM_ACCOUNT_MESSAGE: &str = "No active upstream account is available"; const NO_AVAILABLE_RESPONSES_ACCOUNTS_MESSAGE: &str = diff --git a/backend/src/api/client/responses.rs b/backend/src/api/client/responses.rs index e5cd4b369..1c490d4a8 100644 --- a/backend/src/api/client/responses.rs +++ b/backend/src/api/client/responses.rs @@ -8,7 +8,7 @@ use axum::{ response::{IntoResponse, Response}, Extension, Json, }; -use serde_json::{json, Map, Value}; +use serde_json::{Map, Value}; use crate::{ api::middleware::request_id::{ClientIp, RequestId}, @@ -16,7 +16,7 @@ use crate::{ dispatch::responses::{errors::ResponseDispatchError, service::ResponseDispatchStream}, upstream::openai::protocol::{ responses::{CodexCompactRequest, CodexResponsesRequest}, - sse::{encode_sse_event, sse_body_has_done}, + sse::sse_body_has_done, }, }; @@ -191,41 +191,6 @@ impl ContextField { } } -/// 编码 OpenAI Responses `response.failed` SSE 事件。 -pub fn response_failed_sse_event(error_type: &str, code: &str, message: &str) -> String { - response_failed_sse_event_with_id(None, error_type, code, message) -} - -/// 使用指定 response id 编码 OpenAI Responses `response.failed` SSE 事件。 -pub fn response_failed_sse_event_with_id( - response_id: Option<&str>, - error_type: &str, - code: &str, - message: &str, -) -> String { - let error = json!({ - "type": error_type, - "code": code, - "message": message, - }); - let response_id = response_id - .filter(|value| !value.trim().is_empty()) - .map_or_else( - || format!("resp_proxy_{}", uuid::Uuid::new_v4().simple()), - ToString::to_string, - ); - let data = json!({ - "type": "response.failed", - "response": { - "id": response_id, - "status": "failed", - "error": error, - }, - "error": error, - }); - encode_sse_event("response.failed", &data.to_string()) -} - // ==================================================================== // HTTP 处理器 // ==================================================================== diff --git a/backend/src/bootstrap/import_sqlite/core.rs b/backend/src/bootstrap/import_sqlite/core.rs index 7860d097d..4c49a5aee 100644 --- a/backend/src/bootstrap/import_sqlite/core.rs +++ b/backend/src/bootstrap/import_sqlite/core.rs @@ -112,7 +112,7 @@ async fn import_runtime_settings( let aliases: String = row.try_get("model_aliases_json")?; let admin_key = row .try_get::, _>("admin_api_key")? - .map(|key| hash_credential(&key)); + .and_then(|key| (!key.trim().is_empty()).then(|| hash_credential(&key))); sqlx::query( "insert into runtime_settings (id, model_aliases_json, refresh_margin_seconds, refresh_concurrency, diff --git a/backend/src/bootstrap/services.rs b/backend/src/bootstrap/services.rs index 39d6a3088..19186f975 100644 --- a/backend/src/bootstrap/services.rs +++ b/backend/src/bootstrap/services.rs @@ -29,13 +29,14 @@ use crate::{ service::{ModelService, ModelServiceError}, store::{ModelSnapshotStore, RedisModelSnapshotStore}, }, - settings::{service::RuntimeSettingsService, SettingsSnapshot}, + settings::{service::RuntimeSettingsService, SettingsError, SettingsPatch, SettingsSnapshot}, telemetry::{ account_usage::query::AccountUsageQueryService, account_usage::store::PgAccountUsageStore, buckets::store::PgRequestBucketStore, ops::query::OpsQueryService, ops::store::PgOpsErrorLogStore, + recorder::Recorder, usage::query::UsageQueryService, usage::store::{PgUsageRecordStore, DEFAULT_USAGE_RECORD_CAPTURE_BODY}, }, @@ -109,6 +110,7 @@ pub struct Services { pub installation_id: Option, pub background_tasks: BackgroundTaskStores, account_pool_static: AccountPoolStaticSettings, + settings_update_lock: Arc>, } #[derive(Debug, Clone)] @@ -203,13 +205,12 @@ impl Services { Arc::new(client) } }; - let usage_records = Arc::new(UsageQueryService::new( + let usage_records = Arc::new(UsageQueryService::new(stores.usage_records.clone())); + let ops_errors = Arc::new(OpsQueryService::new(stores.ops_errors.clone())); + let recorder = Arc::new(Recorder::new( stores.usage_records.clone(), - usage_record_options.enabled, - usage_record_options.capture_body, - )); - let ops_errors = Arc::new(OpsQueryService::new( stores.ops_errors.clone(), + usage_record_options.enabled, usage_record_options.capture_body, )); let account_pool_static = AccountPoolStaticSettings { @@ -287,8 +288,7 @@ impl Services { models: models.clone(), codex: codex.clone(), session_affinity: session_affinity.clone(), - usage_records: usage_records.clone(), - ops_errors: ops_errors.clone(), + recorder, installation_id: installation_id.clone(), cloudflare: cloudflare_recovery, })); @@ -317,6 +317,7 @@ impl Services { installation_id, background_tasks: stores, account_pool_static, + settings_update_lock: Arc::new(tokio::sync::Mutex::new(())), }) } @@ -326,8 +327,18 @@ impl Services { Ok(()) } - /// 将已持久化的设置快照同步到各运行时消费者。 - pub async fn apply_settings(&self, settings: &SettingsSnapshot) { + /// 持久化设置并同步到所有运行时消费者。 + pub async fn update_settings( + &self, + patch: SettingsPatch, + ) -> Result, SettingsError> { + let _guard = self.settings_update_lock.lock().await; + let settings = self.settings.update(patch).await?; + self.apply_settings(&settings).await; + Ok(settings) + } + + async fn apply_settings(&self, settings: &SettingsSnapshot) { self.models.update_config(ModelConfig { model_aliases: settings.model_aliases.clone(), }); diff --git a/backend/src/dispatch/responses/event_recording.rs b/backend/src/dispatch/responses/event_recording.rs index d3aa55412..a2a95b81c 100644 --- a/backend/src/dispatch/responses/event_recording.rs +++ b/backend/src/dispatch/responses/event_recording.rs @@ -9,12 +9,11 @@ use crate::{ }, infra::time::elapsed_millis_i64, telemetry::{ - ops::query::OpsQueryService, ops::types::OpsErrorLog, recorder::{ enrich_event_route_metadata, enrich_usage_record_identity, event_kind as response_event_kind, reasoning_effort_from_request, - record_dispatch_error_event, DispatchErrorLogRecord, + record_dispatch_error_event, DispatchErrorLogRecord, Recorder, }, usage::types::UsageRecordLevel, }, @@ -37,7 +36,7 @@ use super::{ }; pub(super) struct ResponseUpstreamErrorEventRecord<'a> { - pub(super) ops_errors: &'a OpsQueryService, + pub(super) recorder: &'a Recorder, pub(super) request_id: &'a str, pub(super) account_id: &'a str, pub(super) account_email: Option<&'a str>, @@ -53,7 +52,7 @@ pub(super) struct ResponseUpstreamErrorEventRecord<'a> { } pub(super) struct ResponseStreamFailureEventRecord<'a> { - pub(super) ops_errors: &'a OpsQueryService, + pub(super) recorder: &'a Recorder, pub(super) request_id: &'a str, pub(super) account_id: &'a str, pub(super) route: &'a str, @@ -71,7 +70,7 @@ pub(super) struct ResponseStreamFailureEventRecord<'a> { } pub(super) struct ResponseDispatchErrorEventRecord<'a> { - pub(super) ops_errors: &'a OpsQueryService, + pub(super) recorder: &'a Recorder, pub(super) request_id: &'a str, pub(super) client_api_key_id: Option<&'a str>, pub(super) account_id: Option<&'a str>, @@ -111,7 +110,7 @@ pub(super) async fn record_response_dispatch_error_event( ); } record_dispatch_error_event(DispatchErrorLogRecord { - ops_errors: record.ops_errors, + recorder: record.recorder, request_id: record.request_id, client_api_key_id: record.client_api_key_id, provider: Some("openai"), @@ -177,7 +176,7 @@ pub(super) async fn record_response_upstream_error_event( record.request.service_tier(), ); event.metadata = metadata; - if let Err(error) = record.ops_errors.record(event).await { + if let Err(error) = record.recorder.record_error(event).await { tracing::warn!(account_id = %record.account_id, error = %error, "failed to record upstream error event"); } } @@ -244,7 +243,7 @@ pub(super) async fn record_prefetched_response_stream_failure_event( } } event.metadata = metadata; - if let Err(error) = record.ops_errors.record(event).await { + if let Err(error) = record.recorder.record_error(event).await { tracing::warn!(account_id = %record.account_id, error = %error, "failed to record prefetched stream error event"); } } @@ -506,7 +505,7 @@ pub(super) async fn record_live_response_stream_event( if is_success_usage_event(level, status_code) { crate::telemetry::recorder::record_response_event( crate::telemetry::usage::types::ResponseUsageRecord { - usage_records: &context.usage_records, + recorder: &context.recorder, request_id: &context.request_id, client_api_key_id: context.request.client_api_key_id.as_deref(), account_id: &context.account_id, @@ -536,7 +535,7 @@ pub(super) async fn record_live_response_stream_event( event.status_code = Some(status_code); event.latency_ms = Some(elapsed_millis_i64(context.started_at)); event.metadata = metadata; - if let Err(error) = context.ops_errors.record(event).await { + if let Err(error) = context.recorder.record_error(event).await { tracing::warn!(account_id = %context.account_id, error = %error, "failed to record live response stream error event"); } } diff --git a/backend/src/dispatch/responses/service.rs b/backend/src/dispatch/responses/service.rs index 33a9f9b93..4573de2a2 100644 --- a/backend/src/dispatch/responses/service.rs +++ b/backend/src/dispatch/responses/service.rs @@ -50,12 +50,10 @@ use crate::{ }, models::service::ModelService, telemetry::{ - ops::query::OpsQueryService, recorder::{ reasoning_effort_from_compact_request, reasoning_effort_from_request, - record_response_event, + record_response_event, Recorder, }, - usage::query::UsageQueryService, usage::types::ResponseUsageRecord, }, upstream::openai::{ @@ -107,8 +105,7 @@ pub struct ResponseDispatchService { codex: Arc, session_affinity: Arc, reasoning_replay: Arc>, - usage_records: Arc, - ops_errors: Arc, + recorder: Arc, installation_id: Option, cloudflare: CloudflareRecovery, } @@ -118,8 +115,7 @@ pub(crate) struct ResponseDispatchServiceParts { pub models: Arc, pub codex: Arc, pub session_affinity: Arc, - pub usage_records: Arc, - pub ops_errors: Arc, + pub recorder: Arc, pub installation_id: Option, pub cloudflare: CloudflareRecovery, } @@ -175,8 +171,7 @@ impl ResponseDispatchService { reasoning_replay: Arc::new(Mutex::new(ReasoningReplayCache::new(Duration::seconds( DEFAULT_REASONING_REPLAY_TTL_SECS, )))), - usage_records: parts.usage_records, - ops_errors: parts.ops_errors, + recorder: parts.recorder, installation_id: parts.installation_id, cloudflare: parts.cloudflare, } diff --git a/backend/src/dispatch/responses/service/compact.rs b/backend/src/dispatch/responses/service/compact.rs index 1e1e4a2a7..7b4e81fd5 100644 --- a/backend/src/dispatch/responses/service/compact.rs +++ b/backend/src/dispatch/responses/service/compact.rs @@ -138,7 +138,7 @@ impl ResponseDispatchService { insert_response_upstream_diagnostics(&mut metadata, &response.diagnostics); insert_response_trace_metadata(&mut metadata, &trace, Some(&attempt)); record_response_event(ResponseUsageRecord { - usage_records: &self.usage_records, + recorder: &self.recorder, request_id, client_api_key_id: request.client_api_key_id.as_deref(), account_id: &account.id, @@ -273,7 +273,7 @@ impl ResponseDispatchService { error: &ResponseDispatchError, ) { record_response_dispatch_error_event(ResponseDispatchErrorEventRecord { - ops_errors: &self.ops_errors, + recorder: &self.recorder, request_id, client_api_key_id, account_id, diff --git a/backend/src/dispatch/responses/service/complete.rs b/backend/src/dispatch/responses/service/complete.rs index 12ca29dab..547edc597 100644 --- a/backend/src/dispatch/responses/service/complete.rs +++ b/backend/src/dispatch/responses/service/complete.rs @@ -397,7 +397,7 @@ impl ResponseDispatchService { } Err(error) => { record_response_upstream_error_event(ResponseUpstreamErrorEventRecord { - ops_errors: &self.ops_errors, + recorder: &self.recorder, request_id, account_id: &release_account_id, account_email: account.email.as_deref(), @@ -456,7 +456,7 @@ impl ResponseDispatchService { insert_response_trace_metadata(&mut metadata, &trace, Some(&attempt)); insert_websocket_pool_decision(&mut metadata, response.websocket_pool_decision); record_response_event(ResponseUsageRecord { - usage_records: &self.usage_records, + recorder: &self.recorder, request_id, client_api_key_id: request.client_api_key_id.as_deref(), account_id: &account.id, @@ -546,7 +546,7 @@ impl ResponseDispatchService { error: &ResponseDispatchError, ) { record_response_dispatch_error_event(ResponseDispatchErrorEventRecord { - ops_errors: &self.ops_errors, + recorder: &self.recorder, request_id, client_api_key_id: details.client_api_key_id, account_id: details.account_id, diff --git a/backend/src/dispatch/responses/service/stream.rs b/backend/src/dispatch/responses/service/stream.rs index 4013fd253..7d12924d8 100644 --- a/backend/src/dispatch/responses/service/stream.rs +++ b/backend/src/dispatch/responses/service/stream.rs @@ -322,7 +322,7 @@ impl ResponseDispatchService { if let ResponseDispatchError::Upstream(upstream_error) = &error { record_response_upstream_error_event( ResponseUpstreamErrorEventRecord { - ops_errors: &self.ops_errors, + recorder: &self.recorder, request_id, account_id: &release_account_id, account_email: account.email.as_deref(), @@ -422,7 +422,7 @@ impl ResponseDispatchService { self.account_pool.release(&release_account_id).await; record_prefetched_response_stream_failure_event( ResponseStreamFailureEventRecord { - ops_errors: &self.ops_errors, + recorder: &self.recorder, request_id, account_id: &release_account_id, route, @@ -447,8 +447,7 @@ impl ResponseDispatchService { account_pool: Arc::clone(&self.account_pool), session_affinity: Arc::clone(&self.session_affinity), reasoning_replay: Arc::clone(&self.reasoning_replay), - usage_records: Arc::clone(&self.usage_records), - ops_errors: Arc::clone(&self.ops_errors), + recorder: Arc::clone(&self.recorder), cloudflare: self.cloudflare.clone(), account_id: account.id, account_plan_type: account.plan_type, @@ -580,7 +579,7 @@ impl ResponseDispatchService { Err(error) => { self.account_pool.release(&release_account_id).await; record_response_upstream_error_event(ResponseUpstreamErrorEventRecord { - ops_errors: &self.ops_errors, + recorder: &self.recorder, request_id, account_id: &release_account_id, account_email: account.email.as_deref(), diff --git a/backend/src/dispatch/responses/stream_lifecycle.rs b/backend/src/dispatch/responses/stream_lifecycle.rs index 431541c35..1119e101f 100644 --- a/backend/src/dispatch/responses/stream_lifecycle.rs +++ b/backend/src/dispatch/responses/stream_lifecycle.rs @@ -5,7 +5,6 @@ use tokio::sync::Mutex; use crate::{ accounts::pool::RuntimeAccountPoolService, - api::client::responses::response_failed_sse_event_with_id, dispatch::{ affinity::{ resolve::{evict_reasoning_replay, record_response_affinity}, @@ -15,15 +14,12 @@ use crate::{ reasoning_replay::ReasoningReplayCache, }, infra::time::elapsed_millis_i64, - telemetry::{ - ops::query::OpsQueryService, usage::query::UsageQueryService, - usage::types::UsageRecordLevel, - }, + telemetry::{recorder::Recorder, usage::types::UsageRecordLevel}, upstream::openai::{ protocol::{ events::extract_sse_usage, responses::{response_from_codex_sse, CodexResponsesRequest, CollectedResponse}, - sse::parse_sse_events, + sse::{parse_sse_events, response_failed_sse_event_with_id}, }, transport::{ CodexBackendTransport, CodexRateLimitHeaderUpdates, CodexTurnStateUpdate, @@ -50,8 +46,7 @@ pub(super) struct LiveResponseStreamContext { pub(super) account_pool: Arc, pub(super) session_affinity: Arc, pub(super) reasoning_replay: Arc>, - pub(super) usage_records: Arc, - pub(super) ops_errors: Arc, + pub(super) recorder: Arc, pub(super) cloudflare: CloudflareRecovery, pub(super) account_id: String, pub(super) account_plan_type: Option, diff --git a/backend/src/settings/service.rs b/backend/src/settings/service.rs index 2ba53f578..ebf1a7d49 100644 --- a/backend/src/settings/service.rs +++ b/backend/src/settings/service.rs @@ -1,9 +1,9 @@ -//! 运行时设置校验、持久化与快照广播。 +//! 运行时设置校验、持久化与当前快照。 use std::{collections::BTreeMap, sync::Arc}; use subtle::ConstantTimeEq; -use tokio::sync::watch; +use tokio::sync::RwLock; use crate::infra::identity::{generate_admin_api_key, hash_credential}; @@ -48,19 +48,18 @@ impl SettingsService { } } -/// 持久化设置的当前快照与变更广播。 +/// 持久化设置的当前快照。 #[derive(Clone)] pub struct RuntimeSettingsService { store: PgSettingsStore, - sender: watch::Sender>, + snapshot: Arc>>, } impl RuntimeSettingsService { pub fn new(settings: SettingsSnapshot, pool: sqlx::PgPool) -> Self { - let (sender, _) = watch::channel(Arc::new(settings)); Self { store: PgSettingsStore::new(pool), - sender, + snapshot: Arc::new(RwLock::new(Arc::new(settings))), } } @@ -73,23 +72,20 @@ impl RuntimeSettingsService { .await } - pub fn current(&self) -> Arc { - self.sender.borrow().clone() - } - - pub fn subscribe(&self) -> watch::Receiver> { - self.sender.subscribe() + pub async fn current(&self) -> Arc { + self.snapshot.read().await.clone() } pub async fn update( &self, patch: SettingsPatch, ) -> Result, SettingsError> { - let mut next = (*self.current()).clone(); + let mut current = self.snapshot.write().await; + let mut next = (**current).clone(); SettingsService::apply_patch(&mut next, patch)?; self.store.save(&next).await?; let next = Arc::new(next); - self.sender.send_replace(next.clone()); + *current = next.clone(); Ok(next) } @@ -98,7 +94,8 @@ impl RuntimeSettingsService { } pub async fn regenerate_admin_api_key(&self) -> Result { - self.store.ensure(&self.current()).await?; + let current = self.current().await; + self.store.ensure(¤t).await?; let key = generate_admin_api_key(); self.store .set_admin_api_key_hash(&hash_credential(&key)) diff --git a/backend/src/telemetry/account_usage/query.rs b/backend/src/telemetry/account_usage/query.rs index 0d2cb0395..8c6bdeda4 100644 --- a/backend/src/telemetry/account_usage/query.rs +++ b/backend/src/telemetry/account_usage/query.rs @@ -151,7 +151,7 @@ pub struct AccountUsageRecord { } /// 管理端账号用量汇总。 -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct AccountUsageSummary { pub account_count: i64, pub request_count: i64, diff --git a/backend/src/telemetry/ops/query.rs b/backend/src/telemetry/ops/query.rs index 2581dc695..ff53fd165 100644 --- a/backend/src/telemetry/ops/query.rs +++ b/backend/src/telemetry/ops/query.rs @@ -1,53 +1,30 @@ //! 管理端运维错误事件服务。 -use chrono::Utc; use thiserror::Error; use crate::telemetry::{ ops::store::{OpsErrorFilter, PgOpsErrorLogStore}, ops::types::OpsErrorLog, - usage::types::{metadata_i64, metadata_service_tier, metadata_string}, }; /// 管理端运维错误事件错误。 #[derive(Debug, Error)] pub enum OpsQueryError { - /// 写入失败。 - #[error("failed to append ops error log")] - Append, /// 查询失败。 #[error("failed to list ops error logs")] List, - /// 保留期清理失败。 - #[error("failed to trim expired ops error logs")] - Retention, } /// 管理端运维错误事件服务。 #[derive(Clone)] pub struct OpsQueryService { store: PgOpsErrorLogStore, - capture_body: bool, } impl OpsQueryService { /// 构造管理端运维错误事件服务。 - pub fn new(store: PgOpsErrorLogStore, capture_body: bool) -> Self { - Self { - store, - capture_body, - } - } - - /// 记录运维错误事件。 - pub async fn record(&self, mut event: OpsErrorLog) -> Result<(), OpsQueryError> { - lift_error_fact_fields(&mut event); - apply_capture_body_policy(&mut event, self.capture_body); - self.store - .append(&event) - .await - .map_err(|_| OpsQueryError::Append)?; - Ok(()) + pub fn new(store: PgOpsErrorLogStore) -> Self { + Self { store } } pub async fn list( @@ -73,97 +50,4 @@ impl OpsQueryService { .await .map_err(|_| OpsQueryError::List) } - - /// 周期清理超过保留期的错误事实。 - pub async fn trim_to_retention( - &self, - now: chrono::DateTime, - ) -> Result { - self.store - .trim_to_retention(now) - .await - .map_err(|_| OpsQueryError::Retention) - } -} - -fn lift_error_fact_fields(event: &mut OpsErrorLog) { - event.client_status_code = event - .client_status_code - .or_else(|| metadata_i64(&event.metadata, &["clientStatusCode", "client_status_code"])); - event.upstream_status_code = event - .upstream_status_code - .or_else(|| metadata_i64(&event.metadata, &["upstreamStatusCode", "upstreamStatus"])); - event.transport = event - .transport - .take() - .or_else(|| metadata_string(&event.metadata, &["transport"])); - event.attempt_index = event - .attempt_index - .or_else(|| metadata_i64(&event.metadata, &["attemptIndex", "attempt_index"])); - event.failure_class = event - .failure_class - .take() - .or_else(|| metadata_string(&event.metadata, &["failureClass", "failure_class"])); - event.response_id = event - .response_id - .take() - .or_else(|| metadata_string(&event.metadata, &["responseId", "response_id"])); - event.upstream_request_id = event.upstream_request_id.take().or_else(|| { - metadata_string( - &event.metadata, - &[ - "upstreamRequestId", - "upstream_request_id", - "openaiRequestId", - ], - ) - }); - event.service_tier = event - .service_tier - .take() - .or_else(|| metadata_service_tier(&event.metadata).map(ToString::to_string)); - - let Some(metadata) = event.metadata.as_object_mut() else { - return; - }; - for key in [ - "clientStatusCode", - "client_status_code", - "upstreamStatusCode", - "upstreamStatus", - "transport", - "attemptIndex", - "attempt_index", - "failureClass", - "failure_class", - "responseId", - "response_id", - "upstreamRequestId", - "upstream_request_id", - "openaiRequestId", - "serviceTier", - ] { - metadata.remove(key); - } -} - -fn apply_capture_body_policy(event: &mut OpsErrorLog, capture_body: bool) { - let Some(metadata) = event.metadata.as_object_mut() else { - return; - }; - for key in [ - "body", - "rawBody", - "requestBody", - "responseBody", - "upstreamBody", - ] { - if capture_body { - if let Some(serde_json::Value::String(value)) = metadata.get_mut(key) { - value.truncate(4096); - } - } else { - metadata.remove(key); - } - } } diff --git a/backend/src/telemetry/recorder.rs b/backend/src/telemetry/recorder.rs index a2f2e2ae3..ac53af0d2 100644 --- a/backend/src/telemetry/recorder.rs +++ b/backend/src/telemetry/recorder.rs @@ -7,15 +7,86 @@ use serde_json::Value; use crate::{ infra::time::elapsed_millis_i64, telemetry::{ - ops::query::OpsQueryService, + ops::store::{PgOpsErrorLogStore, PgOpsErrorLogStoreError}, ops::types::OpsErrorLog, - usage::types::{ResponseUsageRecord, UsageRecord}, + usage::{ + store::{PgUsageRecordStore, PgUsageRecordStoreError}, + types::{ + metadata_i64, metadata_service_tier, metadata_string as metadata_string_any, + ResponseUsageRecord, UsageRecord, + }, + }, }, upstream::openai::protocol::responses::{CodexCompactRequest, CodexResponsesRequest}, }; +/// 成功与失败事实的唯一写入入口。 +#[derive(Clone)] +pub struct Recorder { + usage_records: PgUsageRecordStore, + ops_errors: PgOpsErrorLogStore, + usage_enabled: bool, + capture_body: bool, +} + +/// 遥测事实写入错误。 +#[derive(Debug, thiserror::Error)] +pub enum RecorderError { + #[error("invalid success usage fact")] + InvalidUsageFact, + #[error(transparent)] + Usage(#[from] PgUsageRecordStoreError), + #[error(transparent)] + Ops(#[from] PgOpsErrorLogStoreError), +} + +impl Recorder { + pub fn new( + usage_records: PgUsageRecordStore, + ops_errors: PgOpsErrorLogStore, + usage_enabled: bool, + capture_body: bool, + ) -> Self { + Self { + usage_records, + ops_errors, + usage_enabled, + capture_body, + } + } + + /// 校验并写入一条成功事实及其时间桶。 + pub async fn record_usage(&self, mut event: UsageRecord) -> Result<(), RecorderError> { + if !self.usage_enabled { + return Ok(()); + } + if !is_usage_fact(&event) { + tracing::warn!( + usage_record_id = %event.id, + request_id = event.request_id.as_deref().unwrap_or(""), + account_id = event.account_id, + model = event.model, + status_code = event.status_code, + "rejected invalid success usage fact" + ); + return Err(RecorderError::InvalidUsageFact); + } + apply_success_capture_body_policy(&mut event, self.capture_body); + self.usage_records.append(&event).await?; + Ok(()) + } + + /// 规范化并写入一条失败事实及其时间桶。 + pub async fn record_error(&self, mut event: OpsErrorLog) -> Result<(), RecorderError> { + lift_error_fact_fields(&mut event); + apply_error_capture_body_policy(&mut event, self.capture_body); + self.ops_errors.append(&event).await?; + Ok(()) + } +} + pub(crate) struct DispatchErrorLogRecord<'a> { - pub ops_errors: &'a OpsQueryService, + pub recorder: &'a Recorder, pub request_id: &'a str, pub client_api_key_id: Option<&'a str>, pub provider: Option<&'a str>, @@ -43,7 +114,7 @@ pub(crate) async fn record_dispatch_error_event(record: DispatchErrorLogRecord<' event.latency_ms = Some(elapsed_millis_i64(record.started_at)); event.metadata = metadata; - if let Err(error) = record.ops_errors.record(event).await { + if let Err(error) = record.recorder.record_error(event).await { tracing::warn!( account_id = record.account_id.unwrap_or(""), error = %error, @@ -91,11 +162,120 @@ pub(crate) async fn record_response_event(record: ResponseUsageRecord<'_>) { lift_success_fact_fields(&mut event, &mut metadata); event.metadata = metadata; - if let Err(error) = record.usage_records.record(event).await { + if let Err(error) = record.recorder.record_usage(event).await { tracing::warn!(account_id = %record.account_id, error = %error, "failed to record response event"); } } +fn is_usage_fact(event: &UsageRecord) -> bool { + (200..=399).contains(&event.status_code) + && !event.provider.trim().is_empty() + && !event.account_id.trim().is_empty() + && !event.model.trim().is_empty() +} + +fn apply_success_capture_body_policy(event: &mut UsageRecord, capture_body: bool) { + if capture_body { + return; + } + remove_body_fields(&mut event.metadata); +} + +fn apply_error_capture_body_policy(event: &mut OpsErrorLog, capture_body: bool) { + let Some(metadata) = event.metadata.as_object_mut() else { + return; + }; + for key in body_fields() { + if capture_body { + if let Some(Value::String(value)) = metadata.get_mut(key) { + value.truncate(4096); + } + } else { + metadata.remove(key); + } + } +} + +fn remove_body_fields(metadata: &mut Value) { + let Some(metadata) = metadata.as_object_mut() else { + return; + }; + for key in body_fields() { + metadata.remove(key); + } +} + +fn body_fields() -> [&'static str; 5] { + [ + "body", + "rawBody", + "requestBody", + "responseBody", + "upstreamBody", + ] +} + +fn lift_error_fact_fields(event: &mut OpsErrorLog) { + event.client_status_code = event + .client_status_code + .or_else(|| metadata_i64(&event.metadata, &["clientStatusCode", "client_status_code"])); + event.upstream_status_code = event + .upstream_status_code + .or_else(|| metadata_i64(&event.metadata, &["upstreamStatusCode", "upstreamStatus"])); + event.transport = event + .transport + .take() + .or_else(|| metadata_string_any(&event.metadata, &["transport"])); + event.attempt_index = event + .attempt_index + .or_else(|| metadata_i64(&event.metadata, &["attemptIndex", "attempt_index"])); + event.failure_class = event + .failure_class + .take() + .or_else(|| metadata_string_any(&event.metadata, &["failureClass", "failure_class"])); + event.response_id = event + .response_id + .take() + .or_else(|| metadata_string_any(&event.metadata, &["responseId", "response_id"])); + event.upstream_request_id = event.upstream_request_id.take().or_else(|| { + metadata_string_any( + &event.metadata, + &[ + "upstreamRequestId", + "upstream_request_id", + "openaiRequestId", + ], + ) + }); + event.service_tier = event + .service_tier + .take() + .or_else(|| metadata_service_tier(&event.metadata).map(ToString::to_string)); + + let Some(metadata) = event.metadata.as_object_mut() else { + return; + }; + for key in [ + "clientStatusCode", + "client_status_code", + "upstreamStatusCode", + "upstreamStatus", + "transport", + "attemptIndex", + "attempt_index", + "failureClass", + "failure_class", + "responseId", + "response_id", + "upstreamRequestId", + "upstream_request_id", + "openaiRequestId", + "serviceTier", + ] { + metadata.remove(key); + } +} + fn lift_success_fact_fields(event: &mut UsageRecord, metadata: &mut Value) { event.transport = metadata_string(metadata, "transport"); event.attempt_index = metadata_nonnegative_i64(metadata, "attemptIndex"); diff --git a/backend/src/telemetry/usage/query.rs b/backend/src/telemetry/usage/query.rs index 102b6dc7d..3b3c76586 100644 --- a/backend/src/telemetry/usage/query.rs +++ b/backend/src/telemetry/usage/query.rs @@ -1,6 +1,6 @@ //! 管理端使用记录服务。 -use std::{collections::HashMap, sync::Arc}; +use std::collections::HashMap; use chrono::{DateTime, Utc}; use thiserror::Error; @@ -110,12 +110,6 @@ pub enum UsageQueryError { /// 清空失败。 #[error("failed to clear usage records")] Clear, - /// 写入失败。 - #[error("failed to append usage record")] - Append, - /// 保留期清理失败。 - #[error("failed to trim expired usage records")] - Retention, /// 账号关联信息读取失败。 #[error("failed to load usage record accounts")] Accounts, @@ -125,25 +119,12 @@ pub enum UsageQueryError { #[derive(Clone)] pub struct UsageQueryService { store: PgUsageRecordStore, - settings: Arc>, -} - -#[derive(Debug, Clone, Copy)] -struct UsageQuerySettings { - enabled: bool, - capture_body: bool, } impl UsageQueryService { /// 构造管理端使用记录服务。 - pub fn new(store: PgUsageRecordStore, enabled: bool, capture_body: bool) -> Self { - Self { - store, - settings: Arc::new(tokio::sync::RwLock::new(UsageQuerySettings { - enabled, - capture_body, - })), - } + pub fn new(store: PgUsageRecordStore) -> Self { + Self { store } } /// 分页查询日志。 @@ -254,72 +235,6 @@ impl UsageQueryService { .map(|cleared| ClearUsageRecords { cleared }) .map_err(|_| UsageQueryError::Clear) } - - /// 记录使用记录。 - pub async fn record(&self, event: UsageRecord) -> Result<(), UsageQueryError> { - let settings = *self.settings.read().await; - if !settings.enabled { - return Ok(()); - } - if !is_usage_log_event(&event) { - tracing::warn!( - usage_record_id = %event.id, - request_id = event.request_id.as_deref().unwrap_or(""), - account_id = event.account_id, - model = event.model, - status_code = event.status_code, - "rejected invalid success usage fact" - ); - return Err(UsageQueryError::Append); - } - self.append_with_settings(event, settings).await - } - - async fn append_with_settings( - &self, - mut event: UsageRecord, - settings: UsageQuerySettings, - ) -> Result<(), UsageQueryError> { - apply_capture_body_policy(&mut event, settings.capture_body); - self.store - .append(&event) - .await - .map_err(|_| UsageQueryError::Append)?; - Ok(()) - } - - /// 周期清理超过保留期的成功事实。 - pub async fn trim_to_retention(&self, now: DateTime) -> Result { - self.store - .trim_to_retention(now) - .await - .map_err(|_| UsageQueryError::Retention) - } -} - -fn is_usage_log_event(event: &UsageRecord) -> bool { - (200..=399).contains(&event.status_code) - && !event.provider.trim().is_empty() - && !event.account_id.trim().is_empty() - && !event.model.trim().is_empty() -} - -fn apply_capture_body_policy(event: &mut UsageRecord, capture_body: bool) { - if capture_body { - return; - } - let Some(metadata) = event.metadata.as_object_mut() else { - return; - }; - for key in [ - "body", - "rawBody", - "requestBody", - "responseBody", - "upstreamBody", - ] { - metadata.remove(key); - } } pub(crate) fn usage_record_cost_details( diff --git a/backend/src/telemetry/usage/types.rs b/backend/src/telemetry/usage/types.rs index 9f6f9af3a..2877e8281 100644 --- a/backend/src/telemetry/usage/types.rs +++ b/backend/src/telemetry/usage/types.rs @@ -141,7 +141,7 @@ pub(crate) fn metadata_i64(metadata: &Value, keys: &[&str]) -> Option { /// 响应事件记录（供 dispatch 模块使用）。 pub struct ResponseUsageRecord<'a> { - pub usage_records: &'a crate::telemetry::usage::query::UsageQueryService, + pub recorder: &'a crate::telemetry::recorder::Recorder, pub request_id: &'a str, pub client_api_key_id: Option<&'a str>, pub account_id: &'a str, diff --git a/backend/src/upstream/openai/protocol/sse.rs b/backend/src/upstream/openai/protocol/sse.rs index 807b91259..970658f3a 100644 --- a/backend/src/upstream/openai/protocol/sse.rs +++ b/backend/src/upstream/openai/protocol/sse.rs @@ -1,5 +1,6 @@ //! SSE 事件解析与编码。 +use serde_json::json; use thiserror::Error; /// 单条 SSE 事件。 @@ -21,6 +22,41 @@ pub const MAX_SSE_EVENT_BUFFER_BYTES: usize = 64 * 1024 * 1024; /// SSE 流结束标记帧。 pub const DONE_SSE_FRAME: &str = "data: [DONE]\n\n"; +/// 编码 OpenAI Responses `response.failed` SSE 事件。 +pub fn response_failed_sse_event(error_type: &str, code: &str, message: &str) -> String { + response_failed_sse_event_with_id(None, error_type, code, message) +} + +/// 使用指定 response id 编码 OpenAI Responses `response.failed` SSE 事件。 +pub fn response_failed_sse_event_with_id( + response_id: Option<&str>, + error_type: &str, + code: &str, + message: &str, +) -> String { + let error = json!({ + "type": error_type, + "code": code, + "message": message, + }); + let response_id = response_id + .filter(|value| !value.trim().is_empty()) + .map_or_else( + || format!("resp_proxy_{}", uuid::Uuid::new_v4().simple()), + ToString::to_string, + ); + let data = json!({ + "type": "response.failed", + "response": { + "id": response_id, + "status": "failed", + "error": error, + }, + "error": error, + }); + encode_sse_event("response.failed", &data.to_string()) +} + /// 判断 SSE 文本是否已经包含结束标记。 pub fn sse_body_has_done(body: &str) -> bool { body.trim_end_matches(['\r', '\n']) diff --git a/backend/tests/api/admin/monitoring/dashboard.rs b/backend/tests/api/admin/monitoring/dashboard.rs index 84dc21a5a..5535fce1b 100644 --- a/backend/tests/api/admin/monitoring/dashboard.rs +++ b/backend/tests/api/admin/monitoring/dashboard.rs @@ -54,7 +54,7 @@ async fn dashboard_summary_should_render_dash_when_fingerprint_updated_at_is_mis #[tokio::test] async fn dashboard_summary_should_calculate_today_traffic_and_average_first_token_latency_from_time_buckets( ) { - let (app, store, _pool, _dir) = dashboard_test_app( + let (app, store, pool, _dir) = dashboard_test_app( "dashboard-traffic-latency", crate::support::fingerprint::test_fingerprint(), ) @@ -80,22 +80,23 @@ async fn dashboard_summary_should_calculate_today_traffic_and_average_first_toke .append(&usage_record_with_tokens(yesterday_record_at, 30)) .await .unwrap(); + seed_lifetime_usage_summary(&pool, 7, 400, 500, 100).await; let body = dashboard_summary(app).await; assert_eq!(body["data"]["cards"]["traffic"]["todayRequests"], "2"); assert_eq!(body["data"]["cards"]["traffic"]["todayRequestsValue"], 2); - assert_eq!(body["data"]["cards"]["traffic"]["totalRequests"], "3"); + assert_eq!(body["data"]["cards"]["traffic"]["totalRequests"], "7"); assert_eq!( body["data"]["cards"]["traffic"]["yesterdayRequestsValue"], 1 ); assert_eq!(body["data"]["cards"]["tokens"]["todayTokens"], "30"); assert_eq!(body["data"]["cards"]["tokens"]["todayTokensValue"], 30); - assert_eq!(body["data"]["cards"]["tokens"]["totalTokens"], "60"); + assert_eq!(body["data"]["cards"]["tokens"]["totalTokens"], "900"); assert_eq!(body["data"]["cards"]["tokens"]["yesterdayTokensValue"], 30); - assert_eq!(body["data"]["cards"]["cache"]["totalHitRate"], "0.0%"); - assert_eq!(body["data"]["cards"]["cache"]["totalCachedTokens"], "0"); + assert_eq!(body["data"]["cards"]["cache"]["totalHitRate"], "25.0%"); + assert_eq!(body["data"]["cards"]["cache"]["totalCachedTokens"], "100"); assert_eq!( body["data"]["cards"]["cache"]["averageFirstTokenLatencyMs"], "300 ms" @@ -700,6 +701,30 @@ async fn seed_usage_summary(pool: &PgPool, input_tokens: u64, output_tokens: u64 .await; } +async fn seed_lifetime_usage_summary( + pool: &PgPool, + request_count: u64, + input_tokens: u64, + output_tokens: u64, + cached_tokens: u64, +) { + seed_dashboard_account(pool, "acct_lifetime", "lifetime@example.com").await; + sqlx::query( + "insert into account_usage + (account_id, request_count, input_tokens, output_tokens, cached_tokens, total_tokens) + values ($1, $2, $3, $4, $5, $6)", + ) + .bind("acct_lifetime") + .bind(request_count as i64) + .bind(input_tokens as i64) + .bind(output_tokens as i64) + .bind(cached_tokens as i64) + .bind((input_tokens + output_tokens) as i64) + .execute(pool) + .await + .unwrap(); +} + async fn seed_usage_summary_with_last_used( pool: &PgPool, input_tokens: u64, diff --git a/backend/tests/bootstrap/import_sqlite.rs b/backend/tests/bootstrap/import_sqlite.rs index 30fad6f00..ff0918205 100644 --- a/backend/tests/bootstrap/import_sqlite.rs +++ b/backend/tests/bootstrap/import_sqlite.rs @@ -152,6 +152,30 @@ async fn import_sqlite_should_reject_nonempty_target() { assert!(matches!(error, ImportSqliteError::TargetNotEmpty)); } +#[tokio::test] +async fn import_sqlite_should_keep_blank_admin_api_key_disabled() { + let (_source_guard, source_path) = sqlite_v3_source("blank-admin-api-key").await; + let source = open_sqlite(&source_path, false).await; + sqlx::query( + "insert into runtime_settings values + (1, '{}', 3600, 2, 3, 50, 'smart', ' ', '2026-01-01T00:00:00Z')", + ) + .execute(&source) + .await + .unwrap(); + source.close().await; + let (target, _target_guard) = init_test_db("import-sqlite-blank-admin-api-key").await; + + import_sqlite(&target, &source_path).await.unwrap(); + + let admin_api_key_hash: Option = + sqlx::query_scalar("select admin_api_key_hash from runtime_settings where id = 1") + .fetch_one(&target) + .await + .unwrap(); + assert!(admin_api_key_hash.is_none()); +} + async fn sqlite_v3_source(label: &str) -> (tempfile::TempDir, PathBuf) { let guard = tempfile::tempdir().unwrap(); let path = guard.path().join(format!("{label}.sqlite")); diff --git a/backend/tests/upstream/openai/protocol/mod.rs b/backend/tests/upstream/openai/protocol/mod.rs index 1e5f5728e..f5a5606b6 100644 --- a/backend/tests/upstream/openai/protocol/mod.rs +++ b/backend/tests/upstream/openai/protocol/mod.rs @@ -1,7 +1,5 @@ use axum::http::HeaderMap; -use codex_proxy_rs::api::client::responses::{ - build_codex_request, build_compact_request, response_failed_sse_event, -}; +use codex_proxy_rs::api::client::responses::{build_codex_request, build_compact_request}; use codex_proxy_rs::upstream::openai::protocol::events::{ extract_sse_usage, extract_usage, parse_rate_limit_headers, parse_rate_limits_event, rate_limit_quota, retry_after_seconds_from_body, RateLimitWindow, TokenUsage, @@ -11,7 +9,9 @@ use codex_proxy_rs::upstream::openai::protocol::responses::{ response_sse_event_is_terminal, transport_for_request, CodexResponsesRequest, CodexTransport, CollectedResponse, }; -use codex_proxy_rs::upstream::openai::protocol::sse::parse_sse_events; +use codex_proxy_rs::upstream::openai::protocol::sse::{ + parse_sse_events, response_failed_sse_event, +}; use codex_proxy_rs::upstream::openai::protocol::websocket::{ classify_websocket_error_frame, is_terminal_websocket_event, retry_after_seconds_from_wrapped_error_headers, diff --git a/docs/architecture.md b/docs/architecture.md index 607b0b900..02f7b9f8b 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -39,16 +39,16 @@ | --- | --- | --- | --- | | L0 | `infra` | PostgreSQL 连接/迁移、Redis 连接/键前缀、时间、JSON、日志、身份生成、路径、格式化 | 出现任何业务概念 | | L1 | `upstream` | 纯上游客户端：协议编解码、HTTP/SSE/WebSocket 传输、指纹、OAuth token client。**每家 provider 一个子目录**（`openai/`，规划中 `cloudflare/`） | 依赖任何领域模块；出现账号池/调度/遥测概念 | -| L1 | `telemetry` | 遥测域：usage_records / ops_error_logs / request_time_buckets / account_usage 的**写入与查询**、billing、dashboard 聚合、rebuild | 被冠以 admin 语义；HTTP handler | -| L1 | `keys` | 客户端 API key 域：store、内存缓存、鉴权校验、管理操作 | — | +| L1 | `telemetry` | 遥测域：usage_records / ops_error_logs / request_time_buckets / account_usage 的**写入与查询**、billing、rebuild | 被冠以 admin 语义；HTTP handler | +| L1 | `keys` | 客户端 API key 域：store、鉴权校验、管理操作 | — | | L1 | `auth` | 管理端认证域：admin 用户（PG）、会话（Redis，token_hash 键，database.md §4B.1） | 客户端 key 概念（那是 `keys`） | -| L1 | `settings` | 运行时设置域：持久化、快照、变更广播（watch）、admin API key 校验 | 直接调用上层服务推送变更（§2.3） | +| L1 | `settings` | 运行时设置域：校验、持久化、并发串行化的当前快照、admin API key 校验 | 持有账号池、模型或刷新服务引用（§2.3） | | L1 | `update` | 自更新域：release 查询、下载、校验、解压、替换、状态 | — | | L2 | `accounts` | 账号域：实体、池、调度器、store、quota、token 刷新、Cookie、导入导出、管理操作 | — | | L2 | `models` | 模型目录域：catalog、别名、plan 快照 | — | | L3 | `dispatch` | 请求编排域：账号获取、affinity、上游调用、流转发、失败恢复、重试、遥测记录 | HTTP 类型（axum）出现在签名 | | L4 | `api` | 全部入站 HTTP：`api/client`（/v1）、`api/admin`（/api/admin）、路由组合、中间件、SPA 资源 | 写 SQL；承载领域逻辑 | -| L5 | `bootstrap` | 进程装配：启动配置、服务构造、AppState、后台任务、关闭 | 被任何层依赖 | +| 装配层 | `bootstrap` | 进程装配：启动配置、服务构造、AppState、后台任务、关闭；`api` 只读取其 AppState 门面 | 被 L0–L3 领域层依赖 | ### 2.2 依赖方向 @@ -57,8 +57,6 @@ | 允许的同层/特殊依赖 | 理由 | | --- | --- | | `accounts` → `models` | 候选过滤需要模型允许列表；单向 | -| `accounts` → `settings` | 池参数/刷新策略订阅设置快照（watch 类型）；单向 | -| `auth` → `settings` | admin API key 哈希存放在 runtime_settings 表，经 settings 的 store 方法读取；单向 | **全库硬禁**（CI 脚本断言，§9）： @@ -70,7 +68,7 @@ ### 2.3 控制反转两处（消除向上依赖） -1. **设置传播**：现状 `RuntimeSettingsService` 持有账号池与刷新策略的引用、主动推送更新（低层持有高层引用）。终态：`settings` 只暴露 `tokio::sync::watch::Receiver`；`accounts`（池、刷新策略）、`models`（别名）各自订阅。装配在 `bootstrap` 完成。 +1. **设置传播**：`RuntimeSettingsService` 只负责校验、持久化和当前快照，不持有任何上层消费者。`bootstrap::Services::update_settings` 在持久化成功后同步把同一快照应用到账号池、刷新策略与模型配置；更新串行执行，不保留无人消费的 watch 通道，也不存在低层向上回调。 2. **遥测写入**：现状 proxy 热路径 import `admin::monitoring` 的 Admin* 服务。终态：`dispatch` 依赖 `telemetry::Recorder`（L3 → L1，方向正确），admin 侧只是同一域的查询消费者。 --- @@ -79,117 +77,58 @@ ```text backend/src/ -├── main.rs # 子命令入口：serve（默认）/ import-sqlite / rebuild-buckets -├── lib.rs -├── infra/ # L0 -│ ├── mod.rs -│ ├── database.rs # PgPool 连接/池参数、迁移框架（PG 谱系 0001 终态基线）、ping -│ ├── redis.rs # Redis ConnectionManager、键前缀、ping（database.md §1） -│ ├── migrations/ # *.sql（PG） -│ ├── identity.rs time.rs json.rs format.rs logging.rs paths.rs -├── upstream/ # L1 纯上游客户端（每家 provider 一个子目录） -│ ├── mod.rs # 第二家 provider 出现时，UpstreamClient trait 落此 -│ └── openai/ -│ ├── mod.rs -│ ├── token_client.rs -│ ├── protocol/ -│ │ ├── mod.rs schema.rs responses.rs events.rs sse.rs websocket.rs -│ ├── transport/ -│ │ ├── mod.rs client.rs endpoints.rs headers.rs tls.rs -│ │ ├── response_meta.rs usage.rs diagnostics.rs -│ │ ├── websocket.rs websocket_pool.rs websocket_pump.rs -│ └── fingerprint/ -│ ├── mod.rs types.rs store.rs runtime.rs updater.rs -├── telemetry/ # L1 遥测域（原 admin/monitoring + proxy/dispatch/usage_events） -│ ├── mod.rs -│ ├── recorder.rs # 唯一写入口：成功/失败两条 PG 事务（database.md §5.2） -│ ├── billing.rs -│ ├── rebuild.rs # rebuild-buckets（database.md §7） -│ ├── dashboard.rs # Dashboard 聚合查询（无 HTTP；口径 database.md §5.3） -│ ├── usage/ # usage_records -│ │ ├── mod.rs types.rs store.rs query.rs -│ ├── ops/ # ops_error_logs（含 /api/admin/ops/errors 的查询面） -│ │ ├── mod.rs types.rs store.rs query.rs -│ ├── buckets/ # request_time_buckets -│ │ ├── mod.rs store.rs query.rs -│ └── account_usage/ # account_usage / account_model_usage -│ ├── mod.rs store.rs query.rs -├── keys/ # L1 客户端 key 域（原 admin/keys/service.rs 拆分） -│ ├── mod.rs types.rs store.rs service.rs manage.rs -│ │ # 鉴权 = 完整 key → PG unique 点查（database.md §4.3），无进程内鉴权缓存 -├── auth/ # L1 管理端认证域（原 admin/auth/service.rs） -│ ├── mod.rs types.rs store.rs service.rs # store = PG admin_users + Redis 会话键（§4B.1） -├── settings/ # L1 运行时设置域（原 config/settings.rs） -│ ├── mod.rs types.rs store.rs service.rs # service 持 watch 广播 -├── update/ # L1 自更新域（原 admin/system/*） -│ ├── mod.rs types.rs state.rs release.rs download.rs archive.rs service.rs -├── accounts/ # L2 账号域（原 upstream/accounts + upstream/scheduler + admin/accounts/service） -│ ├── mod.rs -│ ├── account.rs # 实体（原 model.rs） -│ ├── window.rs -│ ├── pool.rs # 池：存储态/槽位/过滤（§5.4 行数预算内需拆出 pool/ 子文件） -│ ├── scheduler.rs # 调度门面（原 scheduler/mod.rs 的类型） -│ ├── scheduler/ -│ │ ├── candidates.rs feedback.rs -│ │ └── strategy/ -│ │ ├── mod.rs smart.rs quota_reset.rs round_robin.rs sticky.rs -│ ├── store.rs # AccountStore trait + PG 门面 -│ ├── store/ -│ │ ├── queries.rs rows.rs -│ ├── quota/ -│ │ ├── mod.rs service.rs runtime.rs -│ ├── refresh/ # 原 token_refresh -│ │ ├── mod.rs policy.rs lease.rs service.rs # lease = Redis SET NX PX（§4B.2） -│ ├── cookies/ -│ │ ├── mod.rs store.rs -│ ├── import.rs # 导入 payload 解析（原 upstream/accounts/importing.rs） -│ └── manage/ # 管理操作（原 admin/accounts/service/*） -│ ├── mod.rs types.rs service.rs lifecycle.rs -│ ├── import.rs export.rs oauth.rs probe.rs quota.rs quota_view.rs -├── models/ # L2 模型目录域 -│ ├── mod.rs types.rs catalog.rs service.rs store.rs # store = Redis 快照 HASH（§4B.4） -├── dispatch/ # L3 请求编排域（原 proxy/dispatch） -│ ├── mod.rs -│ ├── service.rs # 编排入口（原 responses/service.rs，按 §5.4 拆分） -│ ├── upstream_call.rs # 原 dispatch/upstream.rs -│ ├── errors.rs # 原 dispatch/errors.rs + responses/errors.rs 合并 -│ ├── recording.rs # 原 responses/event_recording.rs，调 telemetry::recorder -│ ├── affinity/ # 原 session_affinity.rs（683 行三合一）+ responses/affinity.rs -│ │ ├── mod.rs types.rs identity.rs store.rs service.rs resolve.rs -│ │ │ # store = Redis 键 + 二级索引（§4B.3），无内存映射副本 -│ ├── recovery/ -│ │ ├── mod.rs auth.rs cloudflare.rs exhaustion.rs -│ │ ├── implicit_resume.rs reasoning_replay.rs -│ └── stream/ -│ ├── mod.rs live.rs lifecycle.rs prefetch.rs sse_failure.rs trace.rs -├── api/ # L4 全部入站 HTTP（原 http + web + proxy 入口 + admin routes） -│ ├── mod.rs -│ ├── router.rs # 总路由 + healthz（infra 的 PG ping + Redis ping，任一失败 503） -│ ├── assets.rs # SPA 静态资源（原 web/assets.rs） -│ ├── middleware/ -│ │ ├── mod.rs request_id.rs trace.rs -│ ├── client/ # /v1（原 proxy/auth + proxy/openai） -│ │ ├── mod.rs router.rs auth.rs -│ │ ├── responses.rs models.rs errors.rs sse.rs -│ └── admin/ # /api/admin（原 admin 的全部 routes 与 DTO） -│ ├── mod.rs router.rs response.rs session.rs # session = 鉴权提取器/中间件 -│ ├── auth_routes.rs settings_routes.rs system_routes.rs -│ ├── accounts_routes.rs keys_routes.rs -│ ├── usage_routes.rs ops_routes.rs dashboard_routes.rs # ops_routes = /api/admin/ops/errors -└── bootstrap/ # L5 进程装配（原 runtime + config/schema,loader） - ├── mod.rs - ├── config.rs # 启动配置 schema + loader（原 config/schema.rs + loader.rs；含 redis.url） - ├── services.rs state.rs shutdown.rs - ├── import_sqlite.rs # 一次性 SQLite v3 → PG 导入命令（database.md §6；§5.5 的唯一跨域写豁免） - └── tasks/ - ├── mod.rs coordinator.rs periodic.rs cleanup.rs - ├── cookie_cleanup.rs retention_trim.rs # retention_trim 新增（database.md §7） - ├── model_refresh.rs token_refresh.rs quota_refresh.rs - ├── fingerprint_update.rs - # session_cleanup / session_affinity_cleanup 不复存在：Redis TTL 接管（database.md §4B） +├── main.rs lib.rs +├── infra/ # database.rs、redis.rs、migrations/ 与通用基础能力 +├── upstream/openai/ # 纯 OpenAI/Codex client +│ ├── token_client.rs +│ ├── protocol/ # schema / responses / events / SSE / WebSocket +│ ├── transport/ # HTTP、TLS、WebSocket 与连接池；大文件可按职责建子文件 +│ └── fingerprint/ # types / store / runtime / updater +├── telemetry/ # 四类账本与查询 +│ ├── recorder.rs # 成功/失败事实的唯一领域写入口 +│ ├── billing.rs diagnostics.rs rebuild.rs +│ ├── usage/ # types.rs、query.rs、store.rs、store/query.rs +│ ├── ops/ # types.rs、query.rs、store.rs +│ ├── buckets/ # query.rs、store.rs +│ └── account_usage/ # query.rs、store.rs +├── keys/ # types / PG store / verifier / manage +├── auth/ # types / PG+Redis store / session service +├── settings/ # types / PG store / serialized current-snapshot service +├── update/ # types / state / release / download / archive / service +├── accounts/ +│ ├── account.rs service.rs import.rs window.rs +│ ├── pool.rs pool/{runtime,types}.rs +│ ├── scheduler.rs scheduler/{candidates,feedback,strategy/...}.rs +│ ├── store.rs store/{queries,rows,write}.rs +│ ├── quota/{service,runtime}.rs +│ ├── refresh/{lease,service,runtime}.rs refresh/runtime/types.rs +│ ├── cookies/store.rs +│ └── manage/{service,types,lifecycle,import,export,oauth,probe,quota,quota_view}.rs +├── models/ # config / info / snapshot / catalog / service / Redis store +├── dispatch/ +│ ├── errors.rs upstream.rs cloudflare.rs exhaustion.rs +│ ├── implicit_resume.rs reasoning_replay.rs +│ ├── affinity/{types,identity,store,service,resolve}.rs +│ └── responses/ # Responses 编排是一个完整子能力，不做无收益扁平化 +│ ├── service.rs service/{complete,stream,compact}.rs +│ ├── errors.rs event_recording.rs live_stream.rs +│ └── prefetch.rs sse_failure.rs stream_lifecycle.rs trace.rs +├── api/ +│ ├── router.rs assets.rs middleware/ +│ ├── client/ # /v1 路由、鉴权与协议响应 +│ └── admin/ # /api/admin handlers 与 DTO +│ ├── accounts_routes.rs accounts_routes/{actions,stats}.rs +│ ├── dashboard_routes.rs dashboard_routes/data.rs +│ └── keys/ops/settings/system/usage 等 routes +└── bootstrap/ + ├── application.rs config.rs services.rs state.rs shutdown.rs + ├── import_sqlite.rs import_sqlite/{core,telemetry}.rs + └── tasks/ # periodic、retention、token/quota/model refresh 等装配任务 ``` -`backend/tests/` 目录结构**镜像 `src/`**：模块搬家时测试目录同步搬家，`fixtures/`、`support/` 不动。 +这是已落地结构，不是待办蓝图。为控制单文件行数而增加的同名子目录属于同一能力，不构成抽象层。Dashboard 的跨域只读组合保留在 `api/admin/dashboard_routes/data.rs`：它只组装 presentation read model，不写 SQL、不改变领域状态；把它下沉到 `telemetry` 反而会制造 telemetry → accounts 的向上依赖。 + +`backend/tests/` 顶层按 `src/` 领域镜像；子目录可按公开路由或行为场景组织，因此 `monitoring`、`token_refresh` 等历史行为名称可保留在测试中。`fixtures/`、`support/` 是测试基础设施，不要求与生产文件逐一同名。 --- @@ -248,7 +187,7 @@ backend/src/ ### 5.2 存储边界（SQL 与 Redis 同规） -SQL 文本与 Redis 命令只允许出现在 `store.rs` / `store/` / `query.rs`。`api` 层与 `service.rs` 出现 `sqlx::query` 或 `redis::` 调用即为违规（现状 `http/router.rs` 的 healthz 直接拿某个 store 的连接池执行裸 SQL——终态 `infra::database` 提供 `ping(pool)`、`infra::redis` 提供 `ping(conn)`，router 只调它们）。 +业务 SQL 与 Redis 命令只允许出现在 `store.rs` / `store/` / `query.rs`。明确的基础设施/维护例外只有：`infra/database.rs`（迁移与 ping）、`infra/redis.rs`（连接与 ping）、`bootstrap/import_sqlite*`（一次性跨库导入）、`telemetry/rebuild.rs`（账本重建命令）、`accounts/refresh/lease.rs`（Redis 租约适配器）。`api` 层与领域 `service.rs` 出现存储命令即为违规；healthz 只调用 infra 的 ping 门面。 ### 5.3 HTTP 边界 @@ -257,20 +196,20 @@ axum/tower 类型只允许出现在 `api` 与 `bootstrap`。领域函数签名 ### 5.4 文件行数预算 - 软上限 **400 行**（超过即在 review 中说明理由）；硬上限 **800 行**（超过必须拆，CI 直接阻断，§9）。 -- 现状超硬线清单（11 个，全部有既定拆法）： +- 迁移前超硬线清单（11 个）及最终落点： | 文件（现路径） | 行数 | 拆法 | | --- | --- | --- | -| `proxy/dispatch/responses/service.rs` | 1913 | → `dispatch/service.rs`（编排骨架）+ `upstream_call.rs` + `stream/*` + `recovery/*` 已分走的逻辑归位 | -| `upstream/transport/websocket.rs` | 1470 | → `websocket.rs`（连接/握手）+ `websocket_frames.rs`（帧编解码） | -| `admin/accounts/routes.rs` | 1455 | → `api/admin/accounts_routes.rs` 按子资源拆（accounts / oauth / quota / import-export） | -| `upstream/accounts/pool.rs` | 1420 | → `accounts/pool.rs`（门面）+ `pool/state.rs`（槽位/在途）+ `pool/filters.rs` | -| `upstream/transport/client.rs` | 1251 | → `client.rs` + `client_sse.rs`（SSE 请求路径） | -| `upstream/accounts/store/mod.rs` | 1126 | → `store.rs` 门面 + `store/queries.rs` 按读写再分 | -| `admin/monitoring/usage_record_store.rs` | 1114 | PR0 原地重写（拆表后 metadata 聚合全数删除，行数大减），PR1 搬家时拆 `telemetry/usage/{store,query}.rs` | -| `admin/monitoring/dashboard.rs` | 1080 | → `telemetry/dashboard.rs`（聚合）+ `api/admin/dashboard_routes.rs`（HTTP） | -| `upstream/protocol/websocket.rs` | 1033 | → `websocket.rs`（事件转换）+ `websocket_errors.rs`（错误帧解析） | -| `upstream/accounts/token_refresh/runtime.rs` | 953 | → `accounts/refresh/{service,lease,policy}.rs` | +| `proxy/dispatch/responses/service.rs` | 1913 | → `dispatch/responses/service.rs` + `service/{complete,stream,compact}.rs` | +| `upstream/transport/websocket.rs` | 1470 | → `upstream/openai/transport/websocket.rs` + `websocket/{stream,types}.rs` + `websocket_pump.rs` | +| `admin/accounts/routes.rs` | 1455 | → `api/admin/accounts_routes.rs` + `accounts_routes/{actions,stats}.rs`；领域操作进入 `accounts/manage` | +| `upstream/accounts/pool.rs` | 1420 | → `accounts/pool.rs` + `pool/{runtime,types}.rs` | +| `upstream/transport/client.rs` | 1251 | → `upstream/openai/transport/client.rs` + `client/requests.rs` | +| `upstream/accounts/store/mod.rs` | 1126 | → `accounts/store.rs` + `store/{queries,rows,write}.rs` | +| `admin/monitoring/usage_record_store.rs` | 1114 | → `telemetry/usage/store.rs` + `store/query.rs`；查询门面独立为 `usage/query.rs` | +| `admin/monitoring/dashboard.rs` | 1080 | → 账本查询进入 `telemetry`，跨域 presentation 组合进入 `api/admin/dashboard_routes/{data}.rs` | +| `upstream/protocol/websocket.rs` | 1033 | → `upstream/openai/protocol/websocket.rs` + `websocket/validation.rs` | +| `upstream/accounts/token_refresh/runtime.rs` | 953 | → `accounts/refresh/{service,lease,runtime}.rs` + `runtime/types.rs` | | `admin/system/updater/mod.rs` | 811 | → `update/{service,state}.rs` | ### 5.5 一域一账本 @@ -294,7 +233,7 @@ axum/tower 类型只允许出现在 `api` 与 `bootstrap`。领域函数签名 | `cpr:affinity:*` | `dispatch`（affinity/store.rs） | | `cpr:models:plan_snapshots` | `models` | -跨域读通过属主域的 query/service 方法，不直接跨域写。**唯一豁免**：`bootstrap/import_sqlite.rs`（一次性导入命令，database.md §6）按依赖序直写全部 PG 表；v3 源库退役后该文件可删。 +跨域读通过属主域的 query/service 方法，不直接跨域写。**跨域写唯一豁免**：`bootstrap/import_sqlite*`（一次性导入命令，database.md §6）按依赖序直写全部 PG 表；`telemetry/rebuild.rs` 仍只写 telemetry 自有账本，不是跨域豁免。v3 源库退役后导入模块可删。 --- @@ -304,7 +243,7 @@ axum/tower 类型只允许出现在 `api` 与 `bootstrap`。领域函数签名 1. **热路径反向依赖管理端**：`proxy/dispatch/usage_events.rs`、`event_recording.rs` import `admin::monitoring` 的 `Admin*Service` 写账本；`runtime/services.rs` 用 Admin 命名的服务装配核心链路。遥测是领域层，管理端只是它的一个读者。→ 建 `telemetry` 域（PR1）。 2. **核心编排寄居 proxy**：`proxy/dispatch/`（约 5600 行，含 1913 行 service）是系统心脏，却挂在客户端 HTTP 入口下；`SqliteSessionAffinityStore`（DB store）住在 proxy 里。→ 建 `dispatch` 域（PR4）。 -3. **config 模块越权**：`config/settings.rs`（601 行）= 持久化 + 热更新推送 + admin API key 校验，还持有账号池引用向上推送。config 应只剩启动 schema + loader。→ 建 `settings` 域 + watch 反转（PR2）。 +3. **config 模块越权**：`config/settings.rs`（601 行）= 持久化 + 热更新推送 + admin API key 校验，还持有账号池引用向上推送。config 应只剩启动 schema + loader。→ 建 `settings` 域，由 bootstrap 在持久化成功后统一应用快照（PR2）。 4. **admin/keys 四合一**：`admin/keys/service.rs`（698 行）同时装着 proxy 鉴权用的 `ClientKeyService`、运行时缓存、SQLite store、管理端 service——客户端鉴权路径穿过 admin 模块。→ 建 `keys` 域（PR2）。 5. **web/http 假分家**：`web/` 只有 69 行 assets，是路由的 fallback 组成部分。→ 并入 `api`（PR5）。 6. healthz 穿透三层拿 store 连接池执行裸 SQL（§5.2）。 @@ -352,17 +291,17 @@ axum/tower 类型只允许出现在 `api` 与 `bootstrap`。领域函数签名 | `proxy/dispatch/usage_events.rs` | `telemetry/recorder.rs` †（Admin* 类型消失） | | `proxy/dispatch/session_affinity.rs` | `dispatch/affinity/{types,store,service}.rs` † | | `proxy/dispatch/responses/affinity.rs` | `dispatch/affinity/resolve.rs` | -| `proxy/dispatch/responses/service.rs` | `dispatch/service.rs` † | -| `proxy/dispatch/responses/{live_stream,stream_lifecycle,prefetch,sse_failure,trace}.rs` | `dispatch/stream/{live,lifecycle,prefetch,sse_failure,trace}.rs` | -| `proxy/dispatch/responses/event_recording.rs` | `dispatch/recording.rs` | -| `proxy/dispatch/responses/errors.rs` + `proxy/dispatch/errors.rs` | `dispatch/errors.rs` † | -| `proxy/dispatch/{auth_recovery,cloudflare,exhaustion,implicit_resume,reasoning_replay}.rs` | `dispatch/recovery/{auth,cloudflare,exhaustion,implicit_resume,reasoning_replay}.rs` | -| `proxy/dispatch/upstream.rs` | `dispatch/upstream_call.rs` | +| `proxy/dispatch/responses/service.rs` | `dispatch/responses/service.rs` + `service/{complete,stream,compact}.rs` † | +| `proxy/dispatch/responses/{live_stream,stream_lifecycle,prefetch,sse_failure,trace}.rs` | `dispatch/responses/` 下同名文件 | +| `proxy/dispatch/responses/event_recording.rs` | `dispatch/responses/event_recording.rs`（只依赖 `telemetry::Recorder`） | +| `proxy/dispatch/responses/errors.rs` + `proxy/dispatch/errors.rs` | `dispatch/responses/errors.rs` + `dispatch/errors.rs`（协议编排错误与通用调度错误分开） | +| `proxy/dispatch/{auth_recovery,cloudflare,exhaustion,implicit_resume,reasoning_replay}.rs` | 通用逻辑保留在 `dispatch/` 根；Responses 专属恢复留在 `dispatch/responses/` | +| `proxy/dispatch/upstream.rs` | `dispatch/upstream.rs` | | `admin/monitoring/usage_record_{model,store,service}.rs` | `telemetry/usage/{types,store,query}.rs` †（重写已在 PR0 完成，此处纯搬家） | | `admin/monitoring/ops_error_{model,store,service}.rs` | `telemetry/ops/{types,store,query}.rs` † | | `admin/monitoring/account_usage_{store,service}.rs` | `telemetry/account_usage/{store,query}.rs` † | | `admin/monitoring/billing.rs` | `telemetry/billing.rs` | -| `admin/monitoring/dashboard.rs` | `telemetry/dashboard.rs` + `api/admin/dashboard_routes.rs` † | +| `admin/monitoring/dashboard.rs` | 各账本查询进入 `telemetry/*/query.rs`；跨域只读组合进入 `api/admin/dashboard_routes/{data}.rs` † | | `admin/monitoring/diagnostics.rs` | `telemetry/diagnostics.rs` | | `admin/monitoring/usage_record_routes.rs` | `api/admin/usage_routes.rs` | | `admin/keys/service.rs` | `keys/{types,store,service,manage}.rs` †（进程内鉴权缓存消亡，database.md §4.3） | @@ -372,7 +311,7 @@ axum/tower 类型只允许出现在 `api` 与 `bootstrap`。领域函数签名 | `admin/accounts/routes.rs` | `api/admin/accounts_routes.rs` †（按子资源拆） | | `admin/accounts/quota_view.rs` | `accounts/manage/quota_view.rs` | | `admin/accounts/service/*` | `accounts/manage/*`（`importing`→`import`、`exporting`→`export`、`testing`→`probe`、`contracts`→`types`） | -| `admin/{response,update_payload}.rs` | `api/admin/response.rs`、`update/types.rs` | +| `admin/{response,update_payload}.rs` | `api/admin/{response,update_payload}.rs` | | `admin/router.rs` | `api/admin/router.rs` | | `admin/settings/routes.rs` | `api/admin/settings_routes.rs` | | `admin/system/{routes,state}.rs`、`admin/system/updater/*` | `api/admin/system_routes.rs`、`update/{state,service,release,download,archive}.rs` † | @@ -380,12 +319,12 @@ axum/tower 类型只允许出现在 `api` 与 `bootstrap`。领域函数签名 | `upstream/accounts/{pool,window}.rs` | `accounts/{pool,window}.rs`（pool 拆 †） | | `upstream/accounts/store/*` | `accounts/store.rs` + `accounts/store/{queries,rows}.rs` † | | `upstream/accounts/quota/*` | `accounts/quota/{service,runtime}.rs` † | -| `upstream/accounts/token_refresh/*` | `accounts/refresh/{policy,lease,service}.rs` † | +| `upstream/accounts/token_refresh/*` | `accounts/refresh/{lease,service,runtime}.rs` + `runtime/types.rs` † | | `upstream/accounts/cookies.rs` | `accounts/cookies/store.rs` | | `upstream/accounts/importing.rs` | `accounts/import.rs` | -| `upstream/accounts/service.rs` | 并入 `accounts/mod.rs` 门面导出（32 行） | +| `upstream/accounts/service.rs` | `accounts/service.rs` | | `upstream/scheduler/*` | `accounts/scheduler.rs` + `accounts/scheduler/*`（mod.rs 门面类型移出） | -| `upstream/models/*` | `models/{types,catalog,service,store}.rs`（`backend_entry`+`info`+`snapshot`+`config` 并入 `types.rs`/`service.rs`） | +| `upstream/models/*` | `models/{config,info,snapshot,catalog,service,store}.rs`（各文件保留真实职责） | | `upstream/fingerprint.rs` | `upstream/openai/fingerprint/{types,store,runtime,updater}.rs` †（Repository→Store） | | `upstream/{protocol,transport}/*`、`upstream/token_client.rs` | `upstream/openai/` 下同名路径（websocket/client 按 §5.4 拆） | | `infra/*` | 原地保留 | @@ -401,9 +340,9 @@ axum/tower 类型只允许出现在 `api` 与 `bootstrap`。领域函数签名 | --- | --- | --- | | **PR0** | **存储终态切换**（database.md §7 全部 12 条，在现有目录结构内完成） | Cargo：sqlx 加 postgres、新增 redis；`infra/database.rs` 改 PG + 0001 终态基线、新增 `infra/redis.rs`；全部 store 换 PG 方言 / 运行态四键空间改 Redis；成功/失败事实拆表 + provider 维度 + 两条写入事务；client key 可取回、session/admin-key 哈希化 + `client_api_key_id` 归因贯通；Dashboard 口径换源；trim 周期任务化；`import-sqlite` / `rebuild-buckets` 子命令；tests support 换 PG+Redis;deploy compose 加双服务。验收对齐 database.md §6.4 | | **PR1** | `telemetry` 域成立 | monitoring 全部迁出 admin；`usage_events.rs` → `recorder.rs`；Admin* 遥测类型按 §4.3 更名；dispatch/api 改依赖 telemetry；启用"禁止 `crate::admin::monitoring`"检查 | -| **PR2** | `keys` / `auth` / `settings` / `update` 域成立 | 四个域从 admin/config 拆出；settings 改 watch 广播（§2.3）；config 只剩 schema+loader | +| **PR2** | `keys` / `auth` / `settings` / `update` 域成立 | 四个域从 admin/config 拆出；settings 只保留持久化与串行快照，消费者传播由 bootstrap 统一执行（§2.3）；config 只剩启动 schema+loader | | **PR3** | `accounts` / `models` 域成立 | `upstream/accounts`、`upstream/scheduler`、`admin/accounts/service` 三处合并为 `accounts`；`upstream` 缩为纯客户端并落成 `openai/` 子目录；动名词文件更名；Repository→Store | -| **PR4** | `dispatch` 域成立 | proxy/dispatch 整体迁出并按目标树重组；affinity store 独立成文件；`proxy` 只剩 api/client 的料 | +| **PR4** | `dispatch` 域成立 | proxy/dispatch 整体迁出；保留内聚的 `responses` 子能力并按行数拆 service 子文件；affinity store 独立成文件；`proxy` 只剩 api/client 的料 | | **PR5** | `api` / `bootstrap` 壳成立 | http + web + proxy 入口 + admin routes 合并为 `api`；runtime + config 装配部分改名 `bootstrap`；删除 `crate::{admin,proxy,runtime,web,http,config}` 六个旧根模块；启用全部依赖检查 | | **PR6** | 行数预算清账 | §5.4 清单内 11 个文件拆到硬线以下；启用行数预算 CI 阻断 | @@ -413,62 +352,12 @@ axum/tower 类型只允许出现在 `api` 与 `bootstrap`。领域函数签名 ## 9. CI 强制 -新增 `backend/scripts/check_architecture.sh`，进 `.github/workflows/ci.yml`（在 clippy 之后）： +`backend/scripts/check_architecture.sh` 是可执行规则的唯一来源，并由 `.github/workflows/ci.yml` 在 clippy 后调用；本文不复制脚本，避免规则双写漂移。当前强制项包括： -```bash -#!/usr/bin/env bash -# 架构依赖与命名断言。违规即失败。 -set -euo pipefail -cd "$(dirname "$0")/.." -fail=0 -deny() { # deny <描述> <路径...> - if grep -rn --include='*.rs' -E "$2" "${@:3}" ; then - echo "ARCH VIOLATION: $1" >&2; fail=1 - fi -} - -# 分层：领域层不得依赖 api / bootstrap -deny "domain -> api/bootstrap" 'crate::(api|bootstrap)\b' \ - src/infra src/upstream src/telemetry src/keys src/auth src/settings src/update \ - src/accounts src/models src/dispatch -# upstream 是纯客户端 -deny "upstream -> domain" 'crate::(accounts|dispatch|telemetry|keys|auth|settings|update|models)\b' src/upstream -# HTTP 类型只在 api / bootstrap -deny "axum outside api" '\baxum(::|\b)' \ - src/infra src/upstream src/telemetry src/keys src/auth src/settings src/update \ - src/accounts src/models src/dispatch -# SQL 与 Redis 命令只在 store / query -deny "sqlx outside store/query" '\bsqlx(::|\b)' \ - $(find src/api -name '*.rs') -deny "redis outside store" '\bredis::' \ - $(find src/api -name '*.rs') -# 旧模块根不得复活 -deny "legacy module path" 'crate::(admin|proxy|runtime|web|http|config)::' src -# 领域类型与命名纪律 -deny "Admin-prefixed domain type" '\bAdmin[A-Z][A-Za-z0-9]*\b' \ - src/infra src/upstream src/telemetry src/keys src/auth src/settings src/update \ - src/accounts src/models src/dispatch -deny "Repository naming" '\b[A-Za-z0-9]*Repository\b' src -deny "persistence repository naming" '\b[A-Za-z0-9_]*repository[A-Za-z0-9_]*\b' \ - src/accounts src/auth src/dispatch src/keys src/models src/settings src/telemetry \ - src/upstream/openai/fingerprint -deny "Arc alias" '\bArc[[:space:]]+as[[:space:]]+' src -# 禁用文件名 -banned=$(find src -name 'importing.rs' -o -name 'exporting.rs' -o -name 'testing.rs' \ - -o -name 'util.rs' -o -name 'helper.rs' -o -name 'common.rs' -o -name 'misc.rs') -[ -n "$banned" ] && { echo "ARCH VIOLATION: banned file names: $banned" >&2; fail=1; } -# mod.rs 纪律 -while read -r f; do - lines=$(wc -l < "$f") - [ "$lines" -gt 50 ] && { echo "ARCH VIOLATION: $f has $lines lines (mod.rs <= 50)" >&2; fail=1; } -done < <(find src -name mod.rs) -# 行数硬上限（PR6 已完成，违规即失败） -while read -r f; do - lines=$(wc -l < "$f") - [ "$lines" -gt 800 ] && { echo "ARCH VIOLATION: $f has $lines lines (> 800)" >&2; fail=1; } -done < <(find src -name '*.rs' ! -name mod.rs) -exit $fail -``` +- 领域层不得依赖 `api` / `bootstrap`，且同时识别 `crate::api::...` 与 `use crate::{ api::... }` 两种写法；`upstream` 不得依赖任何领域模块。 +- axum 只进入 `api` / `bootstrap`；API 不得执行 SQL 或 Redis 命令。 +- 存储命令只能位于 §5.2 列出的 adapter/query/维护入口；文件名、类型前缀、旧模块路径、Repository 命名与 Arc 别名均有阻断。 +- `mod.rs` 不超过 50 行，其他 Rust 文件不超过 800 行。 全部规则已启用；后续新增代码一旦恢复旧路径、越层依赖、违规命名或突破硬行数上限，CI 直接失败。 @@ -476,10 +365,10 @@ exit $fail ## 10. 仓库级约定 -- `backend/tests/` 镜像 `src/`，不在 `src/` 写测试模块。测试依赖本机（或 CI 服务容器）的 PostgreSQL 与 Redis：`CPR_TEST_DATABASE_URL` 下每测试建独立数据库、`CPR_TEST_REDIS_URL` 下每测试随机键前缀（database.md §7）。 -- `docs/` 只保留权威文档：`architecture.md`（本文）、`database.md`。行为细节（调度策略、WS 池、更新流程等）以模块文档注释为准（`scheduler/mod.rs` 是范本），不再在 docs 里维护第二份会腐化的副本。 +- `backend/tests/` 顶层镜像 `src/` 领域，不在 `src/` 写测试模块。测试依赖本机（或 CI 服务容器）的 PostgreSQL 与 Redis：`CPR_TEST_DATABASE_URL` 下每测试建独立数据库、`CPR_TEST_REDIS_URL` 下每测试随机键前缀（database.md §7）。 +- `architecture.md`（本文）与 `database.md` 是架构/存储权威；`docs/` 可保留故障复盘或修复记录，但不得另行定义架构、schema 或运行契约。行为实现细节以模块文档注释为准。 - `deploy/` 随 PR0 扩容（compose 增 postgres/redis 服务与健康检查依赖）；`release/`、`skills/` 现状合格，不动。 -- 前端联动（全部随 PR0）：keys 列表继续返回完整 key，并长期提供复制与 CCSwitch 导入；admin API key 状态不再有脱敏值（哈希不可逆，只余"已启用"）；成功记录响应不再有 `level` 字段；错误明细/筛选改走 `/api/admin/ops/errors`；Dashboard "总请求/总 token" 卡片改为 account_usage 来源并标注"成功累计"。 +- 前端联动（全部随 PR0）：keys 列表继续返回完整 key，复制与 CCSwitch 导入长期保持可用且交互/文案不变；admin API key 状态不再有脱敏值（哈希不可逆，只余"已启用"）；成功记录响应不再有 `level` 字段；错误明细/筛选改走 `/api/admin/ops/errors`；Dashboard 只修正取数来源，既有卡片文案不变。 --- diff --git a/docs/database.md b/docs/database.md index 5c3ec8e3d..39f8c635a 100644 --- a/docs/database.md +++ b/docs/database.md @@ -66,7 +66,7 @@ | 布尔 | boolean | — | | 快照/模板/调试 JSON | jsonb | 见 §2.7 | -**枚举用 text + check,不用 PG 原生 enum**(有意修订旧文档 §8 的一句话设想):原生 enum 增删值要 `alter type` 且值永远无法删除;check 约束换一条 `drop/add constraint` 即完成枚举演进(SQLite 0002 改名先例证明枚举确实会变)。**不使用 real/double 存业务数据,不存储金额**。成本一律查询期由 token × 单价计算(`admin/monitoring/billing.rs`):单价会变、需追溯重算,落库的浮点成本既漂移又过期。 +**枚举用 text + check,不用 PG 原生 enum**(有意修订旧文档 §8 的一句话设想):原生 enum 增删值要 `alter type` 且值永远无法删除;check 约束换一条 `drop/add constraint` 即完成枚举演进(SQLite 0002 改名先例证明枚举确实会变)。**不使用 real/double 存业务数据,不存储金额**。成本一律查询期由 token × 单价计算(`telemetry/billing.rs`):单价会变、需追溯重算,落库的浮点成本既漂移又过期。 ### 2.3 时间戳 @@ -114,7 +114,7 @@ 2. **可丢弃性是准入条件**:FLUSHALL 之后系统必须无数据损失地继续运行(会话重登、租约重竞争、亲和退化为普通调度、缓存重拉)。不满足即不属于 Redis。 3. **互斥必须原子**:获取用 `SET NX PX`,续约/释放用 Lua 比较 owner 后操作,禁止 GET-判断-SET 三段式。 4. 二级索引(亲和的 conversation ZSET / account SET)由写入方维护,读取方容忍成员悬垂(主键已过期),遇悬垂惰性清理。 -5. 值统一 JSON 文本;不用 Redis Hash 拆业务对象字段(整读整写,无字段级更新需求)。 +5. 单个业务对象统一编码为完整 JSON 文本,不把对象属性拆成 Redis Hash 字段。集合缓存可使用 HASH（如 `cpr:models:plan_snapshots`），但每个 field 的 value 仍是一个完整 JSON 对象，整读整写。 ### 2.10 保留与可重建性(诚实边界) @@ -699,8 +699,8 @@ errors: 错误分布 / 排查 ← ops_error ### 5.3 Dashboard 口径决策(钉死) -- 所有 traffic 卡片(今日请求、区间请求、错误率、QPS 趋势)**一律来自 request_time_buckets**,语义为"bucket 保留期内"。卡片文案如实标注区间,**不提供也不暗示"历史总量"**——bucket 有 90 天保留期,任何"total"字样都是口径谎言。 -- 生命周期累计仅有一处合法来源:`account_usage` 累计列(成功口径),使用时必须标注"成功累计"。 +- 周期 traffic 卡片(今日请求、区间请求、错误率、QPS 趋势)**一律来自 request_time_buckets**,语义为"bucket 保留期内"；生命周期累计卡片不从 bucket 推导。 +- 生命周期累计仅有一处合法来源:`account_usage` 累计列(成功口径)。这是后端数据语义，不要求改动既有前端文案。 - 同一张卡片禁止混排两种口径的数字(v3 的 `todayRequests` 来自 bucket、`totalRequests` 来自 usage_records 全表扫描就是这个病)。 - 错误明细与错误分布来自 ops_error_logs;成功事件列表不再混排错误行。 - 真正的长期趋势需求 → 日粒度归档表 `request_day_buckets`(§8),不是延长 15 分钟桶保留期。 @@ -761,7 +761,7 @@ codex-proxy-rs import-sqlite <旧库路径.sqlite> | 3 | 抽查 `select count(*) from client_api_keys where key is null or key = ''` = 0 | 旧 key 值不变；鉴权、管理端复制与 CCSwitch 导入均可用 | | 3a | 抽查 `select count(*) from accounts where status = 'refreshing'` = 0 | 旧瞬态状态已规范化为 `expired`，且 `next_refresh_at is null` | | 4 | 执行 `rebuild-buckets`(§7),用新口径重算事实保留期内的桶 | 保留期内桶 token 列 = usage_records 聚合值 | -| 5 | 核对 Dashboard 卡片全部走 bucket 口径,生命周期数字标注"成功累计" | 同卡片无混合口径数字 | +| 5 | 核对 Dashboard 周期卡片走 bucket、生命周期累计走 account_usage，前端既有文案不变 | 同卡片无混合口径数字 | --- @@ -774,7 +774,7 @@ codex-proxy-rs import-sqlite <旧库路径.sqlite> 3. **鉴权归因**(`keys/{store,service,manage}.rs`、`api/client/auth.rs`):key 创建落唯一 `key` 列；鉴权按完整 key 做 PG unique 点查(删除进程内鉴权缓存);鉴权返回 key `id`,调用链把 `client_api_key_id` 装进请求上下文直至事件写入。管理端列表持续返回完整 key，前端长期保留复制与 CCSwitch 导入入口。 4. **事件结构**(`telemetry/{usage,ops}`、`telemetry/recorder.rs`、`dispatch/responses/event_recording.rs`):成功/失败事件携带 `client_api_key_id` 与 `provider`;token、requested/upstream_model、service_tier、first_token_ms 从 metadata 移到一等字段;metadata 不再写已提升字段;UsageRecord 删除 `level`(成功事实无等级)。 5. **写入路径**(`telemetry/usage/store.rs`、`telemetry/ops/store.rs`、`telemetry/buckets/store.rs`):§5.2 的两条 PG 事务;错误路径对桶只 `error_count + 1`;`__unknown__` sentinel;min/max 延迟使用 nullable-safe 的 `least()`/`greatest()`;内联 trim 移除。 -6. **查询路径**(`telemetry/{usage,ops,account_usage}`、`api/admin/{usage,ops,dashboard,accounts}_routes*`):summary/分布/趋势全部走列,删除所有 `->>` 聚合与 metadata LIKE;新增 `/api/admin/ops/errors` 错误明细查询面;Dashboard 按 §5.3 改卡片来源与文案。 +6. **查询路径**(`telemetry/{usage,ops,account_usage}`、`api/admin/{usage,ops,dashboard,accounts}_routes*`):summary/分布/趋势全部走列,删除所有 `->>` 聚合与 metadata LIKE;新增 `/api/admin/ops/errors` 错误明细查询面;Dashboard 按 §5.3 只改取数来源，不改既有文案。 7. **运行态改 Redis**(`auth/store.rs` 会话、`accounts/refresh/lease.rs`、`dispatch/affinity/store.rs`、`models/store.rs`):按 §4B 契约实现;删除 admin 会话/亲和清理任务与亲和重启恢复;账号删除路径挂 Redis 亲和级联(§4B.3)。 8. **维护命令**(`main.rs` CLI):`import-sqlite `(§6);`rebuild-buckets`——删除事实保留期内的桶并从两张事实表重算(15 分钟槽对齐在 SQL 侧 `floor(epoch/900)*900`,§2.3);保留期外只读不动。范围明确**不含** `account_usage` / `account_model_usage`(§2.10,不可重建)。 9. **清理任务**(`bootstrap/tasks/`):三张增长表的周期 trim(读 `runtime_settings` 三列);cookie 清理保留;admin 会话与亲和清理任务删除(TTL 接管)。