mirror of
https://github.com/openai/codex.git
synced 2026-10-23 18:08:45 +00:00
fix: use stale cloud requirements cache fallback
Co-authored-by: Codex noreply@openai.com
This commit is contained in:
co-authored by
Codex noreply@openai.com
parent
0131f99fd5
commit
553bd29d67
Generated
+6
@@ -1667,6 +1667,8 @@ dependencies = [
|
||||
"anyhow",
|
||||
"assert_cmd",
|
||||
"assert_matches",
|
||||
"base64 0.22.1",
|
||||
"chrono",
|
||||
"clap",
|
||||
"clap_complete",
|
||||
"codex-app-server",
|
||||
@@ -1695,13 +1697,17 @@ dependencies = [
|
||||
"codex-utils-cargo-bin",
|
||||
"codex-utils-cli",
|
||||
"codex-utils-path",
|
||||
"codex-utils-pty",
|
||||
"codex-windows-sandbox",
|
||||
"hmac",
|
||||
"libc",
|
||||
"owo-colors",
|
||||
"predicates",
|
||||
"pretty_assertions",
|
||||
"regex-lite",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"sha2",
|
||||
"sqlx",
|
||||
"supports-color 3.0.2",
|
||||
"tempfile",
|
||||
|
||||
@@ -69,7 +69,13 @@ codex_windows_sandbox = { package = "codex-windows-sandbox", path = "../windows-
|
||||
[dev-dependencies]
|
||||
assert_cmd = { workspace = true }
|
||||
assert_matches = { workspace = true }
|
||||
base64 = { workspace = true }
|
||||
chrono = { workspace = true, features = ["serde"] }
|
||||
codex-utils-cargo-bin = { workspace = true }
|
||||
codex-utils-pty = { workspace = true }
|
||||
hmac = { workspace = true }
|
||||
predicates = { workspace = true }
|
||||
pretty_assertions = { workspace = true }
|
||||
serde = { workspace = true, features = ["derive"] }
|
||||
sha2 = { workspace = true }
|
||||
sqlx = { workspace = true }
|
||||
|
||||
@@ -0,0 +1,336 @@
|
||||
use base64::Engine;
|
||||
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
|
||||
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
|
||||
use chrono::DateTime;
|
||||
use chrono::Duration as ChronoDuration;
|
||||
use chrono::Utc;
|
||||
use hmac::Hmac;
|
||||
use hmac::Mac;
|
||||
use serde::Serialize;
|
||||
use serde_json::json;
|
||||
use sha2::Sha256;
|
||||
use std::collections::HashMap;
|
||||
use std::io::Read;
|
||||
use std::io::Write;
|
||||
use std::net::SocketAddr;
|
||||
use std::net::TcpListener;
|
||||
use std::net::TcpStream;
|
||||
use std::path::Path;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::AtomicBool;
|
||||
use std::sync::atomic::AtomicUsize;
|
||||
use std::sync::atomic::Ordering;
|
||||
use std::thread::JoinHandle;
|
||||
use std::time::Duration;
|
||||
use tokio::select;
|
||||
use tokio::time::timeout;
|
||||
|
||||
const CLOUD_REQUIREMENTS_CACHE_FILENAME: &str = "cloud-requirements-cache.json";
|
||||
const CLOUD_REQUIREMENTS_CACHE_HMAC_KEY: &[u8] =
|
||||
b"codex-cloud-requirements-cache-v3-064f8542-75b4-494c-a294-97d3ce597271";
|
||||
|
||||
type HmacSha256 = Hmac<Sha256>;
|
||||
|
||||
#[tokio::test]
|
||||
async fn cloud_requirements_cli_error_includes_actionable_context() -> anyhow::Result<()> {
|
||||
if cfg!(windows) {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let tmp = tempfile::tempdir()?;
|
||||
let codex_home = tmp.path();
|
||||
let cwd = std::env::current_dir()?;
|
||||
let backend = FailingBackend::start()?;
|
||||
write_chatgpt_business_auth(codex_home)?;
|
||||
write_startup_config(codex_home, &cwd, backend.url())?;
|
||||
|
||||
let CodexCliOutput { exit_code, output } = run_codex_cli(codex_home, &cwd).await?;
|
||||
|
||||
assert_ne!(0, exit_code, "Codex CLI should exit nonzero.");
|
||||
assert!(
|
||||
output.contains("Error loading configuration: failed to load your workspace-managed config after 5 attempts"),
|
||||
"expected cloud requirements load error in output, got: {output}"
|
||||
);
|
||||
assert!(
|
||||
output.contains("last backend status: 503"),
|
||||
"expected backend status in output, got: {output}"
|
||||
);
|
||||
assert!(
|
||||
output.contains("valid cached copy"),
|
||||
"expected cache guidance in output, got: {output}"
|
||||
);
|
||||
assert!(
|
||||
backend.request_count() > 0,
|
||||
"expected CLI to request cloud requirements from fake backend"
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn stale_cloud_requirements_cache_allows_actual_cli_startup_when_remote_fails()
|
||||
-> anyhow::Result<()> {
|
||||
if cfg!(windows) {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let tmp = tempfile::tempdir()?;
|
||||
let codex_home = tmp.path();
|
||||
let cwd = std::env::current_dir()?;
|
||||
let backend = FailingBackend::start()?;
|
||||
std::fs::write(
|
||||
codex_home.join("rules"),
|
||||
"rules should be a directory not a file",
|
||||
)?;
|
||||
write_chatgpt_business_auth(codex_home)?;
|
||||
write_stale_cloud_requirements_cache(
|
||||
codex_home,
|
||||
r#"allowed_approval_policies = ["on-request"]"#,
|
||||
)?;
|
||||
write_startup_config(codex_home, &cwd, backend.url())?;
|
||||
|
||||
let CodexCliOutput { exit_code, output } = run_codex_cli(codex_home, &cwd).await?;
|
||||
|
||||
assert_ne!(0, exit_code, "Codex CLI should exit nonzero.");
|
||||
assert!(
|
||||
backend.request_count() > 0,
|
||||
"expected CLI to retry cloud requirements before using stale cache"
|
||||
);
|
||||
assert!(
|
||||
output.contains("Error loading rules:"),
|
||||
"expected startup to proceed beyond config loading, got: {output}"
|
||||
);
|
||||
assert!(
|
||||
output.contains("failed to read rules files"),
|
||||
"expected controlled post-config startup error in output, got: {output}"
|
||||
);
|
||||
assert!(
|
||||
!output.contains("failed to load your workspace-managed config"),
|
||||
"stale cloud requirements cache should avoid config-load failure, got: {output}"
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
struct CodexCliOutput {
|
||||
exit_code: i32,
|
||||
output: String,
|
||||
}
|
||||
|
||||
async fn run_codex_cli(
|
||||
codex_home: impl AsRef<Path>,
|
||||
cwd: impl AsRef<Path>,
|
||||
) -> anyhow::Result<CodexCliOutput> {
|
||||
let codex_cli = codex_utils_cargo_bin::cargo_bin("codex")?;
|
||||
let mut env = HashMap::new();
|
||||
env.insert(
|
||||
"CODEX_HOME".to_string(),
|
||||
codex_home.as_ref().display().to_string(),
|
||||
);
|
||||
|
||||
let args = vec!["-c".to_string(), "analytics.enabled=false".to_string()];
|
||||
let spawned = codex_utils_pty::spawn_pty_process(
|
||||
codex_cli.to_string_lossy().as_ref(),
|
||||
&args,
|
||||
cwd.as_ref(),
|
||||
&env,
|
||||
&None,
|
||||
codex_utils_pty::TerminalSize::default(),
|
||||
)
|
||||
.await?;
|
||||
let mut output = Vec::new();
|
||||
let codex_utils_pty::SpawnedProcess {
|
||||
session,
|
||||
stdout_rx,
|
||||
stderr_rx,
|
||||
exit_rx,
|
||||
} = spawned;
|
||||
let mut output_rx = codex_utils_pty::combine_output_receivers(stdout_rx, stderr_rx);
|
||||
let mut exit_rx = exit_rx;
|
||||
let writer_tx = session.writer_sender();
|
||||
let exit_code_result = timeout(Duration::from_secs(10), async {
|
||||
loop {
|
||||
select! {
|
||||
result = output_rx.recv() => match result {
|
||||
Ok(chunk) => {
|
||||
if chunk.windows(4).any(|window| window == b"\x1b[6n") {
|
||||
let _ = writer_tx.send(b"\x1b[1;1R".to_vec()).await;
|
||||
}
|
||||
output.extend_from_slice(&chunk);
|
||||
}
|
||||
Err(tokio::sync::broadcast::error::RecvError::Closed) => break exit_rx.await,
|
||||
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {}
|
||||
},
|
||||
result = &mut exit_rx => break result,
|
||||
}
|
||||
}
|
||||
})
|
||||
.await;
|
||||
let exit_code = match exit_code_result {
|
||||
Ok(Ok(code)) => code,
|
||||
Ok(Err(err)) => return Err(err.into()),
|
||||
Err(_) => {
|
||||
session.terminate();
|
||||
anyhow::bail!("timed out waiting for codex CLI to exit");
|
||||
}
|
||||
};
|
||||
while let Ok(chunk) = output_rx.try_recv() {
|
||||
output.extend_from_slice(&chunk);
|
||||
}
|
||||
|
||||
let output = String::from_utf8_lossy(&output);
|
||||
Ok(CodexCliOutput {
|
||||
exit_code,
|
||||
output: output.to_string(),
|
||||
})
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct CloudRequirementsCacheFile {
|
||||
signed_payload: CloudRequirementsCacheSignedPayload,
|
||||
signature: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Serialize)]
|
||||
struct CloudRequirementsCacheSignedPayload {
|
||||
cached_at: DateTime<Utc>,
|
||||
expires_at: DateTime<Utc>,
|
||||
chatgpt_user_id: Option<String>,
|
||||
account_id: Option<String>,
|
||||
contents: Option<String>,
|
||||
}
|
||||
|
||||
struct FailingBackend {
|
||||
addr: SocketAddr,
|
||||
request_count: Arc<AtomicUsize>,
|
||||
shutdown: Arc<AtomicBool>,
|
||||
handle: Option<JoinHandle<()>>,
|
||||
}
|
||||
|
||||
impl FailingBackend {
|
||||
fn start() -> anyhow::Result<Self> {
|
||||
let listener = TcpListener::bind("127.0.0.1:0")?;
|
||||
let addr = listener.local_addr()?;
|
||||
listener.set_nonblocking(true)?;
|
||||
let request_count = Arc::new(AtomicUsize::new(0));
|
||||
let shutdown = Arc::new(AtomicBool::new(false));
|
||||
let thread_request_count = request_count.clone();
|
||||
let thread_shutdown = shutdown.clone();
|
||||
let handle = std::thread::spawn(move || {
|
||||
while !thread_shutdown.load(Ordering::SeqCst) {
|
||||
match listener.accept() {
|
||||
Ok((mut stream, _)) => {
|
||||
let mut buffer = [0; 4096];
|
||||
let _ = stream.read(&mut buffer);
|
||||
thread_request_count.fetch_add(1, Ordering::SeqCst);
|
||||
let _ = stream.write_all(
|
||||
b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
|
||||
);
|
||||
}
|
||||
Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => {
|
||||
std::thread::sleep(Duration::from_millis(10));
|
||||
}
|
||||
Err(_) => break,
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
Ok(Self {
|
||||
addr,
|
||||
request_count,
|
||||
shutdown,
|
||||
handle: Some(handle),
|
||||
})
|
||||
}
|
||||
|
||||
fn url(&self) -> String {
|
||||
format!("http://{}/backend-api", self.addr)
|
||||
}
|
||||
|
||||
fn request_count(&self) -> usize {
|
||||
self.request_count.load(Ordering::SeqCst)
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for FailingBackend {
|
||||
fn drop(&mut self) {
|
||||
self.shutdown.store(true, Ordering::SeqCst);
|
||||
let _ = TcpStream::connect(self.addr);
|
||||
if let Some(handle) = self.handle.take() {
|
||||
let _ = handle.join();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn write_startup_config(
|
||||
codex_home: &Path,
|
||||
cwd: impl AsRef<Path>,
|
||||
chatgpt_base_url: String,
|
||||
) -> anyhow::Result<()> {
|
||||
let cwd = cwd.as_ref();
|
||||
let config_contents = format!(
|
||||
r#"
|
||||
chatgpt_base_url = "{chatgpt_base_url}"
|
||||
cli_auth_credentials_store = "file"
|
||||
model_provider = "ollama"
|
||||
|
||||
[projects]
|
||||
"{cwd}" = {{ trust_level = "trusted" }}
|
||||
"#,
|
||||
cwd = cwd.display()
|
||||
);
|
||||
std::fs::write(codex_home.join("config.toml"), config_contents)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn write_chatgpt_business_auth(codex_home: &Path) -> anyhow::Result<()> {
|
||||
let header_b64 = URL_SAFE_NO_PAD.encode(serde_json::to_vec(&json!({
|
||||
"alg": "none",
|
||||
"typ": "JWT",
|
||||
}))?);
|
||||
let payload_b64 = URL_SAFE_NO_PAD.encode(serde_json::to_vec(&json!({
|
||||
"email": "user@example.com",
|
||||
"https://api.openai.com/auth": {
|
||||
"chatgpt_plan_type": "business",
|
||||
"chatgpt_user_id": "user-12345",
|
||||
"user_id": "user-12345",
|
||||
},
|
||||
}))?);
|
||||
let signature_b64 = URL_SAFE_NO_PAD.encode(b"sig");
|
||||
let fake_jwt = format!("{header_b64}.{payload_b64}.{signature_b64}");
|
||||
let auth_json = json!({
|
||||
"OPENAI_API_KEY": null,
|
||||
"tokens": {
|
||||
"id_token": fake_jwt,
|
||||
"access_token": "test-access-token",
|
||||
"refresh_token": "test-refresh-token",
|
||||
"account_id": "account-12345",
|
||||
},
|
||||
"last_refresh": "2025-01-01T00:00:00Z",
|
||||
});
|
||||
std::fs::write(
|
||||
codex_home.join("auth.json"),
|
||||
serde_json::to_vec_pretty(&auth_json)?,
|
||||
)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn write_stale_cloud_requirements_cache(codex_home: &Path, contents: &str) -> anyhow::Result<()> {
|
||||
let signed_payload = CloudRequirementsCacheSignedPayload {
|
||||
cached_at: Utc::now() - ChronoDuration::minutes(31),
|
||||
expires_at: Utc::now() - ChronoDuration::minutes(1),
|
||||
chatgpt_user_id: Some("user-12345".to_string()),
|
||||
account_id: Some("account-12345".to_string()),
|
||||
contents: Some(contents.to_string()),
|
||||
};
|
||||
let payload_bytes = serde_json::to_vec(&signed_payload)?;
|
||||
let mut mac = HmacSha256::new_from_slice(CLOUD_REQUIREMENTS_CACHE_HMAC_KEY)?;
|
||||
mac.update(&payload_bytes);
|
||||
let cache_file = CloudRequirementsCacheFile {
|
||||
signed_payload,
|
||||
signature: BASE64_STANDARD.encode(mac.finalize().into_bytes()),
|
||||
};
|
||||
std::fs::write(
|
||||
codex_home.join(CLOUD_REQUIREMENTS_CACHE_FILENAME),
|
||||
serde_json::to_vec_pretty(&cache_file)?,
|
||||
)?;
|
||||
Ok(())
|
||||
}
|
||||
@@ -47,6 +47,7 @@ const CLOUD_REQUIREMENTS_MAX_ATTEMPTS: usize = 5;
|
||||
const CLOUD_REQUIREMENTS_CACHE_FILENAME: &str = "cloud-requirements-cache.json";
|
||||
const CLOUD_REQUIREMENTS_CACHE_REFRESH_INTERVAL: Duration = Duration::from_secs(5 * 60);
|
||||
const CLOUD_REQUIREMENTS_CACHE_TTL: Duration = Duration::from_secs(30 * 60);
|
||||
const CLOUD_REQUIREMENTS_STALE_CACHE_TTL: Duration = Duration::from_secs(4 * 60 * 60);
|
||||
const CLOUD_REQUIREMENTS_FETCH_ATTEMPT_METRIC: &str = "codex.cloud_requirements.fetch_attempt";
|
||||
const CLOUD_REQUIREMENTS_FETCH_FINAL_METRIC: &str = "codex.cloud_requirements.fetch_final";
|
||||
const CLOUD_REQUIREMENTS_LOAD_METRIC: &str = "codex.cloud_requirements.load";
|
||||
@@ -136,6 +137,51 @@ impl CloudRequirementsCacheSignedPayload {
|
||||
.as_deref()
|
||||
.and_then(|contents| parse_cloud_requirements(contents).ok().flatten())
|
||||
}
|
||||
|
||||
fn is_fresh(&self, now: DateTime<Utc>) -> bool {
|
||||
self.expires_at > now
|
||||
}
|
||||
|
||||
fn can_use_stale(&self, now: DateTime<Utc>) -> bool {
|
||||
if self.is_fresh(now) {
|
||||
return false;
|
||||
}
|
||||
|
||||
let Ok(stale_ttl) = ChronoDuration::from_std(CLOUD_REQUIREMENTS_STALE_CACHE_TTL) else {
|
||||
return false;
|
||||
};
|
||||
self.expires_at
|
||||
.checked_add_signed(stale_ttl)
|
||||
.is_some_and(|stale_expires_at| stale_expires_at > now)
|
||||
}
|
||||
}
|
||||
|
||||
fn is_cloud_requirements_eligible_auth(auth: &CodexAuth) -> bool {
|
||||
let Some(plan_type) = auth.account_plan_type() else {
|
||||
return false;
|
||||
};
|
||||
|
||||
auth.is_chatgpt_auth()
|
||||
&& (plan_type.is_business_like() || matches!(plan_type, PlanType::Enterprise))
|
||||
}
|
||||
|
||||
fn format_cloud_requirements_load_failed_message(
|
||||
attempt_count: usize,
|
||||
status_code: Option<u16>,
|
||||
) -> String {
|
||||
let status = status_code
|
||||
.map(|status_code| format!("last backend status: {status_code}"))
|
||||
.unwrap_or_else(|| "last failure did not include an HTTP status".to_string());
|
||||
format!(
|
||||
"{CLOUD_REQUIREMENTS_LOAD_FAILED_MESSAGE} after {attempt_count} attempts ({status}). Codex could not use a valid cached copy for this account, so startup stopped to avoid running without managed requirements. Please try again."
|
||||
)
|
||||
}
|
||||
|
||||
fn format_cloud_requirements_timeout_message(timeout: Duration) -> String {
|
||||
format!(
|
||||
"timed out waiting for workspace-managed config after {}s. Codex could not use a valid cached copy for this account, so startup stopped to avoid running without managed requirements. Please try again.",
|
||||
timeout.as_secs()
|
||||
)
|
||||
}
|
||||
fn sign_cache_payload(payload_bytes: &[u8]) -> Option<String> {
|
||||
let mut mac = HmacSha256::new_from_slice(CLOUD_REQUIREMENTS_CACHE_WRITE_HMAC_KEY).ok()?;
|
||||
@@ -274,26 +320,29 @@ impl CloudRequirementsService {
|
||||
let _timer =
|
||||
codex_otel::start_global_timer("codex.cloud_requirements.fetch.duration_ms", &[]);
|
||||
let started_at = Instant::now();
|
||||
let fetch_result = timeout(self.timeout, self.fetch())
|
||||
.await
|
||||
.inspect_err(|_| {
|
||||
let message = format!(
|
||||
"Timed out waiting for cloud requirements after {}s",
|
||||
self.timeout.as_secs()
|
||||
);
|
||||
tracing::error!("{message}");
|
||||
emit_load_metric("startup", "error");
|
||||
})
|
||||
.map_err(|_| {
|
||||
CloudRequirementsLoadError::new(
|
||||
let fetch_result = match timeout(self.timeout, self.fetch()).await {
|
||||
Ok(fetch_result) => fetch_result,
|
||||
Err(_) => {
|
||||
let err = CloudRequirementsLoadError::new(
|
||||
CloudRequirementsLoadErrorCode::Timeout,
|
||||
/*status_code*/ None,
|
||||
format!(
|
||||
"timed out waiting for cloud requirements after {}s",
|
||||
self.timeout.as_secs()
|
||||
),
|
||||
)
|
||||
})?;
|
||||
format_cloud_requirements_timeout_message(self.timeout),
|
||||
);
|
||||
if let Some(stale_cache) = self.load_stale_cache_for_current_auth().await {
|
||||
tracing::warn!(
|
||||
path = %self.cache_path.display(),
|
||||
expires_at = %stale_cache.expires_at,
|
||||
error = %err,
|
||||
"Using stale cloud requirements cache after remote load timed out"
|
||||
);
|
||||
Ok(stale_cache.requirements())
|
||||
} else {
|
||||
tracing::error!("{err}");
|
||||
emit_load_metric("startup", "error");
|
||||
return Err(err);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
let result = match fetch_result {
|
||||
Ok(result) => result,
|
||||
@@ -328,33 +377,57 @@ impl CloudRequirementsService {
|
||||
let Some(auth) = self.auth_manager.auth().await else {
|
||||
return Ok(None);
|
||||
};
|
||||
let Some(plan_type) = auth.account_plan_type() else {
|
||||
return Ok(None);
|
||||
};
|
||||
if !auth.is_chatgpt_auth()
|
||||
|| !(plan_type.is_business_like() || matches!(plan_type, PlanType::Enterprise))
|
||||
{
|
||||
if !is_cloud_requirements_eligible_auth(&auth) {
|
||||
return Ok(None);
|
||||
}
|
||||
let (chatgpt_user_id, account_id) = auth_identity(&auth);
|
||||
|
||||
match self
|
||||
.load_cache(chatgpt_user_id.as_deref(), account_id.as_deref())
|
||||
let cache_payload = match self
|
||||
.load_cache_payload(chatgpt_user_id.as_deref(), account_id.as_deref())
|
||||
.await
|
||||
{
|
||||
Ok(signed_payload) => {
|
||||
tracing::info!(
|
||||
path = %self.cache_path.display(),
|
||||
"Using cached cloud requirements"
|
||||
);
|
||||
return Ok(signed_payload.requirements());
|
||||
let now = Utc::now();
|
||||
if signed_payload.is_fresh(now) {
|
||||
tracing::info!(
|
||||
path = %self.cache_path.display(),
|
||||
"Using cached cloud requirements"
|
||||
);
|
||||
return Ok(signed_payload.requirements());
|
||||
}
|
||||
self.log_cache_load_status(&CacheLoadStatus::CacheExpired);
|
||||
Some(signed_payload)
|
||||
}
|
||||
Err(cache_load_status) => {
|
||||
self.log_cache_load_status(&cache_load_status);
|
||||
None
|
||||
}
|
||||
};
|
||||
let stale_cache =
|
||||
cache_payload.filter(|signed_payload| signed_payload.can_use_stale(Utc::now()));
|
||||
|
||||
match self.fetch_with_retries(auth, "startup").await {
|
||||
Ok(requirements) => Ok(requirements),
|
||||
Err(err) => {
|
||||
if err.code() == CloudRequirementsLoadErrorCode::RequestFailed
|
||||
&& let Some(stale_cache) = stale_cache
|
||||
{
|
||||
tracing::warn!(
|
||||
path = %self.cache_path.display(),
|
||||
expires_at = %stale_cache.expires_at,
|
||||
error = %err,
|
||||
"Using stale cloud requirements cache after remote load failed"
|
||||
);
|
||||
return Ok(stale_cache.requirements());
|
||||
}
|
||||
tracing::error!(
|
||||
path = %self.cache_path.display(),
|
||||
error = %err,
|
||||
"Failed to load cloud requirements"
|
||||
);
|
||||
Err(err)
|
||||
}
|
||||
}
|
||||
|
||||
self.fetch_with_retries(auth, "startup").await
|
||||
}
|
||||
|
||||
async fn fetch_with_retries(
|
||||
@@ -517,17 +590,41 @@ impl CloudRequirementsService {
|
||||
CLOUD_REQUIREMENTS_MAX_ATTEMPTS,
|
||||
last_status_code,
|
||||
);
|
||||
tracing::error!(
|
||||
path = %self.cache_path.display(),
|
||||
"{CLOUD_REQUIREMENTS_LOAD_FAILED_MESSAGE}"
|
||||
);
|
||||
Err(CloudRequirementsLoadError::new(
|
||||
CloudRequirementsLoadErrorCode::RequestFailed,
|
||||
last_status_code,
|
||||
CLOUD_REQUIREMENTS_LOAD_FAILED_MESSAGE,
|
||||
format_cloud_requirements_load_failed_message(
|
||||
CLOUD_REQUIREMENTS_MAX_ATTEMPTS,
|
||||
last_status_code,
|
||||
),
|
||||
))
|
||||
}
|
||||
|
||||
async fn load_stale_cache_for_current_auth(
|
||||
&self,
|
||||
) -> Option<CloudRequirementsCacheSignedPayload> {
|
||||
let auth = self.auth_manager.auth().await?;
|
||||
if !is_cloud_requirements_eligible_auth(&auth) {
|
||||
return None;
|
||||
}
|
||||
|
||||
let (chatgpt_user_id, account_id) = auth_identity(&auth);
|
||||
let cache_payload = match self
|
||||
.load_cache_payload(chatgpt_user_id.as_deref(), account_id.as_deref())
|
||||
.await
|
||||
{
|
||||
Ok(cache_payload) => cache_payload,
|
||||
Err(cache_load_status) => {
|
||||
self.log_cache_load_status(&cache_load_status);
|
||||
return None;
|
||||
}
|
||||
};
|
||||
|
||||
cache_payload
|
||||
.can_use_stale(Utc::now())
|
||||
.then_some(cache_payload)
|
||||
}
|
||||
|
||||
async fn refresh_cache_in_background(&self) {
|
||||
loop {
|
||||
sleep(CLOUD_REQUIREMENTS_CACHE_REFRESH_INTERVAL).await;
|
||||
@@ -548,12 +645,7 @@ impl CloudRequirementsService {
|
||||
let Some(auth) = self.auth_manager.auth().await else {
|
||||
return false;
|
||||
};
|
||||
let Some(plan_type) = auth.account_plan_type() else {
|
||||
return false;
|
||||
};
|
||||
if !auth.is_chatgpt_auth()
|
||||
|| !(plan_type.is_business_like() || matches!(plan_type, PlanType::Enterprise))
|
||||
{
|
||||
if !is_cloud_requirements_eligible_auth(&auth) {
|
||||
return false;
|
||||
}
|
||||
|
||||
@@ -571,7 +663,7 @@ impl CloudRequirementsService {
|
||||
true
|
||||
}
|
||||
|
||||
async fn load_cache(
|
||||
async fn load_cache_payload(
|
||||
&self,
|
||||
chatgpt_user_id: Option<&str>,
|
||||
account_id: Option<&str>,
|
||||
@@ -619,10 +711,6 @@ impl CloudRequirementsService {
|
||||
return Err(CacheLoadStatus::CacheIdentityMismatch);
|
||||
}
|
||||
|
||||
if cache_file.signed_payload.expires_at <= Utc::now() {
|
||||
return Err(CacheLoadStatus::CacheExpired);
|
||||
}
|
||||
|
||||
Ok(cache_file.signed_payload)
|
||||
}
|
||||
|
||||
@@ -846,6 +934,20 @@ mod tests {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn write_signed_cache(codex_home: &Path, signed_payload: CloudRequirementsCacheSignedPayload) {
|
||||
let payload_bytes = cache_payload_bytes(&signed_payload).expect("payload");
|
||||
let signature = sign_cache_payload(&payload_bytes).expect("sign payload");
|
||||
let cache_file = CloudRequirementsCacheFile {
|
||||
signed_payload,
|
||||
signature,
|
||||
};
|
||||
std::fs::write(
|
||||
codex_home.join(CLOUD_REQUIREMENTS_CACHE_FILENAME),
|
||||
serde_json::to_vec_pretty(&cache_file).expect("serialize cache"),
|
||||
)
|
||||
.expect("write cache");
|
||||
}
|
||||
|
||||
fn auth_manager_with_api_key() -> Arc<AuthManager> {
|
||||
let tmp = tempdir().expect("tempdir");
|
||||
let auth_json = json!({
|
||||
@@ -1007,6 +1109,19 @@ mod tests {
|
||||
FetchAttemptError::Retryable(RetryableFailureKind::Request { status_code: None })
|
||||
}
|
||||
|
||||
fn request_error_with_status(status_code: u16) -> FetchAttemptError {
|
||||
FetchAttemptError::Retryable(RetryableFailureKind::Request {
|
||||
status_code: Some(status_code),
|
||||
})
|
||||
}
|
||||
|
||||
fn approval_policy_requirements(policy: AskForApproval) -> ConfigRequirementsToml {
|
||||
ConfigRequirementsToml {
|
||||
allowed_approval_policies: Some(vec![policy]),
|
||||
..Default::default()
|
||||
}
|
||||
}
|
||||
|
||||
struct StaticFetcher {
|
||||
contents: Option<String>,
|
||||
}
|
||||
@@ -1322,7 +1437,11 @@ enabled = false
|
||||
let err = result.expect_err("cloud requirements timeout should fail closed");
|
||||
assert!(
|
||||
err.to_string()
|
||||
.contains("timed out waiting for cloud requirements")
|
||||
.contains("timed out waiting for workspace-managed config")
|
||||
);
|
||||
assert!(
|
||||
err.to_string()
|
||||
.contains("Codex could not use a valid cached copy")
|
||||
);
|
||||
}
|
||||
|
||||
@@ -1951,6 +2070,125 @@ enabled = false
|
||||
assert_eq!(fetcher.request_count.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn fetch_cloud_requirements_uses_stale_cache_when_remote_fetch_fails() {
|
||||
let codex_home = tempdir().expect("tempdir");
|
||||
write_signed_cache(
|
||||
codex_home.path(),
|
||||
CloudRequirementsCacheSignedPayload {
|
||||
cached_at: Utc::now() - ChronoDuration::minutes(31),
|
||||
expires_at: Utc::now() - ChronoDuration::minutes(1),
|
||||
chatgpt_user_id: Some("user-12345".to_string()),
|
||||
account_id: Some("account-12345".to_string()),
|
||||
contents: Some(r#"allowed_approval_policies = ["on-request"]"#.to_string()),
|
||||
},
|
||||
);
|
||||
|
||||
let retryable_error = Err(request_error_with_status(503));
|
||||
let fetcher = Arc::new(SequenceFetcher::new(vec![
|
||||
retryable_error;
|
||||
CLOUD_REQUIREMENTS_MAX_ATTEMPTS
|
||||
]));
|
||||
let service = CloudRequirementsService::new(
|
||||
auth_manager_with_plan("business"),
|
||||
fetcher.clone(),
|
||||
codex_home.path().to_path_buf(),
|
||||
CLOUD_REQUIREMENTS_TIMEOUT,
|
||||
);
|
||||
|
||||
let handle = tokio::spawn(async move { service.fetch().await });
|
||||
tokio::task::yield_now().await;
|
||||
tokio::time::advance(Duration::from_secs(5)).await;
|
||||
tokio::task::yield_now().await;
|
||||
|
||||
assert_eq!(
|
||||
handle.await.expect("cloud requirements task"),
|
||||
Ok(Some(approval_policy_requirements(
|
||||
AskForApproval::OnRequest
|
||||
)))
|
||||
);
|
||||
assert_eq!(
|
||||
fetcher.request_count.load(Ordering::SeqCst),
|
||||
CLOUD_REQUIREMENTS_MAX_ATTEMPTS
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn fetch_cloud_requirements_rejects_stale_cache_after_grace_period() {
|
||||
let codex_home = tempdir().expect("tempdir");
|
||||
let stale_ttl =
|
||||
ChronoDuration::from_std(CLOUD_REQUIREMENTS_STALE_CACHE_TTL).expect("stale cache TTL");
|
||||
write_signed_cache(
|
||||
codex_home.path(),
|
||||
CloudRequirementsCacheSignedPayload {
|
||||
cached_at: Utc::now() - stale_ttl - ChronoDuration::minutes(31),
|
||||
expires_at: Utc::now() - stale_ttl - ChronoDuration::minutes(1),
|
||||
chatgpt_user_id: Some("user-12345".to_string()),
|
||||
account_id: Some("account-12345".to_string()),
|
||||
contents: Some(r#"allowed_approval_policies = ["on-request"]"#.to_string()),
|
||||
},
|
||||
);
|
||||
|
||||
let retryable_error = Err(request_error_with_status(503));
|
||||
let fetcher = Arc::new(SequenceFetcher::new(vec![
|
||||
retryable_error;
|
||||
CLOUD_REQUIREMENTS_MAX_ATTEMPTS
|
||||
]));
|
||||
let service = CloudRequirementsService::new(
|
||||
auth_manager_with_plan("business"),
|
||||
fetcher.clone(),
|
||||
codex_home.path().to_path_buf(),
|
||||
CLOUD_REQUIREMENTS_TIMEOUT,
|
||||
);
|
||||
|
||||
let handle = tokio::spawn(async move { service.fetch().await });
|
||||
tokio::task::yield_now().await;
|
||||
tokio::time::advance(Duration::from_secs(5)).await;
|
||||
tokio::task::yield_now().await;
|
||||
|
||||
let err = handle
|
||||
.await
|
||||
.expect("cloud requirements task")
|
||||
.expect_err("too-stale cache should not be used");
|
||||
let err_text = err.to_string();
|
||||
assert!(err_text.contains("failed to load your workspace-managed config after 5 attempts"));
|
||||
assert!(err_text.contains("last backend status: 503"));
|
||||
assert!(err_text.contains("valid cached copy"));
|
||||
assert_eq!(err.code(), CloudRequirementsLoadErrorCode::RequestFailed);
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn fetch_cloud_requirements_uses_stale_cache_when_remote_fetch_times_out() {
|
||||
let codex_home = tempdir().expect("tempdir");
|
||||
write_signed_cache(
|
||||
codex_home.path(),
|
||||
CloudRequirementsCacheSignedPayload {
|
||||
cached_at: Utc::now() - ChronoDuration::minutes(31),
|
||||
expires_at: Utc::now() - ChronoDuration::minutes(1),
|
||||
chatgpt_user_id: Some("user-12345".to_string()),
|
||||
account_id: Some("account-12345".to_string()),
|
||||
contents: Some(r#"allowed_approval_policies = ["on-request"]"#.to_string()),
|
||||
},
|
||||
);
|
||||
|
||||
let service = CloudRequirementsService::new(
|
||||
auth_manager_with_plan("business"),
|
||||
Arc::new(PendingFetcher),
|
||||
codex_home.path().to_path_buf(),
|
||||
CLOUD_REQUIREMENTS_TIMEOUT,
|
||||
);
|
||||
|
||||
let handle = tokio::spawn(async move { service.fetch_with_timeout().await });
|
||||
tokio::time::advance(CLOUD_REQUIREMENTS_TIMEOUT + Duration::from_millis(1)).await;
|
||||
|
||||
assert_eq!(
|
||||
handle.await.expect("cloud requirements task"),
|
||||
Ok(Some(approval_policy_requirements(
|
||||
AskForApproval::OnRequest
|
||||
)))
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn fetch_cloud_requirements_writes_signed_cache() {
|
||||
let codex_home = tempdir().expect("tempdir");
|
||||
@@ -2048,10 +2286,10 @@ enabled = false
|
||||
.await
|
||||
.expect("cloud requirements task")
|
||||
.expect_err("cloud requirements retry exhaustion should fail closed");
|
||||
assert_eq!(
|
||||
err.to_string(),
|
||||
"failed to load your workspace-managed config"
|
||||
);
|
||||
let err_text = err.to_string();
|
||||
assert!(err_text.contains("failed to load your workspace-managed config after 5 attempts"));
|
||||
assert!(err_text.contains("last failure did not include an HTTP status"));
|
||||
assert!(err_text.contains("valid cached copy"));
|
||||
assert_eq!(err.code(), CloudRequirementsLoadErrorCode::RequestFailed);
|
||||
assert_eq!(
|
||||
fetcher.request_count.load(Ordering::SeqCst),
|
||||
|
||||
Reference in New Issue
Block a user