diff --git a/backend/crates/gateway-admin/src/use_case/credential_mutation.rs b/backend/crates/gateway-admin/src/use_case/credential_mutation.rs index e405dcbef..b42e8987d 100644 --- a/backend/crates/gateway-admin/src/use_case/credential_mutation.rs +++ b/backend/crates/gateway-admin/src/use_case/credential_mutation.rs @@ -19,30 +19,60 @@ use gateway_core::{account::ProviderAccountId, routing::ProviderKind, runtime::S use super::{map_store_error, publish_committed}; pub(super) async fn publish_credentials_and_observe_quota( - peer: Option<&dyn crate::ports::peer::PeerUpstreamService>, + peer: Option<&std::sync::Arc>, provider: &std::sync::Arc, + accounts: &std::sync::Arc, snapshot: &dyn SnapshotControl, revision: crate::model::Revision, account_ids: &[ProviderAccountId], request_id: &str, ) -> Result<(), AdminError> { - notify_account_facts_changed(peer, provider.as_ref(), account_ids).await; + notify_account_facts_changed(peer.map(AsRef::as_ref), provider.as_ref(), account_ids).await; publish_committed(snapshot, revision).await?; // 凭据已经提交；额度只是可重建的观察，不能拖住管理请求或回滚提交结果 let provider = provider.clone(); + let peer = peer.cloned(); + let accounts = accounts.clone(); let account_ids = account_ids.to_vec(); let request_id = request_id.to_owned(); tokio::spawn(async move { for account_id in account_ids { - if let Err(error) = provider - .quota(ProviderQuotaRequest { - account_id: account_id.clone(), - refresh: true, - rolling_usage: None, - }) + let request = ProviderQuotaRequest { + account_id: account_id.clone(), + refresh: true, + rolling_usage: None, + }; + let result = if let Some(peer) = peer.as_deref() { + let details = match required_credential( + accounts.as_ref(), + provider.provider_kind(), + &account_id, + "initial quota observation", + ) .await - { + { + Ok(details) => details, + Err(error) => { + tracing::warn!(request_id, account_id = %account_id, error = %error, "initial quota account lookup failed"); + continue; + } + }; + if !details.credential.enabled { + continue; + } + // 首查与手动刷新共用认证能力分派，互联协议由 Admin 解释 + super::peer_link::upstream::read_quota( + Some(peer), + provider.as_ref(), + &details.credential, + request, + ) + .await + } else { + provider.quota(request).await + }; + if let Err(error) = result { tracing::warn!( request_id, provider = %provider.provider_kind().as_str(), diff --git a/backend/crates/gateway-admin/src/use_case/credentials/flow.rs b/backend/crates/gateway-admin/src/use_case/credentials/flow.rs index 1457d4b6b..74a9acd77 100644 --- a/backend/crates/gateway-admin/src/use_case/credentials/flow.rs +++ b/backend/crates/gateway-admin/src/use_case/credentials/flow.rs @@ -92,8 +92,9 @@ impl ProviderCredentials { .map_err(|error| map_store_error(error, "provider credential import"))?; drop(proxy_reservation); publish_credentials_and_observe_quota( - self.upstream_info.as_deref(), + self.upstream_info.as_ref(), &self.provider, + &self.accounts, self.snapshot.as_ref(), result.config_revision, &result.credential_ids, @@ -249,8 +250,9 @@ impl super::CredentialsService { let result = outcome.result; if outcome.newly_committed { publish_credentials_and_observe_quota( - self.upstream_info.as_deref(), + self.upstream_info.as_ref(), provider, + &self.accounts, self.snapshot.as_ref(), result.config_revision, std::slice::from_ref(&result.account_id), diff --git a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/service.rs b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/service.rs index 946adca42..bd05defe6 100644 --- a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/service.rs +++ b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/service.rs @@ -61,7 +61,40 @@ impl PeerInfoService { } pub(crate) async fn account_changed(&self, id: &ProviderAccountId) -> Result<(), StoreError> { - self.forget(id); + let has_observation = self + .observations + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .contains_key(id); + let has_attempt = self + .attempts + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .contains_key(id); + if !has_observation && !has_attempt { + return Ok(()); + } + let Some(account) = self + .accounts + .get_account(id) + .await? + .filter(|account| self.supported(account)) + else { + self.forget(id); + return Ok(()); + }; + let connection = UpstreamConnection::for_account(&account); + // 普通信息编辑保留观测与退避，连接变化由刷新状态中的身份比较处理 + let mut observations = self + .observations + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if observations + .get(id) + .is_some_and(|cached| cached.connection != connection) + { + observations.remove(id); + } Ok(()) } diff --git a/backend/crates/gateway-admin/src/use_case/plugin_accounts.rs b/backend/crates/gateway-admin/src/use_case/plugin_accounts.rs index 7256626af..2a425b045 100644 --- a/backend/crates/gateway-admin/src/use_case/plugin_accounts.rs +++ b/backend/crates/gateway-admin/src/use_case/plugin_accounts.rs @@ -173,8 +173,9 @@ impl PluginAccountAccess for DefaultPluginAccountAccess { .credential .credential_revision; publish_credentials_and_observe_quota( - self.upstream_info.as_deref(), + self.upstream_info.as_ref(), &provider, + &self.accounts, self.snapshot.as_ref(), result.config_revision, std::slice::from_ref(&account_id), @@ -217,8 +218,9 @@ impl PluginAccountAccess for DefaultPluginAccountAccess { .credential_revision .ok_or_else(|| AdminError::internal("账号保存未返回凭据版本"))?; publish_credentials_and_observe_quota( - self.upstream_info.as_deref(), + self.upstream_info.as_ref(), &provider, + &self.accounts, self.snapshot.as_ref(), result.config_revision, std::slice::from_ref(&result.account_id), diff --git a/backend/crates/gateway-admin/tests/use_case/accounts/mod.rs b/backend/crates/gateway-admin/tests/use_case/accounts/mod.rs index a05a1ec03..b1795df03 100644 --- a/backend/crates/gateway-admin/tests/use_case/accounts/mod.rs +++ b/backend/crates/gateway-admin/tests/use_case/accounts/mod.rs @@ -279,6 +279,10 @@ impl FakeProviderAdmin { #[async_trait] impl ProviderAdmin for FakeProviderAdmin { + fn peer_authentication_kinds(&self) -> &[&str] { + &["api_key"] + } + fn account_capabilities( &self, _account_id: &ProviderAccountId, diff --git a/backend/crates/gateway-admin/tests/use_case/credentials/mod.rs b/backend/crates/gateway-admin/tests/use_case/credentials/mod.rs index 78ac04fec..1c77a5cad 100644 --- a/backend/crates/gateway-admin/tests/use_case/credentials/mod.rs +++ b/backend/crates/gateway-admin/tests/use_case/credentials/mod.rs @@ -1,4 +1,5 @@ //! 按 Provider 组织凭据导入、授权与提交边界测试 mod openai; +mod peer; mod xai; diff --git a/backend/crates/gateway-admin/tests/use_case/credentials/peer.rs b/backend/crates/gateway-admin/tests/use_case/credentials/peer.rs new file mode 100644 index 000000000..a80736b8c --- /dev/null +++ b/backend/crates/gateway-admin/tests/use_case/credentials/peer.rs @@ -0,0 +1,184 @@ +//! 验证凭据提交后的互联首查、原生额度分派与异步失败边界 + +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use async_trait::async_trait; +use gateway_admin::{ + model::provider_credentials::{ImportCredentials, ProviderQuota, ProviderQuotaReadRequest}, + ports::{ + peer::PeerUpstreamService, + provider::{ProviderAdminError, ProviderAdminErrorKind}, + }, +}; +use gateway_core::{ + account::{ProviderAccount, ProviderAccountId, UpstreamInfoSource, UpstreamSchedulingSignal}, + error::StoreError, + routing::ProviderKind, +}; + +use super::super::{ + AdminHarness, + accounts::{FakeAccountStore, FakeProviderAdmin, account_record, context, document, events}, +}; + +#[tokio::test] +async fn imported_peer_accounts_should_start_observation_without_waiting_for_network() { + let events = events(); + let provider = FakeProviderAdmin::new("openai", events.clone()); + provider.set_import_account_ids(&["acct_first", "acct_second"]); + let mut first = account_record("openai"); + first.id = "acct_first".into(); + first.authentication_kind = "api_key".into(); + let mut second = first.clone(); + second.id = "acct_second".into(); + let store = FakeAccountStore::with_account(first.clone(), events); + store.set_accounts(vec![first, second]); + let peer = Arc::new(PendingPeer::default()); + let services = AdminHarness::new() + .provider(provider.clone()) + .accounts(store) + .upstream_info(peer.clone()) + .build() + .await; + + let result = tokio::time::timeout( + Duration::from_secs(1), + services + .credentials() + .for_provider(&ProviderKind::new("openai").unwrap()) + .unwrap() + .import_document(ImportCredentials { + outbound_proxy_id: None, + settings: None, + context: context("import-peer"), + document: document(), + }), + ) + .await + .expect("import must not wait for peer response") + .expect("import committed"); + peer.wait_for_requests(1).await; + assert_eq!(peer.changed.lock().unwrap().len(), 2); + assert_eq!(result.credential_ids.len(), 2); + assert!(provider.quota_requests().is_empty()); + peer.gate.add_permits(2); + peer.wait_for_requests(2).await; + assert_eq!( + *peer.requests.lock().unwrap(), + ["acct_first", "acct_second"] + ); + assert!(provider.quota_requests().is_empty()); +} + +#[tokio::test] +async fn native_import_should_use_provider_quota_when_peer_service_is_available() { + let events = events(); + let provider = FakeProviderAdmin::new("openai", events.clone()); + provider.set_import_account_ids(&["acct_test"]); + let store = FakeAccountStore::new("openai", events); + let peer = Arc::new(PendingPeer::default()); + let services = AdminHarness::new() + .provider(provider.clone()) + .accounts(store) + .upstream_info(peer.clone()) + .build() + .await; + + services + .credentials() + .for_provider(&ProviderKind::new("openai").unwrap()) + .unwrap() + .import_document(ImportCredentials { + outbound_proxy_id: None, + settings: None, + context: context("import-native"), + document: document(), + }) + .await + .expect("native import"); + provider.wait_for_quota_requests(1).await; + assert!(peer.requests.lock().unwrap().is_empty()); + assert_eq!( + provider.quota_requests()[0].account_id.as_str(), + "acct_test" + ); +} + +struct PendingPeer { + changed: Mutex>, + requests: Mutex>, + started: tokio::sync::Notify, + gate: tokio::sync::Semaphore, +} + +impl Default for PendingPeer { + fn default() -> Self { + Self { + changed: Mutex::default(), + requests: Mutex::default(), + started: tokio::sync::Notify::new(), + gate: tokio::sync::Semaphore::new(0), + } + } +} + +impl PendingPeer { + async fn wait_for_requests(&self, count: usize) { + tokio::time::timeout(Duration::from_secs(1), async { + loop { + let notified = self.started.notified(); + if self.requests.lock().unwrap().len() >= count { + return; + } + notified.await; + } + }) + .await + .expect("initial peer observation started"); + } +} + +impl UpstreamInfoSource for PendingPeer { + fn signal( + &self, + _: &ProviderAccount, + _: &str, + _: std::time::SystemTime, + ) -> Option { + None + } +} + +#[async_trait] +impl PeerUpstreamService for PendingPeer { + async fn account_changed(&self, id: &ProviderAccountId) -> Result<(), StoreError> { + self.changed.lock().unwrap().push(id.as_str().into()); + Ok(()) + } + fn forget(&self, _: &ProviderAccountId) {} + async fn refresh_quota( + &self, + id: &ProviderAccountId, + ) -> Result { + self.requests.lock().unwrap().push(id.as_str().into()); + self.started.notify_one(); + self.gate.acquire().await.unwrap().forget(); + Err(ProviderAdminError::new(ProviderAdminErrorKind::Unavailable)) + } + async fn read_quota(&self, _: &ProviderAccountId) -> Result, StoreError> { + panic!("unexpected cached read") + } + async fn read_summary( + &self, + _: &ProviderAccountId, + ) -> Result, StoreError> { + panic!("unexpected summary read") + } + async fn read_quotas( + &self, + _: &[ProviderQuotaReadRequest<'_>], + ) -> Vec> { + panic!("unexpected batch read") + } +} diff --git a/backend/crates/gateway-admin/tests/use_case/mod.rs b/backend/crates/gateway-admin/tests/use_case/mod.rs index 8eaee14c2..c0322b8c9 100644 --- a/backend/crates/gateway-admin/tests/use_case/mod.rs +++ b/backend/crates/gateway-admin/tests/use_case/mod.rs @@ -112,6 +112,7 @@ pub(super) struct AdminHarness { plugin_store: Arc, plugin_inspector: Arc, client_key_verifier: Arc, + upstream_info: Option>, service_middleware: gateway_admin::public_service::PlanSource, } @@ -142,6 +143,7 @@ impl AdminHarness { plugin_store: Arc::new(plugins::TestPluginPorts), plugin_inspector: Arc::new(plugins::TestPluginPorts), client_key_verifier: Arc::new(UnavailableClientKeyVerifier), + upstream_info: None, service_middleware: Arc::new(|| None), } } @@ -161,6 +163,14 @@ impl AdminHarness { self } + pub(super) fn upstream_info( + mut self, + peer: Arc, + ) -> Self { + self.upstream_info = Some(peer); + self + } + pub(super) fn account_runtime(mut self, store: Arc) -> Self { self.account_runtime = store; self @@ -287,7 +297,7 @@ impl AdminHarness { Arc::new(plugins::TestPluginPorts), ), gateway_admin::AdminRuntimePorts { - upstream_info: None, + upstream_info: self.upstream_info, timezone: Default::default(), service_middleware: self.service_middleware, plugin_preparation: Arc::new(plugins::TestPluginPorts), diff --git a/backend/crates/gateway-admin/tests/use_case/peer_link/upstream/mod.rs b/backend/crates/gateway-admin/tests/use_case/peer_link/upstream/mod.rs index 1452509dd..b64f2e838 100644 --- a/backend/crates/gateway-admin/tests/use_case/peer_link/upstream/mod.rs +++ b/backend/crates/gateway-admin/tests/use_case/peer_link/upstream/mod.rs @@ -33,21 +33,22 @@ async fn discovery_uses_provider_capabilities_and_publishes_neutral_signals() { }); let registry = ProviderAdminRegistry::new(providers).unwrap(); let accounts = Arc::new(TestAccounts { - accounts: vec![ + accounts: Mutex::new(vec![ account("acct_first", "openai", "peer_test"), account("acct_second", "xai", "peer_test"), account("acct_native", "xai", "oauth"), - ], + ]), records: Mutex::default(), failed_provider, }); let transport = Arc::new(TestTransport(Mutex::new(Vec::new()))); let mut bundle = initialize_peer_upstream(accounts.clone(), ®istry, transport.clone()).unwrap(); + let initial_accounts = accounts.accounts.lock().unwrap().clone(); let reader = UpstreamInfoReader::default(); assert!( reader - .signal(&accounts.accounts[0], "model", SystemTime::now()) + .signal(&initial_accounts[0], "model", SystemTime::now()) .is_none() ); let source = bundle.scheduling_source(); @@ -77,7 +78,7 @@ async fn discovery_uses_provider_capabilities_and_publishes_neutral_signals() { vec!["acct_first", "acct_second"] } ); - for account in &accounts.accounts[..2] { + for account in &initial_accounts[..2] { if Some(account.provider().as_str()) == failed_provider { assert!(reader.signal(account, "model", SystemTime::now()).is_none()); continue; @@ -101,19 +102,65 @@ async fn discovery_uses_provider_capabilities_and_publishes_neutral_signals() { } assert!( reader - .signal(&accounts.accounts[2], "model", SystemTime::now()) + .signal(&initial_accounts[2], "model", SystemTime::now()) .is_none() ); drop(worker); drop(bundle); assert!( reader - .signal(&accounts.accounts[0], "model", SystemTime::now()) + .signal(&initial_accounts[0], "model", SystemTime::now()) .is_none() ); } } +#[tokio::test] +async fn account_edits_preserve_observation_and_refresh_budget_until_connection_changes() { + let mut provider = UnavailableProvider::new("openai"); + provider.peer_enabled = true; + let registry = + ProviderAdminRegistry::new([Arc::new(provider) as Arc]).unwrap(); + let initial = account("acct_peer", "openai", "peer_test"); + let accounts = Arc::new(TestAccounts { + accounts: Mutex::new(vec![initial.clone()]), + records: Mutex::default(), + failed_provider: None, + }); + let transport = Arc::new(TestTransport(Mutex::new(Vec::new()))); + let bundle = initialize_peer_upstream(accounts.clone(), ®istry, transport.clone()).unwrap(); + let service = bundle.service(); + service.refresh_quota(initial.id()).await.unwrap(); + let before = service.read_summary(initial.id()).await.unwrap().unwrap(); + + let edited = initial.clone().with_notes(Some("仅修改备注".into())); + *accounts.accounts.lock().unwrap() = vec![edited.clone()]; + service.account_changed(edited.id()).await.unwrap(); + assert!( + bundle + .scheduling_source() + .signal(&edited, "model", SystemTime::now()) + .is_some() + ); + let after = service.refresh_quota(edited.id()).await.unwrap(); + assert_eq!(after.upstream, before.upstream); + assert_eq!(transport.0.lock().unwrap().len(), 1); + + let reconnected = edited.with_outbound_proxy(Some( + OutboundProxy::parse("http://proxy.invalid:8080").unwrap(), + )); + *accounts.accounts.lock().unwrap() = vec![reconnected.clone()]; + service.account_changed(reconnected.id()).await.unwrap(); + assert!( + bundle + .scheduling_source() + .signal(&reconnected, "model", SystemTime::now()) + .is_none() + ); + service.refresh_quota(reconnected.id()).await.unwrap(); + assert_eq!(transport.0.lock().unwrap().len(), 2); +} + fn account(id: &str, provider: &str, authentication: &str) -> ProviderAccount { ProviderAccount::new( ProviderAccountId::new(id).unwrap(), @@ -165,7 +212,7 @@ impl PeerTransport for TestTransport { } struct TestAccounts { - accounts: Vec, + accounts: Mutex>, records: Mutex>, failed_provider: Option<&'static str>, } @@ -188,6 +235,8 @@ impl ProviderAccountStore for TestAccounts { ) -> Result, StoreError> { Ok(self .accounts + .lock() + .unwrap() .iter() .find(|account| account.id() == account_id) .cloned()) @@ -204,6 +253,8 @@ impl ProviderAccountStore for TestAccounts { } Ok(self .accounts + .lock() + .unwrap() .iter() .filter(|account| account.provider() == provider) .cloned() @@ -227,6 +278,8 @@ impl ProviderAccountStore for TestAccounts { id: &ProviderAccountId, ) -> Result { self.accounts + .lock() + .unwrap() .iter() .find(|account| account.id() == id) .map(loaded) diff --git a/docs/architecture.md b/docs/architecture.md index d1412a3cf..ba40c83b1 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1012,7 +1012,8 @@ Provider 声明可连接的认证种类，并从本轮凭据中提取连接材 Host 执行有界 HTTP 请求，复用组合根注入的证书策略和账号出口；Store 通过通用账号额度端口保存不透明观测，不解释互联或 Provider 协议。 互联与原生额度共用 `provider_quota_json`、`quota_observed_at`、凭据版本和额度访问事实，每个账号仅保存当前来源的快照。 OAuth 额度解析属于 Provider，CPR 发现、解析与投影属于 Admin；账号用例按能力分派读写与刷新，不将两种协议合并为同一解析服务。 -互联刷新沿用运行期排程和 Host Worker 租约，分批轮转账号，同一账号的并发刷新通过运行期锁合并；重启后根据已有快照恢复查询，不持久化领取或请求代次。 +凭据新建、导入或授权提交后，后台首查复用同一额度分派，不等待周期同步或阻塞保存请求。 +互联周期刷新沿用运行期排程和 Host Worker 租约，分批轮转账号，同一账号的并发刷新通过运行期锁合并；重启后根据已有快照恢复查询，不持久化领取或请求代次。 额度时间戳支持轻量缓存校验，普通账号和凭据查询不加载额度正文，完整观测分批读取 共享端按 Core 的预算规则从本地账本生成明确的访问状态，与本地准入共用口径；金额字段只供接入端展示。 @@ -1023,7 +1024,8 @@ OAuth 额度解析属于 Provider，CPR 发现、解析与投影属于 Admin； Admin 按连接身份及额度观察时间缓存管理摘要与调度投影，解析和投影在锁外完成。 Core 定义中立调度信号与只读端口，组合根绑定观测来源；Provider 选号不持有控制面服务，弱引用避免互相保活。 -账号和凭据提交后的共用通知同时失效原生 Provider 与互联派生状态，插件账号写入复用同一流程。 +账号和凭据提交后的共用通知更新原生 Provider 与互联派生状态，插件账号写入复用同一流程； +互联按当前连接身份失效观测，普通信息编辑保留已有摘要与刷新退避，停用或删除时释放运行态。 账号列表、概览与代理关联账号核对观测版本后复用预算与池摘要，详情读取逐节点窗口；手动刷新通过共用账号额度接口等待查询完成并返回结果。 界面复用现有额度窗口，保留上游原始观测时间；Key 金额、原生账号百分比和本机 Token 用量分别解释。 请求与字段合同见 [CPR 互联 API](api.md#cpr-互联只读协议) diff --git a/frontend/src/components/account/AccountPlanBadge.vue b/frontend/src/components/account/AccountPlanBadge.vue index bf041fd33..6fa0c8d29 100644 --- a/frontend/src/components/account/AccountPlanBadge.vue +++ b/frontend/src/components/account/AccountPlanBadge.vue @@ -2,6 +2,7 @@ import type { AccountUpstream } from '@/api' import { computed } from 'vue' +import { accountUpstreamPending } from '@/utils/account-upstream' import { stableVisualIndex } from '@/utils/color' import AccountUpstreamBadge from './AccountUpstreamBadge.vue' @@ -62,7 +63,7 @@ const paletteClass = computed(() => {