From ef8cfe5e96f7782b38745f3f86da61af3ba35e10 Mon Sep 17 00:00:00 2001 From: Anton Panasenko Date: Fri, 2 Oct 2026 23:42:58 +0000 Subject: [PATCH] Retry registry authentication outages and jitter executor reconnects (#50465) ## Why Remote executors need to recover from confirmed authentication-service outages and spread reconnect attempts after a shared outage. Registration retries must remain limited to confirmed failures before a write, since replaying an ambiguous request can replace a newer registration. ## What changed - Retry registry responses with HTTP 502 and `authentication_service_unavailable`, alongside the existing HTTP 503 `registration_conflict` retries. - Accept structured registry errors in `detail`, keeping the canonical `error` authoritative when both are present. - Randomize reconnect delays between half and all of the current exponential backoff, retaining the 30-second cap. ## Testing Add coverage for reconnect jitter bounds, authentication-outage retry cancellation, error-envelope precedence, and preservation of Noise identity and initialized sessions across retries. Make the HTTP-header helper cleanup test wait for the helper PID before cancellation and verify process termination when the provider is dropped. GitOrigin-RevId: 35fdb57732d17b0bbfeca8a4dc557a0988e29a31 --- codex-rs/exec-server/src/remote.rs | 20 ++++++-- .../src/remote/reconnect_backoff.rs | 13 +++++ .../src/remote/reconnect_backoff_tests.rs | 25 +++++++++ .../src/remote/registration_retry.rs | 17 +++++-- .../src/remote/registration_retry_tests.rs | 36 ++++++++++--- .../tests/relay/registration_retry_tests.rs | 18 +++++-- .../rmcp-client/src/http_headers_tests.rs | 51 ++++++++++++++++--- 7 files changed, 156 insertions(+), 24 deletions(-) create mode 100644 codex-rs/exec-server/src/remote/reconnect_backoff.rs create mode 100644 codex-rs/exec-server/src/remote/reconnect_backoff_tests.rs diff --git a/codex-rs/exec-server/src/remote.rs b/codex-rs/exec-server/src/remote.rs index ce4a7fcd53..dbc29968f8 100644 --- a/codex-rs/exec-server/src/remote.rs +++ b/codex-rs/exec-server/src/remote.rs @@ -58,6 +58,7 @@ use direct::run_direct_environment; const ERROR_BODY_PREVIEW_BYTES: usize = 4096; const NOISE_RELAY_SECURITY_PROFILE: &str = "noise_hybrid_ik_v1"; +mod reconnect_backoff; mod registration_retry; /// Wire transport used after registering a remote exec-server. @@ -777,7 +778,11 @@ async fn run_remote_environment_connections( } } - sleep(backoff).await; + sleep(reconnect_backoff::reconnect_delay( + backoff, + uuid::Uuid::new_v4().as_u64_pair().1, + )) + .await; backoff = (backoff * 2).min(Duration::from_secs(30)); } } @@ -831,6 +836,15 @@ fn normalize_environment_id(environment_id: String) -> Result, + detail: Option, +} + +impl RegistryErrorBody { + fn into_error(self) -> Option { + // Keep the canonical error authoritative when both envelopes are present. + self.error + .or_else(|| serde_json::from_value(self.detail?).ok()) + } } #[derive(Deserialize)] @@ -863,7 +877,7 @@ fn environment_registry_auth_error(status: StatusCode, body: &str) -> ExecServer fn environment_registry_http_error(status: StatusCode, body: &str) -> ExecServerError { let parsed = serde_json::from_str::(body).ok(); let (code, message) = parsed - .and_then(|body| body.error) + .and_then(RegistryErrorBody::into_error) .map(|error| { ( error.code, @@ -889,7 +903,7 @@ fn environment_registry_http_error(status: StatusCode, body: &str) -> ExecServer fn registry_error_message(body: &str) -> Option { serde_json::from_str::(body) .ok() - .and_then(|body| body.error) + .and_then(RegistryErrorBody::into_error) .and_then(|error| error.message) .or_else(|| preview_error_body(body)) } diff --git a/codex-rs/exec-server/src/remote/reconnect_backoff.rs b/codex-rs/exec-server/src/remote/reconnect_backoff.rs new file mode 100644 index 0000000000..9f0638ae67 --- /dev/null +++ b/codex-rs/exec-server/src/remote/reconnect_backoff.rs @@ -0,0 +1,13 @@ +//! Spread executor reconnects after a shared outage without exceeding the existing backoff cap. + +use std::time::Duration; + +pub(super) fn reconnect_delay(backoff: Duration, random_sample: u64) -> Duration { + let upper_ms = backoff.as_millis() as u64; + let lower_ms = upper_ms / 2; + Duration::from_millis(lower_ms + random_sample % (upper_ms - lower_ms + 1)) +} + +#[cfg(test)] +#[path = "reconnect_backoff_tests.rs"] +mod tests; diff --git a/codex-rs/exec-server/src/remote/reconnect_backoff_tests.rs b/codex-rs/exec-server/src/remote/reconnect_backoff_tests.rs new file mode 100644 index 0000000000..990db1cff6 --- /dev/null +++ b/codex-rs/exec-server/src/remote/reconnect_backoff_tests.rs @@ -0,0 +1,25 @@ +//! Reconnect delays vary across samples while retaining a nonzero floor and the original cap. + +use std::time::Duration; + +use pretty_assertions::assert_eq; + +use super::reconnect_delay; + +#[test] +fn reconnect_jitter_stays_within_each_backoff_window() { + for seconds in [1, 2, 4, 8, 16, 30] { + let backoff = Duration::from_secs(seconds); + let half_ms = seconds * 500; + assert_eq!(reconnect_delay(backoff, /*random_sample*/ 0), backoff / 2); + assert_eq!(reconnect_delay(backoff, half_ms), backoff); + for sample in [1, 1234, 987654, u64::MAX] { + let delay = reconnect_delay(backoff, sample); + assert!((backoff / 2..=backoff).contains(&delay)); + } + assert_ne!( + reconnect_delay(backoff, /*random_sample*/ 1), + reconnect_delay(backoff, /*random_sample*/ 2) + ); + } +} diff --git a/codex-rs/exec-server/src/remote/registration_retry.rs b/codex-rs/exec-server/src/remote/registration_retry.rs index 1a2dc989da..92b09f076b 100644 --- a/codex-rs/exec-server/src/remote/registration_retry.rs +++ b/codex-rs/exec-server/src/remote/registration_retry.rs @@ -1,4 +1,4 @@ -//! Retry only explicit registration conflicts after the registry's write-retry loop has finished. +//! Retry explicit registration conflicts and authentication outages before registration writes. //! Ambiguous failures must not be replayed: a timed-out request can still replace a newer registration. //! The enclosing remote-transport future owns cancellation; retries spawn no background work. @@ -28,8 +28,16 @@ impl EnvironmentRegistryClient { { Ok(response) => return Ok(response), Err(ExecServerError::EnvironmentRegistryHttp { status, code, .. }) - if status == StatusCode::SERVICE_UNAVAILABLE - && code.as_deref() == Some("registration_conflict") => + if matches!( + (status, code.as_deref()), + ( + StatusCode::SERVICE_UNAVAILABLE, + Some("registration_conflict") + ) | ( + StatusCode::BAD_GATEWAY, + Some("authentication_service_unavailable") + ) + ) => { let delay = registry_recovery_retry_delay(&retry_key, attempt); attempt = attempt.saturating_add(1); @@ -39,7 +47,8 @@ impl EnvironmentRegistryClient { noise_outcome = "retry", retry_attempt = attempt, retry_delay_ms = delay.as_millis() as u64, - "Noise executor retrying registry registration conflict" + status = status.as_u16(), + "Noise executor retrying uncommitted registry registration" ); sleep(delay).await; } diff --git a/codex-rs/exec-server/src/remote/registration_retry_tests.rs b/codex-rs/exec-server/src/remote/registration_retry_tests.rs index 2c280c4947..1e9fdcd3a2 100644 --- a/codex-rs/exec-server/src/remote/registration_retry_tests.rs +++ b/codex-rs/exec-server/src/remote/registration_retry_tests.rs @@ -1,4 +1,4 @@ -//! Only confirmed conflicts permit replay; ambiguous registration results remain terminal. +//! Confirmed conflicts and pre-write auth outages permit replay; ambiguous results remain terminal. use std::sync::Arc; use std::sync::atomic::AtomicUsize; @@ -29,8 +29,12 @@ use crate::NoiseChannelIdentity; const ENVIRONMENT_ID: &str = "registration-retry-test"; const CONFLICT_BODY: &str = r#"{"error":{"code":"registration_conflict","message":"registration unavailable"}}"#; +const CONFLICT_WITH_STRING_DETAIL_BODY: &str = r#"{"error":{"code":"registration_conflict","message":"registration unavailable"},"detail":"additional diagnostics"}"#; +const CONFLICT_WITH_STRUCTURED_DETAIL_BODY: &str = r#"{"error":{"code":"registration_conflict","message":"registration unavailable"},"detail":{"code":"registration_denied","message":"additional diagnostics"}}"#; +const AUTH_UNAVAILABLE_BODY: &str = r#"{"detail":{"code":"authentication_service_unavailable","message":"Authentication service unavailable"}}"#; const ERROR_BODY: &str = r#"{"error":{"code":"registration_denied","message":"registration unavailable"}}"#; +const ERROR_WITH_AUTH_DETAIL_BODY: &str = r#"{"error":{"code":"registration_denied","message":"registration unavailable"},"detail":{"code":"authentication_service_unavailable","message":"Authentication service unavailable"}}"#; const SUCCESS_BODY: &str = r#"{"environment_id":"registration-retry-test","url":"ws://localhost/relay","security_profile":"noise_hybrid_ik_v1","executor_registration_id":"committed-registration"}"#; #[derive(Debug, Default)] @@ -62,11 +66,19 @@ impl tracing_subscriber::Layer for RetryObserved { #[test_case::test_case(200, "not JSON", Duration::ZERO; "malformed_success")] #[test_case::test_case(200, SUCCESS_BODY, Duration::from_secs(60); "committed_success_body_timeout")] #[test_case::test_case(503, CONFLICT_BODY, Duration::ZERO; "confirmed_conflict_backoff_is_cancellable")] +#[test_case::test_case(503, CONFLICT_WITH_STRING_DETAIL_BODY, Duration::ZERO; "conflict_with_string_detail")] +#[test_case::test_case(503, CONFLICT_WITH_STRUCTURED_DETAIL_BODY, Duration::ZERO; "conflict_with_structured_detail")] #[test_case::test_case(503, CONFLICT_BODY, Duration::from_secs(60); "conflict_code_not_received")] #[test_case::test_case(503, "not JSON", Duration::ZERO; "malformed_unavailable")] #[test_case::test_case(503, "{}", Duration::ZERO; "missing_conflict_code")] #[test_case::test_case(503, ERROR_BODY, Duration::ZERO; "different_error_code")] #[test_case::test_case(502, CONFLICT_BODY, Duration::ZERO; "gateway_error")] +#[test_case::test_case(502, AUTH_UNAVAILABLE_BODY, Duration::ZERO; "auth_outage_backoff_is_cancellable")] +#[test_case::test_case(502, ERROR_WITH_AUTH_DETAIL_BODY, Duration::ZERO; "canonical_error_overrides_auth_detail")] +#[test_case::test_case(502, AUTH_UNAVAILABLE_BODY, Duration::from_secs(60); "auth_outage_code_not_received")] +#[test_case::test_case(502, "not JSON", Duration::ZERO; "unconfirmed_gateway_error")] +#[test_case::test_case(502, r#"{"detail":"Authentication service unavailable"}"#, Duration::ZERO; "legacy_auth_outage")] +#[test_case::test_case(503, AUTH_UNAVAILABLE_BODY, Duration::ZERO; "auth_outage_wrong_status")] #[test_case::test_case(408, CONFLICT_BODY, Duration::ZERO; "request_timeout")] #[test_case::test_case(429, CONFLICT_BODY, Duration::ZERO; "too_many_requests")] #[test_case::test_case(401, ERROR_BODY, Duration::from_secs(60); "unauthorized_stalled_body")] @@ -76,7 +88,7 @@ impl tracing_subscriber::Layer for RetryObserved { #[test_case::test_case(403, ERROR_BODY, Duration::from_millis(50); "delayed_forbidden_details")] #[test_case::test_case(404, ERROR_BODY, Duration::from_millis(50); "delayed_error_details")] #[tokio::test] -async fn registration_requires_a_confirmed_conflict_before_replay( +async fn registration_requires_a_confirmed_pre_write_failure_before_replay( status: u16, body: &'static str, body_delay: Duration, @@ -126,11 +138,20 @@ async fn registration_requires_a_confirmed_conflict_before_replay( .with_subscriber(subscriber), ); - if status == 503 && body == CONFLICT_BODY && body_delay.is_zero() { + if body_delay.is_zero() + && ((status == 503 + && matches!( + body, + CONFLICT_BODY + | CONFLICT_WITH_STRING_DETAIL_BODY + | CONFLICT_WITH_STRUCTURED_DETAIL_BODY + )) + || (status == 502 && body == AUTH_UNAVAILABLE_BODY)) + { timeout(Duration::from_secs(1), async { tokio::select! { _ = retry.notified() => anyhow::Ok(()), - _ = &mut registration => anyhow::bail!("a confirmed conflict must enter backoff"), + _ = &mut registration => anyhow::bail!("a confirmed pre-write failure must enter backoff"), } }) .await??; @@ -163,12 +184,15 @@ async fn registration_requires_a_confirmed_conflict_before_replay( message, } if !matches!(status, 200 | 401 | 403) => { let expected_code = match (body_delay < client.connect_timeout, body) { - (true, ERROR_BODY) => Some("registration_denied"), + (true, ERROR_BODY | ERROR_WITH_AUTH_DETAIL_BODY) => Some("registration_denied"), (true, CONFLICT_BODY) => Some("registration_conflict"), + (true, AUTH_UNAVAILABLE_BODY) => Some("authentication_service_unavailable"), _ => None, }; assert_eq!((actual.as_u16(), code.as_deref()), (status, expected_code)); - if expected_code.is_some() { + if expected_code == Some("authentication_service_unavailable") { + assert_eq!(message, "Authentication service unavailable"); + } else if expected_code.is_some() { assert_eq!(message, "registration unavailable"); } } diff --git a/codex-rs/exec-server/tests/relay/registration_retry_tests.rs b/codex-rs/exec-server/tests/relay/registration_retry_tests.rs index 792d909fa0..2df73491ed 100644 --- a/codex-rs/exec-server/tests/relay/registration_retry_tests.rs +++ b/codex-rs/exec-server/tests/relay/registration_retry_tests.rs @@ -1,4 +1,4 @@ -//! Confirmed registration conflicts retry without replacing the Noise identity or session handler. +//! Pre-write registration failures retry without replacing the Noise identity or session handler. use std::time::Duration; @@ -43,6 +43,10 @@ impl RegistryFixture { "security_profile": "noise_hybrid_ik_v1", "executor_registration_id": format!("registration-{attempt}"), })); + } else if status == 502 { + response = response.set_body_json(serde_json::json!({ + "detail": {"code": "authentication_service_unavailable", "message": "Authentication service unavailable"}, + })); } else { response = response.set_body_json(serde_json::json!({ "error": {"code": "registration_conflict", "message": "registration conflicted"}, @@ -139,9 +143,17 @@ impl RegistryFixture { } } +#[test_case::test_case(503; "registration_conflict")] +#[test_case::test_case(502; "authentication_service_unavailable")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn registration_retries_preserve_noise_identity_and_initialized_session() -> Result<()> { - let fixture = RegistryFixture::new(vec![503, 200, 503, 503, 200], Duration::ZERO).await?; +async fn registration_retries_preserve_noise_identity_and_initialized_session( + retry_status: u16, +) -> Result<()> { + let fixture = RegistryFixture::new( + vec![retry_status, 200, retry_status, retry_status, 200], + Duration::ZERO, + ) + .await?; let (shutdown, remote) = fixture.start()?; let harness_identity = NoiseChannelIdentity::generate()?; let (client, first_relay) = fixture diff --git a/codex-rs/rmcp-client/src/http_headers_tests.rs b/codex-rs/rmcp-client/src/http_headers_tests.rs index 292e1b6410..8caf0c0859 100644 --- a/codex-rs/rmcp-client/src/http_headers_tests.rs +++ b/codex-rs/rmcp-client/src/http_headers_tests.rs @@ -76,21 +76,56 @@ async fn helper_attempt_is_shared_after_cancellation() { let dropped = HttpHeadersProvider::new( "https://example.com", &format!( - "printf x > '{}'; sleep 1; printf x > '{}'", + "sleep 0.6; printf '%s\\n' \"$$\" > '{}'; sleep 30; printf x > '{}'", dropped_started.display(), dropped_finished.display(), ), cwd, ) .expect("dropped provider"); - assert!( - tokio::time::timeout(Duration::from_millis(500), dropped.headers()) - .await - .is_err() - ); - assert!(dropped_started.exists()); + // A slow spawn must not race cancellation. Wait for the helper's PID before + // dropping the caller future, leaving the provider as the attempt's owner. + let dropped_pid = tokio::time::timeout(Duration::from_secs(5), async { + tokio::select! { + _ = dropped.headers() => panic!("helper completed before cancellation"), + pid = async { + loop { + match std::fs::read_to_string(&dropped_started) { + Ok(pid) if pid.ends_with('\n') => { + break pid.trim().parse::().expect("helper PID"); + } + Ok(_) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => panic!("failed to read helper PID: {error}"), + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + } => pid, + } + }) + .await + .expect("helper started before cancellation"); + assert!(dropped_pid > 0); drop(dropped); - tokio::time::sleep(Duration::from_millis(1_100)).await; + // Observe cleanup before HELPER_TIMEOUT or the helper's sleep could finish. + tokio::time::timeout(Duration::from_secs(2), async { + loop { + // SAFETY: signal 0 only checks whether the known child PID exists. + let result = unsafe { + libc::kill(dropped_pid, /* sig */ 0) + }; + if result == -1 { + assert_eq!( + std::io::Error::last_os_error().raw_os_error(), + Some(libc::ESRCH) + ); + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("dropping the provider terminated the helper"); assert!(!dropped_finished.exists()); }