github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-19 08:16:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

Make remote-control conflict retry tests deterministic (#54647)

Browse Source 
## What changed

Wait for the rejected socket to close before advancing Tokio's clock through
capped retry delays. Keep socket I/O on real time so pending handshakes cannot
auto-advance their own timeouts.

Return bodyless HTTP 409 responses in the test server, check that neither
rejection restarts fast retries, and assert that recovery preserves the expected
reconnect attempt count. Both cursor and cursorless cases use the updated helper.

GitOrigin-RevId: da167312f28b7e24b864e6929be749865ea70081
This commit is contained in:
jif
2026-10-16 16:07:25 +00:00
committed by copyberry
parent b3cc59afcf
commit fa46ef3af6
1 changed files with 24 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-transport/src/transport/remote_control/websocket_retry_tests.rs
+24 -18
View File
@@ -13,6 +13,7 @@ use super::*;
use crate::transport::remote_control::protocol::normalize_remote_control_url;
use pretty_assertions::assert_eq;
use tempfile::TempDir;
use tokio::io::AsyncReadExt;
use tokio::net::TcpListener;
use tokio::time::Duration;
use tokio::time::Instant;
@@ -62,8 +63,8 @@ async fn assert_conflict_recovery(subscribe_cursor: Option<&str>) {
shutdown_token.clone(),
Arc::new(enabled_desired_state_sender()),
);
// Start at the cap to exercise two consecutive capped waits without spending
// another minute reaching it. Use the real handshake and reconnect loop.
// Start at the cap and advance the retry waits below. Keep socket I/O on
// real time so a pending handshake cannot auto-advance its own timeout.
websocket.reconnect_attempt = 9;
websocket.state.lock().await.subscribe_cursor = subscribe_cursor.map(str::to_string);
let connect_task = tokio::spawn(async move {
@@ -75,7 +76,7 @@ async fn assert_conflict_recovery(subscribe_cursor: Option<&str>) {
let mut rejected_at = None;
for attempt in 0..3 {
let (stream, _) = timeout(Duration::from_secs(40), listener.accept())
let (mut stream, _) = timeout(Duration::from_secs(40), listener.accept())
.await
.expect("connection should retry within the capped delay")
.unwrap();
@@ -86,7 +87,7 @@ async fn assert_conflict_recovery(subscribe_cursor: Option<&str>) {
);
}
rejected_at = Some(Instant::now());
let result = accept_hdr_async(stream, |request: &Request, response: Response| {
let result = accept_hdr_async(&mut stream, |request: &Request, response: Response| {
assert_eq!(
request.uri().path(),
"/backend-api/wham/remote/control/server"
@@ -104,7 +105,7 @@ async fn assert_conflict_recovery(subscribe_cursor: Option<&str>) {
if attempt < 2 {
Err(tungstenite::http::Response::builder()
.status(/*status*/ 409)
.body(Some("Remote app server already online".to_string()))
.body(/*body*/ None)
.unwrap())
} else {
Ok(response)
@@ -113,23 +114,28 @@ async fn assert_conflict_recovery(subscribe_cursor: Option<&str>) {
.await;
if attempt < 2 {
assert!(result.is_err());
if attempt == 1 {
assert!(
timeout(Duration::from_secs(1), listener.accept())
.await
.is_err(),
"sustained HTTP 409 responses must not restart fast retries"
);
// The real loop is sleeping after the second rejection. Skip the
// middle of this wait, retaining its boundary assertion below.
tokio::time::pause();
tokio::time::advance(Duration::from_secs(13)).await;
tokio::time::resume();
}
drop(result);
// The peer closes the rejected socket while processing the 409.
// On this current-thread runtime it then enters the retry sleep
// before we can observe EOF, so the clock advance cannot race it.
assert_eq!(stream.read(&mut [0]).await.unwrap(), 0);
tokio::time::pause();
tokio::time::advance(Duration::from_secs(13)).await;
tokio::time::resume();
assert!(
timeout(Duration::from_secs(1), listener.accept())
.await
.is_err(),
"sustained HTTP 409 responses must not restart fast retries"
);
tokio::time::pause();
tokio::time::advance(Duration::from_secs(17)).await;
tokio::time::resume();
} else {
let _server_connection = result.expect("available ownership should allow recovery");
let (mut websocket, outcome) = connect_task.await.unwrap();
assert!(matches!(outcome, ConnectOutcome::Connected(_)));
assert_eq!(websocket.reconnect_attempt, 11);
assert!(
next_reconnect_delay(&mut websocket.reconnect_attempt) >= Duration::from_secs(15),
"a successful handshake must not reset backoff before the connection stays healthy"