diff --git a/codex-rs/cloud-config/src/cache.rs b/codex-rs/cloud-config/src/cache.rs index 3ecaf6d905..9b306ce4c2 100644 --- a/codex-rs/cloud-config/src/cache.rs +++ b/codex-rs/cloud-config/src/cache.rs @@ -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)] diff --git a/codex-rs/cloud-config/src/cache_tests.rs b/codex-rs/cloud-config/src/cache_tests.rs index 2616077479..609f67baf2 100644 --- a/codex-rs/cloud-config/src/cache_tests.rs +++ b/codex-rs/cloud-config/src/cache_tests.rs @@ -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] diff --git a/codex-rs/cloud-config/src/service.rs b/codex-rs/cloud-config/src/service.rs index aebbe33f20..31288dd288 100644 --- a/codex-rs/cloud-config/src/service.rs +++ b/codex-rs/cloud-config/src/service.rs @@ -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 { + 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 { + 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 { 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 { + 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)); diff --git a/codex-rs/cloud-config/src/service_tests.rs b/codex-rs/cloud-config/src/service_tests.rs index 4f09ecccf3..a9d363275b 100644 --- a/codex-rs/cloud-config/src/service_tests.rs +++ b/codex-rs/cloud-config/src/service_tests.rs @@ -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 { 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!(