diff --git a/backend/crates/gateway-admin/src/model/upstream_info.rs b/backend/crates/gateway-admin/src/model/upstream_info.rs index 073e0bed6..f50198fa1 100644 --- a/backend/crates/gateway-admin/src/model/upstream_info.rs +++ b/backend/crates/gateway-admin/src/model/upstream_info.rs @@ -4,6 +4,7 @@ use chrono::{DateTime, Utc}; #[derive(Debug, Clone, PartialEq, Eq)] pub struct ProviderUpstreamInfo { + /// 当前连接曾成功识别互联，失权时保留身份但清除观测内容 pub discovered: bool, pub observed_at: Option>, pub stale: bool, diff --git a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/cache.rs b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/cache.rs index e9b420090..7ad146e7d 100644 --- a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/cache.rs +++ b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/cache.rs @@ -57,6 +57,7 @@ impl PeerInfoService { scheduling: None, summary: None, unsupported: false, + unavailable: None, }, )); } @@ -92,6 +93,7 @@ impl PeerInfoService { ( connection.clone(), cached.and_then(|cached| cached.summary.clone()), + cached.and_then(|cached| cached.unavailable), cached.is_some_and(|cached| cached.unsupported), ) }) @@ -99,10 +101,11 @@ impl PeerInfoService { }; Ok(summaries .into_iter() - .map(|(connection, peer, unsupported)| { + .map(|(connection, peer, unavailable, unsupported)| { let quota = Self::project_peer( &connection, peer.as_deref(), + unavailable, self.last_error(&connection) .or(unsupported.then_some(UpstreamInfoFailure::Unsupported)), now, @@ -137,6 +140,8 @@ impl PeerInfoService { summary, unsupported: observation .is_some_and(|value| snapshot::is_unsupported(&value.quota, connection)), + unavailable: observation + .and_then(|value| snapshot::unavailable_reason(&value.quota, connection)), }; let previous = self .observations diff --git a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/projection.rs b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/projection.rs index 3eb1f04a7..5ec6076ef 100644 --- a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/projection.rs +++ b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/projection.rs @@ -20,6 +20,7 @@ impl PeerInfoService { pub(super) fn project_peer( connection: &UpstreamConnection, peer: Option<&PeerLink>, + unavailable: Option, last_error: Option, now: SystemTime, ) -> ProviderQuota { @@ -30,11 +31,13 @@ impl PeerInfoService { !snapshot::component_fresh(Some(&peer.observed_at), Some(&peer.expires_at), now) }); let info = ProviderUpstreamInfo { - discovered: peer.is_some(), + discovered: peer.is_some() || unavailable.is_some(), observed_at, stale, connection_revision: connection.revision.clone(), - last_error: last_error.map(|error| error.as_str().to_owned()), + last_error: last_error + .or(unavailable) + .map(|error| error.as_str().to_owned()), pool: peer .as_ref() .and_then(|peer| peer.pool.as_ref()) diff --git a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/refresh.rs b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/refresh.rs index becad8fe7..d382d8b2a 100644 --- a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/refresh.rs +++ b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/refresh.rs @@ -135,21 +135,30 @@ impl PeerInfoService { } .max(retry_after.unwrap_or_default()); state.next_attempt = Instant::now() + delay; - state.manual_after = Instant::now() - + (if failure.is_some_and(|error| error != UpstreamInfoFailure::Unsupported) { - delay - } else { - Duration::from_secs(15) - }) - .max(retry_after.unwrap_or_default()); + // 管理员修复授权后可主动复核，不继承自动重试的长退避，仍遵守上游限流 + state.manual_after = + Instant::now() + Duration::from_secs(15).max(retry_after.unwrap_or_default()); state.last_error = failure; let document = match result { Ok(document) => Some(document), - // 空观测清除已撤销内容，并用共有观察时间阻止更早请求重新写回 - Err(UpstreamInfoFailure::Unauthorized) => { - Some(OpaqueProviderData::new(Default::default())) + Err(error @ (UpstreamInfoFailure::Unauthorized | UpstreamInfoFailure::Unsupported)) => { + let discovered = self + .read_summary(id) + .await + .map_err(store_error)? + .and_then(|quota| quota.upstream) + .is_some_and(|info| { + info.connection_revision == connection.revision && info.discovered + }); + // 首次探测失败不能证明对方是互联节点；已确认的连接只留下无共享内容的识别记录 + Some(if discovered { + snapshot::unavailable(&connection, error) + } else if error == UpstreamInfoFailure::Unsupported { + snapshot::unsupported(&connection) + } else { + OpaqueProviderData::new(Default::default()) + }) } - Err(UpstreamInfoFailure::Unsupported) => Some(snapshot::unsupported(&connection)), Err(_) => None, }; if let Some(document) = document { 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 a81805893..91722c281 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 @@ -35,6 +35,7 @@ pub(super) struct CachedObservation { pub scheduling: Option>, pub summary: Option>, pub unsupported: bool, + pub unavailable: Option, } pub(crate) struct PeerInfoService { @@ -175,12 +176,16 @@ impl PeerInfoService { let unsupported = observation .as_ref() .is_some_and(|value| snapshot::is_unsupported(&value.quota, &connection)); + let unavailable = observation + .as_ref() + .and_then(|value| snapshot::unavailable_reason(&value.quota, &connection)); if self.current_connection(id).await?.as_ref() != Some(&connection) { return Ok(None); } Ok(Some(Self::project_peer( &connection, peer.as_ref(), + unavailable, self.last_error(&connection) .or(unsupported.then_some(UpstreamInfoFailure::Unsupported)), SystemTime::now(), diff --git a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/snapshot.rs b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/snapshot.rs index 2b192d2c0..0f77155ae 100644 --- a/backend/crates/gateway-admin/src/use_case/peer_link/upstream/snapshot.rs +++ b/backend/crates/gateway-admin/src/use_case/peer_link/upstream/snapshot.rs @@ -30,7 +30,7 @@ pub(super) fn unsupported(connection: &UpstreamConnection) -> OpaqueProviderData pub(super) fn is_unsupported(data: &OpaqueProviderData, connection: &UpstreamConnection) -> bool { let fields = data.expose_to_provider(); - fields.get("object").and_then(serde_json::Value::as_str) == Some("cpr.peer.unsupported") + (fields.get("object").and_then(serde_json::Value::as_str) == Some("cpr.peer.unsupported") && fields .get("schema_version") .and_then(serde_json::Value::as_u64) @@ -38,7 +38,42 @@ pub(super) fn is_unsupported(data: &OpaqueProviderData, connection: &UpstreamCon && fields .get("connection_revision") .and_then(serde_json::Value::as_str) - == Some(connection.revision.as_str()) + == Some(connection.revision.as_str())) + || unavailable_reason(data, connection) == Some(UpstreamInfoFailure::Unsupported) +} + +// 失权后只保留同一连接已确认互联的事实，不保留已撤销的池、预算与模型数据 +pub(super) fn unavailable( + connection: &UpstreamConnection, + reason: UpstreamInfoFailure, +) -> OpaqueProviderData { + OpaqueProviderData::new(serde_json::Map::from_iter([ + ("object".into(), "cpr.peer.unavailable".into()), + ("schema_version".into(), 1.into()), + ( + "connection_revision".into(), + connection.revision.clone().into(), + ), + ("reason".into(), reason.as_str().into()), + ])) +} + +pub(super) fn unavailable_reason( + data: &OpaqueProviderData, + connection: &UpstreamConnection, +) -> Option { + let fields = data.expose_to_provider(); + if fields.get("object")?.as_str()? != "cpr.peer.unavailable" + || fields.get("schema_version")?.as_u64()? != 1 + || fields.get("connection_revision")?.as_str()? != connection.revision + { + return None; + } + match fields.get("reason")?.as_str()? { + "unauthorized" => Some(UpstreamInfoFailure::Unauthorized), + "unsupported" => Some(UpstreamInfoFailure::Unsupported), + _ => None, + } } pub(super) fn timestamp(value: &str) -> Option { diff --git a/backend/crates/gateway-admin/tests/use_case/peer_link/upstream/refresh.rs b/backend/crates/gateway-admin/tests/use_case/peer_link/upstream/refresh.rs index feee7630f..99c1ef5f9 100644 --- a/backend/crates/gateway-admin/tests/use_case/peer_link/upstream/refresh.rs +++ b/backend/crates/gateway-admin/tests/use_case/peer_link/upstream/refresh.rs @@ -175,6 +175,174 @@ async fn uncertain_discovery_failures_retry_without_becoming_unsupported() { } } +#[tokio::test(start_paused = true)] +async fn known_peer_failures_keep_identity_and_only_revocation_discards_shared_data() { + for failure in [ + UpstreamInfoFailure::Unauthorized, + UpstreamInfoFailure::Unsupported, + UpstreamInfoFailure::Unavailable, + UpstreamInfoFailure::RateLimited, + UpstreamInfoFailure::InvalidResponse, + ] { + let (bundle, accounts, transport, id) = fixture(failure); + *transport.failure.lock().unwrap() = None; + let before = bundle.service().refresh_quota(&id).await.unwrap(); + assert!(!before.windows.is_empty()); + tokio::time::advance(Duration::from_secs(15)).await; + *transport.failure.lock().unwrap() = Some(failure.into()); + let result = bundle.service().refresh_quota(&id).await; + assert_eq!(result.is_ok(), failure == UpstreamInfoFailure::Unsupported); + let clears_data = matches!( + failure, + UpstreamInfoFailure::Unauthorized | UpstreamInfoFailure::Unsupported + ); + let quota = bundle.service().read_quota(&id).await.unwrap().unwrap(); + let info = quota.upstream.as_ref().unwrap(); + assert!(info.discovered); + assert_eq!(info.last_error.as_deref(), Some(failure.as_str())); + assert_eq!(quota.windows.is_empty(), clears_data); + assert_eq!(info.observed_at.is_none(), clears_data); + assert_eq!( + bundle + .service() + .read_summary(&id) + .await + .unwrap() + .unwrap() + .upstream, + quota.upstream + ); + let account = accounts.accounts.lock().unwrap()[0].clone(); + assert_eq!( + bundle + .scheduling_source() + .signal(&account, "model", SystemTime::now()) + .is_none(), + clears_data + ); + + drop(bundle); + let restarted = initialize(accounts, transport); + let restored = restarted.service().read_quota(&id).await.unwrap().unwrap(); + assert!(restored.upstream.as_ref().unwrap().discovered); + assert_eq!(restored.windows.is_empty(), clears_data); + if clears_data { + assert_eq!( + restored.upstream.as_ref().unwrap().last_error.as_deref(), + Some(failure.as_str()) + ); + } + assert_eq!( + restarted + .service() + .read_summary(&id) + .await + .unwrap() + .unwrap() + .upstream, + restored.upstream + ); + } +} + +#[tokio::test(start_paused = true)] +async fn known_peer_identity_is_scoped_to_the_connection_after_revocation() { + for failure in [ + UpstreamInfoFailure::Unauthorized, + UpstreamInfoFailure::Unsupported, + ] { + let (bundle, accounts, transport, id) = fixture(failure); + *transport.failure.lock().unwrap() = None; + bundle.service().refresh_quota(&id).await.unwrap(); + tokio::time::advance(Duration::from_secs(15)).await; + *transport.failure.lock().unwrap() = Some(failure.into()); + let _result = bundle.service().refresh_quota(&id).await; + let initial = accounts.accounts.lock().unwrap()[0].clone(); + let edited = initial.clone().with_notes(Some("保留连接身份".into())); + *accounts.accounts.lock().unwrap() = vec![edited.clone()]; + bundle.service().account_changed(&id).await.unwrap(); + assert!( + bundle + .service() + .read_summary(&id) + .await + .unwrap() + .unwrap() + .upstream + .unwrap() + .discovered + ); + + // 地址和密钥修改推进凭据版本，出口变化则改变同一版本的连接身份 + for replacement in [ + edited.with_outbound_proxy(Some( + OutboundProxy::parse("http://other.invalid:8080").unwrap(), + )), + ProviderAccount::new( + id.clone(), + initial.provider().clone(), + "替换连接".into(), + None, + "peer_test".into(), + CredentialRevision::new(2).unwrap(), + None, + ) + .with_account_facts( + true, + CredentialState::Ready, + QuotaState::unknown(), + None, + None, + ), + ] { + *accounts.accounts.lock().unwrap() = vec![replacement]; + bundle.service().account_changed(&id).await.unwrap(); + let detail = bundle.service().read_quota(&id).await.unwrap().unwrap(); + assert!(!detail.upstream.as_ref().unwrap().discovered); + assert!(detail.upstream.as_ref().unwrap().last_error.is_none()); + assert_eq!( + bundle + .service() + .read_summary(&id) + .await + .unwrap() + .unwrap() + .upstream, + detail.upstream + ); + } + } +} + +#[tokio::test(start_paused = true)] +async fn manual_rediscovery_does_not_wait_for_authorization_backoff_but_respects_retry_after() { + for retry_after in [None, Some(Duration::from_secs(90))] { + let (mut bundle, _, transport, id) = fixture(UpstreamInfoFailure::Unauthorized); + *transport.failure.lock().unwrap() = None; + bundle.service().refresh_quota(&id).await.unwrap(); + tokio::time::advance(Duration::from_secs(15)).await; + *transport.failure.lock().unwrap() = Some(PeerFetchFailure { + kind: UpstreamInfoFailure::Unauthorized, + retry_after, + }); + assert!(bundle.service().refresh_quota(&id).await.is_err()); + let worker = worker(&mut bundle); + *transport.failure.lock().unwrap() = None; + let delay = retry_after.unwrap_or(Duration::from_secs(15)); + tokio::time::advance(delay - Duration::from_secs(1)).await; + assert!(bundle.service().refresh_quota(&id).await.is_err()); + assert_eq!(transport.requests.load(Ordering::SeqCst), 2); + tokio::time::advance(Duration::from_secs(1)).await; + cycle(&worker).await; + assert_eq!(transport.requests.load(Ordering::SeqCst), 2); + let recovered = bundle.service().refresh_quota(&id).await.unwrap(); + assert_eq!(transport.requests.load(Ordering::SeqCst), 3); + assert!(recovered.upstream.as_ref().unwrap().discovered); + assert!(recovered.upstream.as_ref().unwrap().last_error.is_none()); + assert!(!recovered.windows.is_empty()); + } +} + fn fixture( failure: UpstreamInfoFailure, ) -> ( diff --git a/docs/api.md b/docs/api.md index c7ba2e0bc..284f98b4f 100644 --- a/docs/api.md +++ b/docs/api.md @@ -1147,11 +1147,11 @@ OpenAI 主动额度刷新和正常响应携带的明确套餐会同步到账号 | 字段 | 含义 | | --- | --- | -| `discovered` | 当前连接存在有效解析的互联快照；首次识别前或授权撤销后为 `false` | +| `discovered` | 当前连接曾成功识别为互联；首次识别前为 `false`，识别后的同步失败或授权撤销不清除此身份，也不代表当前仍有共享授权或可用快照 | | `observedAt`、`observedAtDisplay`、`observedAtRelative` | 上游观测时间及服务端格式化结果，不代表本次页面查询时间 | | `stale` | 快照是否已过期；过期值仅供查看 | | `connectionRevision` | 当前连接身份的不可解释标识，用于隔离地址、密钥或出口变化前的展示结果 | -| `lastError` | 当前实例最近同步失败分类：`unauthorized`、`unsupported`、`rate_limited`、`invalid_response`、`unavailable`；已保存的不支持结论在重启后仍返回 `unsupported`，同步成功时为 `null` | +| `lastError` | 当前实例最近同步失败分类：`unauthorized`、`unsupported`、`rate_limited`、`invalid_response`、`unavailable`；已保存的不支持结论及已确认互联的失权记录重启后仍返回对应分类，同步成功时为 `null` | | `pool` | 上游授权公开的账号数量、状态数量及套餐数量，未公开时为 `null` | 互联额度复用 `quota.windows`。上游 Key 账本按实际提供的周期展示，不补造缺失窗口；不限额也保留已用金额。 @@ -1164,11 +1164,16 @@ Key 消费窗口的 `localUsage` 按各自 `[resetAt - windowSeconds, resetAt)` 互联窗口返回独立的 `observedAtDisplay` 和 `stale` 手动刷新复用 `POST /api/admin/accounts/quota/refresh`，请求 `{ accountId }`，与原生额度刷新一样等待查询结束并返回账号视图。 -手动刷新允许重新探测已确认不支持的连接；不支持协议作为正常识别结果返回账号视图，`discovered: false`、`lastError: unsupported`，不作为后台刷新失败报警。 -其他查询失败通过现有管理错误返回；并发刷新合并，连续刷新遵守最小间隔和上游重试时间。 +手动刷新允许重新探测已确认不支持的连接；不支持协议作为正常识别结果返回账号视图和 `lastError: unsupported`，不作为后台刷新失败报警。 +其他查询失败通过现有管理错误返回；失败后可回读账号取得更新后的身份与观测。 +并发刷新合并，连续手动刷新至少间隔 15 秒，并遵守上游 `Retry-After`，不继承自动重试的长退避。 自动刷新由后台任务执行，刷新节流与临时失败提示属于当前实例的运行期状态，重启后重新建立。 -网络错误、限流或无效响应保留旧观测；401/403 清除已撤销内容。这些失败不写入不支持结论，待识别及已确认支持的连接仍退避重试。 -账号凭据或出口变更使旧观测失效。 +网络错误、限流或无效响应保留旧观测；401/403 清除已撤销内容，但已确认的互联连接保留识别身份和失败分类。 +已确认互联后变为不支持协议时，也只保留识别身份和失败分类，不保留池、预算或模型数据。 +无观测内容时 `observedAt` 与 `pool` 为 `null`，互联额度窗口为空，识别身份不作为调度或额度恢复证据。 +首次探测失败不能证明上游支持互联，管理端继续按普通 API Key 展示，不显示互联告警；已确认的连接显示同步失败原因，可通过现有刷新额度入口重试。 +除明确不支持协议外，待识别及已确认支持的连接仍退避重试。 +账号地址、密钥或出口变更使旧观测与识别身份失效，重新识别当前连接。 账号停用后停止同步，重新启用后按当前连接的识别结果恢复后台任务 成功同步后，明确的预算访问结论决定账号是否进入 `quota_exhausted`； diff --git a/docs/architecture.md b/docs/architecture.md index 5abda7dae..45d3a6251 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1035,6 +1035,8 @@ Provider 声明可连接的认证种类，并从本轮凭据中提取连接材 Host 执行有界 HTTP 请求，复用组合根注入的证书策略和账号出口；Store 通过通用账号额度端口保存不透明观测，不解释互联或 Provider 协议。 互联与原生额度共用 `provider_quota_json`、`quota_observed_at`、凭据版本和额度访问事实，每个账号仅保存当前来源的观测。 Admin 将明确不支持互联的结果绑定连接身份保存于同一观测，不伪造额度快照或改变额度访问状态。 +已确认互联的连接失权或变为不支持时，同一存储只保留连接识别记录和失败分类，清除共享内容； +列表与详情复用该记录展示身份并保留额度刷新入口，调度只消费实际快照。首次探测失败不建立互联身份。 OAuth 额度解析属于 Provider，CPR 发现、解析与投影属于 Admin；账号用例按能力分派读写与刷新，不将两种协议合并为同一解析服务。 凭据新建、导入或授权提交后，后台首查复用同一额度分派，不等待周期同步或阻塞保存请求。 互联首查、手动刷新和周期刷新共用服务级并发上限，排队后读取当前凭据；失败或取消释放名额。 @@ -1046,7 +1048,7 @@ OAuth 额度解析属于 Provider，CPR 发现、解析与投影属于 Admin； 共享端按 Core 的预算规则从本地账本生成明确的访问状态，与本地准入共用口径；金额字段只供接入端展示。 接入端 Admin 的独立额度模块解释默认公开的访问状态，展示、调度投影与同步共用，不要求公开账号池额度；未知或过期事实保留已有状态，Provider 不新增 CPR 错误码分类。 通用额度写入按凭据版本和查询开始时间提交快照，并独立保护访问事实的观察时间；网络查询可附带连接校验，在同一事务核对出口与启用状态。 -账号重新授权或出口变更使旧快照失效，正常 token 续期保留额度；授权撤销提交空观测和时间屏障，阻止更早响应重新发布旧内容。 +账号重新授权或出口变更使旧快照及互联识别记录失效，正常 token 续期保留额度；授权撤销提交不含共享内容的观测和时间屏障，阻止更早响应重新发布旧内容。 互联解析只提交有效且明确的预算访问事实，额度同步不改变凭据或独立冷却 Admin 按连接身份及额度观察时间缓存管理摘要与调度投影，解析和投影在锁外完成。 diff --git a/frontend/src/api/modules/accounts/types.ts b/frontend/src/api/modules/accounts/types.ts index 0e596b5d9..486e83d1d 100644 --- a/frontend/src/api/modules/accounts/types.ts +++ b/frontend/src/api/modules/accounts/types.ts @@ -181,6 +181,7 @@ export interface Account { } export interface AccountUpstream { + // 同一连接曾确认支持互联，失权时仍为 true，观测内容则清空 discovered: boolean observedAt: string | null observedAtDisplay: string diff --git a/frontend/src/components/account/AccountPlanBadge.vue b/frontend/src/components/account/AccountPlanBadge.vue index 930552bae..68afe5f88 100644 --- a/frontend/src/components/account/AccountPlanBadge.vue +++ b/frontend/src/components/account/AccountPlanBadge.vue @@ -2,7 +2,6 @@ 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' @@ -65,7 +64,7 @@ const paletteClass = computed(() => {