mirror of
https://github.com/openai/codex.git
synced 2026-11-05 16:29:31 +00:00
Decouple HTTP retry backoff from overload integration testing (#41313)
## Why The overload integration test coupled terminal error behavior to live retry timing and tracing events. ## What changed - Verify retry attempts and exponential backoff directly with `run_with_retry` under paused Tokio time. - Keep the overload integration test focused on request count and terminal error events. - Add timeouts while waiting for retry telemetry, resumed requests, and turn completion so failures cannot hang the test. GitOrigin-RevId: 571a6a6a36a6aded038000a63bff1958c7209913
This commit is contained in:
1 parent
f98649cde9
commit
a73bf25d17
1 file changed
+94
-60
@@ -1,4 +1,9 @@
|
|||||||
use anyhow::Result;
|
use anyhow::Result;
|
||||||
|
use codex_client::RetryOn;
|
||||||
|
use codex_client::RetryPolicy;
|
||||||
|
use codex_client::run_with_retry;
|
||||||
|
use codex_http_client::Request;
|
||||||
|
use codex_http_client::TransportError;
|
||||||
use codex_login::CodexAuth;
|
use codex_login::CodexAuth;
|
||||||
use codex_protocol::protocol::CodexErrorInfo;
|
use codex_protocol::protocol::CodexErrorInfo;
|
||||||
use codex_protocol::protocol::EventMsg;
|
use codex_protocol::protocol::EventMsg;
|
||||||
@@ -10,6 +15,8 @@ use core_test_support::skip_if_no_network;
|
|||||||
use core_test_support::test_codex::TestCodex;
|
use core_test_support::test_codex::TestCodex;
|
||||||
use core_test_support::test_codex::test_codex;
|
use core_test_support::test_codex::test_codex;
|
||||||
use core_test_support::wait_for_event;
|
use core_test_support::wait_for_event;
|
||||||
|
use http::Method;
|
||||||
|
use http::StatusCode;
|
||||||
use pretty_assertions::assert_eq;
|
use pretty_assertions::assert_eq;
|
||||||
use serde_json::json;
|
use serde_json::json;
|
||||||
use std::net::TcpListener;
|
use std::net::TcpListener;
|
||||||
@@ -182,10 +189,9 @@ impl RetryTelemetryCapture {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async fn next_retry(&mut self) -> RetryTelemetryEvent {
|
async fn next_retry(&mut self) -> RetryTelemetryEvent {
|
||||||
let retry = self
|
let retry = tokio::time::timeout(Duration::from_secs(10), self.events.recv())
|
||||||
.events
|
|
||||||
.recv()
|
|
||||||
.await
|
.await
|
||||||
|
.expect("timed out waiting for retry telemetry")
|
||||||
.expect("retry telemetry subscriber should remain installed");
|
.expect("retry telemetry subscriber should remain installed");
|
||||||
// Parallel tests may first register request callsites without our thread-local subscriber.
|
// Parallel tests may first register request callsites without our thread-local subscriber.
|
||||||
tracing::callsite::rebuild_interest_cache();
|
tracing::callsite::rebuild_interest_cache();
|
||||||
@@ -197,11 +203,13 @@ async fn wait_for_retry(
|
|||||||
telemetry: &mut RetryTelemetryCapture,
|
telemetry: &mut RetryTelemetryCapture,
|
||||||
retry: &RetryTelemetryEvent,
|
retry: &RetryTelemetryEvent,
|
||||||
) -> Duration {
|
) -> Duration {
|
||||||
let elapsed = telemetry
|
let elapsed = tokio::time::timeout(
|
||||||
.resumptions
|
retry.delay + Duration::from_secs(10),
|
||||||
.recv()
|
telemetry.resumptions.recv(),
|
||||||
.await
|
)
|
||||||
.expect("retry should start another request after its sleep");
|
.await
|
||||||
|
.expect("timed out waiting for the request after a retry")
|
||||||
|
.expect("retry should start another request after its sleep");
|
||||||
assert!(
|
assert!(
|
||||||
elapsed >= retry.delay,
|
elapsed >= retry.delay,
|
||||||
"{} {} retry waited {elapsed:?}, less than its selected {:?} delay",
|
"{} {} retry waited {elapsed:?}, less than its selected {:?} delay",
|
||||||
@@ -285,12 +293,63 @@ async fn responses_http_uses_local_backoff_despite_retry_after() -> Result<()> {
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Check backoff without a live server or tracing events coordinating the retry loop.
|
||||||
|
#[tokio::test(start_paused = true)]
|
||||||
|
async fn http_retry_backoff_exhausts_attempts() {
|
||||||
|
let attempts = Mutex::new(Vec::new());
|
||||||
|
let result = run_with_retry(
|
||||||
|
RetryPolicy {
|
||||||
|
max_attempts: 2,
|
||||||
|
base_delay: Duration::from_millis(200),
|
||||||
|
retry_on: RetryOn {
|
||||||
|
retry_429: false,
|
||||||
|
retry_5xx: true,
|
||||||
|
retry_transport: false,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
|| Request::new(Method::POST, "http://localhost/v1/responses".into()),
|
||||||
|
|_, attempt| {
|
||||||
|
attempts
|
||||||
|
.lock()
|
||||||
|
.expect("retry attempts should not be poisoned")
|
||||||
|
.push((attempt, tokio::time::Instant::now()));
|
||||||
|
std::future::ready(Err::<(), _>(TransportError::Http {
|
||||||
|
status: StatusCode::SERVICE_UNAVAILABLE,
|
||||||
|
url: None,
|
||||||
|
headers: None,
|
||||||
|
body: None,
|
||||||
|
}))
|
||||||
|
},
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
|
||||||
|
assert!(
|
||||||
|
matches!(result, Err(TransportError::Http { status, .. }) if status == StatusCode::SERVICE_UNAVAILABLE)
|
||||||
|
);
|
||||||
|
let attempts = attempts
|
||||||
|
.into_inner()
|
||||||
|
.expect("retry attempts should not be poisoned");
|
||||||
|
assert_eq!(
|
||||||
|
attempts
|
||||||
|
.iter()
|
||||||
|
.map(|(attempt, _)| *attempt)
|
||||||
|
.collect::<Vec<_>>(),
|
||||||
|
vec![0, 1, 2]
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
(FIRST_RETRY_MIN_DELAY..=FIRST_RETRY_MAX_DELAY).contains(&(attempts[1].1 - attempts[0].1))
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
(SECOND_RETRY_MIN_DELAY..=SECOND_RETRY_MAX_DELAY)
|
||||||
|
.contains(&(attempts[2].1 - attempts[1].1))
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
/// Headerless HTTP overloads currently exhaust request retries before emitting one terminal error.
|
/// Headerless HTTP overloads currently exhaust request retries before emitting one terminal error.
|
||||||
#[tokio::test(flavor = "current_thread")]
|
#[tokio::test(flavor = "current_thread")]
|
||||||
async fn responses_http_overload_without_retry_after_exhausts_request_retries() -> Result<()> {
|
async fn responses_http_overload_without_retry_after_exhausts_request_retries() -> Result<()> {
|
||||||
skip_if_no_network!(Ok(()));
|
skip_if_no_network!(Ok(()));
|
||||||
|
|
||||||
let mut telemetry = RetryTelemetryCapture::install();
|
|
||||||
let server = responses::start_mock_server().await;
|
let server = responses::start_mock_server().await;
|
||||||
Mock::given(method("POST"))
|
Mock::given(method("POST"))
|
||||||
.and(path("/v1/responses"))
|
.and(path("/v1/responses"))
|
||||||
@@ -309,57 +368,36 @@ async fn responses_http_overload_without_retry_after_exhausts_request_retries()
|
|||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
submit_user_input(&test, "reject the disabled model").await?;
|
submit_user_input(&test, "reject the disabled model").await?;
|
||||||
let first_retry = telemetry.next_retry().await;
|
|
||||||
assert!((FIRST_RETRY_MIN_DELAY..FIRST_RETRY_MAX_DELAY).contains(&first_retry.delay));
|
|
||||||
assert_eq!(
|
|
||||||
first_retry,
|
|
||||||
RetryTelemetryEvent {
|
|
||||||
attempt: 1,
|
|
||||||
delay: first_retry.delay,
|
|
||||||
layer: "http".into(),
|
|
||||||
operation: "request".into(),
|
|
||||||
}
|
|
||||||
);
|
|
||||||
wait_for_retry(&mut telemetry, &first_retry).await;
|
|
||||||
let second_retry = telemetry.next_retry().await;
|
|
||||||
assert!((SECOND_RETRY_MIN_DELAY..SECOND_RETRY_MAX_DELAY).contains(&second_retry.delay));
|
|
||||||
assert_eq!(
|
|
||||||
second_retry,
|
|
||||||
RetryTelemetryEvent {
|
|
||||||
attempt: 2,
|
|
||||||
delay: second_retry.delay,
|
|
||||||
layer: "http".into(),
|
|
||||||
operation: "request".into(),
|
|
||||||
}
|
|
||||||
);
|
|
||||||
wait_for_retry(&mut telemetry, &second_retry).await;
|
|
||||||
|
|
||||||
let mut error_events = 0;
|
let mut error_events = 0;
|
||||||
let mut stream_error_events = 0;
|
let mut stream_error_events = 0;
|
||||||
loop {
|
tokio::time::timeout(Duration::from_secs(10), async {
|
||||||
match wait_for_event(&test.codex, |_| true).await {
|
loop {
|
||||||
EventMsg::Error(error) => {
|
match wait_for_event(&test.codex, |_| true).await {
|
||||||
error_events += 1;
|
EventMsg::Error(error) => {
|
||||||
assert_eq!(
|
error_events += 1;
|
||||||
error.codex_error_info,
|
assert_eq!(
|
||||||
Some(CodexErrorInfo::ServerOverloaded)
|
error.codex_error_info,
|
||||||
);
|
Some(CodexErrorInfo::ServerOverloaded)
|
||||||
assert_eq!(
|
);
|
||||||
error.message,
|
assert_eq!(
|
||||||
"Selected model is at capacity. Please try a different model."
|
error.message,
|
||||||
);
|
"Selected model is at capacity. Please try a different model."
|
||||||
|
);
|
||||||
|
}
|
||||||
|
EventMsg::StreamError(_) => stream_error_events += 1,
|
||||||
|
EventMsg::TurnComplete(event) => {
|
||||||
|
assert_eq!(
|
||||||
|
event.error.and_then(|error| error.codex_error_info),
|
||||||
|
Some(CodexErrorInfo::ServerOverloaded)
|
||||||
|
);
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
_ => {}
|
||||||
}
|
}
|
||||||
EventMsg::StreamError(_) => stream_error_events += 1,
|
|
||||||
EventMsg::TurnComplete(event) => {
|
|
||||||
assert_eq!(
|
|
||||||
event.error.and_then(|error| error.codex_error_info),
|
|
||||||
Some(CodexErrorInfo::ServerOverloaded)
|
|
||||||
);
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
_ => {}
|
|
||||||
}
|
}
|
||||||
}
|
})
|
||||||
|
.await
|
||||||
|
.expect("overload retries should finish the turn within 10 seconds");
|
||||||
|
|
||||||
assert_eq!(error_events, 1);
|
assert_eq!(error_events, 1);
|
||||||
assert_eq!(stream_error_events, 0);
|
assert_eq!(stream_error_events, 0);
|
||||||
@@ -374,10 +412,6 @@ async fn responses_http_overload_without_retry_after_exhausts_request_retries()
|
|||||||
request_count, 3,
|
request_count, 3,
|
||||||
"headerless overload should exhaust the configured request retries"
|
"headerless overload should exhaust the configured request retries"
|
||||||
);
|
);
|
||||||
assert_eq!(
|
|
||||||
telemetry.events.try_recv(),
|
|
||||||
Err(mpsc::error::TryRecvError::Empty)
|
|
||||||
);
|
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in new issue
Block a user