mirror of
https://github.com/zyycn/codex-proxy-rs.git
synced 2026-11-13 00:07:32 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1294e50b76 | ||
|
|
a0e540642b | ||
|
|
14d6a7de31 |
@@ -310,16 +310,6 @@ pub(crate) fn usage_range(
|
||||
domain::TimeRange::new(start, end).map_err(|_| WireValidationError::new("timeRange"))
|
||||
}
|
||||
|
||||
pub(crate) fn dashboard_range(
|
||||
start: Option<&str>,
|
||||
end: Option<&str>,
|
||||
) -> Result<domain::TimeRange, WireValidationError> {
|
||||
let end = parse_datetime(end)?.unwrap_or_else(Utc::now);
|
||||
let start =
|
||||
parse_datetime(start)?.unwrap_or_else(|| domain::china_day_start(end) - Duration::days(1));
|
||||
domain::TimeRange::new(start, end).map_err(|_| WireValidationError::new("timeRange"))
|
||||
}
|
||||
|
||||
pub(crate) fn dashboard_today_range(
|
||||
start: Option<&str>,
|
||||
end: Option<&str>,
|
||||
|
||||
@@ -38,7 +38,8 @@ where
|
||||
S: AdminSessionState + Send + Sync,
|
||||
{
|
||||
let kind = query.trend_kind().map_err(map_wire_error)?;
|
||||
let range = dashboard_range(query.start_time.as_deref(), query.end_time.as_deref())
|
||||
// 概览默认按中国时区当日统计,与单独趋势接口保持同一口径。
|
||||
let range = dashboard_today_range(query.start_time.as_deref(), query.end_time.as_deref())
|
||||
.map_err(map_wire_error)?;
|
||||
let result = state
|
||||
.admin_services()
|
||||
|
||||
@@ -83,6 +83,7 @@ pub(super) struct AdminTestFixture {
|
||||
pub diagnostics: Arc<Mutex<Vec<DiagnosticObservation>>>,
|
||||
pub ops_errors: Arc<Mutex<Vec<OpsError>>>,
|
||||
pub dashboard_observation: Arc<Mutex<Option<DashboardObservation>>>,
|
||||
pub dashboard_summary_range: Arc<Mutex<Option<TimeRange>>>,
|
||||
}
|
||||
|
||||
impl AdminTestFixture {
|
||||
@@ -100,12 +101,14 @@ impl AdminTestFixture {
|
||||
let diagnostics = Arc::new(Mutex::new(Vec::new()));
|
||||
let ops_errors = Arc::new(Mutex::new(Vec::new()));
|
||||
let dashboard_observation = Arc::new(Mutex::new(None));
|
||||
let dashboard_summary_range = Arc::new(Mutex::new(None));
|
||||
let unused = Arc::new(UnusedStore {
|
||||
usage_records: Arc::clone(&usage_records),
|
||||
usage_detail: Arc::clone(&usage_detail),
|
||||
diagnostics: Arc::clone(&diagnostics),
|
||||
ops_errors: Arc::clone(&ops_errors),
|
||||
dashboard_observation: Arc::clone(&dashboard_observation),
|
||||
dashboard_summary_range: Arc::clone(&dashboard_summary_range),
|
||||
});
|
||||
let stores = AdminStorePorts::new(
|
||||
unused.clone(),
|
||||
@@ -142,6 +145,7 @@ impl AdminTestFixture {
|
||||
diagnostics,
|
||||
ops_errors,
|
||||
dashboard_observation,
|
||||
dashboard_summary_range,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -413,6 +417,7 @@ struct UnusedStore {
|
||||
diagnostics: Arc<Mutex<Vec<DiagnosticObservation>>>,
|
||||
ops_errors: Arc<Mutex<Vec<OpsError>>>,
|
||||
dashboard_observation: Arc<Mutex<Option<DashboardObservation>>>,
|
||||
dashboard_summary_range: Arc<Mutex<Option<TimeRange>>>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
@@ -523,7 +528,11 @@ impl AccountStore for UnusedStore {
|
||||
|
||||
#[async_trait]
|
||||
impl ObservabilityStore for UnusedStore {
|
||||
async fn dashboard_summary(&self, _: TimeRange) -> AdminStoreResult<DashboardObservation> {
|
||||
async fn dashboard_summary(&self, range: TimeRange) -> AdminStoreResult<DashboardObservation> {
|
||||
*self
|
||||
.dashboard_summary_range
|
||||
.lock()
|
||||
.expect("dashboard summary range") = Some(range);
|
||||
self.dashboard_observation
|
||||
.lock()
|
||||
.expect("dashboard observation")
|
||||
|
||||
@@ -270,6 +270,61 @@ async fn dashboard_summary_should_include_quota_exhaustion_in_unavailable_headli
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn dashboard_summary_should_default_to_current_china_day() {
|
||||
use axum::{
|
||||
body::Body,
|
||||
http::{Request, StatusCode, header},
|
||||
};
|
||||
use chrono::{Duration, Utc};
|
||||
use gateway_admin::model::observability::{
|
||||
AccountPoolMetrics, AttemptMetrics, DashboardObservation, RequestMetrics, TimeRange,
|
||||
china_day_start,
|
||||
};
|
||||
use gateway_api::admin::observability;
|
||||
use tower::ServiceExt as _;
|
||||
|
||||
use crate::admin::{AdminTestFixture, AdminTestState};
|
||||
|
||||
let fixture = AdminTestFixture::new().await;
|
||||
fixture.auth.insert_session("valid-session");
|
||||
let now = Utc::now();
|
||||
*fixture
|
||||
.dashboard_observation
|
||||
.lock()
|
||||
.expect("dashboard observation") = Some(DashboardObservation {
|
||||
range: TimeRange::new(now - Duration::hours(1), now).expect("dashboard range"),
|
||||
requests: RequestMetrics::default(),
|
||||
attempts: AttemptMetrics::default(),
|
||||
provider_accounts: AccountPoolMetrics::default(),
|
||||
trend: Vec::new(),
|
||||
account_usage: Vec::new(),
|
||||
recent_requests: Vec::new(),
|
||||
});
|
||||
|
||||
let response = observability::router::<AdminTestState>()
|
||||
.with_state(fixture.state())
|
||||
.oneshot(
|
||||
Request::builder()
|
||||
.uri("/api/admin/dashboard/summary")
|
||||
.header(header::COOKIE, "cpr_admin_session=valid-session")
|
||||
.header("x-request-id", "req_dashboard_today_range")
|
||||
.body(Body::empty())
|
||||
.expect("dashboard request"),
|
||||
)
|
||||
.await
|
||||
.expect("dashboard response");
|
||||
let range = fixture
|
||||
.dashboard_summary_range
|
||||
.lock()
|
||||
.expect("dashboard summary range")
|
||||
.take()
|
||||
.expect("recorded dashboard summary range");
|
||||
|
||||
assert_eq!(response.status(), StatusCode::OK);
|
||||
assert_eq!(range.start, china_day_start(range.end));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn usage_detail_should_keep_attempt_snapshot_contract() {
|
||||
use axum::{
|
||||
|
||||
@@ -215,6 +215,20 @@ impl AccountAvailability {
|
||||
}
|
||||
}
|
||||
|
||||
/// 返回账号进入不可恢复错误态、但 Provider 未给出更具体上游原因时的稳定兜底。
|
||||
///
|
||||
/// Provider 应优先持久化已脱敏的上游 message 或原因码;该值只避免错误态沿用
|
||||
/// 上一次无关的错误信息。
|
||||
#[must_use]
|
||||
pub const fn fallback_error_reason(self) -> Option<&'static str> {
|
||||
match self {
|
||||
Self::Expired => Some("credential_expired"),
|
||||
Self::Banned => Some("account_banned"),
|
||||
Self::Invalid => Some("credential_invalid"),
|
||||
Self::Unknown | Self::Ready | Self::QuotaExhausted => None,
|
||||
}
|
||||
}
|
||||
|
||||
/// 解析数据库稳定值。
|
||||
#[must_use]
|
||||
pub fn parse(value: &str) -> Option<Self> {
|
||||
|
||||
@@ -36,7 +36,7 @@ use crate::transport::{CodexBackendClient, CodexClientError, CodexRequestContext
|
||||
|
||||
use super::agent_identity::{CodexAgentIdentityTaskService, PreparedCodexRuntimeCredential};
|
||||
use super::repository::{CodexCredentialRepository, CredentialRepositoryError};
|
||||
use recovery::{quota_state_transition, quota_success_state};
|
||||
use recovery::{quota_state_reason, quota_state_transition, quota_success_state};
|
||||
use snapshot::{
|
||||
parse_account_quota_snapshot, quota_projection_ttl, quota_snapshot_from_observation,
|
||||
scheduling_signals_from_snapshot,
|
||||
@@ -548,21 +548,23 @@ impl CodexCredentialQuotaService {
|
||||
// 402:真额度耗尽,写入 QuotaExhausted(worker 按 reset_at 自动恢复)。
|
||||
Some(AccountAvailability::QuotaExhausted) => {
|
||||
summary.exhausted += 1;
|
||||
let _ = self
|
||||
.repository
|
||||
.apply_state(
|
||||
&account,
|
||||
AccountAvailability::QuotaExhausted,
|
||||
observed_at,
|
||||
)
|
||||
.await;
|
||||
self.persist_upstream_state(
|
||||
&account,
|
||||
AccountAvailability::QuotaExhausted,
|
||||
&error,
|
||||
observed_at,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
Some(AccountAvailability::Banned) => {
|
||||
summary.banned += 1;
|
||||
let _ = self
|
||||
.repository
|
||||
.apply_state(&account, AccountAvailability::Banned, observed_at)
|
||||
.await;
|
||||
self.persist_upstream_state(
|
||||
&account,
|
||||
AccountAvailability::Banned,
|
||||
&error,
|
||||
observed_at,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
// 429/503 等临时拒绝:不写 availability(限流不改变账号可用性),仅记失败待重试。
|
||||
None | Some(_) => {
|
||||
@@ -913,9 +915,7 @@ impl CodexCredentialQuotaService {
|
||||
Err(CodexQuotaFetchError::Upstream { account, error }) => {
|
||||
// 402/deactivated → 写终态;429/503 等临时拒绝不写 availability(限流不改变账号可用性)。
|
||||
if let Some(availability) = quota_state_transition(&error) {
|
||||
let _ = self
|
||||
.repository
|
||||
.apply_state(&account, availability, observed_at)
|
||||
self.persist_upstream_state(&account, availability, &error, observed_at)
|
||||
.await;
|
||||
}
|
||||
return Err(CodexCredentialQuotaError::Upstream {
|
||||
@@ -996,6 +996,29 @@ impl CodexCredentialQuotaService {
|
||||
Ok(snapshot)
|
||||
}
|
||||
|
||||
async fn persist_upstream_state(
|
||||
&self,
|
||||
account: &ProviderAccount,
|
||||
availability: AccountAvailability,
|
||||
upstream_error: &CodexClientError,
|
||||
observed_at: SystemTime,
|
||||
) {
|
||||
let reason = quota_state_reason(upstream_error).map(str::to_owned);
|
||||
if let Err(error) = self
|
||||
.repository
|
||||
.apply_state_with_message(account, availability, observed_at, reason.clone())
|
||||
.await
|
||||
{
|
||||
tracing::warn!(
|
||||
account_id = %account.id(),
|
||||
?availability,
|
||||
reason = reason.as_deref().unwrap_or_default(),
|
||||
error = %error,
|
||||
"OpenAI quota upstream state write failed"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
async fn fetch_usage_with_recovery(
|
||||
&self,
|
||||
client: &CodexBackendClient,
|
||||
|
||||
@@ -50,6 +50,18 @@ pub(crate) fn quota_state_transition(error: &CodexClientError) -> Option<Account
|
||||
}
|
||||
}
|
||||
|
||||
/// 配额接口将账号变为非 ready 状态时可安全持久化的上游原因码。
|
||||
pub(crate) fn quota_state_reason(error: &CodexClientError) -> Option<&'static str> {
|
||||
match quota_state_transition(error)? {
|
||||
AccountAvailability::Banned => Some("deactivated_workspace"),
|
||||
AccountAvailability::QuotaExhausted => Some("payment_required"),
|
||||
AccountAvailability::Unknown
|
||||
| AccountAvailability::Ready
|
||||
| AccountAvailability::Expired
|
||||
| AccountAvailability::Invalid => None,
|
||||
}
|
||||
}
|
||||
|
||||
fn is_deactivated_workspace(body: &str) -> bool {
|
||||
serde_json::from_str::<Value>(body)
|
||||
.ok()
|
||||
|
||||
@@ -11,6 +11,7 @@ use gateway_core::provider_ports::{
|
||||
ProviderCredentialStatePort, ProviderLeaseAcquisition, ProviderLeasePort, ProviderLeaseRequest,
|
||||
ProviderRefreshCapacityRequest, ProviderRefreshLeaseRequest, ProviderRefreshPolicy,
|
||||
ProviderRuntimePolicyPort, ProviderStoreError, provider_refresh_backoff_at,
|
||||
provider_refresh_retry_at,
|
||||
};
|
||||
use secrecy::{ExposeSecret, SecretString};
|
||||
use thiserror::Error;
|
||||
@@ -27,13 +28,88 @@ const PROVIDER_NAME: &str = "openai";
|
||||
const MAX_REFRESH_BATCH: u32 = 1_000;
|
||||
/// 连续失败计数窗口;每次瞬态失败刷新该 TTL,静默满窗后计数过期归零。
|
||||
const REFRESH_BACKOFF_WINDOW: Duration = Duration::from_secs(30 * 60);
|
||||
/// 窗口内先执行五次指数退避,再进入固定恢复周期。
|
||||
const REFRESH_BACKOFF_MAX_ATTEMPTS: u32 = 5;
|
||||
/// 耗尽指数退避后的 OAuth 恢复周期。
|
||||
const REFRESH_RECOVERY_DELAY: Duration = Duration::from_secs(10 * 60);
|
||||
/// 过期 AT 仍允许 RT 恢复的最长窗口。
|
||||
const REFRESH_RECOVERY_WINDOW: Duration = Duration::from_secs(24 * 60 * 60);
|
||||
|
||||
pub(crate) fn refresh_recovery_deadline(
|
||||
access_token_expires_at: Option<SystemTime>,
|
||||
) -> Option<SystemTime> {
|
||||
access_token_expires_at.and_then(|expires_at| expires_at.checked_add(REFRESH_RECOVERY_WINDOW))
|
||||
}
|
||||
|
||||
fn refresh_recovery_window_exhausted(
|
||||
access_token_expires_at: Option<SystemTime>,
|
||||
now: SystemTime,
|
||||
) -> bool {
|
||||
refresh_recovery_deadline(access_token_expires_at).is_some_and(|deadline| deadline <= now)
|
||||
}
|
||||
|
||||
fn bounded_refresh_retry_at(
|
||||
retry_at: SystemTime,
|
||||
access_token_expires_at: Option<SystemTime>,
|
||||
observed_at: SystemTime,
|
||||
) -> SystemTime {
|
||||
let mut bounded = retry_at;
|
||||
if let Some(expires_at) = access_token_expires_at
|
||||
&& expires_at > observed_at
|
||||
&& bounded > expires_at
|
||||
{
|
||||
bounded = expires_at;
|
||||
}
|
||||
if let Some(deadline) = refresh_recovery_deadline(access_token_expires_at)
|
||||
&& bounded > deadline
|
||||
{
|
||||
bounded = deadline;
|
||||
}
|
||||
bounded
|
||||
}
|
||||
|
||||
fn oauth_refresh_retry_at(
|
||||
account_id: &ProviderAccountId,
|
||||
access_token_expires_at: Option<SystemTime>,
|
||||
observed_at: SystemTime,
|
||||
attempt: u32,
|
||||
reason: &'static str,
|
||||
) -> Result<SystemTime, ProviderStoreError> {
|
||||
let retry_at = if attempt <= REFRESH_BACKOFF_MAX_ATTEMPTS {
|
||||
provider_refresh_backoff_at(account_id, observed_at, attempt, reason)?
|
||||
} else {
|
||||
provider_refresh_retry_at(account_id, observed_at, REFRESH_RECOVERY_DELAY, reason)?
|
||||
};
|
||||
Ok(bounded_refresh_retry_at(
|
||||
retry_at,
|
||||
access_token_expires_at,
|
||||
observed_at,
|
||||
))
|
||||
}
|
||||
|
||||
fn log_refresh_deferred(
|
||||
account_id: &ProviderAccountId,
|
||||
access_token_expires_at: Option<SystemTime>,
|
||||
attempt: u32,
|
||||
reason: &'static str,
|
||||
retry_at: SystemTime,
|
||||
) {
|
||||
tracing::warn!(
|
||||
account_id = %account_id,
|
||||
attempt,
|
||||
reason,
|
||||
retry_at = %DateTime::<Utc>::from(retry_at),
|
||||
access_token_expires_at = ?access_token_expires_at.map(DateTime::<Utc>::from),
|
||||
recovery_deadline = ?refresh_recovery_deadline(access_token_expires_at)
|
||||
.map(DateTime::<Utc>::from),
|
||||
"OpenAI OAuth refresh deferred"
|
||||
);
|
||||
}
|
||||
|
||||
fn refresh_due_at(account: &ProviderAccount, now: SystemTime) -> Option<SystemTime> {
|
||||
if let Some(expires_at) = account.access_token_expires_at()
|
||||
&& expires_at <= now
|
||||
{
|
||||
// AT 已过期时不能继续等待持久退避;后续额度/请求探测都必须先用 RT 续出可用 AT。
|
||||
return Some(expires_at);
|
||||
if refresh_recovery_window_exhausted(account.access_token_expires_at(), now) {
|
||||
// 即使旧 next_refresh_at 异常地落在 deadline 之后,也必须进入终态化路径。
|
||||
return refresh_recovery_deadline(account.access_token_expires_at());
|
||||
}
|
||||
account.next_refresh_at().filter(|next| *next <= now)
|
||||
}
|
||||
@@ -145,7 +221,14 @@ impl CodexCredentialRefreshService {
|
||||
for (account_id, result) in refreshed {
|
||||
match result {
|
||||
Ok(outcome) => outcomes.push(outcome),
|
||||
Err(_) => outcomes.push(CodexCredentialRefreshOutcome::Failed { account_id }),
|
||||
Err(error) => {
|
||||
tracing::warn!(
|
||||
account_id = %account_id,
|
||||
error = %error,
|
||||
"OpenAI OAuth refresh attempt failed"
|
||||
);
|
||||
outcomes.push(CodexCredentialRefreshOutcome::Failed { account_id });
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(outcomes)
|
||||
@@ -157,6 +240,19 @@ impl CodexCredentialRefreshService {
|
||||
policy: ProviderRefreshPolicy,
|
||||
) -> Result<CodexCredentialRefreshOutcome, CodexCredentialRefreshError> {
|
||||
let account_id = due.account.id().to_string();
|
||||
if refresh_recovery_window_exhausted(
|
||||
due.account.access_token_expires_at(),
|
||||
SystemTime::now(),
|
||||
) {
|
||||
return self
|
||||
.persist_terminal(
|
||||
&due.account,
|
||||
AccountAvailability::Expired,
|
||||
"refresh_recovery_window_exhausted",
|
||||
CodexCredentialRefreshOutcome::Invalidated { account_id },
|
||||
)
|
||||
.await;
|
||||
}
|
||||
let capacity = self
|
||||
.leases
|
||||
.try_acquire(ProviderLeaseRequest::RefreshCapacity(
|
||||
@@ -192,6 +288,7 @@ impl CodexCredentialRefreshService {
|
||||
self.persist_terminal(
|
||||
&due.account,
|
||||
AccountAvailability::Expired,
|
||||
"refresh_invalid_grant",
|
||||
CodexCredentialRefreshOutcome::Invalidated { account_id },
|
||||
)
|
||||
.await
|
||||
@@ -200,6 +297,7 @@ impl CodexCredentialRefreshService {
|
||||
self.persist_terminal(
|
||||
&due.account,
|
||||
AccountAvailability::Banned,
|
||||
"account_banned",
|
||||
CodexCredentialRefreshOutcome::Banned { account_id },
|
||||
)
|
||||
.await
|
||||
@@ -328,6 +426,7 @@ impl CodexCredentialRefreshService {
|
||||
.persist_terminal(
|
||||
&due.account,
|
||||
AccountAvailability::Invalid,
|
||||
"identity_rejected",
|
||||
CodexCredentialRefreshOutcome::Invalidated {
|
||||
account_id: due.account.id().to_string(),
|
||||
},
|
||||
@@ -359,8 +458,11 @@ impl CodexCredentialRefreshService {
|
||||
.record_refresh_backoff(due.account.id(), REFRESH_BACKOFF_WINDOW)
|
||||
.await
|
||||
.unwrap_or(1);
|
||||
let retry_at = provider_refresh_backoff_at(
|
||||
let access_token_expires_at =
|
||||
Some(SystemTime::from(signed.access_token_expires_at()));
|
||||
let retry_at = oauth_refresh_retry_at(
|
||||
due.account.id(),
|
||||
access_token_expires_at,
|
||||
SystemTime::now(),
|
||||
attempt,
|
||||
"identity-completion",
|
||||
@@ -371,6 +473,13 @@ impl CodexCredentialRefreshService {
|
||||
.await
|
||||
{
|
||||
Ok(_) => {
|
||||
log_refresh_deferred(
|
||||
due.account.id(),
|
||||
access_token_expires_at,
|
||||
attempt,
|
||||
"identity-completion",
|
||||
retry_at,
|
||||
);
|
||||
return Ok(CodexCredentialRefreshOutcome::Transient {
|
||||
account_id: due.account.id().to_string(),
|
||||
});
|
||||
@@ -384,6 +493,7 @@ impl CodexCredentialRefreshService {
|
||||
}
|
||||
}
|
||||
};
|
||||
let access_token_expires_at = profile.access_token_expires_at.map(SystemTime::from);
|
||||
let next_refresh_at = policy
|
||||
.next_attempt_at(
|
||||
due.account.id(),
|
||||
@@ -412,6 +522,13 @@ impl CodexCredentialRefreshService {
|
||||
.credential_state
|
||||
.clear_refresh_backoff(due.account.id())
|
||||
.await;
|
||||
tracing::info!(
|
||||
account_id = %due.account.id(),
|
||||
credential_revision = revision.get(),
|
||||
access_token_expires_at = ?access_token_expires_at.map(DateTime::<Utc>::from),
|
||||
next_refresh_at = %next_refresh_at,
|
||||
"OpenAI OAuth refresh succeeded"
|
||||
);
|
||||
Ok(CodexCredentialRefreshOutcome::Refreshed {
|
||||
account_id: due.account.id().to_string(),
|
||||
credential_revision: revision.get(),
|
||||
@@ -430,6 +547,7 @@ impl CodexCredentialRefreshService {
|
||||
&self,
|
||||
account: &ProviderAccount,
|
||||
availability: AccountAvailability,
|
||||
reason: &'static str,
|
||||
outcome: CodexCredentialRefreshOutcome,
|
||||
) -> Result<CodexCredentialRefreshOutcome, CodexCredentialRefreshError> {
|
||||
match self.repository.load_runtime_credential(account).await {
|
||||
@@ -443,10 +561,27 @@ impl CodexCredentialRefreshService {
|
||||
}
|
||||
match self
|
||||
.repository
|
||||
.apply_state(account, availability, SystemTime::now())
|
||||
.apply_state_with_message(
|
||||
account,
|
||||
availability,
|
||||
SystemTime::now(),
|
||||
Some(reason.to_owned()),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(()) => Ok(outcome),
|
||||
Ok(()) => {
|
||||
tracing::warn!(
|
||||
account_id = %account.id(),
|
||||
?availability,
|
||||
reason,
|
||||
access_token_expires_at = ?account.access_token_expires_at()
|
||||
.map(DateTime::<Utc>::from),
|
||||
recovery_deadline = ?refresh_recovery_deadline(account.access_token_expires_at())
|
||||
.map(DateTime::<Utc>::from),
|
||||
"OpenAI OAuth refresh marked account terminal"
|
||||
);
|
||||
Ok(outcome)
|
||||
}
|
||||
Err(CredentialRepositoryError::RevisionConflict) => {
|
||||
Ok(CodexCredentialRefreshOutcome::Stale {
|
||||
account_id: account.id().to_string(),
|
||||
@@ -466,10 +601,24 @@ impl CodexCredentialRefreshService {
|
||||
.record_refresh_backoff(account.id(), REFRESH_BACKOFF_WINDOW)
|
||||
.await
|
||||
.unwrap_or(1);
|
||||
let retry_at =
|
||||
provider_refresh_backoff_at(account.id(), SystemTime::now(), attempt, reason)?;
|
||||
let retry_at = oauth_refresh_retry_at(
|
||||
account.id(),
|
||||
account.access_token_expires_at(),
|
||||
SystemTime::now(),
|
||||
attempt,
|
||||
reason,
|
||||
)?;
|
||||
match self.repository.defer_refresh(account, retry_at).await {
|
||||
Ok(_) => Ok(true),
|
||||
Ok(_) => {
|
||||
log_refresh_deferred(
|
||||
account.id(),
|
||||
account.access_token_expires_at(),
|
||||
attempt,
|
||||
reason,
|
||||
retry_at,
|
||||
);
|
||||
Ok(true)
|
||||
}
|
||||
Err(CredentialRepositoryError::RevisionConflict) => Ok(false),
|
||||
Err(error) => Err(error.into()),
|
||||
}
|
||||
|
||||
@@ -239,7 +239,8 @@ impl CodexCredentialRepository {
|
||||
.await
|
||||
}
|
||||
|
||||
/// 带上游错误描述的状态写入;`message` 为 `Some` 时持久化最近错误,恢复 `Ready` 时清空。
|
||||
/// 带上游错误描述的状态写入;上游 message 优先,终态缺失 message 时写稳定兜底,
|
||||
/// 恢复 `Ready` 时清空。
|
||||
pub async fn apply_state_with_message(
|
||||
&self,
|
||||
account: &ProviderAccount,
|
||||
@@ -250,6 +251,9 @@ impl CodexCredentialRepository {
|
||||
if !account.enabled() {
|
||||
return Ok(());
|
||||
}
|
||||
let message = message
|
||||
.filter(|value| !value.trim().is_empty())
|
||||
.or_else(|| availability.fallback_error_reason().map(str::to_owned));
|
||||
self.store
|
||||
.apply_state_change(AccountStateChange {
|
||||
account_id: account.id().clone(),
|
||||
|
||||
@@ -25,11 +25,12 @@ use super::agent_identity::CodexAgentIdentityTaskService;
|
||||
use super::catalog::CodexCredentialCatalogService;
|
||||
use super::cookie::CodexCookiePolicy;
|
||||
use super::quota::CodexCredentialQuotaService;
|
||||
use super::refresh::refresh_recovery_deadline;
|
||||
use super::repository::{CodexCredentialRepository, CredentialRepositoryError};
|
||||
use super::security::CodexRuntimeAuthentication;
|
||||
use super::types::{
|
||||
CODEX_AUTHENTICATION_KIND_AGENT_IDENTITY, CodexCookie, CodexCookieCaptureOutcome,
|
||||
RuntimeCodexCookie,
|
||||
CODEX_AUTHENTICATION_KIND_AGENT_IDENTITY, CODEX_AUTHENTICATION_KIND_OAUTH, CodexCookie,
|
||||
CodexCookieCaptureOutcome, RuntimeCodexCookie,
|
||||
};
|
||||
|
||||
const CLOUDFLARE_RECOVERY_STALE_AFTER: Duration = Duration::from_secs(60 * 60);
|
||||
@@ -73,6 +74,21 @@ pub enum CodexAccountFailure {
|
||||
CloudflarePathBlocked,
|
||||
}
|
||||
|
||||
impl CodexAccountFailure {
|
||||
/// 上游未提供可展示错误文本时,保留已分类的稳定原因。
|
||||
const fn fallback_error_reason(self) -> Option<&'static str> {
|
||||
match self {
|
||||
Self::CredentialExpired => Some("upstream_credential_expired"),
|
||||
Self::IdentityVerificationRequired => Some("upstream_identity_verification_required"),
|
||||
Self::Banned => Some("upstream_account_banned"),
|
||||
Self::QuotaExhausted => Some("upstream_quota_exhausted"),
|
||||
Self::UsageLimitExhausted { .. } => Some("upstream_usage_limit_exhausted"),
|
||||
Self::CloudflarePathBlocked => Some("upstream_cloudflare_path_blocked"),
|
||||
Self::RateLimited { .. } | Self::CloudflareChallenge { .. } => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
struct RiskRecoveryState {
|
||||
challenge_count: u32,
|
||||
@@ -733,8 +749,29 @@ impl CodexCredentialSelector {
|
||||
message: Option<String>,
|
||||
) -> Result<(), CredentialSelectionError> {
|
||||
let now = SystemTime::now();
|
||||
let message = message
|
||||
.filter(|value| !value.trim().is_empty())
|
||||
.or_else(|| failure.fallback_error_reason().map(str::to_owned));
|
||||
match failure {
|
||||
CodexAccountFailure::CredentialExpired => {
|
||||
if account.authentication_kind() == CODEX_AUTHENTICATION_KIND_OAUTH
|
||||
&& account.has_refresh_token()
|
||||
&& account
|
||||
.access_token_expires_at()
|
||||
.is_some_and(|expires_at| expires_at <= now)
|
||||
&& refresh_recovery_deadline(account.access_token_expires_at())
|
||||
.is_some_and(|deadline| deadline > now)
|
||||
{
|
||||
tracing::info!(
|
||||
account_id = %account.id(),
|
||||
access_token_expires_at = ?account.access_token_expires_at()
|
||||
.map(chrono::DateTime::<chrono::Utc>::from),
|
||||
recovery_deadline = ?refresh_recovery_deadline(account.access_token_expires_at())
|
||||
.map(chrono::DateTime::<chrono::Utc>::from),
|
||||
"OpenAI access token expired; retaining account for bounded OAuth refresh recovery"
|
||||
);
|
||||
return Ok(());
|
||||
}
|
||||
self.apply_state_with_message(account, AccountAvailability::Expired, now, message)
|
||||
.await
|
||||
}
|
||||
|
||||
@@ -2435,11 +2435,10 @@ impl ScheduledTask for OpenAiOAuthRefreshTask {
|
||||
if context.cancellation().is_cancelled() {
|
||||
return Ok(());
|
||||
}
|
||||
let outcomes = self
|
||||
.service
|
||||
.refresh_due()
|
||||
.await
|
||||
.map_err(|_| WorkerTaskError::safe("OpenAI OAuth refresh failed"))?;
|
||||
let outcomes = self.service.refresh_due().await.map_err(|error| {
|
||||
tracing::error!(error = %error, "OpenAI OAuth refresh cycle failed");
|
||||
WorkerTaskError::safe("OpenAI OAuth refresh failed")
|
||||
})?;
|
||||
let mut refreshed = 0_u64;
|
||||
let mut invalidated = 0_u64;
|
||||
let mut banned = 0_u64;
|
||||
@@ -2468,6 +2467,18 @@ impl ScheduledTask for OpenAiOAuthRefreshTask {
|
||||
}
|
||||
}
|
||||
}
|
||||
if !outcomes.is_empty() {
|
||||
tracing::info!(
|
||||
refreshed,
|
||||
invalidated,
|
||||
banned,
|
||||
transient,
|
||||
lease_unavailable,
|
||||
stale,
|
||||
failed,
|
||||
"OpenAI OAuth refresh cycle completed"
|
||||
);
|
||||
}
|
||||
if transient > 0 || failed > 0 {
|
||||
tracing::warn!(
|
||||
refreshed,
|
||||
|
||||
@@ -688,6 +688,43 @@ fn credential_expired_failure_marks_unified_account_expired() {
|
||||
.availability(),
|
||||
AccountAvailability::Expired
|
||||
);
|
||||
assert_eq!(
|
||||
store.last_error_message("acct_primary").as_deref(),
|
||||
Some("upstream_credential_expired")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn credential_expired_failure_keeps_expired_oauth_for_bounded_refresh_recovery() {
|
||||
let store = Arc::new(MemoryAccountStore::default());
|
||||
let mut expired_profile = profile("chatgpt-acct_primary");
|
||||
expired_profile.access_token_expires_at =
|
||||
Some(chrono::Utc::now() - chrono::Duration::minutes(1));
|
||||
block_on(store.seed_oauth_credential(ImportCodexOAuthCredential {
|
||||
account_id: "acct_primary".to_owned(),
|
||||
name: "acct_primary".to_owned(),
|
||||
secret: secret("at-primary"),
|
||||
verified_account: expired_profile,
|
||||
next_refresh_at: Some(chrono::Utc::now() + chrono::Duration::minutes(10)),
|
||||
enabled: true,
|
||||
}));
|
||||
let account = store.account("acct_primary").expect("account");
|
||||
let selector = selector(&store, Arc::new(TestLeaseCoordinator::default()));
|
||||
|
||||
block_on(selector.record_failure(
|
||||
&account,
|
||||
CodexAccountFailure::CredentialExpired,
|
||||
Some("token_expired".to_owned()),
|
||||
))
|
||||
.expect("record credential expiry");
|
||||
|
||||
assert_eq!(
|
||||
store
|
||||
.account("acct_primary")
|
||||
.expect("account retained for refresh")
|
||||
.availability(),
|
||||
AccountAvailability::Ready
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -211,3 +211,47 @@ async fn expired_old_reset_with_fresh_allowed_snapshot_recovers_ready() {
|
||||
"old reset elapsed + fresh allowed snapshot must recover QuotaExhausted"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn deactivated_workspace_quota_response_persists_the_upstream_reason() {
|
||||
let store = Arc::new(MemoryAccountStore::default());
|
||||
let account_id = "acct_deactivated_workspace";
|
||||
create_account(&store, account_id).await;
|
||||
let account = store.account(account_id).expect("created account");
|
||||
let server = MockServer::start().await;
|
||||
Mock::given(method("GET"))
|
||||
.and(path("/api/codex/usage"))
|
||||
.and(header(
|
||||
"authorization",
|
||||
format!("Bearer token-{account_id}"),
|
||||
))
|
||||
.respond_with(ResponseTemplate::new(402).set_body_json(json!({
|
||||
"detail": {
|
||||
"code": "deactivated_workspace",
|
||||
"message": "workspace disabled"
|
||||
}
|
||||
})))
|
||||
.mount(&server)
|
||||
.await;
|
||||
let service = quota_service_with_base_url(
|
||||
&store,
|
||||
reqwest::Client::builder()
|
||||
.no_proxy()
|
||||
.build()
|
||||
.expect("client"),
|
||||
server.uri(),
|
||||
);
|
||||
|
||||
assert!(service.refresh_account(account.id()).await.is_err());
|
||||
assert_eq!(
|
||||
store
|
||||
.account(account_id)
|
||||
.expect("account after deactivation")
|
||||
.availability(),
|
||||
AccountAvailability::Banned
|
||||
);
|
||||
assert_eq!(
|
||||
store.last_error_message(account_id).as_deref(),
|
||||
Some("deactivated_workspace")
|
||||
);
|
||||
}
|
||||
|
||||
@@ -19,6 +19,8 @@ use secrecy::{ExposeSecret, SecretString};
|
||||
|
||||
use crate::support::{MemoryAccountStore, profile, runtime_policy, secret};
|
||||
|
||||
const OAUTH_BACKOFF_ATTEMPTS: u32 = 5;
|
||||
|
||||
struct Refresher {
|
||||
outcomes: Mutex<VecDeque<Result<TokenPair, RefreshFailure>>>,
|
||||
seen: Mutex<Vec<String>>,
|
||||
@@ -553,6 +555,10 @@ async fn upstream_ban_marks_unified_account_banned() {
|
||||
.availability(),
|
||||
AccountAvailability::Banned
|
||||
);
|
||||
assert_eq!(
|
||||
store.last_error_message("acct_refresh").as_deref(),
|
||||
Some("account_banned")
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -597,7 +603,7 @@ async fn transient_refresh_defers_without_invalidating_account() {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn expired_access_token_refreshes_even_when_next_refresh_is_deferred() {
|
||||
async fn expired_access_token_respects_persisted_refresh_backoff() {
|
||||
let store = Arc::new(MemoryAccountStore::default());
|
||||
seed_due_account(&store).await;
|
||||
let original = store.account("acct_refresh").expect("seeded account");
|
||||
@@ -615,16 +621,6 @@ async fn expired_access_token_refreshes_even_when_next_refresh_is_deferred() {
|
||||
})
|
||||
.await
|
||||
.expect("seed expired access token with deferred refresh");
|
||||
let deferred = store.account("acct_refresh").expect("deferred account");
|
||||
store
|
||||
.repository()
|
||||
.apply_state(
|
||||
&deferred,
|
||||
AccountAvailability::QuotaExhausted,
|
||||
SystemTime::now(),
|
||||
)
|
||||
.await
|
||||
.expect("mark account quota exhausted");
|
||||
let refresher = Arc::new(Refresher::new(Ok(success_tokens())));
|
||||
let service = CodexCredentialRefreshService::new(
|
||||
store.repository(),
|
||||
@@ -640,31 +636,19 @@ async fn expired_access_token_refreshes_even_when_next_refresh_is_deferred() {
|
||||
|
||||
let outcomes = service.refresh_due().await.expect("refresh due");
|
||||
|
||||
assert!(matches!(
|
||||
outcomes.as_slice(),
|
||||
[CodexCredentialRefreshOutcome::Refreshed {
|
||||
account_id,
|
||||
credential_revision: 3,
|
||||
}] if account_id == "acct_refresh"
|
||||
));
|
||||
assert_eq!(
|
||||
refresher.seen.lock().expect("seen tokens lock").as_slice(),
|
||||
["rt-old-access"]
|
||||
);
|
||||
let refreshed = store.account("acct_refresh").expect("refreshed account");
|
||||
assert_eq!(
|
||||
refreshed.availability(),
|
||||
AccountAvailability::QuotaExhausted
|
||||
);
|
||||
assert!(outcomes.is_empty());
|
||||
assert!(refresher.seen.lock().expect("seen tokens lock").is_empty());
|
||||
assert!(
|
||||
refreshed
|
||||
.access_token_expires_at()
|
||||
.is_some_and(|expires_at| expires_at > SystemTime::now())
|
||||
store
|
||||
.account("acct_refresh")
|
||||
.expect("deferred account")
|
||||
.next_refresh_at()
|
||||
.is_some_and(|next_refresh_at| next_refresh_at > SystemTime::now())
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn expired_access_token_refreshes_during_active_cooldown() {
|
||||
async fn expired_access_token_refreshes_when_persisted_refresh_is_due() {
|
||||
let store = Arc::new(MemoryAccountStore::default());
|
||||
seed_due_account(&store).await;
|
||||
let original = store.account("acct_refresh").expect("seeded account");
|
||||
@@ -678,7 +662,7 @@ async fn expired_access_token_refreshes_during_active_cooldown() {
|
||||
expected_credential_revision: original.revision().get(),
|
||||
secret: secret("old-access"),
|
||||
verified_account: expired_profile,
|
||||
next_refresh_at: Some(chrono::Utc::now() + chrono::Duration::hours(12)),
|
||||
next_refresh_at: Some(chrono::Utc::now() - chrono::Duration::seconds(1)),
|
||||
})
|
||||
.await
|
||||
.expect("seed expired access token with deferred refresh");
|
||||
@@ -825,6 +809,15 @@ async fn seed_due_account(store: &MemoryAccountStore) {
|
||||
.await;
|
||||
}
|
||||
|
||||
async fn force_due_account(store: &Arc<MemoryAccountStore>) {
|
||||
let account = store.account("acct_refresh").expect("account to reset");
|
||||
store
|
||||
.repository()
|
||||
.defer_refresh(&account, SystemTime::now() - Duration::from_secs(1))
|
||||
.await
|
||||
.expect("reset next_refresh_at to the past");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn refresh_backoff_grows_exponentially_across_attempts() {
|
||||
let store = Arc::new(MemoryAccountStore::default());
|
||||
@@ -891,6 +884,175 @@ async fn refresh_backoff_grows_exponentially_across_attempts() {
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn exhausted_oauth_backoff_continues_after_access_token_expiry() {
|
||||
let store = Arc::new(MemoryAccountStore::default());
|
||||
let mut expired_profile = profile("chatgpt-acct_refresh");
|
||||
expired_profile.access_token_expires_at =
|
||||
Some(chrono::Utc::now() - chrono::Duration::minutes(1));
|
||||
store
|
||||
.seed_oauth_credential(ImportCodexOAuthCredential {
|
||||
account_id: "acct_refresh".to_owned(),
|
||||
name: "refresh".to_owned(),
|
||||
secret: secret("old-access"),
|
||||
verified_account: expired_profile,
|
||||
next_refresh_at: Some(chrono::Utc::now() - chrono::Duration::seconds(1)),
|
||||
enabled: true,
|
||||
})
|
||||
.await;
|
||||
let credential_state = Arc::new(CountingCredentialState::default());
|
||||
let state_port: Arc<dyn ProviderCredentialStatePort> = credential_state.clone();
|
||||
let refresher = Arc::new(Refresher::scripted([
|
||||
Err(RefreshFailure::Transport),
|
||||
Err(RefreshFailure::Transport),
|
||||
Err(RefreshFailure::Transport),
|
||||
Err(RefreshFailure::Transport),
|
||||
Err(RefreshFailure::Transport),
|
||||
Err(RefreshFailure::Transport),
|
||||
]));
|
||||
let service = CodexCredentialRefreshService::new(
|
||||
store.repository(),
|
||||
refresher.clone(),
|
||||
Arc::new(VerifiedIdentity),
|
||||
Arc::new(RefreshLeases {
|
||||
available: true,
|
||||
requests: Mutex::new(Vec::new()),
|
||||
}),
|
||||
state_port,
|
||||
runtime_policy(),
|
||||
);
|
||||
|
||||
for _ in 0..OAUTH_BACKOFF_ATTEMPTS {
|
||||
service.refresh_due().await.expect("backoff attempt");
|
||||
force_due_account(&store).await;
|
||||
}
|
||||
|
||||
let observed_at = SystemTime::now();
|
||||
service.refresh_due().await.expect("recovery attempt");
|
||||
let recovery_delay = store
|
||||
.account("acct_refresh")
|
||||
.expect("account after recovery defer")
|
||||
.next_refresh_at()
|
||||
.expect("recovery scheduled")
|
||||
.duration_since(observed_at)
|
||||
.expect("recovery is in the future");
|
||||
assert!(
|
||||
(Duration::from_secs(8 * 60)..=Duration::from_secs(12 * 60 + 1)).contains(&recovery_delay),
|
||||
"recovery delay should be about ten minutes, got {recovery_delay:?}"
|
||||
);
|
||||
assert_eq!(
|
||||
credential_state.count("acct_refresh"),
|
||||
OAUTH_BACKOFF_ATTEMPTS + 1
|
||||
);
|
||||
assert_eq!(refresher.seen.lock().expect("seen tokens lock").len(), 6);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn oauth_backoff_never_passes_an_unexpired_access_token_expiry() {
|
||||
let store = Arc::new(MemoryAccountStore::default());
|
||||
let mut account_profile = profile("chatgpt-acct_refresh");
|
||||
account_profile.access_token_expires_at =
|
||||
Some(chrono::Utc::now() + chrono::Duration::seconds(30));
|
||||
let access_token_expires_at: SystemTime = account_profile
|
||||
.access_token_expires_at
|
||||
.expect("access expiry")
|
||||
.into();
|
||||
store
|
||||
.seed_oauth_credential(ImportCodexOAuthCredential {
|
||||
account_id: "acct_refresh".to_owned(),
|
||||
name: "refresh".to_owned(),
|
||||
secret: secret("old-access"),
|
||||
verified_account: account_profile,
|
||||
next_refresh_at: Some(chrono::Utc::now() - chrono::Duration::seconds(1)),
|
||||
enabled: true,
|
||||
})
|
||||
.await;
|
||||
let credential_state = Arc::new(CountingCredentialState::default());
|
||||
let state_port: Arc<dyn ProviderCredentialStatePort> = credential_state.clone();
|
||||
let service = CodexCredentialRefreshService::new(
|
||||
store.repository(),
|
||||
Arc::new(Refresher::scripted([
|
||||
Err(RefreshFailure::Transport),
|
||||
Err(RefreshFailure::Transport),
|
||||
Err(RefreshFailure::Transport),
|
||||
Err(RefreshFailure::Transport),
|
||||
Err(RefreshFailure::Transport),
|
||||
Err(RefreshFailure::Transport),
|
||||
])),
|
||||
Arc::new(VerifiedIdentity),
|
||||
Arc::new(RefreshLeases {
|
||||
available: true,
|
||||
requests: Mutex::new(Vec::new()),
|
||||
}),
|
||||
state_port,
|
||||
runtime_policy(),
|
||||
);
|
||||
|
||||
for attempt in 0..=OAUTH_BACKOFF_ATTEMPTS {
|
||||
service.refresh_due().await.expect("backoff attempt");
|
||||
if attempt < OAUTH_BACKOFF_ATTEMPTS {
|
||||
force_due_account(&store).await;
|
||||
}
|
||||
}
|
||||
|
||||
assert_eq!(
|
||||
store
|
||||
.account("acct_refresh")
|
||||
.expect("account after capped retry")
|
||||
.next_refresh_at(),
|
||||
Some(access_token_expires_at)
|
||||
);
|
||||
assert_eq!(
|
||||
credential_state.count("acct_refresh"),
|
||||
OAUTH_BACKOFF_ATTEMPTS + 1
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn recovery_window_exhaustion_marks_account_expired_without_token_exchange() {
|
||||
let store = Arc::new(MemoryAccountStore::default());
|
||||
let mut expired_profile = profile("chatgpt-acct_refresh");
|
||||
expired_profile.access_token_expires_at =
|
||||
Some(chrono::Utc::now() - chrono::Duration::hours(25));
|
||||
store
|
||||
.seed_oauth_credential(ImportCodexOAuthCredential {
|
||||
account_id: "acct_refresh".to_owned(),
|
||||
name: "refresh".to_owned(),
|
||||
secret: secret("old-access"),
|
||||
verified_account: expired_profile,
|
||||
next_refresh_at: Some(chrono::Utc::now() + chrono::Duration::hours(12)),
|
||||
enabled: true,
|
||||
})
|
||||
.await;
|
||||
let refresher = Arc::new(Refresher::new(Ok(success_tokens())));
|
||||
let service = CodexCredentialRefreshService::new(
|
||||
store.repository(),
|
||||
refresher.clone(),
|
||||
Arc::new(VerifiedIdentity),
|
||||
Arc::new(RefreshLeases {
|
||||
available: true,
|
||||
requests: Mutex::new(Vec::new()),
|
||||
}),
|
||||
Arc::new(CountingCredentialState::default()),
|
||||
runtime_policy(),
|
||||
);
|
||||
|
||||
let outcomes = service.refresh_due().await.expect("refresh due");
|
||||
|
||||
assert!(matches!(
|
||||
outcomes.as_slice(),
|
||||
[CodexCredentialRefreshOutcome::Invalidated { account_id }] if account_id == "acct_refresh"
|
||||
));
|
||||
assert!(refresher.seen.lock().expect("seen tokens lock").is_empty());
|
||||
assert_eq!(
|
||||
store
|
||||
.account("acct_refresh")
|
||||
.expect("expired account")
|
||||
.availability(),
|
||||
AccountAvailability::Expired
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn successful_refresh_clears_backoff_counter() {
|
||||
let store = Arc::new(MemoryAccountStore::default());
|
||||
|
||||
@@ -38,6 +38,7 @@ struct StoredAccount {
|
||||
credential: gateway_core::engine::credential::PlaintextCredential,
|
||||
quota: Option<QuotaObservation>,
|
||||
state_observed_at: Option<SystemTime>,
|
||||
last_error_message: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
@@ -70,6 +71,15 @@ impl MemoryAccountStore {
|
||||
.map(|stored| stored.account.clone())
|
||||
}
|
||||
|
||||
pub(crate) fn last_error_message(&self, id: &str) -> Option<String> {
|
||||
let id = ProviderAccountId::new(id).ok()?;
|
||||
self.accounts
|
||||
.lock()
|
||||
.expect("account store lock")
|
||||
.get(&id)
|
||||
.and_then(|stored| stored.last_error_message.clone())
|
||||
}
|
||||
|
||||
pub(crate) async fn advance_credential_revision(&self, id: &str) -> CredentialRevision {
|
||||
let account_id = ProviderAccountId::new(id).expect("test account id");
|
||||
let current = self
|
||||
@@ -148,6 +158,7 @@ impl ProviderAccountStore for MemoryAccountStore {
|
||||
credential: input.credential,
|
||||
quota: None,
|
||||
state_observed_at: None,
|
||||
last_error_message: None,
|
||||
},
|
||||
);
|
||||
Ok(())
|
||||
@@ -321,6 +332,11 @@ impl ProviderAccountStore for MemoryAccountStore {
|
||||
},
|
||||
);
|
||||
stored.state_observed_at = Some(change.observed_at);
|
||||
if change.availability == AccountAvailability::Ready {
|
||||
stored.last_error_message = None;
|
||||
} else if let Some(message) = change.message {
|
||||
stored.last_error_message = Some(message);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -221,7 +221,7 @@ impl GrokOAuthClient {
|
||||
/// # Errors
|
||||
///
|
||||
/// 返回已分类的 OAuth 错误。refresh token 可能已轮换,Ambiguous 类
|
||||
/// transport 失败不得自动重试。
|
||||
/// transport 失败不在本次 exchange 内重试,后续交由 refresh scheduler 退避协调。
|
||||
pub async fn refresh(
|
||||
&self,
|
||||
discovery: &DiscoveryDocument,
|
||||
|
||||
@@ -37,7 +37,7 @@ pub enum OAuthErrorCode {
|
||||
pub enum FailureClass {
|
||||
/// 瞬时失败,允许后续独立协调的重试。
|
||||
Transient,
|
||||
/// 发送状态不确定,一次性凭据不得重放。
|
||||
/// 发送状态不确定,不应在同一 exchange 内重放一次性凭据。
|
||||
Ambiguous,
|
||||
/// 当前 revision 的凭据被永久拒绝。
|
||||
CredentialPermanent,
|
||||
|
||||
@@ -12,10 +12,10 @@ use gateway_core::engine::credential::{
|
||||
AccountAvailability, CredentialRevision, LoadedCredential, ProviderAccountId,
|
||||
};
|
||||
use gateway_core::provider_ports::{
|
||||
ProviderCooldown, ProviderCooldownPort, ProviderCredentialStatePort, ProviderLeaseAcquisition,
|
||||
ProviderLeasePort, ProviderLeaseRequest, ProviderRefreshCapacityRequest,
|
||||
ProviderRefreshLeaseRequest, ProviderRefreshPolicy, ProviderRuntimePolicyPort,
|
||||
ProviderStoreError, provider_refresh_backoff_at,
|
||||
ProviderCredentialStatePort, ProviderLeaseAcquisition, ProviderLeasePort, ProviderLeaseRequest,
|
||||
ProviderRefreshCapacityRequest, ProviderRefreshLeaseRequest, ProviderRefreshPolicy,
|
||||
ProviderRuntimePolicyPort, ProviderStoreError, provider_refresh_backoff_at,
|
||||
provider_refresh_retry_at,
|
||||
};
|
||||
|
||||
use super::catalog::GrokCredentialCatalogService;
|
||||
@@ -38,12 +38,49 @@ const MAX_SECRET_BYTES: usize = 64 * 1_024;
|
||||
const DISCOVERY_NEGATIVE_CACHE_TTL: Duration = Duration::from_secs(5);
|
||||
/// 连续失败计数窗口;静默满窗后计数过期归零。
|
||||
const REFRESH_BACKOFF_WINDOW: Duration = Duration::from_secs(30 * 60);
|
||||
const REFRESH_AMBIGUITY_COOLDOWN: Duration = Duration::from_secs(24 * 60 * 60);
|
||||
/// 窗口内先执行五次指数退避,再进入固定恢复周期。
|
||||
const REFRESH_BACKOFF_MAX_ATTEMPTS: u32 = 5;
|
||||
/// 耗尽指数退避后的 OAuth 恢复周期。
|
||||
const REFRESH_RECOVERY_DELAY: Duration = Duration::from_secs(10 * 60);
|
||||
/// 过期 AT 仍允许 RT 恢复的最长窗口。
|
||||
const REFRESH_RECOVERY_WINDOW: Duration = Duration::from_secs(24 * 60 * 60);
|
||||
|
||||
fn refresh_recovery_deadline(access_token_expires_at: Option<SystemTime>) -> Option<SystemTime> {
|
||||
access_token_expires_at.and_then(|expires_at| expires_at.checked_add(REFRESH_RECOVERY_WINDOW))
|
||||
}
|
||||
|
||||
fn refresh_recovery_window_exhausted(
|
||||
access_token_expires_at: Option<SystemTime>,
|
||||
now: SystemTime,
|
||||
) -> bool {
|
||||
refresh_recovery_deadline(access_token_expires_at).is_some_and(|deadline| deadline <= now)
|
||||
}
|
||||
|
||||
fn bounded_refresh_retry_at(
|
||||
retry_at: SystemTime,
|
||||
access_token_expires_at: Option<SystemTime>,
|
||||
observed_at: SystemTime,
|
||||
) -> SystemTime {
|
||||
let mut bounded = retry_at;
|
||||
if let Some(expires_at) = access_token_expires_at
|
||||
&& expires_at > observed_at
|
||||
&& bounded > expires_at
|
||||
{
|
||||
bounded = expires_at;
|
||||
}
|
||||
if let Some(deadline) = refresh_recovery_deadline(access_token_expires_at)
|
||||
&& bounded > deadline
|
||||
{
|
||||
bounded = deadline;
|
||||
}
|
||||
bounded
|
||||
}
|
||||
|
||||
/// 一个到期且已按 revision 读取明文 RT 的 xAI account。
|
||||
pub struct DueGrokCredential {
|
||||
account_id: ProviderAccountId,
|
||||
credential_revision: CredentialRevision,
|
||||
access_token_expires_at: Option<SystemTime>,
|
||||
refresh_token: SecretValue,
|
||||
id_token: Option<SecretValue>,
|
||||
scope: String,
|
||||
@@ -77,6 +114,7 @@ impl fmt::Debug for DueGrokCredential {
|
||||
.debug_struct("DueGrokCredential")
|
||||
.field("account_id", &self.account_id)
|
||||
.field("credential_revision", &self.credential_revision)
|
||||
.field("access_token_expires_at", &self.access_token_expires_at)
|
||||
.field("refresh_token", &"[REDACTED]")
|
||||
.field("id_token", &self.id_token.as_ref().map(|_| "[REDACTED]"))
|
||||
.field("scope", &"[REDACTED]")
|
||||
@@ -277,7 +315,6 @@ pub struct GrokCredentialRefreshService {
|
||||
refresher: Arc<dyn GrokCredentialRefresher>,
|
||||
catalog: Arc<GrokCredentialCatalogService>,
|
||||
leases: Arc<dyn ProviderLeasePort>,
|
||||
cooldowns: Arc<dyn ProviderCooldownPort>,
|
||||
credential_state: Arc<dyn ProviderCredentialStatePort>,
|
||||
runtime_policy: Arc<dyn ProviderRuntimePolicyPort>,
|
||||
}
|
||||
@@ -288,7 +325,6 @@ impl GrokCredentialRefreshService {
|
||||
refresher: Arc<dyn GrokCredentialRefresher>,
|
||||
catalog: Arc<GrokCredentialCatalogService>,
|
||||
leases: Arc<dyn ProviderLeasePort>,
|
||||
cooldowns: Arc<dyn ProviderCooldownPort>,
|
||||
credential_state: Arc<dyn ProviderCredentialStatePort>,
|
||||
runtime_policy: Arc<dyn ProviderRuntimePolicyPort>,
|
||||
) -> Self {
|
||||
@@ -297,7 +333,6 @@ impl GrokCredentialRefreshService {
|
||||
refresher,
|
||||
catalog,
|
||||
leases,
|
||||
cooldowns,
|
||||
credential_state,
|
||||
runtime_policy,
|
||||
}
|
||||
@@ -331,6 +366,7 @@ impl GrokCredentialRefreshService {
|
||||
let credential = DueGrokCredential {
|
||||
account_id: account_id.clone(),
|
||||
credential_revision,
|
||||
access_token_expires_at: loaded.account.access_token_expires_at(),
|
||||
refresh_token: loaded.refresh_token,
|
||||
id_token: loaded.id_token,
|
||||
scope: loaded.scope,
|
||||
@@ -447,12 +483,7 @@ impl GrokCredentialRefreshService {
|
||||
.into_iter()
|
||||
.map(|account_id| GrokCredentialRefreshOutcome::Failed { account_id })
|
||||
.collect::<Vec<_>>();
|
||||
let mut credentials = Vec::with_capacity(batch.credentials.len());
|
||||
for credential in batch.credentials {
|
||||
if !self.runtime_cooldown_active(&credential).await {
|
||||
credentials.push(credential);
|
||||
}
|
||||
}
|
||||
let credentials = batch.credentials;
|
||||
if credentials.is_empty() {
|
||||
return Ok(outcomes);
|
||||
}
|
||||
@@ -474,7 +505,14 @@ impl GrokCredentialRefreshService {
|
||||
for (account_id, result) in refreshed {
|
||||
match result {
|
||||
Ok(outcome) => outcomes.push(outcome),
|
||||
Err(_) => outcomes.push(GrokCredentialRefreshOutcome::Failed { account_id }),
|
||||
Err(error) => {
|
||||
tracing::warn!(
|
||||
account_id = %account_id,
|
||||
error = %error,
|
||||
"xAI OAuth refresh attempt failed"
|
||||
);
|
||||
outcomes.push(GrokCredentialRefreshOutcome::Failed { account_id });
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(outcomes)
|
||||
@@ -486,9 +524,10 @@ impl GrokCredentialRefreshService {
|
||||
policy: ProviderRefreshPolicy,
|
||||
) -> Result<GrokCredentialRefreshOutcome, GrokCredentialRefreshError> {
|
||||
let account_id = credential.account_id.clone();
|
||||
let observed_at = SystemTime::now();
|
||||
if credential
|
||||
.refresh_token_expires_at
|
||||
.is_some_and(|expires_at| expires_at <= Utc::now())
|
||||
.is_some_and(|expires_at| expires_at <= DateTime::<Utc>::from(observed_at))
|
||||
{
|
||||
return self
|
||||
.persist_terminal_failure(
|
||||
@@ -498,6 +537,15 @@ impl GrokCredentialRefreshService {
|
||||
)
|
||||
.await;
|
||||
}
|
||||
if refresh_recovery_window_exhausted(credential.access_token_expires_at, observed_at) {
|
||||
return self
|
||||
.persist_terminal_failure(
|
||||
credential,
|
||||
GrokCredentialAvailability::Expired,
|
||||
"refresh_recovery_window_exhausted",
|
||||
)
|
||||
.await;
|
||||
}
|
||||
let capacity = self
|
||||
.leases
|
||||
.try_acquire(ProviderLeaseRequest::RefreshCapacity(
|
||||
@@ -545,9 +593,14 @@ impl GrokCredentialRefreshService {
|
||||
.await
|
||||
}
|
||||
Err(GrokRefreshFailure::Ambiguous) => {
|
||||
self.record_runtime_cooldown(&credential, REFRESH_AMBIGUITY_COOLDOWN)
|
||||
.await;
|
||||
Ok(GrokCredentialRefreshOutcome::Ambiguous { account_id })
|
||||
if self
|
||||
.persist_backoff(&credential, "refresh-ambiguous")
|
||||
.await?
|
||||
{
|
||||
Ok(GrokCredentialRefreshOutcome::Ambiguous { account_id })
|
||||
} else {
|
||||
Ok(GrokCredentialRefreshOutcome::Stale { account_id })
|
||||
}
|
||||
}
|
||||
Err(GrokRefreshFailure::Transient) => {
|
||||
if self
|
||||
@@ -644,35 +697,19 @@ impl GrokCredentialRefreshService {
|
||||
.credential_state
|
||||
.clear_refresh_backoff(&account_id)
|
||||
.await;
|
||||
tracing::info!(
|
||||
account_id = %account_id,
|
||||
credential_revision = record.credential_revision.get(),
|
||||
access_token_expires_at = %access_expires_at,
|
||||
next_refresh_at = %next_refresh_at,
|
||||
"xAI OAuth refresh succeeded"
|
||||
);
|
||||
Ok(GrokCredentialRefreshOutcome::Refreshed {
|
||||
account_id,
|
||||
credential_revision: record.credential_revision,
|
||||
})
|
||||
}
|
||||
|
||||
async fn runtime_cooldown_active(&self, credential: &DueGrokCredential) -> bool {
|
||||
// runtime cooldown 是账号/model 级生命周期,与 credential revision 无关:
|
||||
// 轮换不清除、不因 revision 变化判失效(与 selector 的读取语义一致)。
|
||||
let Ok(Some(cooldown)) = self.cooldowns.read(credential.account_id()).await else {
|
||||
return false;
|
||||
};
|
||||
cooldown.until() > SystemTime::now()
|
||||
}
|
||||
|
||||
async fn record_runtime_cooldown(&self, credential: &DueGrokCredential, duration: Duration) {
|
||||
let cooldown_until = SystemTime::now()
|
||||
.checked_add(duration)
|
||||
.unwrap_or_else(SystemTime::now);
|
||||
let _ = self
|
||||
.cooldowns
|
||||
.put_if_later(ProviderCooldown::new(
|
||||
credential.account_id.clone(),
|
||||
credential.credential_revision,
|
||||
cooldown_until,
|
||||
))
|
||||
.await;
|
||||
}
|
||||
|
||||
async fn persist_terminal_failure(
|
||||
&self,
|
||||
credential: DueGrokCredential,
|
||||
@@ -702,7 +739,19 @@ impl GrokCredentialRefreshService {
|
||||
})
|
||||
.await
|
||||
{
|
||||
Ok(()) => Ok(GrokCredentialRefreshOutcome::Invalidated { account_id }),
|
||||
Ok(()) => {
|
||||
tracing::warn!(
|
||||
account_id = %account_id,
|
||||
?availability,
|
||||
reason,
|
||||
access_token_expires_at = ?credential.access_token_expires_at
|
||||
.map(DateTime::<Utc>::from),
|
||||
recovery_deadline = ?refresh_recovery_deadline(credential.access_token_expires_at)
|
||||
.map(DateTime::<Utc>::from),
|
||||
"xAI OAuth refresh marked account terminal"
|
||||
);
|
||||
Ok(GrokCredentialRefreshOutcome::Invalidated { account_id })
|
||||
}
|
||||
Err(error) if stale_repository_error(&error) => {
|
||||
Ok(GrokCredentialRefreshOutcome::Stale { account_id })
|
||||
}
|
||||
@@ -720,12 +769,19 @@ impl GrokCredentialRefreshService {
|
||||
.record_refresh_backoff(&credential.account_id, REFRESH_BACKOFF_WINDOW)
|
||||
.await
|
||||
.unwrap_or(1);
|
||||
let retry_at = provider_refresh_backoff_at(
|
||||
&credential.account_id,
|
||||
SystemTime::now(),
|
||||
attempt,
|
||||
reason,
|
||||
)?;
|
||||
let observed_at = SystemTime::now();
|
||||
let retry_at = if attempt <= REFRESH_BACKOFF_MAX_ATTEMPTS {
|
||||
provider_refresh_backoff_at(&credential.account_id, observed_at, attempt, reason)?
|
||||
} else {
|
||||
provider_refresh_retry_at(
|
||||
&credential.account_id,
|
||||
observed_at,
|
||||
REFRESH_RECOVERY_DELAY,
|
||||
reason,
|
||||
)?
|
||||
};
|
||||
let retry_at =
|
||||
bounded_refresh_retry_at(retry_at, credential.access_token_expires_at, observed_at);
|
||||
match self
|
||||
.repository
|
||||
.defer_refresh(
|
||||
@@ -735,7 +791,20 @@ impl GrokCredentialRefreshService {
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(()) => Ok(true),
|
||||
Ok(()) => {
|
||||
tracing::warn!(
|
||||
account_id = %credential.account_id,
|
||||
attempt,
|
||||
reason,
|
||||
retry_at = %DateTime::<Utc>::from(retry_at),
|
||||
access_token_expires_at = ?credential.access_token_expires_at
|
||||
.map(DateTime::<Utc>::from),
|
||||
recovery_deadline = ?refresh_recovery_deadline(credential.access_token_expires_at)
|
||||
.map(DateTime::<Utc>::from),
|
||||
"xAI OAuth refresh deferred"
|
||||
);
|
||||
Ok(true)
|
||||
}
|
||||
Err(error) if stale_repository_error(&error) => Ok(false),
|
||||
Err(error) => Err(error.into()),
|
||||
}
|
||||
@@ -814,6 +883,7 @@ impl GrokCredentialRepository {
|
||||
due.push(DueGrokCredential {
|
||||
account_id: account.id().clone(),
|
||||
credential_revision: account.revision(),
|
||||
access_token_expires_at: account.access_token_expires_at(),
|
||||
refresh_token: loaded.refresh_token,
|
||||
id_token: loaded.id_token,
|
||||
scope: loaded.scope,
|
||||
@@ -844,7 +914,8 @@ fn account_due(
|
||||
| AccountAvailability::Banned
|
||||
| AccountAvailability::Invalid
|
||||
)
|
||||
&& account.next_refresh_at().is_some_and(|next| next <= now)
|
||||
&& (refresh_recovery_window_exhausted(account.access_token_expires_at(), now)
|
||||
|| account.next_refresh_at().is_some_and(|next| next <= now))
|
||||
}
|
||||
|
||||
fn refreshed_tokens(tokens: RefreshedTokenSet) -> Result<GrokRefreshTokens, GrokRefreshFailure> {
|
||||
|
||||
@@ -6,10 +6,10 @@ use std::time::SystemTime;
|
||||
|
||||
use chrono::{DateTime, Utc};
|
||||
use gateway_core::engine::credential::{
|
||||
AccountStateChange, CredentialCasOutcome, CredentialCasUpdate, CredentialRevision,
|
||||
LoadedCredential, NewProviderAccount, OpaqueProviderData, PlaintextCredential, ProviderAccount,
|
||||
ProviderAccountId, ProviderAccountStore, ProviderAccountUpdate, QuotaObservation,
|
||||
QuotaWriteOutcome,
|
||||
AccountAvailability, AccountStateChange, CredentialCasOutcome, CredentialCasUpdate,
|
||||
CredentialRevision, LoadedCredential, NewProviderAccount, OpaqueProviderData,
|
||||
PlaintextCredential, ProviderAccount, ProviderAccountId, ProviderAccountStore,
|
||||
ProviderAccountUpdate, QuotaObservation, QuotaWriteOutcome,
|
||||
};
|
||||
use gateway_core::error::StoreErrorKind;
|
||||
use gateway_core::provider_ports::ProviderRefreshPolicy;
|
||||
@@ -428,15 +428,21 @@ impl GrokCredentialRepository {
|
||||
&self,
|
||||
input: &UpdateGrokCredentialState,
|
||||
) -> Result<(), GrokCredentialRepositoryError> {
|
||||
validate_reason(input.availability_reason.as_deref())?;
|
||||
let availability = AccountAvailability::from(input.availability);
|
||||
let availability_reason = input
|
||||
.availability_reason
|
||||
.clone()
|
||||
.filter(|value| !value.trim().is_empty())
|
||||
.or_else(|| availability.fallback_error_reason().map(str::to_owned));
|
||||
validate_reason(availability_reason.as_deref())?;
|
||||
self.ensure_xai_account(&input.account_id).await?;
|
||||
self.store
|
||||
.apply_state_change(AccountStateChange {
|
||||
account_id: input.account_id.clone(),
|
||||
expected_revision: input.expected_revision,
|
||||
availability: input.availability.into(),
|
||||
availability,
|
||||
observed_at: to_system_time(input.observed_at),
|
||||
message: input.availability_reason.clone(),
|
||||
message: availability_reason,
|
||||
})
|
||||
.await
|
||||
.map_err(map_store_error)
|
||||
|
||||
@@ -171,7 +171,6 @@ pub async fn initialize(
|
||||
refresher,
|
||||
Arc::clone(&catalog),
|
||||
Arc::clone(&leases),
|
||||
cooldowns.clone(),
|
||||
ports.credential_state(),
|
||||
Arc::clone(&runtime_policy),
|
||||
));
|
||||
|
||||
@@ -2040,11 +2040,10 @@ impl ScheduledTask for XaiOAuthRefreshTask {
|
||||
if context.cancellation().is_cancelled() {
|
||||
return Ok(());
|
||||
}
|
||||
let outcomes = self
|
||||
.service
|
||||
.refresh_due()
|
||||
.await
|
||||
.map_err(|_| WorkerTaskError::safe("xAI OAuth refresh failed"))?;
|
||||
let outcomes = self.service.refresh_due().await.map_err(|error| {
|
||||
tracing::error!(error = %error, "xAI OAuth refresh cycle failed");
|
||||
WorkerTaskError::safe("xAI OAuth refresh failed")
|
||||
})?;
|
||||
let failures = outcomes
|
||||
.iter()
|
||||
.filter(|outcome| {
|
||||
@@ -2056,6 +2055,13 @@ impl ScheduledTask for XaiOAuthRefreshTask {
|
||||
)
|
||||
})
|
||||
.count();
|
||||
if !outcomes.is_empty() {
|
||||
tracing::info!(
|
||||
accounts = outcomes.len(),
|
||||
failures,
|
||||
"xAI OAuth refresh cycle completed"
|
||||
);
|
||||
}
|
||||
if failures > 0 {
|
||||
tracing::warn!(failures, "xAI OAuth refresh cycle contained failures");
|
||||
}
|
||||
|
||||
@@ -233,6 +233,29 @@ async fn state_update_uses_credential_revision_fence() {
|
||||
assert_eq!(account.availability(), AccountAvailability::Ready);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn terminal_state_without_upstream_message_uses_stable_error_reason() {
|
||||
let (store, repository) = repository();
|
||||
let input = create_input("state-fallback", "subject-state-fallback");
|
||||
seed_input(&store, &input).await.expect("create");
|
||||
|
||||
repository
|
||||
.update_state(&UpdateGrokCredentialState {
|
||||
account_id: input.account_id.clone(),
|
||||
expected_revision: CredentialRevision::new(1).expect("revision"),
|
||||
availability: GrokCredentialAvailability::Banned,
|
||||
availability_reason: None,
|
||||
observed_at: Utc::now(),
|
||||
})
|
||||
.await
|
||||
.expect("state update");
|
||||
|
||||
assert_eq!(
|
||||
store.last_error_message(&input.account_id).as_deref(),
|
||||
Some("account_banned")
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn admin_prepare_does_not_mutate_provider_account_store() {
|
||||
let (store, _) = repository();
|
||||
|
||||
@@ -10,8 +10,8 @@ use gateway_core::engine::credential::{
|
||||
ProviderAccountId, ProviderAccountStore, ProviderAccountUpdate,
|
||||
};
|
||||
use gateway_core::provider_ports::{
|
||||
ProviderCooldown, ProviderCooldownPort, ProviderCredentialState, ProviderCredentialStatePort,
|
||||
ProviderLeaseAcquisition, ProviderLeasePort, ProviderLeaseRequest, ProviderStoreError,
|
||||
ProviderCredentialState, ProviderCredentialStatePort, ProviderLeaseAcquisition,
|
||||
ProviderLeasePort, ProviderLeaseRequest, ProviderStoreError,
|
||||
};
|
||||
use provider_xai::{
|
||||
GrokCredentialCatalogCache, GrokCredentialRecovery, GrokCredentialRecoveryOutcome,
|
||||
@@ -22,12 +22,13 @@ use provider_xai::{
|
||||
};
|
||||
|
||||
use crate::support::{
|
||||
MemoryCooldownPort, MemoryGrokCatalogCache, MemoryProviderAccountStore, create_input,
|
||||
credential_object, runtime_policy, seed_input,
|
||||
MemoryGrokCatalogCache, MemoryProviderAccountStore, create_input, credential_object,
|
||||
runtime_policy, seed_input,
|
||||
};
|
||||
|
||||
const OFFICIAL_FIXTURE: &[u8] =
|
||||
include_bytes!("../transport/catalog/fixtures/official_grok_models_snapshot.json");
|
||||
const OAUTH_BACKOFF_ATTEMPTS: u32 = 5;
|
||||
|
||||
struct StaticCatalogTransport;
|
||||
|
||||
@@ -143,48 +144,6 @@ async fn fixture(
|
||||
fixture_many([input], responses, lease_available).await
|
||||
}
|
||||
|
||||
/// 带外部注入 cooldown 端口的 fixture(AUD-05 组合测试用)。
|
||||
#[allow(clippy::type_complexity)]
|
||||
async fn fixture_with_cooldowns(
|
||||
input: provider_xai::CreateGrokCredential,
|
||||
responses: impl IntoIterator<Item = Result<GrokRefreshTokens, GrokRefreshFailure>>,
|
||||
lease_available: bool,
|
||||
cooldowns: Arc<MemoryCooldownPort>,
|
||||
) -> (
|
||||
Arc<MemoryProviderAccountStore>,
|
||||
GrokCredentialRepository,
|
||||
Arc<QueueRefresher>,
|
||||
GrokCredentialRefreshService,
|
||||
Arc<MemoryCooldownPort>,
|
||||
) {
|
||||
let store = MemoryProviderAccountStore::shared();
|
||||
let account_store: Arc<dyn ProviderAccountStore> = store.clone();
|
||||
let repository = GrokCredentialRepository::new(account_store);
|
||||
seed_input(&store, &input).await.expect("create account");
|
||||
let refresher = QueueRefresher::new(responses);
|
||||
let refresher_port: Arc<dyn GrokCredentialRefresher> = refresher.clone();
|
||||
let cache: Arc<dyn GrokCredentialCatalogCache> = MemoryGrokCatalogCache::shared();
|
||||
let catalog = Arc::new(crate::support::grok_catalog_service(
|
||||
repository.clone(),
|
||||
Arc::new(StaticCatalogTransport),
|
||||
cache,
|
||||
));
|
||||
let leases = Arc::new(TestRefreshLeases {
|
||||
available: lease_available,
|
||||
calls: AtomicUsize::new(0),
|
||||
});
|
||||
let service = GrokCredentialRefreshService::new(
|
||||
repository.clone(),
|
||||
refresher_port,
|
||||
catalog,
|
||||
leases,
|
||||
cooldowns.clone(),
|
||||
Arc::new(CountingCredentialState::default()),
|
||||
runtime_policy(),
|
||||
);
|
||||
(store, repository, refresher, service, cooldowns)
|
||||
}
|
||||
|
||||
/// 真实累加连续失败计数的测试 double,用于断言退避的指数增长与成功清零。
|
||||
#[derive(Default)]
|
||||
struct CountingCredentialState {
|
||||
@@ -313,13 +272,11 @@ async fn fixture_many_with_state(
|
||||
available: lease_available,
|
||||
calls: AtomicUsize::new(0),
|
||||
});
|
||||
let cooldowns = Arc::new(MemoryCooldownPort::default());
|
||||
let service = GrokCredentialRefreshService::new(
|
||||
repository.clone(),
|
||||
refresher_port,
|
||||
catalog,
|
||||
leases,
|
||||
cooldowns,
|
||||
credential_state,
|
||||
runtime_policy(),
|
||||
);
|
||||
@@ -354,6 +311,38 @@ async fn force_due(store: &MemoryProviderAccountStore, id: &ProviderAccountId) {
|
||||
));
|
||||
}
|
||||
|
||||
async fn set_token_deadlines(
|
||||
store: &MemoryProviderAccountStore,
|
||||
id: &ProviderAccountId,
|
||||
access_token_expires_at: SystemTime,
|
||||
next_refresh_at: SystemTime,
|
||||
) {
|
||||
let account = store.account(id).expect("account to update");
|
||||
let credential = store.credential(id).expect("credential to update");
|
||||
let update = CredentialCasUpdate::new(
|
||||
id.clone(),
|
||||
account.revision(),
|
||||
ProviderAccountUpdate {
|
||||
account_id: id.clone(),
|
||||
name: account.name().to_owned(),
|
||||
email: account.email().map(str::to_owned),
|
||||
plan_type: account.plan_type().map(str::to_owned),
|
||||
},
|
||||
credential,
|
||||
true,
|
||||
Some(access_token_expires_at),
|
||||
Some(next_refresh_at),
|
||||
)
|
||||
.expect("valid token deadline update");
|
||||
assert!(matches!(
|
||||
store
|
||||
.compare_and_swap_credential(update)
|
||||
.await
|
||||
.expect("update token deadlines"),
|
||||
CredentialCasOutcome::Updated(_)
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn successful_refresh_rotates_plaintext_tokens_once() {
|
||||
let input = due_input("success");
|
||||
@@ -415,6 +404,10 @@ async fn unauthorized_recovery_marks_a_permanently_rejected_refresh_expired() {
|
||||
store.account(&id).expect("account").availability(),
|
||||
AccountAvailability::Expired
|
||||
);
|
||||
assert_eq!(
|
||||
store.last_error_message(&id).as_deref(),
|
||||
Some("refresh_invalid_grant")
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -573,14 +566,25 @@ async fn banned_failure_marks_account_banned() {
|
||||
store.account(&id).expect("account").availability(),
|
||||
AccountAvailability::Banned
|
||||
);
|
||||
assert_eq!(
|
||||
store.last_error_message(&id).as_deref(),
|
||||
Some("account_banned")
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ambiguous_refresh_uses_runtime_cooldown_without_persisting_account_state() {
|
||||
async fn ambiguous_refresh_applies_bounded_oauth_backoff() {
|
||||
let input = due_input("ambiguous");
|
||||
let id = input.account_id.clone();
|
||||
let (store, _, refresher, service) =
|
||||
fixture(input, [Err(GrokRefreshFailure::Ambiguous)], true).await;
|
||||
let counting = Arc::new(CountingCredentialState::default());
|
||||
let (store, _, refresher, service) = fixture_many_with_state(
|
||||
[input],
|
||||
[Err(GrokRefreshFailure::Ambiguous)],
|
||||
true,
|
||||
counting.clone(),
|
||||
)
|
||||
.await;
|
||||
let started_at = SystemTime::now();
|
||||
let outcomes = service.refresh_due().await.expect("refresh");
|
||||
assert!(matches!(
|
||||
outcomes.as_slice(),
|
||||
@@ -590,7 +594,18 @@ async fn ambiguous_refresh_uses_runtime_cooldown_without_persisting_account_stat
|
||||
store.account(&id).expect("account").availability(),
|
||||
AccountAvailability::Unknown
|
||||
);
|
||||
assert_eq!(store.account(&id).expect("account").revision().get(), 1);
|
||||
let account = store.account(&id).expect("account");
|
||||
assert_eq!(account.revision().get(), 2);
|
||||
let retry_delay = account
|
||||
.next_refresh_at()
|
||||
.expect("OAuth retry scheduled")
|
||||
.duration_since(started_at)
|
||||
.expect("OAuth retry is in the future");
|
||||
assert!(
|
||||
retry_delay <= Duration::from_secs(7),
|
||||
"first retry must use bounded OAuth backoff, got {retry_delay:?}"
|
||||
);
|
||||
assert_eq!(counting.count(&id), 1);
|
||||
assert!(
|
||||
service
|
||||
.refresh_due()
|
||||
@@ -602,7 +617,7 @@ async fn ambiguous_refresh_uses_runtime_cooldown_without_persisting_account_stat
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn pre_send_transient_failure_applies_bounded_cooldown() {
|
||||
async fn pre_send_transient_failure_applies_bounded_oauth_backoff() {
|
||||
let input = due_input("transient");
|
||||
let id = input.account_id.clone();
|
||||
let (store, _, _, service) = fixture(input, [Err(GrokRefreshFailure::Transient)], true).await;
|
||||
@@ -672,6 +687,121 @@ async fn refresh_backoff_grows_exponentially_across_attempts() {
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn exhausted_oauth_backoff_continues_after_access_token_expiry() {
|
||||
let input = due_input("backoff-recovery");
|
||||
let id = input.account_id.clone();
|
||||
let counting = Arc::new(CountingCredentialState::default());
|
||||
let (store, _, refresher, service) = fixture_many_with_state(
|
||||
[input],
|
||||
[
|
||||
Err(GrokRefreshFailure::Transient),
|
||||
Err(GrokRefreshFailure::Transient),
|
||||
Err(GrokRefreshFailure::Transient),
|
||||
Err(GrokRefreshFailure::Transient),
|
||||
Err(GrokRefreshFailure::Transient),
|
||||
Err(GrokRefreshFailure::Transient),
|
||||
],
|
||||
true,
|
||||
counting.clone(),
|
||||
)
|
||||
.await;
|
||||
set_token_deadlines(
|
||||
&store,
|
||||
&id,
|
||||
(Utc::now() - chrono::Duration::minutes(1)).into(),
|
||||
SystemTime::now() - Duration::from_secs(1),
|
||||
)
|
||||
.await;
|
||||
|
||||
for _ in 0..OAUTH_BACKOFF_ATTEMPTS {
|
||||
service.refresh_due().await.expect("backoff attempt");
|
||||
force_due(&store, &id).await;
|
||||
}
|
||||
|
||||
let observed_at = SystemTime::now();
|
||||
service.refresh_due().await.expect("recovery attempt");
|
||||
let recovery_delay = store
|
||||
.account(&id)
|
||||
.expect("account after recovery defer")
|
||||
.next_refresh_at()
|
||||
.expect("recovery scheduled")
|
||||
.duration_since(observed_at)
|
||||
.expect("recovery is in the future");
|
||||
assert!(
|
||||
(Duration::from_secs(8 * 60)..=Duration::from_secs(12 * 60 + 1)).contains(&recovery_delay),
|
||||
"recovery delay should be about ten minutes, got {recovery_delay:?}"
|
||||
);
|
||||
assert_eq!(counting.count(&id), OAUTH_BACKOFF_ATTEMPTS + 1);
|
||||
assert_eq!(refresher.prepare_calls.load(Ordering::SeqCst), 6);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn oauth_backoff_never_passes_an_unexpired_access_token_expiry() {
|
||||
let mut input = due_input("backoff-expiry-cap");
|
||||
input.account.access_token_expires_at = Utc::now() + chrono::Duration::seconds(30);
|
||||
let access_expires_at: SystemTime = input.account.access_token_expires_at.into();
|
||||
let id = input.account_id.clone();
|
||||
let counting = Arc::new(CountingCredentialState::default());
|
||||
let (store, _, _, service) = fixture_many_with_state(
|
||||
[input],
|
||||
[
|
||||
Err(GrokRefreshFailure::Transient),
|
||||
Err(GrokRefreshFailure::Transient),
|
||||
Err(GrokRefreshFailure::Transient),
|
||||
Err(GrokRefreshFailure::Transient),
|
||||
Err(GrokRefreshFailure::Transient),
|
||||
Err(GrokRefreshFailure::Transient),
|
||||
],
|
||||
true,
|
||||
counting.clone(),
|
||||
)
|
||||
.await;
|
||||
|
||||
for attempt in 0..=OAUTH_BACKOFF_ATTEMPTS {
|
||||
service.refresh_due().await.expect("backoff attempt");
|
||||
if attempt < OAUTH_BACKOFF_ATTEMPTS {
|
||||
force_due(&store, &id).await;
|
||||
}
|
||||
}
|
||||
|
||||
assert_eq!(
|
||||
store
|
||||
.account(&id)
|
||||
.expect("account after capped retry")
|
||||
.next_refresh_at(),
|
||||
Some(access_expires_at)
|
||||
);
|
||||
assert_eq!(counting.count(&id), OAUTH_BACKOFF_ATTEMPTS + 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn recovery_window_exhaustion_marks_account_expired_without_token_exchange() {
|
||||
let input = due_input("recovery-window-exhausted");
|
||||
let id = input.account_id.clone();
|
||||
let (store, _, refresher, service) = fixture(input, [Ok(success_tokens(None))], true).await;
|
||||
set_token_deadlines(
|
||||
&store,
|
||||
&id,
|
||||
(Utc::now() - chrono::Duration::hours(25)).into(),
|
||||
(Utc::now() + chrono::Duration::hours(12)).into(),
|
||||
)
|
||||
.await;
|
||||
|
||||
let outcomes = service.refresh_due().await.expect("refresh due");
|
||||
|
||||
assert!(matches!(
|
||||
outcomes.as_slice(),
|
||||
[GrokCredentialRefreshOutcome::Invalidated { account_id }] if account_id == &id
|
||||
));
|
||||
assert_eq!(
|
||||
store.account(&id).expect("expired account").availability(),
|
||||
AccountAvailability::Expired
|
||||
);
|
||||
assert_eq!(refresher.prepare_calls.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(refresher.responses.lock().expect("queue").len(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn successful_refresh_clears_backoff_counter() {
|
||||
let input = due_input("backoff-clear");
|
||||
@@ -776,52 +906,33 @@ async fn malformed_account_refresh_does_not_stop_later_accounts() {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn runtime_cooldown_survives_credential_rotation_and_blocks_refresh() {
|
||||
// cooldown 写于 revision N,轮换到 N+1 后 refresh worker 不得清除
|
||||
// 或失效它;cooldown 活跃期间账号被跳过刷新。
|
||||
let input = create_input(
|
||||
"cooldown-survives-rotation",
|
||||
"subject-cooldown-survives-rotation",
|
||||
);
|
||||
async fn scheduled_oauth_refresh_runs_after_credential_rotation() {
|
||||
let input = due_input("refresh-after-rotation");
|
||||
let id = input.account_id.clone();
|
||||
let cooldowns = Arc::new(MemoryCooldownPort::default());
|
||||
let (store, _, _, service, cooldowns) = fixture_with_cooldowns(
|
||||
let (store, _, refresher, service) = fixture(
|
||||
input,
|
||||
[Ok(success_tokens(Some("rotated-refresh")))],
|
||||
[
|
||||
Ok(success_tokens(Some("recovered-refresh"))),
|
||||
Ok(success_tokens(Some("scheduled-refresh"))),
|
||||
],
|
||||
true,
|
||||
cooldowns,
|
||||
)
|
||||
.await;
|
||||
// 先写 account cooldown(revision 1,未来到期)。
|
||||
let until = SystemTime::now() + Duration::from_secs(3600);
|
||||
cooldowns
|
||||
.put_if_later(ProviderCooldown::new(
|
||||
id.clone(),
|
||||
CredentialRevision::new(1).expect("revision"),
|
||||
until,
|
||||
))
|
||||
.await
|
||||
.expect("write cooldown");
|
||||
|
||||
// 轮换到 revision 2。
|
||||
let outcome = service
|
||||
.recover_unauthorized(&id, CredentialRevision::new(1).expect("revision"))
|
||||
.await;
|
||||
assert_eq!(outcome, GrokCredentialRecoveryOutcome::Recovered);
|
||||
assert_eq!(store.account(&id).expect("account").revision().get(), 2);
|
||||
|
||||
// cooldown 仍有效(轮换不清除)。
|
||||
assert!(
|
||||
cooldowns
|
||||
.cooldown(&id)
|
||||
.is_some_and(|cooldown| cooldown.until() > SystemTime::now()),
|
||||
"rotation must not clear account runtime cooldown"
|
||||
);
|
||||
// cooldown 活跃期间 refresh worker 跳过该账号(不清除、不刷新)。
|
||||
force_due(&store, &id).await;
|
||||
let outcomes = service.refresh_due().await.expect("refresh due");
|
||||
assert!(
|
||||
outcomes.is_empty(),
|
||||
"cooldown-active account must be skipped by refresh worker"
|
||||
);
|
||||
assert_eq!(store.account(&id).expect("account").revision().get(), 2);
|
||||
assert!(matches!(
|
||||
outcomes.as_slice(),
|
||||
[GrokCredentialRefreshOutcome::Refreshed {
|
||||
account_id,
|
||||
credential_revision,
|
||||
}] if account_id == &id && credential_revision.get() == 4
|
||||
));
|
||||
assert_eq!(refresher.prepare_calls.load(Ordering::SeqCst), 2);
|
||||
}
|
||||
|
||||
@@ -194,6 +194,7 @@ struct StoredAccount {
|
||||
account: ProviderAccount,
|
||||
credential: PlaintextCredential,
|
||||
quota: Option<QuotaObservation>,
|
||||
last_error_message: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
@@ -214,6 +215,12 @@ impl MemoryProviderAccountStore {
|
||||
.map(|stored| stored.account.clone())
|
||||
}
|
||||
|
||||
pub fn last_error_message(&self, id: &ProviderAccountId) -> Option<String> {
|
||||
lock(&self.accounts)
|
||||
.get(id)
|
||||
.and_then(|stored| stored.last_error_message.clone())
|
||||
}
|
||||
|
||||
pub fn credential(&self, id: &ProviderAccountId) -> Option<PlaintextCredential> {
|
||||
lock(&self.accounts)
|
||||
.get(id)
|
||||
@@ -246,6 +253,7 @@ impl ProviderAccountStore for MemoryProviderAccountStore {
|
||||
account: account.account,
|
||||
credential: account.credential,
|
||||
quota: None,
|
||||
last_error_message: None,
|
||||
},
|
||||
);
|
||||
Ok(())
|
||||
@@ -388,6 +396,11 @@ impl ProviderAccountStore for MemoryProviderAccountStore {
|
||||
plan_type: stored.account.plan_type().map(str::to_owned),
|
||||
},
|
||||
);
|
||||
if change.availability == AccountAvailability::Ready {
|
||||
stored.last_error_message = None;
|
||||
} else if let Some(message) = change.message {
|
||||
stored.last_error_message = Some(message);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -116,7 +116,7 @@ Provider 独占以下职责:
|
||||
- OpenAI 是逐字节透明代理:请求 body 的未知字段和字段顺序保持不变(受控模型映射除外),上游 wire 是客户端可见的事实来源,HTTP SSE 帧以原始字节原样下发,WebSocket 上游 JSON 文本原样下发;response ID 以 opaque bytes 存储,不假设 UUID 或固定长度。canonical facts 只从同一帧旁路解析,用于观测、亲和与计费。仅网关内部帧(rate limits、metadata)被就地消费,不向客户端转发。
|
||||
- xAI 是翻译层:把 Grok wire 解码后重新合成 Responses wire 事件,不承诺字节透明;上游错误按结构化 message/code/type 透出,message 中的 UUID 账号指纹替换为占位符。
|
||||
|
||||
credential 更新使用 `credential_revision` CAS。认证永久失败、封禁、额度耗尽和带截止时间的 cooldown 是账号运行时事实。cooldown 到期后,Core 的有效调度谓词会自然允许账号重新参与选择;Redis cooldown 只是热缓存。成功调用或成功额度观测才可把账号状态重新观测为 `ready`。自动 refresh 的瞬态失败(限流、5xx、超时、畸形响应)保留现有凭据并按指数退避推迟下一次刷新,不把账号标记为失效;只有上游明确的永久错误(`invalid_grant`、封禁)才写入终态失效。
|
||||
credential 更新使用 `credential_revision` CAS。认证永久失败、封禁、额度耗尽和带截止时间的 cooldown 是账号运行时事实。cooldown 到期后,Core 的有效调度谓词会自然允许账号重新参与选择;Redis cooldown 只是热缓存。成功调用或成功额度观测才可把账号状态重新观测为 `ready`。推理请求的账号/model cooldown 只约束数据面调度,不参与 OAuth refresh;OpenAI 与 xAI 的 OAuth refresh 请求状态不明确时只按 OAuth 自身的有界退避推进该 credential 的 `next_refresh_at`:前五次约为 5/15/45/135/300 秒,之后约每 10 分钟恢复;仍有效 AT 时不晚于其到期时刻,AT 到期后仍按该节奏恢复,至 `AT 到期 + 24h` 才终态化。自动 refresh 的瞬态失败(限流、5xx、超时、畸形响应)保留现有凭据并按指数退避推迟下一次刷新,不把账号标记为失效;上游明确的永久错误(`invalid_grant`、封禁)仍立即写入终态失效。
|
||||
|
||||
`refresh_token_expires_at` 不是公共 SQL 列或 Core 权威状态。xAI 可在 `provider_credentials_json` 内保存它作为 Provider 私有提示;真正失效以 refresh endpoint 返回的永久错误为准。
|
||||
|
||||
|
||||
@@ -1 +1 @@
|
||||
version: "3.1.9"
|
||||
version: "3.1.10"
|
||||
|
||||
Reference in New Issue
Block a user