openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-10 20:26:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add cloud config cache fallback

Browse Source
This commit is contained in:
Joe Florencio
2026-07-09 13:03:29 -07:00
parent 086b803120
commit 8cafe7639e
4 changed files with 478 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
codex-rs/cloud-config/src/cache.rs
View File

@@ -1,7 +1,8 @@
//! Signed on-disk cache for cloud config bundles.
//!
//! The cache is scoped to the authenticated ChatGPT user and account, has a
//! short TTL, and is HMAC-signed so malformed or edited files fail closed.
//! The cache is scoped to the authenticated ChatGPT user and account. Entries
//! refresh after a short interval but remain eligible as a bounded fallback,
//! and are HMAC-signed so malformed or edited files fail closed.
use base64::Engine;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
@@ -24,7 +25,9 @@ use tokio::fs;
const CLOUD_CONFIG_BUNDLE_CACHE_VERSION: u32 = 1;
pub(super) const CLOUD_CONFIG_BUNDLE_CACHE_FILENAME: &str = "cloud-config-bundle-cache.json";
pub(super) const CLOUD_CONFIG_BUNDLE_CACHE_LOCK_FILENAME: &str = "cloud-config-bundle-cache.lock";
pub(super) const CLOUD_CONFIG_BUNDLE_CACHE_TTL: Duration = Duration::from_secs(15 * 60);
pub(super) const CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL: Duration =
Duration::from_secs(15 * 60);
pub(super) const CLOUD_CONFIG_BUNDLE_CACHE_HARD_TTL: Duration = Duration::from_secs(24 * 60 * 60);
const CLOUD_CONFIG_BUNDLE_CACHE_WRITE_HMAC_KEY: &[u8] =
b"codex-cloud-config-bundle-cache-v1-6160ae70-bcfd-4ca8-a99b-40f73b3b072e";
const CLOUD_CONFIG_BUNDLE_CACHE_READ_HMAC_KEYS: &[&[u8]] =
@@ -141,7 +144,8 @@ impl CloudConfigBundleCache {
.signed_duration_since(cache_file.signed_payload.cached_at)
.to_std()
.map_err(|_| CacheLoadStatus::CacheExpired)?;
if cache_file.signed_payload.expires_at <= now || cache_age >= CLOUD_CONFIG_BUNDLE_CACHE_TTL
if cache_file.signed_payload.expires_at <= now
|| cache_age >= CLOUD_CONFIG_BUNDLE_CACHE_HARD_TTL
{
return Err(CacheLoadStatus::CacheExpired);
}
@@ -152,13 +156,19 @@ impl CloudConfigBundleCache {
.signed_duration_since(now)
.to_std()
.map_err(|_| CacheLoadStatus::CacheExpired)?;
let ttl_remaining = CLOUD_CONFIG_BUNDLE_CACHE_TTL
let hard_ttl_remaining = CLOUD_CONFIG_BUNDLE_CACHE_HARD_TTL
.checked_sub(cache_age)
.ok_or(CacheLoadStatus::CacheExpired)?;
let freshness = match CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL.checked_sub(cache_age) {
Some(refresh_in) => CacheFreshness::Fresh {
refresh_in: refresh_in.min(expires_in).min(hard_ttl_remaining),
},
None => CacheFreshness::Stale,
};
Ok(LoadedCloudConfigBundleCache {
signed_payload: cache_file.signed_payload,
refresh_in: expires_in.min(ttl_remaining),
freshness,
})
}
@@ -190,7 +200,7 @@ impl CloudConfigBundleCache {
let now = Utc::now();
let expires_at = now
.checked_add_signed(
ChronoDuration::from_std(CLOUD_CONFIG_BUNDLE_CACHE_TTL)
ChronoDuration::from_std(CLOUD_CONFIG_BUNDLE_CACHE_HARD_TTL)
.map_err(|_| CloudConfigBundleCacheError)?,
)
.ok_or(CloudConfigBundleCacheError)?;
@@ -219,7 +229,13 @@ impl CloudConfigBundleCache {
fs::write(&self.path, serialized)
.await
.map_err(|_| CloudConfigBundleCacheError)?;
expires_at
let refresh_at = now
.checked_add_signed(
ChronoDuration::from_std(CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL)
.map_err(|_| CloudConfigBundleCacheError)?,
)
.ok_or(CloudConfigBundleCacheError)?;
refresh_at
.signed_duration_since(Utc::now())
.to_std()
.map_err(|_| CloudConfigBundleCacheError)
@@ -257,7 +273,15 @@ pub(super) struct CloudConfigBundleCacheError;
#[derive(Clone, Debug, Eq, PartialEq)]
pub(super) struct LoadedCloudConfigBundleCache {
pub(super) signed_payload: CloudConfigBundleCacheSignedPayload,
pub(super) refresh_in: Duration,
pub(super) freshness: CacheFreshness,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub(super) enum CacheFreshness {
/// The entry may be used immediately and refreshed after this delay.
Fresh { refresh_in: Duration },
/// The entry must be refreshed, but remains eligible as a failure fallback.
Stale,
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]

39
codex-rs/cloud-config/src/cache_tests.rs
View File

@@ -44,7 +44,7 @@ fn valid_signed_payload() -> CloudConfigBundleCacheSignedPayload {
CloudConfigBundleCacheSignedPayload {
version: CLOUD_CONFIG_BUNDLE_CACHE_VERSION,
cached_at,
expires_at: cached_at + ChronoDuration::minutes(15),
expires_at: cached_at + ChronoDuration::hours(24),
chatgpt_user_id: Some("user-12345".to_string()),
account_id: Some("account-12345".to_string()),
bundle: test_bundle(),
@@ -83,7 +83,7 @@ async fn save_writes_signed_payload_and_loads_for_matching_identity() {
.expect("parse cache");
assert!(
cache_file.signed_payload.expires_at
<= cache_file.signed_payload.cached_at + ChronoDuration::minutes(15)
<= cache_file.signed_payload.cached_at + ChronoDuration::hours(24)
);
assert!(cache_file.signed_payload.expires_at > cache_file.signed_payload.cached_at);
assert_eq!(
@@ -103,7 +103,11 @@ async fn save_writes_signed_payload_and_loads_for_matching_identity() {
.await
.expect("load cache");
assert_eq!(loaded_cache.signed_payload, cache_file.signed_payload);
assert!(loaded_cache.refresh_in <= CLOUD_CONFIG_BUNDLE_CACHE_TTL);
assert!(matches!(
loaded_cache.freshness,
CacheFreshness::Fresh { refresh_in }
if refresh_in <= CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL
));
}
#[tokio::test]
@@ -196,13 +200,29 @@ async fn load_rejects_expired_cache() {
}
#[tokio::test]
async fn load_rejects_cache_older_than_ttl_even_when_expiry_is_later() {
async fn load_returns_stale_cache_after_refresh_interval() {
let codex_home = tempdir().expect("tempdir");
let cache = create_test_cache(codex_home.path());
let mut signed_payload = valid_signed_payload();
signed_payload.cached_at =
Utc::now() - ChronoDuration::minutes(15) - ChronoDuration::seconds(1);
signed_payload.expires_at = Utc::now() + ChronoDuration::minutes(15);
signed_payload.expires_at = Utc::now() + ChronoDuration::hours(23);
write_cache_file(&cache, &signed_cache_file(signed_payload));
let loaded_cache = cache
.load(Some("user-12345"), Some("account-12345"))
.await
.expect("load stale cache");
assert_eq!(loaded_cache.freshness, CacheFreshness::Stale);
}
#[tokio::test]
async fn load_rejects_cache_older_than_hard_ttl_even_when_expiry_is_later() {
let codex_home = tempdir().expect("tempdir");
let cache = create_test_cache(codex_home.path());
let mut signed_payload = valid_signed_payload();
signed_payload.cached_at = Utc::now() - ChronoDuration::hours(24) - ChronoDuration::seconds(1);
signed_payload.expires_at = Utc::now() + ChronoDuration::hours(1);
write_cache_file(&cache, &signed_cache_file(signed_payload));
assert_eq!(
@@ -212,7 +232,7 @@ async fn load_rejects_cache_older_than_ttl_even_when_expiry_is_later() {
}
#[tokio::test]
async fn load_uses_ttl_cap_for_refresh_delay() {
async fn load_uses_refresh_interval_cap_for_refresh_delay() {
let codex_home = tempdir().expect("tempdir");
let cache = create_test_cache(codex_home.path());
let now = Utc::now();
@@ -226,8 +246,11 @@ async fn load_uses_ttl_cap_for_refresh_delay() {
.await
.expect("load cache");
assert!(loaded_cache.refresh_in <= Duration::from_secs(10 * 60));
assert!(loaded_cache.refresh_in >= Duration::from_secs(10 * 60 - 5));
let CacheFreshness::Fresh { refresh_in } = loaded_cache.freshness else {
panic!("cache should be fresh");
};
assert!(refresh_in <= Duration::from_secs(10 * 60));
assert!(refresh_in >= Duration::from_secs(10 * 60 - 5));
}
#[tokio::test]

155
codex-rs/cloud-config/src/service.rs
View File

@@ -6,6 +6,7 @@
use crate::backend::BundleClient;
use crate::backend::BundleRequestError;
use crate::backend::RetryableFailureKind;
use crate::cache::CacheFreshness;
use crate::cache::CacheLoadStatus;
use crate::cache::CacheLockAttempt;
use crate::cache::CloudConfigBundleCache;
@@ -84,6 +85,18 @@ enum CacheRefreshSchedule {
ContinueAfter(Duration),
}
enum CachedBundle {
Fresh(LoadedBundle),
Fallback(LoadedBundle),
Miss,
}
#[derive(Clone, Copy)]
enum StaleCachePolicy {
FallbackOnError,
RefreshRequired,
}
enum UnauthorizedRecoveryAction {
RetrySameAttempt,
RetryNextAttempt,
@@ -123,7 +136,7 @@ where
let _timer =
codex_otel::start_global_timer("codex.cloud_config_bundle.fetch.duration_ms", &[]);
let started_at = Instant::now();
let load_result = timeout(self.timeout, async {
let load_result = match timeout(self.timeout, async {
let Some(auth) = self.auth_manager.auth().await else {
return Ok(StartupLoad::Inactive);
};
@@ -131,29 +144,35 @@ where
return Ok(StartupLoad::Inactive);
}
self.load_bundle(auth, "startup")
self.load_bundle(auth, "startup", StaleCachePolicy::FallbackOnError)
.await
.map(StartupLoad::Active)
})
.await
.inspect_err(|_| {
let message = format!(
"Timed out waiting for cloud config bundle after {}s",
self.timeout.as_secs()
);
tracing::error!("{message}");
emit_load_metric("startup", "error", /*bundle*/ None);
})
.map_err(|_| {
CloudConfigBundleLoadError::new(
CloudConfigBundleLoadErrorCode::Timeout,
/*status_code*/ None,
format!(
"timed out waiting for cloud config bundle after {}s",
self.timeout.as_secs()
),
)
})?;
{
Ok(load_result) => load_result,
Err(_) => {
let fallback = self.load_cached_fallback_for_current_auth().await;
if let Some(loaded) = fallback {
tracing::warn!(
path = %self.cache.path().display(),
"Timed out refreshing cloud config bundle; using cached fallback"
);
Ok(StartupLoad::Active(loaded))
} else {
let message = format!(
"timed out waiting for cloud config bundle after {}s",
self.timeout.as_secs()
);
tracing::error!("{message}");
Err(CloudConfigBundleLoadError::new(
CloudConfigBundleLoadErrorCode::Timeout,
/*status_code*/ None,
message,
))
}
}
};
let result = match load_result {
Ok(result) => result,
@@ -188,7 +207,7 @@ where
Ok(result)
}
async fn load_valid_cached_bundle(&self, auth: &CodexAuth) -> Option<LoadedBundle> {
async fn load_cached_bundle(&self, auth: &CodexAuth) -> CachedBundle {
let (chatgpt_user_id, account_id) = auth_identity(auth);
match self
.cache
@@ -206,33 +225,55 @@ where
);
self.cache
.log_load_status(&CacheLoadStatus::CacheInvalidBundle);
None
CachedBundle::Miss
} else {
tracing::info!(
path = %self.cache.path().display(),
"Using cached cloud config bundle"
);
Some(LoadedBundle {
bundle: optional_bundle(loaded_cache.signed_payload.bundle),
refresh_in: loaded_cache.refresh_in,
})
let bundle = optional_bundle(loaded_cache.signed_payload.bundle);
match loaded_cache.freshness {
CacheFreshness::Fresh { refresh_in } => {
tracing::info!(
path = %self.cache.path().display(),
"Using cached cloud config bundle"
);
CachedBundle::Fresh(LoadedBundle { bundle, refresh_in })
}
CacheFreshness::Stale => CachedBundle::Fallback(LoadedBundle {
bundle,
refresh_in: CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_RETRY_INTERVAL,
}),
}
}
}
Err(cache_load_status) => {
self.cache.log_load_status(&cache_load_status);
None
CachedBundle::Miss
}
}
}
async fn load_cached_fallback_for_current_auth(&self) -> Option<LoadedBundle> {
let auth = self.auth_manager.auth_cached()?;
if !cloud_config_eligible_auth(&auth) {
return None;
}
match self.load_cached_bundle(&auth).await {
CachedBundle::Fresh(loaded) | CachedBundle::Fallback(loaded) => Some(loaded),
CachedBundle::Miss => None,
}
}
async fn load_bundle(
&self,
auth: CodexAuth,
trigger: &'static str,
stale_cache_policy: StaleCachePolicy,
) -> Result<LoadedBundle, CloudConfigBundleLoadError> {
loop {
if let Some(loaded) = self.load_valid_cached_bundle(&auth).await {
return Ok(loaded);
// Fresh cache entries satisfy the load immediately. A soft-stale
// entry continues through coordination and a blocking fetch; only
// startup may use it after that refresh fails.
match self.load_cached_bundle(&auth).await {
CachedBundle::Fresh(loaded) => return Ok(loaded),
CachedBundle::Fallback(_) | CachedBundle::Miss => {}
}
// This is a cross-process single-flight lock, not a cache-file
@@ -241,11 +282,12 @@ where
match self.cache.try_acquire_lock().await {
Ok(CacheLockAttempt::Acquired(_cache_lock)) => {
// Close the race between the cache read and lock acquisition.
if let Some(loaded) = self.load_valid_cached_bundle(&auth).await {
return Ok(loaded);
match self.load_cached_bundle(&auth).await {
CachedBundle::Fresh(loaded) => return Ok(loaded),
CachedBundle::Fallback(_) | CachedBundle::Miss => {}
}
return self
.fetch_remote_bundle_and_update_cache_with_retries(auth, trigger)
.fetch_remote_bundle_with_fallback(auth, trigger, stale_cache_policy)
.await;
}
Ok(CacheLockAttempt::Contended) => {
@@ -258,13 +300,47 @@ where
"Failed to acquire cloud config bundle cache lock; fetching without coordination"
);
return self
.fetch_remote_bundle_and_update_cache_with_retries(auth, trigger)
.fetch_remote_bundle_with_fallback(auth, trigger, stale_cache_policy)
.await;
}
}
}
}
async fn fetch_remote_bundle_with_fallback(
&self,
auth: CodexAuth,
trigger: &'static str,
stale_cache_policy: StaleCachePolicy,
) -> Result<LoadedBundle, CloudConfigBundleLoadError> {
match self
.fetch_remote_bundle_and_update_cache_with_retries(auth, trigger)
.await
{
Ok(loaded) => Ok(loaded),
Err(err)
if matches!(stale_cache_policy, StaleCachePolicy::FallbackOnError)
&& err.code() != CloudConfigBundleLoadErrorCode::Auth =>
{
// Auth recovery may have changed identities during the fetch.
// Re-read the cache against the current identity before using it.
let fallback = self.load_cached_fallback_for_current_auth().await;
match fallback {
Some(loaded) => {
tracing::warn!(
path = %self.cache.path().display(),
error = %err,
"Failed to refresh cloud config bundle; using cached fallback"
);
Ok(loaded)
}
None => Err(err),
}
}
Err(err) => Err(err),
}
}
async fn fetch_remote_bundle_and_update_cache_with_retries(
&self,
mut auth: CodexAuth,
@@ -547,7 +623,10 @@ where
return CacheRefreshSchedule::Stop;
}
match self.load_bundle(auth, "refresh").await {
match self
.load_bundle(auth, "refresh", StaleCachePolicy::RefreshRequired)
.await
{
Ok(loaded) => {
emit_load_metric("refresh", "success", loaded.bundle.as_ref());
publisher.publish(Ok(loaded.bundle));

334
codex-rs/cloud-config/src/service_tests.rs
View File

@@ -4,8 +4,9 @@ use crate::backend::BundleRequestError;
use crate::backend::RetryableFailureKind;
use crate::backend::bundle_from_response;
use crate::cache::CLOUD_CONFIG_BUNDLE_CACHE_FILENAME;
use crate::cache::CLOUD_CONFIG_BUNDLE_CACHE_HARD_TTL;
use crate::cache::CLOUD_CONFIG_BUNDLE_CACHE_LOCK_FILENAME;
use crate::cache::CLOUD_CONFIG_BUNDLE_CACHE_TTL;
use crate::cache::CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL;
use crate::cache::CloudConfigBundleCache;
use crate::cache::CloudConfigBundleCacheFile;
use crate::cache::cache_payload_bytes;
@@ -63,6 +64,27 @@ fn shift_cache_timestamps(cache: &CloudConfigBundleCache, offset: chrono::Durati
.expect("write cache");
}
async fn save_test_cache_with_age(
codex_home: &Path,
bundle: CloudConfigBundle,
age: Duration,
) -> CloudConfigBundleCache {
let cache = create_test_cache(codex_home);
cache
.save(
Some("user-12345".to_string()),
Some("account-12345".to_string()),
bundle,
)
.await
.expect("save cache");
shift_cache_timestamps(
&cache,
-chrono::Duration::from_std(age).expect("cache age should fit chrono duration"),
);
cache
}
async fn auth_manager_with_api_key() -> Arc<AuthManager> {
let tmp = tempdir().expect("tempdir");
let auth_json = json!({
@@ -533,7 +555,10 @@ async fn get_bundle_allows_agent_identity_business_plan() {
CLOUD_CONFIG_BUNDLE_TIMEOUT,
);
assert_eq!(service.load_startup_bundle().await, Ok(Some(bundle)));
assert_eq!(
expect_active(service.load_startup_bundle().await).bundle,
Some(bundle)
);
assert_eq!(fetcher.request_count.load(Ordering::SeqCst), 1);
assert!(
codex_home
@@ -684,23 +709,27 @@ async fn get_bundle_fetches_and_caches_when_cache_lock_fails() {
}
#[tokio::test]
async fn get_bundle_refetches_cache_older_than_ttl() {
let bundle = test_bundle();
async fn get_bundle_refetches_cache_older_than_refresh_interval() {
let cached_bundle = test_bundle();
let replacement_bundle = CloudConfigBundle {
config_toml: CloudConfigTomlBundle::default(),
requirements_toml: CloudRequirementsTomlBundle {
enterprise_managed: vec![CloudRequirementsFragment {
id: "req_2".to_string(),
name: "Replacement requirements".to_string(),
contents: "allowed_approval_policies = [\"on-request\"]".to_string(),
}],
},
};
let codex_home = tempdir().expect("tempdir");
let prime_service = CloudConfigBundleService::new(
auth_manager_with_plan("business").await,
Arc::new(StaticBundleClient::new(bundle.clone())),
codex_home.path().to_path_buf(),
CLOUD_CONFIG_BUNDLE_TIMEOUT,
);
expect_active(prime_service.load_startup_bundle().await);
shift_cache_timestamps(
&create_test_cache(codex_home.path()),
-chrono::Duration::from_std(CLOUD_CONFIG_BUNDLE_CACHE_TTL + Duration::from_secs(1))
.expect("cache age should fit chrono duration"),
);
save_test_cache_with_age(
codex_home.path(),
cached_bundle,
CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL + Duration::from_secs(1),
)
.await;
let fetcher = Arc::new(StaticBundleClient::new(bundle.clone()));
let fetcher = Arc::new(StaticBundleClient::new(replacement_bundle.clone()));
let service = CloudConfigBundleService::new(
auth_manager_with_plan("business").await,
fetcher.clone(),
@@ -710,11 +739,112 @@ async fn get_bundle_refetches_cache_older_than_ttl() {
assert_eq!(
expect_active(service.load_startup_bundle().await).bundle,
Some(bundle)
Some(replacement_bundle)
);
assert_eq!(fetcher.request_count.load(Ordering::SeqCst), 1);
}
#[tokio::test(start_paused = true)]
async fn startup_uses_stale_cache_when_refresh_retries_are_exhausted() {
let cached_bundle = test_bundle();
let codex_home = tempdir().expect("tempdir");
save_test_cache_with_age(
codex_home.path(),
cached_bundle.clone(),
CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL + Duration::from_secs(1),
)
.await;
let fetcher = Arc::new(SequenceBundleClient::new(vec![
Err(request_error());
CLOUD_CONFIG_BUNDLE_MAX_ATTEMPTS
]));
let service = CloudConfigBundleService::new(
auth_manager_with_plan("business").await,
fetcher.clone(),
codex_home.path().to_path_buf(),
CLOUD_CONFIG_BUNDLE_TIMEOUT,
);
let handle = tokio::spawn(async move { service.load_startup_bundle().await });
tokio::task::yield_now().await;
tokio::time::advance(Duration::from_secs(5)).await;
tokio::task::yield_now().await;
assert_eq!(
handle.await.expect("startup task"),
Ok(StartupLoad::Active(LoadedBundle {
bundle: Some(cached_bundle),
refresh_in: CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_RETRY_INTERVAL,
}))
);
assert_eq!(
fetcher.request_count.load(Ordering::SeqCst),
CLOUD_CONFIG_BUNDLE_MAX_ATTEMPTS
);
}
#[tokio::test(start_paused = true)]
async fn startup_uses_stale_cache_when_refresh_times_out() {
let cached_bundle = test_bundle();
let codex_home = tempdir().expect("tempdir");
save_test_cache_with_age(
codex_home.path(),
cached_bundle.clone(),
CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL + Duration::from_secs(1),
)
.await;
let service = CloudConfigBundleService::new(
auth_manager_with_plan("business").await,
Arc::new(PendingBundleClient),
codex_home.path().to_path_buf(),
CLOUD_CONFIG_BUNDLE_TIMEOUT,
);
let handle = tokio::spawn(async move { service.load_startup_bundle().await });
tokio::task::yield_now().await;
tokio::time::advance(CLOUD_CONFIG_BUNDLE_TIMEOUT + Duration::from_millis(1)).await;
assert_eq!(
handle.await.expect("startup task"),
Ok(StartupLoad::Active(LoadedBundle {
bundle: Some(cached_bundle),
refresh_in: CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_RETRY_INTERVAL,
}))
);
}
#[tokio::test(start_paused = true)]
async fn startup_does_not_use_cache_past_hard_ttl() {
let codex_home = tempdir().expect("tempdir");
save_test_cache_with_age(
codex_home.path(),
test_bundle(),
CLOUD_CONFIG_BUNDLE_CACHE_HARD_TTL + Duration::from_secs(1),
)
.await;
let fetcher = Arc::new(SequenceBundleClient::new(vec![
Err(request_error());
CLOUD_CONFIG_BUNDLE_MAX_ATTEMPTS
]));
let service = CloudConfigBundleService::new(
auth_manager_with_plan("business").await,
fetcher,
codex_home.path().to_path_buf(),
CLOUD_CONFIG_BUNDLE_TIMEOUT,
);
let handle = tokio::spawn(async move { service.load_startup_bundle().await });
tokio::task::yield_now().await;
tokio::time::advance(Duration::from_secs(5)).await;
tokio::task::yield_now().await;
let err = handle
.await
.expect("startup task")
.expect_err("hard-expired cache must not be used");
assert_eq!(err.code(), CloudConfigBundleLoadErrorCode::RequestFailed);
}
#[tokio::test]
async fn get_bundle_ignores_cache_for_different_auth_identity() {
let codex_home = tempdir().expect("tempdir");
@@ -932,8 +1062,86 @@ async fn get_bundle_recovers_after_unauthorized_reload_updates_cache_identity()
assert_eq!(fetcher.request_count.load(Ordering::SeqCst), 2);
}
#[tokio::test(start_paused = true)]
async fn request_failure_after_auth_identity_change_does_not_use_old_stale_cache() {
let auth_home = tempdir().expect("tempdir");
write_auth_json(
auth_home.path(),
chatgpt_auth_json_with_last_refresh(
"business",
Some("user-12345"),
Some("account-12345"),
"stale-access-token",
"test-refresh-token",
"3025-01-01T00:00:00Z",
),
)
.expect("write initial auth");
let auth_manager = Arc::new(
AuthManager::new(
auth_home.path().to_path_buf(),
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
/*forced_chatgpt_workspace_id*/ None,
/*chatgpt_base_url*/ None,
AuthKeyringBackendKind::default(),
/*auth_route_config*/ None,
)
.await,
);
write_auth_json(
auth_home.path(),
chatgpt_auth_json_with_last_refresh(
"business",
Some("user-99999"),
Some("account-12345"),
"fresh-access-token",
"test-refresh-token",
"3025-01-01T00:00:00Z",
),
)
.expect("write refreshed auth");
let codex_home = tempdir().expect("tempdir");
save_test_cache_with_age(
codex_home.path(),
test_bundle(),
CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL + Duration::from_secs(1),
)
.await;
let mut responses = vec![Err(BundleRequestError::Unauthorized {
status_code: Some(401),
message: "GET /config/bundle failed: 401".to_string(),
})];
responses.extend(
std::iter::repeat_with(|| Err(request_error())).take(CLOUD_CONFIG_BUNDLE_MAX_ATTEMPTS),
);
let fetcher = Arc::new(SequenceBundleClient::new(responses));
let service = CloudConfigBundleService::new(
auth_manager,
fetcher.clone(),
codex_home.path().to_path_buf(),
CLOUD_CONFIG_BUNDLE_TIMEOUT,
);
let handle = tokio::spawn(async move { service.load_startup_bundle().await });
tokio::task::yield_now().await;
tokio::time::advance(Duration::from_secs(5)).await;
tokio::task::yield_now().await;
let err = handle
.await
.expect("startup task")
.expect_err("old identity cache must not be used");
assert_eq!(err.code(), CloudConfigBundleLoadErrorCode::RequestFailed);
assert_eq!(
fetcher.request_count.load(Ordering::SeqCst),
CLOUD_CONFIG_BUNDLE_MAX_ATTEMPTS + 1
);
}
#[tokio::test]
async fn get_bundle_surfaces_auth_recovery_message() {
async fn auth_recovery_error_does_not_use_stale_cache() {
let auth_home = tempdir().expect("tempdir");
write_auth_json(
auth_home.path(),
@@ -975,6 +1183,12 @@ async fn get_bundle_surfaces_auth_recovery_message() {
request_count: AtomicUsize::new(0),
});
let codex_home = tempdir().expect("tempdir");
save_test_cache_with_age(
codex_home.path(),
test_bundle(),
CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL + Duration::from_secs(1),
)
.await;
let service = CloudConfigBundleService::new(
auth_manager,
fetcher.clone(),
@@ -1048,7 +1262,10 @@ async fn get_bundle_refreshes_external_auth_after_unauthorized() {
CLOUD_CONFIG_BUNDLE_TIMEOUT,
);
assert_eq!(service.load_startup_bundle().await, Ok(Some(test_bundle())));
assert_eq!(
expect_active(service.load_startup_bundle().await).bundle,
Some(test_bundle())
);
assert_eq!(fetcher.request_count.load(Ordering::SeqCst), 2);
assert_eq!(external_auth.refresh_count.load(Ordering::SeqCst), 1);
}
@@ -1173,7 +1390,7 @@ async fn background_refresh_stops_when_loader_is_dropped() {
let (loader, publisher) = pending_loader();
let refresh_task = tokio::spawn(async move {
service
.refresh_cache_in_background(CLOUD_CONFIG_BUNDLE_CACHE_TTL, publisher)
.refresh_cache_in_background(CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL, publisher)
.await;
});
@@ -1215,6 +1432,55 @@ async fn refresh_failure_uses_retry_interval() {
);
}
#[tokio::test(start_paused = true)]
async fn refresh_failure_keeps_newer_in_memory_bundle_instead_of_disk_fallback() {
let stale_bundle = test_bundle();
let in_memory_bundle = CloudConfigBundle {
config_toml: CloudConfigTomlBundle::default(),
requirements_toml: CloudRequirementsTomlBundle {
enterprise_managed: vec![CloudRequirementsFragment {
id: "req_2".to_string(),
name: "Newer requirements".to_string(),
contents: "allowed_approval_policies = [\"on-request\"]".to_string(),
}],
},
};
let codex_home = tempdir().expect("tempdir");
save_test_cache_with_age(
codex_home.path(),
stale_bundle,
CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL + Duration::from_secs(1),
)
.await;
let fetcher = Arc::new(SequenceBundleClient::new(vec![
Err(request_error());
CLOUD_CONFIG_BUNDLE_MAX_ATTEMPTS
]));
let service = CloudConfigBundleService::new(
auth_manager_with_plan("business").await,
fetcher.clone(),
codex_home.path().to_path_buf(),
CLOUD_CONFIG_BUNDLE_TIMEOUT,
);
let (loader, publisher) = CloudConfigBundleLoader::pending();
publisher.publish(Ok(Some(in_memory_bundle.clone())));
let refresh = tokio::spawn(async move { service.refresh_cache_once(&publisher).await });
tokio::task::yield_now().await;
tokio::time::advance(Duration::from_secs(5)).await;
tokio::task::yield_now().await;
assert_eq!(
refresh.await.expect("refresh task"),
CacheRefreshSchedule::ContinueAfter(CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_RETRY_INTERVAL)
);
assert_eq!(loader.get().await, Ok(Some(in_memory_bundle)));
assert_eq!(
fetcher.request_count.load(Ordering::SeqCst),
CLOUD_CONFIG_BUNDLE_MAX_ATTEMPTS
);
}
#[tokio::test]
async fn startup_uses_retry_interval_when_cache_write_fails() {
let codex_home = tempdir().expect("tempdir");
@@ -1307,20 +1573,12 @@ async fn concurrent_startups_make_one_remote_request() {
#[tokio::test]
async fn refresh_fetches_and_caches_when_cache_lock_fails() {
let codex_home = tempdir().expect("tempdir");
let cache = create_test_cache(codex_home.path());
cache
.save(
Some("user-12345".to_string()),
Some("account-12345".to_string()),
test_bundle(),
)
.await
.expect("save cache");
shift_cache_timestamps(
&cache,
-chrono::Duration::from_std(CLOUD_CONFIG_BUNDLE_CACHE_TTL + Duration::from_secs(1))
.expect("cache age should fit chrono duration"),
);
let cache = save_test_cache_with_age(
codex_home.path(),
test_bundle(),
CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL + Duration::from_secs(1),
)
.await;
std::fs::create_dir(
codex_home
.path()
@@ -1386,8 +1644,10 @@ async fn refresh_from_remote_updates_stale_cached_bundle() {
);
shift_cache_timestamps(
&create_test_cache(codex_home.path()),
-chrono::Duration::from_std(CLOUD_CONFIG_BUNDLE_CACHE_TTL + Duration::from_secs(1))
.expect("cache age should fit chrono duration"),
-chrono::Duration::from_std(
CLOUD_CONFIG_BUNDLE_CACHE_REFRESH_INTERVAL + Duration::from_secs(1),
)
.expect("cache age should fit chrono duration"),
);
let (loader, publisher) = pending_loader();
assert!(matches!(