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

@@ -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<H: NoiseStreamHandler>(
             }
         }
 
-        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<String, ExecServer
 #[derive(Deserialize)]
 struct RegistryErrorBody {
     error: Option<RegistryError>,
+    detail: Option<serde_json::Value>,
+}
+
+impl RegistryErrorBody {
+    fn into_error(self) -> Option<RegistryError> {
+        // 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::<RegistryErrorBody>(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<String> {
     serde_json::from_str::<RegistryErrorBody>(body)
         .ok()
-        .and_then(|body| body.error)
+        .and_then(RegistryErrorBody::into_error)
         .and_then(|error| error.message)
         .or_else(|| preview_error_body(body))
 }

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;

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)
+        );
+    }
+}

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;
                 }

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<S: tracing::Subscriber> tracing_subscriber::Layer<S> 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<S: tracing::Subscriber> tracing_subscriber::Layer<S> 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");
             }
         }

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