fix(responses): align heartbeat and websocket connection lifecycles (#548)

Use passive upstream heartbeats and independent downstream liveness checks.
Reclaim unbound sockets after five idle minutes and release client-owned
resources on disconnect, with safe continuation ownership transfer.

Co-authored-by: Zyy <zyycn@users.noreply.github.com>
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-10-29 00:08:40 +08:00
committed by GitHub
co-authored by Zyy Codex
parent d679d5e708
commit 2f2c1787be
44 changed files with 1296 additions and 837 deletions
@@ -10,6 +10,8 @@ pub(super) const DOWNSTREAM_WRITE_TIMEOUT: Duration = Duration::from_secs(5 * 60
pub(super) const DOWNSTREAM_CLOSE_TIMEOUT: Duration = Duration::from_secs(1);
pub(super) const CONNECTION_MAX_AGE: Duration = Duration::from_secs(60 * 60);
pub(super) const DOWNSTREAM_PING_INTERVAL: Duration = Duration::from_secs(25);
pub(super) const DOWNSTREAM_PONG_TIMEOUT: Duration = Duration::from_secs(30);
pub(super) const DOWNSTREAM_PING_WRITE_TIMEOUT: Duration = Duration::from_secs(5);
/// WebSocket pump 的写入与生命周期预算
#[derive(Clone, Copy)]
@@ -124,6 +126,7 @@ pub enum PumpExitReason {
ReadError,
WriteError,
WriteTimeout,
PongTimeout,
LifecycleShutdown,
ConnectionMaxAge,
CoordinatorDropped,
@@ -141,6 +144,7 @@ impl PumpExitReason {
Self::ReadError => "read_error",
Self::WriteError => "write_error",
Self::WriteTimeout => "write_timeout",
Self::PongTimeout => "pong_timeout",
Self::LifecycleShutdown => "lifecycle_shutdown",
Self::ConnectionMaxAge => "connection_max_age",
Self::CoordinatorDropped => "coordinator_dropped",
@@ -30,6 +30,7 @@ use tokio::{
pub struct ResponsesWebSocketConnection {
connection_id: Arc<str>,
opened_at: Instant,
cancellation: CancellationToken,
expired: Arc<AtomicBool>,
commands: Option<mpsc::Sender<ConnectionCommand>>,
incoming: mpsc::Receiver<PendingConnectionEvent>,
@@ -266,7 +267,9 @@ impl ResponsesWebSocketConnection {
}
fn terminate(&mut self, reason: PumpExitReason) {
let reason = self.exited.try_recv().unwrap_or(reason);
self.exit_reason.get_or_insert(reason);
self.cancellation.cancel();
self.commands.take();
if let Some(task) = self.pump_task.take() {
task.abort();
@@ -306,6 +309,7 @@ impl ResponsesWebSocketConnection {
impl Drop for ResponsesWebSocketConnection {
fn drop(&mut self) {
self.cancellation.cancel();
self.commands.take();
if let Some(task) = self.pump_task.take() {
task.abort();
@@ -314,6 +318,7 @@ impl Drop for ResponsesWebSocketConnection {
}
/// 为 WebSocket transport 启动包含冻结中间件计划的单 owner pump
/// cancellation 必须属于本连接,连接退出或句柄释放时会取消它
pub fn spawn_connection<S, E>(
socket: S,
connection_id: Arc<str>,
@@ -341,7 +346,7 @@ where
incoming_tx,
exit_tx,
Arc::clone(&connection_id),
cancellation,
cancellation.clone(),
opened_at,
Arc::clone(&expired),
Arc::clone(&stats),
@@ -353,6 +358,7 @@ where
ResponsesWebSocketConnection {
connection_id,
opened_at,
cancellation,
expired,
commands: Some(command_tx),
incoming: incoming_rx,
@@ -19,7 +19,7 @@ use std::{
},
};
use tokio::{
sync::{Semaphore, mpsc, oneshot},
sync::{Semaphore, mpsc, oneshot, watch},
time::Instant,
};
@@ -42,6 +42,7 @@ pub(super) async fn run_pump<S, E>(
S: Stream<Item = Result<Message, E>> + Sink<Message, Error = E> + Unpin + Send + 'static,
E: fmt::Display + Send + 'static,
{
let _connection_lifetime = middleware::CancelOnDrop(Some(cancellation.clone()));
let context = Arc::new(PumpContext {
extensions,
connection_id,
@@ -57,11 +58,13 @@ pub(super) async fn run_pump<S, E>(
let (writer, mut reader) = socket.split();
let writer = Arc::new(middleware::Writer::new(writer, context.clone()));
let sender: Arc<dyn contract::Sender> = writer.clone();
// 同一个受管任务同时驱动两侧;等待写入不能阻塞入站控制、关闭或取消
let (activity, received_activity) = watch::channel(());
// 心跳独立于业务写入等待,但共享唯一 writer,避免长写入或排队阻止失联检测
let reason = tokio::select! {
biased;
reason = read_connection(&mut reader, &incoming, sender.clone(), &context) => reason,
reason = read_connection(&mut reader, &incoming, sender.clone(), &activity, &context) => reason,
reason = write_connection(writer.clone(), commands, &context) => reason,
reason = heartbeat_connection(writer.clone(), received_activity, &context) => reason,
};
if matches!(reason, PumpExitReason::ClientClose) {
if let Err(error) = writer.finish_peer_close(&cancellation).await {
@@ -78,6 +81,7 @@ async fn read_connection<S, E>(
socket: &mut S,
incoming: &mpsc::Sender<PendingConnectionEvent>,
sender: Arc<dyn contract::Sender>,
activity: &watch::Sender<()>,
context: &PumpContext,
) -> PumpExitReason
where
@@ -103,6 +107,10 @@ where
}
message = socket.next() => {
if matches!(&message, Some(Ok(_))) { context.stats.record_read(context.opened_at); }
if matches!(&message, Some(Ok(message)) if !matches!(message, Message::Close(_))) {
// 链路活动以真实入站为准,插件丢弃消息不应抹掉对端已经响应的事实
activity.send_replace(());
}
let message = match message {
Some(Ok(message)) => match middleware::transform(message, contract::Direction::Incoming, sender.clone(), context).await {
Ok(Some(message)) => Some(Ok(message)),
@@ -145,40 +153,22 @@ where
S: Sink<Message, Error = E> + Unpin + Send + 'static,
E: fmt::Display + Send + 'static,
{
let mut heartbeat = tokio::time::interval_at(
context.opened_at + DOWNSTREAM_PING_INTERVAL,
DOWNSTREAM_PING_INTERVAL,
);
heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
let mut ping_sequence = 0_u64;
loop {
let message = tokio::select! {
let command = tokio::select! {
biased;
() = context.cancellation.cancelled() => return PumpExitReason::LifecycleShutdown,
command = commands.recv() => {
let Some(command) = command else { return PumpExitReason::CoordinatorDropped; };
let ConnectionCommand { message, context: write_context, acknowledged } = command;
let closing = matches!(message, Message::Close(_));
let result = match middleware::transform(message, contract::Direction::Outgoing, writer.clone(), context).await {
Ok(Some(message)) => writer.write(message).await.map(|()| WriteOutcome::Written),
Ok(None) => Ok(WriteOutcome::Suppressed),
Err(error) => Err(ConnectionWriteError::Transport { message: error.to_string() }),
};
let reason = result.as_ref().err().map(|error| write_exit_reason(error, context));
if let Err(error) = &result {
tracing::debug!(websocket_connection_id = %context.connection_id, request_id = write_context.request_id.as_deref().unwrap_or(""), frame_phase = write_context.phase.as_str(), error = %error, "Responses WebSocket pump transport write failed");
}
let _ = acknowledged.send(result);
if let Some(reason) = reason { return reason; }
if closing { return PumpExitReason::ServerClose; }
continue;
}
_ = heartbeat.tick() => {
ping_sequence = ping_sequence.wrapping_add(1);
Message::Ping(ping_sequence.to_be_bytes().to_vec().into())
}
command = commands.recv() => command,
};
let message = match middleware::transform(
let Some(command) = command else {
return PumpExitReason::CoordinatorDropped;
};
let ConnectionCommand {
message,
context: write_context,
acknowledged,
} = command;
let closing = matches!(message, Message::Close(_));
let result = match middleware::transform(
message,
contract::Direction::Outgoing,
writer.clone(),
@@ -186,21 +176,125 @@ where
)
.await
{
Ok(Some(message)) => message,
Ok(None) => continue,
Err(_) => return PumpExitReason::MiddlewareFailed,
Ok(Some(message)) => writer.write(message).await.map(|()| WriteOutcome::Written),
Ok(None) => Ok(WriteOutcome::Suppressed),
Err(error) => Err(ConnectionWriteError::Transport {
message: error.to_string(),
}),
};
if let Err(error) = writer.write(message).await {
tracing::info!(websocket_connection_id = %context.connection_id, error = %error, "Responses WebSocket control write failed");
return write_exit_reason(&error, context);
let reason = result
.as_ref()
.err()
.map(|error| write_exit_reason(error, context));
if let Err(error) = &result {
tracing::debug!(websocket_connection_id = %context.connection_id, request_id = write_context.request_id.as_deref().unwrap_or(""), frame_phase = write_context.phase.as_str(), error = %error, "Responses WebSocket pump transport write failed");
}
let _ = acknowledged.send(result);
if let Some(reason) = reason {
return reason;
}
if closing {
return PumpExitReason::ServerClose;
}
context
.stats
.ping_written_count
.fetch_add(1, Ordering::Relaxed);
}
}
async fn heartbeat_connection<S, E>(
writer: Arc<middleware::Writer<S>>,
mut activity: watch::Receiver<()>,
context: &PumpContext,
) -> PumpExitReason
where
S: Sink<Message, Error = E> + Unpin + Send + 'static,
E: fmt::Display + Send + 'static,
{
let mut heartbeat = tokio::time::interval_at(
context.opened_at + DOWNSTREAM_PING_INTERVAL,
DOWNSTREAM_PING_INTERVAL,
);
heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
let mut ping_sequence = 0_u64;
let pong_deadline = tokio::time::sleep(DOWNSTREAM_PONG_TIMEOUT);
tokio::pin!(pong_deadline);
let mut pending_ping = false;
loop {
tokio::select! {
biased;
() = context.cancellation.cancelled() => return PumpExitReason::LifecycleShutdown,
changed = activity.changed() => {
if changed.is_err() { return PumpExitReason::PumpStopped; }
pending_ping = false;
continue;
}
() = &mut pong_deadline, if pending_ping => return PumpExitReason::PongTimeout,
_ = heartbeat.tick() => {}
}
ping_sequence = ping_sequence.wrapping_add(1);
let write = tokio::time::timeout(
DOWNSTREAM_PING_WRITE_TIMEOUT,
write_heartbeat(&writer, ping_sequence, context),
);
tokio::pin!(write);
let mut received_during_write = false;
loop {
tokio::select! {
biased;
() = context.cancellation.cancelled() => return PumpExitReason::LifecycleShutdown,
changed = activity.changed() => {
if changed.is_err() { return PumpExitReason::PumpStopped; }
pending_ping = false;
received_during_write = true;
}
() = &mut pong_deadline, if pending_ping => return PumpExitReason::PongTimeout,
result = &mut write => {
match result {
Ok(Ok(true)) => {
context.stats.ping_written_count.fetch_add(1, Ordering::Relaxed);
// 重复 Ping 不顺延截止,写入完成前已经收到的响应也不能丢失
if !pending_ping && !received_during_write {
pong_deadline.as_mut().reset(Instant::now() + DOWNSTREAM_PONG_TIMEOUT);
pending_ping = true;
}
}
Ok(Ok(false)) => {}
Ok(Err(reason)) => return reason,
Err(_) => return PumpExitReason::WriteTimeout,
}
break;
}
}
}
}
}
async fn write_heartbeat<S, E>(
writer: &Arc<middleware::Writer<S>>,
sequence: u64,
context: &PumpContext,
) -> Result<bool, PumpExitReason>
where
S: Sink<Message, Error = E> + Unpin + Send + 'static,
E: fmt::Display + Send + 'static,
{
let message = middleware::transform(
Message::Ping(sequence.to_be_bytes().to_vec().into()),
contract::Direction::Outgoing,
writer.clone(),
context,
)
.await
.map_err(|_| PumpExitReason::MiddlewareFailed)?;
let Some(message) = message else {
return Ok(false);
};
let is_ping = matches!(message, Message::Ping(_));
writer.write(message).await.map_err(|error| {
tracing::info!(websocket_connection_id = %context.connection_id, error = %error, "Responses WebSocket control write failed");
write_exit_reason(&error, context)
})?;
Ok(is_ping)
}
fn write_exit_reason(error: &ConnectionWriteError, context: &PumpContext) -> PumpExitReason {
match error {
ConnectionWriteError::Timeout { .. } => PumpExitReason::WriteTimeout,
@@ -181,13 +181,15 @@ async fn serve_responses_websocket(socket: WebSocket, session: ResponsesWebSocke
"Responses WebSocket connected"
);
let _connection_guard = connection_guard;
let cancellation = lifecycle.cancellation();
let cancellation = lifecycle.cancellation().child_token();
let _connection_cancellation =
crate::middleware::websocket::CancelOnDrop(Some(cancellation.clone()));
let request_headers =
request_headers.with_downstream_websocket_connection_id(connection_id.clone());
let mut connection = ResponsesWebSocketConnection::new(
socket,
connection_id,
cancellation,
cancellation.clone(),
middleware,
encode_headers(&raw_headers).into(),
);
@@ -299,7 +301,11 @@ async fn serve_responses_websocket(socket: WebSocket, session: ResponsesWebSocke
}
execution.prepare_execution(client).await
};
let prepared = match preparation.await {
let prepared = match tokio::select! {
biased;
_ = connection.wait_for_exit() => break,
prepared = preparation => prepared,
} {
Ok(prepared) => prepared,
Err(error) => {
trace_rejected_request(
@@ -317,6 +323,7 @@ async fn serve_responses_websocket(socket: WebSocket, session: ResponsesWebSocke
continue;
}
};
let prepared = prepared.with_connection_cancellation(cancellation.clone());
let request_id = Arc::<str>::from(prepared.request_id().to_string());
response_control = prepared.response_control();
let capture = new_replay_capture();
@@ -557,7 +557,7 @@ async fn outbound_commands_are_written_in_acknowledged_order() {
}
#[tokio::test(start_paused = true)]
async fn stalled_write_times_out_at_three_hundred_seconds_and_aborts_the_pump() {
async fn stalled_business_write_cannot_block_the_heartbeat_write_deadline() {
let PumpHarness {
mut connection,
incoming: _incoming,
@@ -566,6 +566,7 @@ async fn stalled_write_times_out_at_three_hundred_seconds_and_aborts_the_pump()
..
} = test_connection(true);
let request_id = Arc::<str>::from("req_test");
let started = tokio::time::Instant::now();
let error = connection
.send_text(
@@ -573,12 +574,14 @@ async fn stalled_write_times_out_at_three_hundred_seconds_and_aborts_the_pump()
WriteContext::request(&request_id, FramePhase::First),
)
.await
.expect_err("stalled write must time out");
.expect_err("heartbeat deadline must interrupt the stalled business write");
tokio::task::yield_now().await;
assert!(
matches!(error, ConnectionWriteError::Timeout { timeout } if timeout == Duration::from_secs(300))
&& dropped.load(Ordering::Acquire)
assert!(matches!(error, ConnectionWriteError::Closed) && dropped.load(Ordering::Acquire));
assert_eq!(started.elapsed(), Duration::from_secs(25 + 5));
assert_eq!(
connection.wait_for_exit().await,
PumpExitReason::WriteTimeout
);
}
@@ -586,10 +589,19 @@ async fn stalled_write_times_out_at_three_hundred_seconds_and_aborts_the_pump()
async fn pump_marks_the_connection_expired_at_sixty_minutes() {
let PumpHarness {
mut connection,
incoming: _incoming,
written: _written,
incoming,
mut written,
..
} = test_connection(false);
let peer = tokio::spawn(async move {
while let Some(message) = written.recv().await {
if let Message::Ping(payload) = message {
incoming
.send(Ok(Message::Pong(payload)))
.expect("reply to Ping");
}
}
});
let event = connection
.next_event()
@@ -597,6 +609,7 @@ async fn pump_marks_the_connection_expired_at_sixty_minutes() {
.expect("deadline must emit an event");
assert!(matches!(event, ConnectionEvent::Expired) && connection.is_expired());
peer.abort();
}
#[tokio::test]
@@ -606,6 +619,7 @@ async fn dropping_the_connection_aborts_and_drops_the_socket_owner() {
incoming: _incoming,
written: _written,
dropped,
cancellation,
..
} = test_connection(false);
@@ -613,6 +627,7 @@ async fn dropping_the_connection_aborts_and_drops_the_socket_owner() {
tokio::task::yield_now().await;
assert!(dropped.load(Ordering::Acquire));
assert!(cancellation.is_cancelled());
}
#[tokio::test(start_paused = true)]
@@ -1122,7 +1137,7 @@ async fn client_close_should_cancel_active_execution_without_starting_queued_req
async fn idle_pump_sends_heartbeat_without_bursting_after_a_delay() {
let PumpHarness {
connection,
incoming: _incoming,
incoming,
mut written,
..
} = test_connection(false);
@@ -1132,6 +1147,10 @@ async fn idle_pump_sends_heartbeat_without_bursting_after_a_delay() {
tokio::time::advance(Duration::from_secs(25)).await;
let first = written.recv().await.expect("first heartbeat");
assert!(matches!(first, Message::Ping(_)));
incoming
.send(Ok(Message::Pong(vec![1].into())))
.expect("reply before scheduler delay");
tokio::task::yield_now().await;
tokio::time::advance(Duration::from_secs(120)).await;
let second = written.recv().await.expect("heartbeat after delay");
@@ -1143,7 +1162,35 @@ async fn idle_pump_sends_heartbeat_without_bursting_after_a_delay() {
}
#[tokio::test(start_paused = true)]
async fn missing_pongs_do_not_cancel_the_connection_or_block_business_frames() {
async fn missing_responses_expire_from_the_first_ping_without_extending_on_repeated_pings() {
let PumpHarness {
mut connection,
incoming,
mut written,
dropped,
cancellation,
..
} = test_connection(false);
let started = tokio::time::Instant::now();
for _ in 0..32 {
incoming
.send(Ok(Message::Text("queued request".into())))
.expect("fill the business queue before probing");
}
assert_eq!(
connection.wait_for_exit().await,
PumpExitReason::PongTimeout
);
assert_eq!(started.elapsed(), Duration::from_secs(25 + 30));
assert!(matches!(written.try_recv(), Ok(Message::Ping(_))));
assert!(matches!(written.try_recv(), Ok(Message::Ping(_))));
assert!(dropped.load(Ordering::Acquire));
assert!(cancellation.is_cancelled());
}
#[tokio::test(start_paused = true)]
async fn business_frames_keep_the_connection_healthy_without_pongs() {
let PumpHarness {
mut connection,
incoming,
@@ -1153,15 +1200,14 @@ async fn missing_pongs_do_not_cancel_the_connection_or_block_business_frames() {
for _ in 0..13 {
tokio::time::advance(Duration::from_secs(25)).await;
assert!(matches!(written.recv().await, Some(Message::Ping(_))));
incoming
.send(Ok(Message::Text("still connected".into())))
.expect("send business frame instead of Pong");
assert!(matches!(
connection.next_event().await,
Some(ConnectionEvent::Text(text)) if text == "still connected"
));
}
incoming
.send(Ok(Message::Text("still connected".into())))
.expect("send business frame after 325 seconds without Pong");
assert!(matches!(
connection.next_event().await,
Some(ConnectionEvent::Text(text)) if text == "still connected"
));
incoming.send(Ok(Message::Close(None))).expect("close peer");
assert_eq!(
connection.wait_for_exit().await,
@@ -1192,7 +1238,7 @@ async fn stalled_heartbeat_can_be_cancelled_without_waiting_for_write_timeout()
}
#[tokio::test(start_paused = true)]
async fn stalled_heartbeat_obeys_the_existing_write_timeout() {
async fn stalled_heartbeat_uses_a_short_write_timeout() {
let PumpHarness {
mut connection,
incoming: _incoming,
@@ -1205,7 +1251,7 @@ async fn stalled_heartbeat_obeys_the_existing_write_timeout() {
connection.wait_for_exit().await,
PumpExitReason::WriteTimeout
);
assert_eq!(started.elapsed(), Duration::from_secs(25 + 300));
assert_eq!(started.elapsed(), Duration::from_secs(25 + 5));
assert!(dropped.load(Ordering::Acquire));
}
@@ -211,6 +211,15 @@ impl PreparedRootExecution {
self.response_control.clone()
}
/// 连接退出同时撤销跨轮次保留的上游资源,不改变单次执行的取消范围
#[must_use]
pub fn with_connection_cancellation(mut self, cancellation: CancellationToken) -> Self {
self.response_control = self
.response_control
.with_connection_cancellation(cancellation);
self
}
pub(super) fn new(client: AuthenticatedClient) -> Result<Self, GatewayError> {
let started_at = SystemTime::now();
let deadline_at = Deadline::from_timeout(started_at, client.execution_timeout())
@@ -3,6 +3,7 @@
use std::fmt;
use std::sync::{Arc, Mutex, Weak};
use crate::lifecycle::CancellationToken;
use async_trait::async_trait;
/// Provider 实现的连接控制通道,消息内容由上游协议解释
@@ -22,6 +23,7 @@ pub struct ResponseControlUnavailable;
#[derive(Clone, Default)]
pub struct ResponseControl {
target: Arc<Mutex<Option<Weak<dyn ResponseControlTransport>>>>,
connection_cancellation: Option<CancellationToken>,
}
/// 某次已交付响应所属的弱连接快照,重绑后不能把旧消息发送到新连接
@@ -50,6 +52,18 @@ impl fmt::Debug for ResponseControl {
}
impl ResponseControl {
/// 下游连接跨多轮存续,不能使用单次响应的取消信号代替
#[must_use]
pub fn with_connection_cancellation(mut self, cancellation: CancellationToken) -> Self {
self.connection_cancellation = Some(cancellation);
self
}
#[must_use]
pub fn connection_cancellation(&self) -> Option<&CancellationToken> {
self.connection_cancellation.as_ref()
}
/// 仅保存弱引用,不延长旧 attempt 的连接生命周期
#[must_use]
pub fn binding(&self) -> Option<ResponseControlBinding> {
@@ -169,15 +169,22 @@ pub struct CodexWebSocketPoolSettings {
pub enabled: bool,
pub max_age_ms: u64,
pub max_connecting: usize,
#[serde(default = "default_websocket_pool_idle_timeout_ms")]
pub idle_timeout_ms: u64,
pub stream_idle_timeout_ms: u64,
}
const fn default_websocket_pool_idle_timeout_ms() -> u64 {
300_000
}
impl Default for CodexWebSocketPoolSettings {
fn default() -> Self {
Self {
enabled: true,
max_age_ms: 55 * 60 * 1000,
max_connecting: 8,
idle_timeout_ms: default_websocket_pool_idle_timeout_ms(),
stream_idle_timeout_ms: 300_000,
}
}
@@ -185,7 +192,7 @@ impl Default for CodexWebSocketPoolSettings {
impl CodexWebSocketPoolSettings {
fn validate(&self) -> Result<(), OpenAiConfigError> {
if self.max_age_ms == 0 || self.max_connecting == 0 {
if self.max_age_ms == 0 || self.max_connecting == 0 || self.idle_timeout_ms == 0 {
return Err(OpenAiConfigError::InvalidField("openai.ws_pool"));
}
Ok(())
@@ -196,6 +203,7 @@ impl CodexWebSocketPoolSettings {
enabled: self.enabled,
max_age: Duration::from_millis(self.max_age_ms),
max_connecting: self.max_connecting,
idle_timeout: Duration::from_millis(self.idle_timeout_ms),
stream_idle_timeout: (self.stream_idle_timeout_ms != 0)
.then(|| Duration::from_millis(self.stream_idle_timeout_ms)),
..CodexWebSocketPoolConfig::default()
@@ -827,7 +827,7 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
&authorization,
cookie_header.as_ref(),
account_selection,
).with_trace(&trace),
).with_trace(&trace).with_response_control(context.response_control()),
active_account.id().as_str(),
context.deadline(),
&cancellation,
@@ -1246,7 +1246,7 @@ fn websocket_diagnostic_message(error: &CodexWebSocketExchangeError) -> Provider
let mut message =
format!("OpenAI WebSocket stream ended before a terminal response ({reason})");
if let Some(timeout) = timeout {
message.push_str(&format!("; local keepalive timeout after {timeout:?}"));
message.push_str(&format!("; local Pong write timeout after {timeout:?}"));
}
if let Some(last_event_type) = last_event_type {
message.push_str("; last event type: ");
@@ -1108,7 +1108,6 @@ impl CodexProvider {
.map_err(|error| map_client_error(error, UpstreamSendState::NotSent, false).error)?
.with_responses_api_base_url(lease.authentication().responses_api_base_url())
.with_connection_budget(context.connection_budget().clone())
.with_response_control(context.response_control().cloned())
.with_middleware_headers(middleware_headers)
.with_outbound_middleware(outbound_middleware)
.with_privacy(context.privacy(), context.cancellation().clone()),
@@ -678,6 +678,7 @@ pub(super) fn codex_request_context<'a>(
account_selection: CodexAccountSelectionTelemetry<'a>,
) -> CodexRequestContext<'a> {
CodexRequestContext {
response_control: None,
trace: None,
authorization: authorization.expose_secret(),
account_id: account.upstream_account_id(),
@@ -464,6 +464,8 @@ impl<'a> CodexAccountSelectionTelemetry<'a> {
/// 单次 Codex 上游请求的上下文
#[derive(Clone, Copy)]
pub struct CodexRequestContext<'a> {
/// 本次根执行的控制与下游连接生命周期,不由共享 HTTP client 持有
pub response_control: Option<&'a gateway_core::engine::response_control::ResponseControl>,
/// 显式传递的请求诊断上下文,跨连接任务共享
pub trace: Option<&'a gateway_core::diagnostics::TraceContext>,
/// Provider 已构造并脱敏持有的完整 Authorization 值
@@ -512,6 +514,7 @@ impl<'a> CodexRequestContext<'a> {
) -> Self {
Self {
trace: None,
response_control: None,
authorization,
account_id,
request_id,
@@ -537,6 +540,15 @@ impl<'a> CodexRequestContext<'a> {
self.trace = Some(trace);
self
}
#[must_use]
pub fn with_response_control(
mut self,
control: Option<&'a gateway_core::engine::response_control::ResponseControl>,
) -> Self {
self.response_control = control;
self
}
}
impl fmt::Debug for CodexRequestContext<'_> {
@@ -689,7 +701,6 @@ pub struct CodexBackendClient {
gateway_core::lifecycle::CancellationToken,
)>,
pub(super) timezone: gateway_core::time::DeploymentTimeZone,
pub(super) response_control: Option<gateway_core::engine::response_control::ResponseControl>,
pub(super) connection_budget: Option<gateway_core::engine::connection::ConnectionBudget>,
pub(super) client: Client,
pub(super) direct_client: Client,
@@ -716,14 +727,6 @@ impl CodexBackendClient {
self
}
pub(crate) fn with_response_control(
mut self,
control: Option<gateway_core::engine::response_control::ResponseControl>,
) -> Self {
self.response_control = control;
self
}
pub(crate) fn with_connection_budget(
mut self,
budget: gateway_core::engine::connection::ConnectionBudget,
@@ -69,7 +69,6 @@ impl CodexBackendClient {
privacy: None,
timezone: Default::default(),
connection_budget: None,
response_control: None,
direct_client: client.clone(),
client,
websocket_origin_key: websocket_origin_key(&base_url),
@@ -115,6 +114,14 @@ impl CodexBackendClient {
upstream_request: &CodexResponsesRequest,
context: CodexRequestContext<'_>,
) -> CodexClientResult<CodexBackendStreamingResponse> {
if let (Some(pool), Some(connection_id)) = (
&self.websocket_pool,
upstream_request
.downstream_websocket_connection_id
.as_deref(),
) {
pool.release_downstream_connections(connection_id).await;
}
let profile = self.profile.snapshot();
let mut headers =
self.request_headers_for_http_response(&profile, upstream_request, context)?;
@@ -395,7 +402,21 @@ impl CodexBackendClient {
let pool_key =
self.websocket_pool_key(request, context, pool_account_id, &connection_profile);
let pool_log_context = pool_key.as_ref().map(WebSocketPoolLogContext::from_key);
let pool = self.websocket_pool.as_deref().map(|pool| (pool, pool_key));
let downstream_cancellation = context.response_control.and_then(
gateway_core::engine::response_control::ResponseControl::connection_cancellation,
);
if pool_key.is_none()
&& let (Some(pool), Some(connection_id)) = (
&self.websocket_pool,
request.downstream_websocket_connection_id.as_deref(),
)
{
pool.release_downstream_connections(connection_id).await;
}
let pool = self
.websocket_pool
.as_deref()
.map(|pool| (pool, pool_key, downstream_cancellation));
let fast_path_budget = match requirement {
TransportRequirement::PersistedContinuation | TransportRequirement::NewChain => {
Some(WEBSOCKET_FAST_PATH_BUDGET)
@@ -553,7 +574,7 @@ impl CodexBackendClient {
let mut exchange = execute_prepared_response_create_request_stream(
&websocket_request,
prepared,
self.response_control.clone(),
context.response_control.cloned(),
context
.trace
.cloned()
@@ -8,6 +8,7 @@ use gateway_core::engine::response_control::{
};
use tokio::sync::{Mutex, mpsc, oneshot};
use tokio_tungstenite::tungstenite::{self, Message};
use tokio_util::sync::CancellationToken;
use super::CodexWebSocketExchangeError;
@@ -25,6 +26,7 @@ pub(super) type SharedMessages = Arc<Mutex<mpsc::Receiver<Result<Message, tungst
pub(super) struct PumpControl {
pub(super) tx_command: mpsc::Sender<PumpCommand>,
pub(super) rx_message: SharedMessages,
pub(super) cancellation: CancellationToken,
}
pub(super) async fn send_message(
@@ -45,17 +47,23 @@ pub(super) async fn send_message(
#[async_trait]
impl ResponseControlTransport for PumpControl {
async fn send(&self, payload: &str) -> Result<(), ResponseControlUnavailable> {
tokio::time::timeout(
WEBSOCKET_SEND_TIMEOUT,
send_message(&self.tx_command, Message::Text(payload.to_owned().into())),
)
.await
.map_err(|_| ResponseControlUnavailable)?
.map_err(|_| ResponseControlUnavailable)
tokio::select! {
biased;
() = self.cancellation.cancelled() => Err(ResponseControlUnavailable),
result = tokio::time::timeout(
WEBSOCKET_SEND_TIMEOUT,
send_message(&self.tx_command, Message::Text(payload.to_owned().into())),
) => result.map_err(|_| ResponseControlUnavailable)?.map_err(|_| ResponseControlUnavailable),
}
}
async fn receive(&self) -> Result<String, ResponseControlUnavailable> {
match self.rx_message.lock().await.recv().await {
let message = tokio::select! {
biased;
() = self.cancellation.cancelled() => return Err(ResponseControlUnavailable),
message = async { self.rx_message.lock().await.recv().await } => message,
};
match message {
Some(Ok(Message::Text(text))) => Ok(text.to_string()),
_ => Err(ResponseControlUnavailable),
}
@@ -29,7 +29,7 @@ use super::{
WebSocketPoolConnectLease, WebSocketPoolConnectOutcome, WebSocketPoolDecision,
WebSocketPoolLease,
},
pump::{PumpKeepalive, PumpLogContext, PumpedWebSocket, transport_metric_reason},
pump::{PumpLogContext, PumpedWebSocket, transport_metric_reason},
};
pub(crate) const WEBSOCKET_FAST_PATH_BUDGET: Duration = Duration::from_millis(800);
@@ -112,7 +112,11 @@ impl PreparedWebSocket {
/// 只建立或租用 WebSocket,不发送 payload
pub(crate) async fn prepare_response_create_request_with_pool(
request: &CodexWebSocketRequest,
pool: Option<(&CodexWebSocketPool, Option<CodexWebSocketPoolKey>)>,
pool: Option<(
&CodexWebSocketPool,
Option<CodexWebSocketPoolKey>,
Option<&gateway_core::lifecycle::CancellationToken>,
)>,
breaker: &WebSocketOriginBreaker,
origin_key: &str,
fast_path_budget: Option<Duration>,
@@ -120,7 +124,7 @@ pub(crate) async fn prepare_response_create_request_with_pool(
fallback_stream_idle_timeout: Option<Duration>,
) -> Result<WebSocketFastPath<PreparedWebSocket>, CodexWebSocketExchangeError> {
let decision_started_at = Instant::now();
let Some((pool, Some(key))) = pool else {
let Some((pool, Some(key), downstream_cancellation)) = pool else {
if require_pool || !request.continuation().permits_fresh_connection() {
return Err(continuation_unavailable(
PreviousResponseUnavailableReason::PoolUnavailable,
@@ -133,7 +137,7 @@ pub(crate) async fn prepare_response_create_request_with_pool(
fast_path_budget,
fallback_stream_idle_timeout,
decision_started_at,
pool.map(|(pool, _)| pool),
pool.map(|(pool, _, _)| pool),
)
.await;
};
@@ -151,6 +155,7 @@ pub(crate) async fn prepare_response_create_request_with_pool(
&key,
required_response_id,
request.connection().outbound_proxy.as_ref(),
downstream_cancellation,
)
.await
{
@@ -237,7 +242,6 @@ async fn prepare_unpooled_websocket(
let context = pump_log_context_from_connection(request.connection());
let connected = connect_with_budget(
request.connection(),
PumpKeepalive::disabled(),
fast_path_budget,
context,
proxy_permit,
@@ -343,7 +347,6 @@ fn start_pooled_websocket_connect(
let task_key = connect_lease.key().clone();
let started_at = connect_lease.started_at();
let cancellation = connect_lease.cancellation_token();
let keepalive = pool.keepalive();
let context = PumpLogContext::new(
Some(task_key.account_id().to_owned()),
Some(task_key.conversation_id_hash()),
@@ -365,7 +368,7 @@ fn start_pooled_websocket_connect(
);
return;
}
result = connect_websocket_connection(&connection, keepalive, context, fast_path, proxy_permit) => result,
result = connect_websocket_connection(&connection, context, fast_path, proxy_permit) => result,
};
match finish_breaker_attempt(permit, connected) {
Ok((connection, connect_elapsed)) => {
@@ -465,7 +468,6 @@ fn pump_log_context_from_connection(connection: &CodexWebSocketConnection) -> Pu
async fn connect_with_budget(
connection: &CodexWebSocketConnection,
keepalive: PumpKeepalive,
fast_path_budget: Option<Duration>,
context: PumpLogContext,
proxy_permit: Option<pool::ProxyConnectionPermit>,
@@ -475,7 +477,6 @@ async fn connect_with_budget(
fast_path_budget,
connect_websocket_connection(
connection,
keepalive,
context,
fast_path_budget.is_some(),
proxy_permit,
@@ -490,14 +491,13 @@ async fn connect_with_budget(
async fn connect_websocket_connection(
connection: &CodexWebSocketConnection,
keepalive: PumpKeepalive,
context: PumpLogContext,
fast_path: bool,
proxy_permit: Option<pool::ProxyConnectionPermit>,
) -> Result<(PooledWebSocketConnection, Duration), CodexWebSocketExchangeError> {
let started_at = Instant::now();
let (websocket, response) =
connect_pumped_websocket(connection, keepalive, context, fast_path, proxy_permit).await?;
connect_pumped_websocket(connection, context, fast_path, proxy_permit).await?;
Ok((
PooledWebSocketConnection {
websocket,
@@ -69,6 +69,10 @@ pub(in crate::transport::websocket) fn stream_websocket_response(
response_control: Option<ResponseControl>,
) -> CodexWebSocketStreamingExchange {
websocket.bind_control(response_control.as_ref());
let downstream_cancellation = response_control
.as_ref()
.and_then(ResponseControl::connection_cancellation)
.cloned();
let websocket_connection_id = websocket.connection_id();
let response_metadata = metadata.clone();
let rate_limit_updates = Arc::new(Mutex::new(Vec::new()));
@@ -94,6 +98,7 @@ pub(in crate::transport::websocket) fn stream_websocket_response(
stream_idle_timeout,
trace,
shutdown,
downstream_cancellation,
rate_limit_updates: rate_limit_updates_for_task,
response_metadata_updates: response_metadata_updates_for_task,
tx,
@@ -133,6 +138,7 @@ struct WebSocketStreamForwardState {
reused_connection: bool,
stream_idle_timeout: Option<Duration>,
shutdown: CancellationToken,
downstream_cancellation: Option<gateway_core::lifecycle::CancellationToken>,
rate_limit_updates: CodexWebSocketRateLimitUpdates,
response_metadata_updates: CodexWebSocketResponseMetadataUpdates,
tx: mpsc::Sender<Result<Bytes, CodexWebSocketExchangeError>>,
@@ -147,6 +153,7 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) {
reused_connection,
stream_idle_timeout,
shutdown,
downstream_cancellation,
rate_limit_updates,
response_metadata_updates,
tx,
@@ -163,6 +170,20 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) {
loop {
let message = tokio::select! {
biased;
() = async {
match &downstream_cancellation {
Some(cancellation) => cancellation.cancelled().await,
None => std::future::pending().await,
}
} => {
trace.record("upstream.cancelled", json!({"reason": "downstream_connection_closed"}));
discard_stream_websocket(
websocket,
pool_return,
StreamWebSocketDiscardReason::ClientCancelled,
).await;
return;
}
// 客户端断开下游 SSE 流:立即丢弃连接并释放池 slot,
// 不再傻等上游 idle 超时(否则同会话后续请求会一直 bypass/busy)
() = tx.closed() => {
@@ -382,10 +403,7 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) {
reason => CodexWebSocketExchangeError::StreamEndedBeforeTerminal {
reason: reason.map_or("stream_eof", PumpExitReason::as_str),
timeout: match reason {
Some(
PumpExitReason::PongTimeout { timeout }
| PumpExitReason::LivenessTimeout { timeout },
) => Some(*timeout),
Some(PumpExitReason::PongWriteTimeout { timeout }) => Some(*timeout),
_ => None,
},
last_event_type,
@@ -414,7 +432,7 @@ async fn finish_stream_websocket(
pool_return: Option<WebSocketStreamPoolReturn>,
) {
let Some(pool_return) = pool_return else {
websocket.close().await;
websocket.close_and_wait().await;
return;
};
pool_return
@@ -30,7 +30,7 @@ use super::{
error::CodexWebSocketExchangeError,
model::{CodexWebSocketConnection, CodexWebSocketRequest, WebSocketContinuationRequirement},
pool::CodexWebSocketConnectionMetadata,
pump::{PumpKeepalive, PumpLogContext, PumpedWebSocket, RawWsStream},
pump::{PumpLogContext, PumpedWebSocket, RawWsStream},
};
use super::control::WEBSOCKET_SEND_TIMEOUT;
@@ -153,16 +153,12 @@ pub(super) async fn connect_sideband_websocket(
pub(super) async fn connect_pumped_websocket(
connection: &CodexWebSocketConnection,
keepalive: PumpKeepalive,
context: PumpLogContext,
fast_path: bool,
proxy_permit: Option<super::pool::ProxyConnectionPermit>,
) -> Result<(PumpedWebSocket, WsResponse<Option<Vec<u8>>>), CodexWebSocketExchangeError> {
let (raw, response) = connect_websocket(connection, fast_path).await?;
Ok((
PumpedWebSocket::new(raw, keepalive, context, proxy_permit),
response,
))
Ok((PumpedWebSocket::new(raw, context, proxy_permit), response))
}
pub(super) async fn send_websocket_request(
@@ -98,7 +98,7 @@ impl CodexWebSocketPool {
.filter_map(|(key, slot)| match slot {
WebSocketPoolSlot::Idle { idle_since, .. }
if key.proxy_capacity_key == proxy_key
&& key.downstream_connection_id.is_empty() =>
&& !state.has_downstream_binding(key) =>
{
Some((key, *idle_since))
}
@@ -0,0 +1,134 @@
//! 下游连接的取消监督与池内归属转移
use std::sync::Arc;
use gateway_core::lifecycle::CancellationToken;
use tokio::time::Instant;
use super::state::{
PooledWebSocketConnection, WebSocketPoolSlot, WebSocketPoolState, close_pooled_connections,
};
use super::{CodexWebSocketPool, CodexWebSocketPoolKey, WebSocketPoolConnectOutcome};
impl CodexWebSocketPool {
pub(super) fn register_downstream(
&self,
key: &CodexWebSocketPoolKey,
cancellation: &CancellationToken,
) {
if key.downstream_connection_id.is_empty() {
return;
}
let connection_id = key.downstream_connection_id.clone();
{
let mut state = self.lock_state();
if state.shutting_down || state.downstream_connections.contains_key(&connection_id) {
return;
}
state
.downstream_connections
.insert(connection_id.clone(), cancellation.clone());
}
let inner = Arc::downgrade(&self.inner);
let cancellation = cancellation.clone();
let shutdown = self.shutdown.clone();
let capacity_changed = Arc::clone(&self.capacity_changed);
drop(self.tasks.spawn(async move {
tokio::select! {
biased;
() = shutdown.cancelled() => return,
() = cancellation.cancelled() => {}
}
let Some(inner) = inner.upgrade() else { return };
let close = {
let mut state = inner
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
let close = state.take_downstream_connections(
&connection_id,
None,
"downstream_connection_closed",
);
state.downstream_connections.remove(&connection_id);
close
};
close_pooled_connections(close).await;
capacity_changed.notify_waiters();
}));
}
/// 实际切到 HTTP 或无池连接时撤销旧 WS 绑定,保留客户端本身的生命周期监督
pub(crate) async fn release_downstream_connections(&self, connection_id: &str) {
let close = self.lock_state().take_downstream_connections(
connection_id,
None,
"downstream_rebound",
);
close_pooled_connections(close).await;
self.capacity_changed.notify_waiters();
}
pub(super) fn close_retired_connections(&self, connections: Vec<PooledWebSocketConnection>) {
if connections.is_empty() {
return;
}
let capacity_changed = Arc::clone(&self.capacity_changed);
drop(self.tasks.spawn(async move {
close_pooled_connections(connections).await;
capacity_changed.notify_waiters();
}));
}
}
impl WebSocketPoolState {
pub(super) fn has_downstream_binding(&self, key: &CodexWebSocketPoolKey) -> bool {
self.downstream_connections
.get(&key.downstream_connection_id)
.is_some_and(|token| !token.is_cancelled())
}
pub(super) fn downstream_closed(&self, key: &CodexWebSocketPoolKey) -> bool {
self.downstream_connections
.get(&key.downstream_connection_id)
.is_some_and(CancellationToken::is_cancelled)
}
pub(super) fn take_downstream_connections(
&mut self,
connection_id: &str,
keep: Option<&CodexWebSocketPoolKey>,
reason: &'static str,
) -> Vec<PooledWebSocketConnection> {
if connection_id.is_empty() {
return Vec::new();
}
let keys = self
.slots
.keys()
.filter(|key| key.downstream_connection_id == connection_id && Some(*key) != keep)
.cloned()
.collect::<Vec<_>>();
let mut close = Vec::new();
for key in keys {
match self.slots.remove(&key) {
Some(WebSocketPoolSlot::Idle { connection, .. }) => {
self.remember_continuation_loss(
&key,
connection.continuation.latest_response_id(),
connection.websocket.observation().with_exit_reason(reason),
Instant::now(),
);
close.push(*connection);
}
Some(WebSocketPoolSlot::Connecting(connecting)) => {
connecting.cancellation.cancel();
connecting
.outcome
.send_replace(WebSocketPoolConnectOutcome::Failed);
}
Some(WebSocketPoolSlot::Busy(_)) | None => {}
}
}
close
}
}
@@ -1,6 +1,7 @@
//! Codex WebSocket 连接池
mod capacity;
mod downstream;
mod lease;
mod state;
mod supervisor;
@@ -18,9 +19,8 @@ use uuid::Uuid;
use self::state::{
WebSocketPoolConnecting, WebSocketPoolSlot, WebSocketPoolState, close_pooled_connection,
close_pooled_connections,
close_pooled_connections, idle_connection_loss,
};
use super::pump::PumpKeepalive;
use super::pump::WebSocketConnectionObservation;
use self::capacity::ProxyCapacity;
@@ -38,10 +38,8 @@ pub(crate) use self::{
const DEFAULT_MAX_CONNECTING: usize = 16;
const DEFAULT_MAX_AGE: Duration = Duration::from_mins(55);
const DEFAULT_MAINTENANCE_INTERVAL: Duration = Duration::from_secs(25);
const DEFAULT_PING_INTERVAL: Duration = Duration::from_secs(25);
// 心跳也覆盖正在生成的连接,给短时链路停顿留出恢复余量
const DEFAULT_PING_TIMEOUT: Duration = Duration::from_secs(30);
const DEFAULT_MAINTENANCE_INTERVAL: Duration = Duration::from_secs(5);
const DEFAULT_IDLE_TIMEOUT: Duration = Duration::from_secs(300);
pub(crate) const DEFAULT_STREAM_IDLE_TIMEOUT: Duration = Duration::from_secs(300);
/// WebSocket 连接池
@@ -74,12 +72,8 @@ pub struct CodexWebSocketPoolConfig {
pub max_connecting: usize,
/// 后台维护间隔;`None` 表示不启动后台任务
pub maintenance_interval: Option<Duration>,
/// 池化连接的探活 ping 间隔(包括正在生成的连接);`None` 表示不主动 ping
pub ping_interval: Option<Duration>,
/// 发送 ping 后等待任意入站帧的超时时间;零值表示不校验 Pong deadline
pub ping_timeout: Duration,
/// idle socket 无活动多久后视为失活
pub liveness_timeout: Option<Duration>,
/// 未绑定下游 WS 的连接归池后可空闲的时间,不由 Ping/Pong 刷新
pub idle_timeout: Duration,
/// 等待下一条上游消息的空闲超时;`None` 或零值使用默认 300 秒
pub stream_idle_timeout: Option<Duration>,
}
@@ -91,35 +85,18 @@ impl Default for CodexWebSocketPoolConfig {
max_age: DEFAULT_MAX_AGE,
max_connecting: DEFAULT_MAX_CONNECTING,
maintenance_interval: Some(DEFAULT_MAINTENANCE_INTERVAL),
ping_interval: Some(DEFAULT_PING_INTERVAL),
ping_timeout: DEFAULT_PING_TIMEOUT,
// idle 连接不设失活截断:靠 ping/pong 保活,只在 max_age(55 分钟)
// 或 ping 失败时关闭,维持跨轮可复用连接
liveness_timeout: None,
idle_timeout: DEFAULT_IDLE_TIMEOUT,
stream_idle_timeout: Some(DEFAULT_STREAM_IDLE_TIMEOUT),
}
}
}
impl CodexWebSocketPoolConfig {
/// pump 后台任务的保活策略:从连接池配置派生出 ping/pong 与 liveness 策略
pub(crate) fn keepalive(&self) -> PumpKeepalive {
PumpKeepalive {
ping_interval: self.ping_interval,
ping_timeout: (!self.ping_timeout.is_zero()).then_some(self.ping_timeout),
liveness_timeout: self.liveness_timeout,
}
}
}
impl CodexWebSocketPool {
/// 使用指定连接总寿命构造连接池,不启动定时保活与清理
/// 使用指定连接总寿命构造连接池,不启动定时清理
pub fn new(max_age: Duration) -> Self {
Self::with_config(CodexWebSocketPoolConfig {
max_age,
maintenance_interval: None,
ping_interval: None,
liveness_timeout: None,
..CodexWebSocketPoolConfig::default()
})
}
@@ -140,11 +117,6 @@ impl CodexWebSocketPool {
pool
}
/// pump 后台任务的保活策略(供建连时传入)
pub(crate) fn keepalive(&self) -> PumpKeepalive {
self.config.keepalive()
}
/// 等待下一条上游消息的空闲超时;`None` 或零值使用默认 300 秒
pub(crate) fn stream_idle_timeout(&self) -> Option<Duration> {
self.config.stream_idle_timeout
@@ -160,17 +132,26 @@ impl CodexWebSocketPool {
key: &CodexWebSocketPoolKey,
required_response_id: Option<&str>,
proxy: Option<&gateway_core::account::OutboundProxy>,
downstream_cancellation: Option<&gateway_core::lifecycle::CancellationToken>,
) -> WebSocketPoolAcquire {
self.spawn_maintenance_task();
if let Some(cancellation) = downstream_cancellation {
self.register_downstream(key, cancellation);
}
let mut proxy_permit = None;
loop {
let mut connections_to_close = Vec::new();
let acquire = {
let mut state = self.lock_state();
let mut continuation_loss = None;
if !self.config.enabled || state.shutting_down {
if !self.config.enabled
|| state.shutting_down
|| downstream_cancellation
.is_some_and(gateway_core::lifecycle::CancellationToken::is_cancelled)
{
return WebSocketPoolAcquire::Bypass(WebSocketPoolBypassReason::Disabled);
}
let requested_key = key;
let key = if let Some(response_id) = required_response_id {
let Some(key) = state.slots.iter().find_map(|(candidate, slot)| {
(candidate.same_logical_connection(key)
@@ -201,35 +182,57 @@ impl CodexWebSocketPool {
});
}
Some(WebSocketPoolSlot::Idle { .. }) => {
let Some(WebSocketPoolSlot::Idle { connection, .. }) =
state.slots.remove(&key)
let Some(WebSocketPoolSlot::Idle {
mut connection,
idle_since,
}) = state.slots.remove(&key)
else {
return WebSocketPoolAcquire::Bypass(WebSocketPoolBypassReason::Busy);
};
// 零成本探活:后台 pump 已实时感知连接死亡(RST/Close/EOF/失活),
// 复用前只需读取 is_closed 标志,避免复用到静默死连接后卡到超时
let expired = connection.created_at.elapsed() >= self.config.max_age;
let closed = connection.websocket.is_closed();
if !expired && !closed {
// 取得时与后台扫描共用到期判断,避免维护周期内重新借出过期连接
let loss = idle_connection_loss(
state.has_downstream_binding(&key),
&connection,
idle_since,
tokio::time::Instant::now(),
&self.config,
)
.or_else(|| {
state.downstream_closed(&key).then(|| {
connection
.websocket
.observation()
.with_exit_reason("downstream_connection_closed")
})
});
let Some(observation) = loss else {
connection.websocket.revoke_control();
let retired = state.take_downstream_connections(
&requested_key.downstream_connection_id,
Some(&key),
"downstream_rebound",
);
// 精确续接可接手另一条下游连接的 socket,池键必须跟随新 owner
let key = key.with_downstream_connection_id(
requested_key.downstream_connection_id.clone(),
);
let lease = WebSocketPoolLease::reserve(
self.clone(),
key.clone(),
connection.continuation.latest_response_id(),
);
state.slots.insert(
key.clone(),
WebSocketPoolSlot::Busy(lease.reservation.clone()),
);
state
.slots
.insert(key, WebSocketPoolSlot::Busy(lease.reservation.clone()));
self.close_retired_connections(retired);
return WebSocketPoolAcquire::Reused { connection, lease };
}
let observation = if expired && !closed {
connection
.websocket
.observation()
.with_exit_reason("max_age_expired")
} else {
connection.websocket.observation()
};
tracing::info!(
websocket_connection_id = %connection.websocket.connection_id(),
exit_reason = observation.exit_reason(),
idle_age_ms = idle_since.elapsed().as_millis(),
"Removed unavailable idle WebSocket before reuse"
);
state.remember_continuation_loss(
&key,
connection.continuation.latest_response_id(),
@@ -244,6 +247,11 @@ impl CodexWebSocketPool {
None => {}
}
connections_to_close.extend(state.take_downstream_connections(
&requested_key.downstream_connection_id,
Some(&key),
"downstream_rebound",
));
let acquire = if let Some(observation) = continuation_loss {
Some(WebSocketPoolAcquire::ContinuationLost(observation))
} else if required_response_id.is_some() {
@@ -285,7 +293,17 @@ impl CodexWebSocketPool {
if let Some(acquire) = acquire {
return acquire;
}
match self.acquire_proxy_capacity(proxy, Some(key)).await {
let capacity = tokio::select! {
biased;
() = async {
match downstream_cancellation {
Some(cancellation) => cancellation.cancelled().await,
None => std::future::pending().await,
}
} => return WebSocketPoolAcquire::Bypass(WebSocketPoolBypassReason::Disabled),
capacity = self.acquire_proxy_capacity(proxy, Some(key)) => capacity,
};
match capacity {
Ok(permit) => proxy_permit = permit,
Err(_) => return WebSocketPoolAcquire::Bypass(WebSocketPoolBypassReason::Disabled),
}
@@ -5,6 +5,7 @@ use std::{
time::Duration,
};
use futures::{StreamExt, stream::FuturesUnordered};
use sha2::{Digest, Sha256};
use tokio::{sync::watch, time::Instant};
use tokio_util::sync::CancellationToken;
@@ -107,6 +108,7 @@ const MAX_CONTINUATION_TOMBSTONES: usize = 4_096;
#[derive(Default)]
pub(super) struct WebSocketPoolState {
pub(super) slots: HashMap<CodexWebSocketPoolKey, WebSocketPoolSlot>,
pub(super) downstream_connections: HashMap<String, gateway_core::lifecycle::CancellationToken>,
continuation_tombstones: VecDeque<WebSocketContinuationTombstone>,
pub(super) shutting_down: bool,
}
@@ -237,18 +239,33 @@ pub(super) async fn close_pooled_connection(connection: PooledWebSocketConnectio
}
pub(super) async fn close_pooled_connections(connections: Vec<PooledWebSocketConnection>) {
for connection in connections {
close_pooled_connection(connection).await;
}
// 每条连接独立执行有界关闭,避免批量回收累计握手等待并拖延后续扫描
let mut closing = connections
.into_iter()
.map(close_pooled_connection)
.collect::<FuturesUnordered<_>>();
while closing.next().await.is_some() {}
}
/// idle 连接是否应从池中摘除:被后台 pump 标记死亡,或已超过 `max_age`
pub(super) fn should_close_idle_connection(
/// 取得连接和维护扫描共用的失效判断,保留原生续接丢失的具体原因
pub(super) fn idle_connection_loss(
has_downstream_binding: bool,
connection: &PooledWebSocketConnection,
idle_since: Instant,
now: Instant,
max_age: Duration,
) -> bool {
connection.websocket.is_closed() || now.duration_since(connection.created_at) >= max_age
config: &super::CodexWebSocketPoolConfig,
) -> Option<WebSocketConnectionObservation> {
if connection.websocket.is_closed() {
return Some(connection.websocket.observation());
}
let reason = if now.duration_since(connection.created_at) >= config.max_age {
"max_age_expired"
} else if !has_downstream_binding && now.duration_since(idle_since) >= config.idle_timeout {
"idle_timeout_expired"
} else {
return None;
};
Some(connection.websocket.observation().with_exit_reason(reason))
}
fn short_sha256<'a>(parts: impl IntoIterator<Item = &'a str>) -> String {
@@ -10,8 +10,7 @@ use tokio::time::Instant;
use super::CodexWebSocketPool;
use super::lease::WebSocketPoolConnectOutcome;
use super::state::{
PooledWebSocketConnection, WebSocketPoolSlot, close_pooled_connections,
should_close_idle_connection,
PooledWebSocketConnection, WebSocketPoolSlot, close_pooled_connections, idle_connection_loss,
};
impl CodexWebSocketPool {
@@ -71,11 +70,7 @@ impl CodexWebSocketPool {
self.tasks.wait().await;
}
/// 维护池 slot:清扫已死亡或超龄的 idle 连接,以及异常残留的 Busy reservation
///
/// 保活(ping/pong)与失活检测已下沉到每条连接的 pump 任务内部,此处不再做
/// 同步 ping 探活,只负责把「后台已标记 closed」或「超过 max_age」的 idle
/// 连接从池中摘除并关闭,避免它们占用 slot
/// 清扫已关闭、超龄或空闲到期的连接,以及超龄残留的 reservation
pub async fn maintain_idle_connections(&self) {
let close = self.take_expired_slots().await;
close_pooled_connections(close).await;
@@ -156,43 +151,50 @@ impl CodexWebSocketPool {
.slots
.iter()
.filter_map(|(key, slot)| match slot {
WebSocketPoolSlot::Idle { connection, .. }
if should_close_idle_connection(connection, now, self.config.max_age) =>
{
Some(key.clone())
}
WebSocketPoolSlot::Idle {
connection,
idle_since,
} => idle_connection_loss(
state.has_downstream_binding(key),
connection,
*idle_since,
now,
&self.config,
)
.map(|observation| (key.clone(), Some(observation))),
WebSocketPoolSlot::Busy(reservation)
if now.duration_since(reservation.reserved_at) >= self.config.max_age =>
{
Some(key.clone())
Some((key.clone(), None))
}
WebSocketPoolSlot::Connecting(connecting)
if now.duration_since(connecting.started_at) >= self.config.max_age =>
{
Some(key.clone())
Some((key.clone(), None))
}
WebSocketPoolSlot::Idle { .. }
| WebSocketPoolSlot::Busy(_)
| WebSocketPoolSlot::Connecting(_) => None,
WebSocketPoolSlot::Busy(_) | WebSocketPoolSlot::Connecting(_) => None,
})
.collect::<Vec<_>>();
for key in keys {
for (key, observation) in keys {
match state.slots.remove(&key) {
Some(WebSocketPoolSlot::Idle { connection, .. }) => {
let observation = if connection.websocket.is_closed() {
connection.websocket.observation()
} else {
connection
.websocket
.observation()
.with_exit_reason("max_age_expired")
};
state.remember_continuation_loss(
&key,
connection.continuation.latest_response_id(),
observation,
now,
);
Some(WebSocketPoolSlot::Idle {
connection,
idle_since,
}) => {
if let Some(observation) = observation {
tracing::info!(
websocket_connection_id = %connection.websocket.connection_id(),
exit_reason = observation.exit_reason(),
idle_age_ms = now.duration_since(idle_since).as_millis(),
"Removed unavailable idle WebSocket during maintenance"
);
state.remember_continuation_loss(
&key,
connection.continuation.latest_response_id(),
observation,
now,
);
}
close.push(*connection);
}
Some(WebSocketPoolSlot::Busy(reservation)) => {
@@ -1,17 +1,7 @@
//! 常驻后台的 WebSocket 泵(对齐官方 Codex CLI 的连接管理)
//! 单 owner 驱动上游 WebSocket,被动回复心跳并在终止后有界释放 socket
//!
//! 每条上游 WebSocket 连接都由一个后台 pump 任务独占:
//! - 持续读取 socket:一旦观察到 `Close` / EOF / 传输错误,立即把连接标记为 `closed`
//! - 自动回应上游 `Ping`,并按 `ping_interval` 主动 `Ping`;收到非关闭入站帧即解除本次心跳 deadline
//! - 可选 `liveness_timeout`:长时间无任何入站活动时判定连接失活并退出
//!
//! 因此空闲连接的“死没死”在后台被实时感知;复用方只需零成本读取 [`PumpedWebSocket::is_closed`],
//! 无需在请求路径上做同步探活 ping(消除“复用到静默死连接才卡住超时”的长尾)
//!
//! 收发都通过 channel 与 pump 任务交互:
//! - 发送:`send` 走 command channel,等待 pump 回执
//! - 接收:`next` 从 message channel 取出 pump 转发的入站帧(`Ping`/`Pong` 已被 pump 吞掉)
//! - 入站缓冲满时暂停读取 socket 和主动探活,仅继续处理发送/关闭命令;消费恢复后按原顺序继续转发
//! 业务消息通过有界队列交付,Ping/Pong 由 pump 消费,不刷新生成期间的消息等待超时
//! 入站背压暂停读取,仍可接收关闭命令;已观察到的终止先释放 socket,再等待终止消息交付
use gateway_core::engine::response_control::{ResponseControl, ResponseControlTransport};
@@ -22,11 +12,7 @@ use std::sync::{
use std::time::Duration;
use futures::{Sink, SinkExt, Stream, StreamExt};
use tokio::{
sync::mpsc,
task::JoinHandle,
time::{Instant, MissedTickBehavior},
};
use tokio::{sync::mpsc, task::JoinHandle, time::Instant};
use tokio_tungstenite::tungstenite::Message;
use uuid::Uuid;
@@ -55,6 +41,8 @@ impl<T> WebSocketIo for T where
pub(crate) type RawWsStream = Box<dyn WebSocketIo>;
const PUMP_COMMAND_BUFFER: usize = 32;
const CLOSE_TIMEOUT: Duration = Duration::from_secs(1);
const PONG_WRITE_TIMEOUT: Duration = Duration::from_secs(5);
/// pump 日志的账号级关联上下文;空闲连接被上游重置等场景用于回溯归属
#[derive(Debug, Clone, Default)]
@@ -72,28 +60,6 @@ impl PumpLogContext {
}
}
/// pump 保活策略
#[derive(Debug, Clone, Copy)]
pub(crate) struct PumpKeepalive {
/// 主动 `Ping` 间隔;`None` 表示 pump 不主动 ping(仅被动读取 + 回应上游 ping)
pub(crate) ping_interval: Option<Duration>,
/// 主动 `Ping` 发出后等待任意入站帧的 deadline;`None` 表示不校验响应
pub(crate) ping_timeout: Option<Duration>,
/// 无入站活动多久后判定失活;`None` 表示只靠显式 close/传输错误发现死亡
pub(crate) liveness_timeout: Option<Duration>,
}
impl PumpKeepalive {
/// 不做主动保活(用于即用即弃的非池化连接)
pub(crate) fn disabled() -> Self {
Self {
ping_interval: None,
ping_timeout: None,
liveness_timeout: None,
}
}
}
const PUMP_MESSAGE_BUFFER: usize = 64;
#[derive(Debug, Clone)]
@@ -112,10 +78,7 @@ pub(crate) enum PumpExitReason {
message: String,
metric_reason: &'static str,
},
PongTimeout {
timeout: Duration,
},
LivenessTimeout {
PongWriteTimeout {
timeout: Duration,
},
KeepaliveTransportError {
@@ -134,8 +97,7 @@ impl PumpExitReason {
Self::UpstreamCloseFrame { .. } => "upstream_close_frame",
Self::UpstreamEof => "upstream_eof",
Self::InboundTransportError { .. } => "inbound_transport_error",
Self::PongTimeout { .. } => "pong_timeout",
Self::LivenessTimeout { .. } => "liveness_timeout",
Self::PongWriteTimeout { .. } => "pong_write_timeout",
Self::KeepaliveTransportError { .. } => "keepalive_transport_error",
Self::MessageReceiverClosed => "message_receiver_closed",
}
@@ -147,9 +109,7 @@ impl PumpExitReason {
| Self::InboundTransportError { message, .. }
| Self::KeepaliveTransportError { message, .. } => Some(message.clone()),
Self::UpstreamCloseFrame { close } => close.as_ref().map(ToString::to_string),
Self::PongTimeout { timeout } | Self::LivenessTimeout { timeout } => {
Some(format!("{timeout:?}"))
}
Self::PongWriteTimeout { timeout } => Some(format!("{timeout:?}")),
Self::CommandChannelClosed
| Self::LocalClose
| Self::UpstreamEof
@@ -172,8 +132,7 @@ impl PumpExitReason {
}
}
Self::UpstreamEof => "eof",
Self::PongTimeout { .. } => "pong_timeout",
Self::LivenessTimeout { .. } => "liveness_timeout",
Self::PongWriteTimeout { .. } => "pong_write_timeout",
Self::MessageReceiverClosed => "message_receiver_closed",
}
}
@@ -191,8 +150,7 @@ impl PumpExitReason {
Self::OutboundTransportError { .. }
| Self::UpstreamEof
| Self::InboundTransportError { .. }
| Self::PongTimeout { .. }
| Self::LivenessTimeout { .. }
| Self::PongWriteTimeout { .. }
| Self::KeepaliveTransportError { .. }
)
}
@@ -202,8 +160,6 @@ impl PumpExitReason {
self,
Self::CommandChannelClosed
| Self::UpstreamCloseFrame { .. }
| Self::PongTimeout { .. }
| Self::LivenessTimeout { .. }
| Self::MessageReceiverClosed
)
}
@@ -214,12 +170,6 @@ struct PendingInbound {
exit_reason: Option<PumpExitReason>,
}
#[derive(Debug, Clone, Copy)]
struct PendingPing {
payload: [u8; 8],
deadline: Instant,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WebSocketConnectionObservation {
connection_id: Uuid,
@@ -283,7 +233,7 @@ pub(crate) struct PumpedWebSocket {
connection_id: Uuid,
tx_command: mpsc::Sender<PumpCommand>,
rx_message: SharedMessages,
control: Option<Arc<dyn ResponseControlTransport>>,
control: Option<Arc<PumpControl>>,
closed: Arc<AtomicBool>,
lifecycle: Arc<Mutex<PumpLifecycleState>>,
pump: Option<JoinHandle<()>>,
@@ -308,7 +258,6 @@ impl PumpedWebSocket {
/// 用底层流启动一个 pump 任务并返回句柄
pub(crate) fn new(
inner: RawWsStream,
keepalive: PumpKeepalive,
context: PumpLogContext,
proxy_permit: Option<super::pool::ProxyConnectionPermit>,
) -> Self {
@@ -335,7 +284,6 @@ impl PumpedWebSocket {
inner,
rx_command,
tx_message,
keepalive,
context,
closed_for_task,
lifecycle_for_task,
@@ -363,16 +311,26 @@ impl PumpedWebSocket {
/// 每轮发送后替换控制 owner,旧执行的弱引用不能控制下一轮连接
pub(crate) fn bind_control(&mut self, control: Option<&ResponseControl>) {
self.revoke_control();
self.control = control.map(|control| {
let transport: Arc<dyn ResponseControlTransport> = Arc::new(PumpControl {
let transport = Arc::new(PumpControl {
tx_command: self.tx_command.clone(),
rx_message: Arc::clone(&self.rx_message),
cancellation: tokio_util::sync::CancellationToken::new(),
});
control.bind(&transport);
let target: Arc<dyn ResponseControlTransport> = transport.clone();
control.bind(&target);
transport
});
}
/// 弱引用升级后的在途读取仍持有强引用,复用前必须主动撤销,释放消息接收锁
pub(crate) fn revoke_control(&mut self) {
if let Some(control) = self.control.take() {
control.cancellation.cancel();
}
}
/// 取出下一帧入站消息(`Ping`/`Pong` 已被 pump 处理,不会到达这里)
///
/// 返回 `None` 表示连接已结束(pump 已退出且缓冲已排空)
@@ -406,7 +364,10 @@ impl PumpedWebSocket {
}
pub(crate) async fn close_and_wait(self) {
self.close().await;
if !self.is_closed() {
// 预算覆盖入队和回执,已阻塞的业务写入不能拖住连接回收
let _ = tokio::time::timeout(CLOSE_TIMEOUT, self.send(Message::Close(None))).await;
}
self.abort_and_wait().await;
}
@@ -418,13 +379,6 @@ impl PumpedWebSocket {
let _ = pump.await;
}
}
/// 主动关闭连接(best-effort 发送 Close 帧)
pub(crate) async fn close(&self) {
if !self.is_closed() {
let _ = self.send(Message::Close(None)).await;
}
}
}
impl Drop for PumpedWebSocket {
@@ -435,38 +389,16 @@ impl Drop for PumpedWebSocket {
}
}
#[expect(
clippy::too_many_arguments,
reason = "pump owns one flat set of channel/state handles and log context"
)]
async fn pump_loop(
connection_id: Uuid,
mut inner: PumpSocket,
mut rx_command: mpsc::Receiver<PumpCommand>,
tx_message: mpsc::Sender<Result<Message, tungstenite::Error>>,
keepalive: PumpKeepalive,
context: PumpLogContext,
closed: Arc<AtomicBool>,
lifecycle: Arc<Mutex<PumpLifecycleState>>,
) {
let mut last_activity = lifecycle
.lock()
.expect("WebSocket pump lifecycle lock poisoned")
.last_activity;
let ping_interval = keepalive.ping_interval.filter(|d| !d.is_zero());
let ping_timeout = keepalive.ping_timeout.filter(|d| !d.is_zero());
let liveness_timeout = keepalive.liveness_timeout.filter(|d| !d.is_zero());
let mut ping_ticker = ping_interval.map(|d| {
// 首个 tick 推迟一整个间隔:tokio::time::interval 默认会让首个 tick 立即就绪,
// 否则连接一建立就会立刻发一帧 Ping,与首个请求 Text 抢跑、打乱帧序
let mut ticker = tokio::time::interval_at(tokio::time::Instant::now() + d, d);
ticker.set_missed_tick_behavior(MissedTickBehavior::Skip);
ticker
});
let mut pending_inbound: Option<PendingInbound> = None;
let mut pending_ping: Option<PendingPing> = None;
let mut ping_sequence = 0_u64;
let mut pending_inbound = None;
let mut backpressure_events = 0_u64;
let reason = 'pump: loop {
@@ -479,12 +411,7 @@ async fn pump_loop(
let pending = pending_inbound
.take()
.expect("pending inbound frame must exist");
permit.send(pending.item);
if let Some(reason) = pending.exit_reason {
break 'pump reason;
}
// 背压期间无法读取 socket,不能把这段本地下游阻塞误算成上游静默
last_activity = Instant::now();
permit.send(pending);
}
command = rx_command.recv() => {
let Some(command) = command else {
@@ -498,8 +425,6 @@ async fn pump_loop(
continue;
}
let pong_deadline = pending_ping.map(|ping| ping.deadline);
let liveness_deadline = liveness_timeout.map(|timeout| last_activity + timeout);
tokio::select! {
command = rx_command.recv() => {
let Some(command) = command else {
@@ -515,17 +440,22 @@ async fn pump_loop(
Some(Ok(message)) => {
// Close 是终止信号,不能刷新链路健康时间或抹掉关闭前的静默时长
if !matches!(message, Message::Close(_)) {
last_activity = Instant::now();
record_activity(&lifecycle, last_activity);
observe_inbound(connection_id, &mut pending_ping, &message);
record_activity(&lifecycle, Instant::now());
}
match message {
Message::Ping(payload) => {
if let Err(error) = inner.stream.send(Message::Pong(payload)).await {
break 'pump PumpExitReason::KeepaliveTransportError {
match tokio::time::timeout(
PONG_WRITE_TIMEOUT,
inner.stream.send(Message::Pong(payload)),
).await {
Ok(Ok(())) => {}
Ok(Err(error)) => break 'pump PumpExitReason::KeepaliveTransportError {
message: error.to_string(),
metric_reason: transport_metric_reason(&error),
};
},
Err(_) => break 'pump PumpExitReason::PongWriteTimeout {
timeout: PONG_WRITE_TIMEOUT,
},
}
}
Message::Pong(_) => {}
@@ -580,28 +510,6 @@ async fn pump_loop(
}
}
}
_ = tick(&mut ping_ticker) => {
if pending_ping.is_none() {
match send_keepalive_ping(
&mut inner.stream,
&mut ping_sequence,
ping_timeout,
).await {
Ok(pending) => pending_ping = pending,
Err(reason) => break 'pump reason,
}
}
}
_ = wait_until(pong_deadline) => {
break 'pump PumpExitReason::PongTimeout {
timeout: ping_timeout.expect("pending ping must have a timeout"),
};
}
_ = wait_until(liveness_deadline) => {
break 'pump PumpExitReason::LivenessTimeout {
timeout: liveness_timeout.expect("liveness deadline must have a timeout"),
};
}
}
};
@@ -626,13 +534,18 @@ async fn pump_loop(
);
if reason.should_send_close() {
let _ = inner.stream.send(Message::Close(None)).await;
let _ = tokio::time::timeout(CLOSE_TIMEOUT, inner.stream.send(Message::Close(None))).await;
}
// 消费方可能仍在排空业务消息,连接和容量不能继续等它读取终止消息
drop(inner);
if let Some(pending) = pending_inbound {
let _ = tx_message.send(pending).await;
}
}
fn enqueue_inbound(
tx_message: &mpsc::Sender<Result<Message, tungstenite::Error>>,
pending_inbound: &mut Option<PendingInbound>,
pending_inbound: &mut Option<Result<Message, tungstenite::Error>>,
connection_id: Uuid,
backpressure_events: &mut u64,
inbound: PendingInbound,
@@ -649,8 +562,8 @@ fn enqueue_inbound(
"Responses WebSocket pump applying inbound backpressure"
);
}
*pending_inbound = Some(PendingInbound { item, exit_reason });
None
*pending_inbound = Some(item);
exit_reason
}
Err(mpsc::error::TrySendError::Closed(_)) => Some(PumpExitReason::MessageReceiverClosed),
}
@@ -687,22 +600,6 @@ async fn handle_command(
}
}
fn observe_inbound(connection_id: Uuid, pending_ping: &mut Option<PendingPing>, message: &Message) {
let Some(pending) = pending_ping else {
return;
};
if let Message::Pong(payload) = message
&& payload.as_ref() != pending.payload
{
tracing::debug!(
websocket_connection_id = %connection_id,
"Responses WebSocket pump received a non-matching Pong"
);
}
// 匹配 Pong 是探针的直接回执;其他入站帧同样证明上游链路仍可达
*pending_ping = None;
}
fn record_activity(lifecycle: &Mutex<PumpLifecycleState>, observed_at: Instant) {
lifecycle
.lock()
@@ -748,26 +645,6 @@ fn duration_millis(duration: Duration) -> u64 {
u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
}
async fn send_keepalive_ping(
inner: &mut RawWsStream,
sequence: &mut u64,
ping_timeout: Option<Duration>,
) -> Result<Option<PendingPing>, PumpExitReason> {
let payload = sequence.to_be_bytes();
*sequence = sequence.wrapping_add(1);
inner
.send(Message::Ping(payload.to_vec().into()))
.await
.map_err(|error| PumpExitReason::KeepaliveTransportError {
message: error.to_string(),
metric_reason: transport_metric_reason(&error),
})?;
Ok(ping_timeout.map(|timeout| PendingPing {
payload,
deadline: Instant::now() + timeout,
}))
}
fn log_pump_exit(
connection_id: Uuid,
context: &PumpLogContext,
@@ -823,21 +700,3 @@ fn log_pump_exit(
);
}
}
/// 等待 ping ticker;`None` 时永远挂起,让 `select!` 分支实际禁用
async fn tick(ticker: &mut Option<tokio::time::Interval>) {
match ticker {
Some(ticker) => {
ticker.tick().await;
}
None => std::future::pending::<()>().await,
}
}
/// 等待可选 deadline;`None` 时永远挂起,让 `select!` 分支实际禁用
async fn wait_until(deadline: Option<Instant>) {
match deadline {
Some(deadline) => tokio::time::sleep_until(deadline).await,
None => std::future::pending::<()>().await,
}
}
@@ -1,6 +1,6 @@
//! 验证 OpenAI 配置解析、路径派生与上游地址约束
use std::path::Path;
use std::{path::Path, time::Duration};
use provider_openai::config::{MAX_STREAM_MAX_RETRIES, OpenAiConfig};
@@ -95,6 +95,42 @@ fn openai_stream_retry_budget_uses_the_official_hard_cap() {
);
}
#[test]
fn websocket_pool_idle_timeout_accepts_existing_configs_and_explicit_overrides() {
for (idle_timeout_ms, expected) in [(None, 300_000), (Some(600_000), 600_000)] {
let mut document = serde_json::json!({
"ws_pool": {
"enabled": true,
"max_age_ms": 3_300_000,
"max_connecting": 8,
"stream_idle_timeout_ms": 300_000
}
});
if let Some(idle_timeout_ms) = idle_timeout_ms {
document["ws_pool"]["idle_timeout_ms"] = serde_json::json!(idle_timeout_ms);
}
let mut config: OpenAiConfig = serde_json::from_value(document).unwrap();
config
.resolve_and_validate(Path::new("/srv/gateway"))
.unwrap();
assert_eq!(
config.websocket_pool_config().idle_timeout,
Duration::from_millis(expected)
);
}
}
#[test]
fn websocket_pool_idle_timeout_rejects_zero() {
let mut config = valid_config();
config.ws_pool.idle_timeout_ms = 0;
assert!(
config
.resolve_and_validate(Path::new("/srv/gateway"))
.is_err()
);
}
fn valid_config() -> OpenAiConfig {
OpenAiConfig::default()
}
@@ -4433,130 +4433,6 @@ async fn websocket_idle_timeout_diagnosis_survives_ambiguous_send_wrapping() {
);
}
#[tokio::test]
async fn websocket_pong_timeout_diagnosis_survives_ambiguous_send_wrapping() {
for reuse in [false, true] {
const ACCOUNT_ID: &str = "acct_abrupt_disconnect";
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, ACCOUNT_ID).await;
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let base_url = format!("http://{}", listener.local_addr().unwrap());
let (ping_seen_tx, ping_seen_rx) = oneshot::channel();
let server = tokio::spawn(async move {
let (stream, _) = listener.accept().await.unwrap();
let mut websocket = accept_codex_test_websocket(stream).await;
websocket.next().await.unwrap().unwrap();
if reuse {
websocket
.send(Message::Text(
json!({
"type": "response.completed",
"response": {
"id": "resp_warmup",
"model": "gpt-5.4",
"status": "completed",
"output": [],
"usage": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2}
}
})
.to_string()
.into(),
))
.await
.unwrap();
websocket.next().await.unwrap().unwrap();
}
for event in [
json!({"type": "response.created", "response": {"id": "resp_pong", "model": "gpt-5.4"}}),
json!({"type": "response.output_text.delta", "delta": "private-response-body"}),
] {
websocket
.send(Message::Text(event.to_string().into()))
.await
.unwrap();
}
std::assert_matches!(websocket.next().await.unwrap().unwrap(), Message::Ping(_));
ping_seen_tx.send(()).unwrap();
// 停止 poll,避免 tungstenite 自动回 Pong,复现本地保活超时
futures::future::pending::<()>().await;
});
let operation = Operation::Generate(generate_with_persisted_session_context(
ACCOUNT_ID,
"conversation-pong-timeout",
"session-pong-timeout",
"turn-pong-timeout",
));
let provider = provider_with_base_url(&store, base_url);
if reuse {
let mut warmup = Arc::clone(&provider)
.execute(
planned_request("openai", operation.clone()),
context("req_pong_warmup", CancellationToken::new()),
)
.await
.unwrap();
while let Some(event) = warmup.next().await {
event.unwrap();
}
}
let mut stream = provider
.execute(
planned_request("openai", operation),
context("req_pong_timeout", CancellationToken::new()),
)
.await
.unwrap();
let mut pool_kind = None;
loop {
let event = stream.next().await.unwrap().unwrap();
if let Some(observation) = event.response_observation() {
pool_kind = observation.websocket_pool().or(pool_kind);
}
if event.has_client_event() {
break;
}
}
assert_eq!(
pool_kind,
Some(if reuse {
WebSocketPoolKind::Reuse
} else {
WebSocketPoolKind::New
})
);
tokio::time::pause();
tokio::time::advance(Duration::from_secs(25)).await;
ping_seen_rx.await.unwrap();
tokio::time::advance(Duration::from_secs(31)).await;
let error = loop {
match stream.next().await {
Some(Ok(_)) => {}
Some(Err(error)) => break error,
None => panic!("Pong timeout must surface a provider error"),
}
};
server.abort();
tokio::time::resume();
assert_eq!(error.send_state(), UpstreamSendState::Ambiguous);
assert!(!error.replay_is_safe());
assert_eq!(error.pre_delivery_retry(), None);
assert_eq!(error.kind(), ProviderErrorKind::Transport);
assert_eq!(
error.connection_observation().unwrap().exit_reason(),
"pong_timeout"
);
let diagnostic = error.diagnostic().unwrap();
assert_eq!(diagnostic.code(), Some("pong_timeout"));
assert_eq!(diagnostic.stage(), Some("receive"));
assert_eq!(
diagnostic.as_str(),
"OpenAI WebSocket stream ended before a terminal response (pong_timeout); local keepalive timeout after 30s; last event type: response.output_text.delta",
);
assert!(error.raw_upstream_error().is_none());
assert!(error.client_visible_upstream_error().is_none());
}
}
#[tokio::test]
async fn ambiguous_websocket_close_reconnects_and_retains_native_continuation() {
const ACCOUNT_ID: &str = "acct_websocket_close";
@@ -1,4 +1,4 @@
//! 验证控制帧写入阻塞时,接收超时和下游取消仍能结束响应并释放连接
//! 验证控制消息写入阻塞时,超时、取消和空闲池回收仍能释放物理连接
use super::*;
use crate::transport::accept_codex_test_websocket_with;
@@ -13,6 +13,88 @@ async fn cancellation_should_abort_stalled_control_write() {
assert_stalled_control_cleanup(true).await;
}
#[tokio::test]
async fn pool_shutdown_should_abort_stalled_idle_control_write() {
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, "acct_provider_contract").await;
let socket = tokio::net::TcpSocket::new_v4().unwrap();
socket.set_recv_buffer_size(8192).unwrap();
socket.bind("127.0.0.1:0".parse().unwrap()).unwrap();
let listener = socket.listen(1).unwrap();
let base_url = format!("http://{}", listener.local_addr().unwrap());
let (started, started_rx) = oneshot::channel();
let (drain, drain_rx) = oneshot::channel();
let server = tokio::spawn(async move {
let (socket, _) = listener.accept().await.unwrap();
let mut ws = accept_codex_test_websocket_with(socket, |_, response| {
response.headers_mut().remove("sec-websocket-extensions");
})
.await;
assert_eq!(next_json(&mut ws).await["type"], "response.create");
// 未协商压缩,直接写入终态,让连接归池后再阻塞下一条控制消息
let terminal = br#"{"type":"response.completed","response":{"id":"resp_idle_control","status":"completed","output":[]}}"#;
assert!(terminal.len() < 126);
ws.get_mut()
.write_all(&[0x81, u8::try_from(terminal.len()).unwrap()])
.await
.unwrap();
ws.get_mut().write_all(terminal).await.unwrap();
ws.get_mut().read_u8().await.unwrap();
started.send(()).unwrap();
drain_rx.await.unwrap();
tokio::io::copy(ws.get_mut(), &mut tokio::io::sink())
.await
.unwrap();
});
let (provider, pool) =
provider_with_cooldowns(&store, base_url, Arc::new(MemoryCooldownPort::new()));
let control = ResponseControl::default();
let request = generate_with_persisted_session_context(
"acct_provider_contract",
"idle-control-close",
"idle-control-session",
"idle-control-thread",
);
let mut stream = provider
.execute(
planned_request("openai", Operation::Generate(request)),
interrupt_context(control.clone(), None),
)
.await
.unwrap();
timeout(Duration::from_secs(3), async {
while let Some(event) = stream.next().await {
event.unwrap();
}
})
.await
.expect("the response must finish and return its connection to the pool");
let payload =
json!({"type": "future.control", "data": "x".repeat(16 * 1024 * 1024)}).to_string();
let pending = control.send(&payload);
tokio::pin!(pending);
tokio::select! {
result = &mut pending => panic!("control write should stall: {result:?}"),
ready = started_rx => ready.unwrap(),
() = tokio::time::sleep(Duration::from_secs(5)) => panic!("control write should start"),
}
timeout(Duration::from_secs(2), pool.shutdown())
.await
.expect("pool shutdown must bound Close queued behind a stalled write");
assert_eq!(
timeout(Duration::from_secs(1), pending)
.await
.expect("pool shutdown must stop the pending control write"),
Err(ResponseControlUnavailable)
);
drain.send(()).unwrap();
timeout(Duration::from_secs(5), server)
.await
.expect("pool shutdown must release the physical socket")
.unwrap();
}
async fn assert_stalled_control_cleanup(cancel_response: bool) {
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, "acct_provider_contract").await;
@@ -71,6 +71,7 @@ async fn controls_reject_closed_socket_without_losing_backpressured_close() {
let base_url = format!("http://{}", listener.local_addr().unwrap());
let (burst, ready) = oneshot::channel();
let (closed, closing) = oneshot::channel();
let (released, released_rx) = oneshot::channel();
let server = tokio::spawn(async move {
let (socket, _) = listener.accept().await.unwrap();
let mut ws = accept_codex_test_websocket(socket).await;
@@ -89,7 +90,7 @@ async fn controls_reject_closed_socket_without_losing_backpressured_close() {
for _ in 0..81 {
ws.send(delta.clone()).await.unwrap();
}
// Pong 证明 pump 已读取前面的突发数据,随后 Close 必须等待下游排空
// Pong 证明 pump 已读取前面的突发数据,Close 的交付仍需等待下游排空
ws.send(Message::Ping(vec![1].into())).await.unwrap();
assert!(matches!(
ws.next().await.unwrap().unwrap(),
@@ -102,8 +103,11 @@ async fn controls_reject_closed_socket_without_losing_backpressured_close() {
.await
.unwrap();
closed.send(()).unwrap();
// 保留 TCP 直到客户端确认关闭,避免把 Close 场景变成连接重置
let _ = ws.next().await;
// 直接观察 TCP EOF,不能只凭客户端回了 Close 就认定物理连接已经释放
tokio::io::copy(ws.get_mut(), &mut tokio::io::sink())
.await
.unwrap();
released.send(()).unwrap();
});
let control = ResponseControl::default();
let mut stream = provider_with_base_url(&store, base_url)
@@ -123,7 +127,10 @@ async fn controls_reject_closed_socket_without_losing_backpressured_close() {
.await
.unwrap()
.unwrap();
tokio::time::sleep(Duration::from_millis(50)).await;
timeout(Duration::from_secs(2), released_rx)
.await
.expect("observed Close must release the socket before the response is drained")
.unwrap();
assert_eq!(
timeout(
Duration::from_secs(2),
@@ -437,6 +437,7 @@ fn profile() -> CodexWireProfileState {
fn context() -> CodexRequestContext<'static> {
CodexRequestContext {
response_control: None,
trace: None,
authorization: "Bearer oauth-access",
account_id: Some("acct_123"),
@@ -27,6 +27,7 @@ fn upstream_error_formatting_should_redact_body() {
#[test]
fn request_context_debug_should_redact_all_identity_material() {
let context = CodexRequestContext {
response_control: None,
trace: None,
authorization: "Bearer access-secret-marker",
account_id: Some("account-secret-marker"),
@@ -215,6 +215,7 @@ async fn backend_http_should_preserve_business_headers_without_downstream_transp
.create_response(
&request,
CodexRequestContext {
response_control: None,
trace: None,
authorization: "Bearer lease-token",
account_id: Some("lease-account"),
@@ -360,6 +361,7 @@ async fn backend_websocket_should_preserve_business_headers_without_downstream_t
.create_response(
&request,
CodexRequestContext {
response_control: None,
trace: None,
authorization: "Bearer lease-token",
account_id: Some("lease-account"),
@@ -172,6 +172,7 @@ async fn backend_websocket_should_forward_context_headers_and_preserve_payload_f
.create_response(
&request,
CodexRequestContext {
response_control: None,
trace: None,
authorization: "Bearer access-token",
account_id: Some("chatgpt-account"),
@@ -289,6 +290,7 @@ async fn backend_http_should_send_codex_context_without_browser_headers() {
.create_response(
&request,
CodexRequestContext {
response_control: None,
trace: None,
authorization: "Bearer access-token",
account_id: Some("chatgpt-account"),
@@ -416,6 +418,7 @@ async fn backend_http_should_ignore_unrepresentable_protocol_headers_without_blo
.create_response(
&request,
CodexRequestContext {
response_control: None,
trace: None,
client_request_id: Some("opaque\nclient-request-id"),
session_id: Some("opaque\nsession"),
@@ -123,6 +123,7 @@ async fn codex_backend_client_should_preserve_oversized_error_response() {
.create_response(
&request,
CodexRequestContext {
response_control: None,
trace: None,
authorization: "Bearer access-token",
account_id: Some("chatgpt-account"),
@@ -223,6 +224,7 @@ async fn codex_backend_client_should_parse_retry_after_from_rate_limit_error_bod
.create_response(
&request,
CodexRequestContext {
response_control: None,
trace: None,
authorization: "Bearer access-token",
account_id: Some("chatgpt-account"),
@@ -408,6 +410,7 @@ async fn codex_backend_client_should_capture_forwardable_response_metadata() {
.create_response(
&request,
CodexRequestContext {
response_control: None,
trace: None,
authorization: "Bearer access-token",
account_id: Some("chatgpt-account"),
@@ -150,6 +150,7 @@ fn request_context<'a>(
account_id: Option<&'a str>,
) -> CodexRequestContext<'a> {
CodexRequestContext {
response_control: None,
trace: None,
authorization: "Bearer access-token",
account_id,
@@ -517,18 +518,12 @@ async fn write_empty_http_response(stream: &mut TcpStream) {
fn websocket_pool_config_for_tests(
maintenance_interval: Option<Duration>,
ping_interval: Option<Duration>,
liveness_timeout: Option<Duration>,
) -> CodexWebSocketPoolConfig {
CodexWebSocketPoolConfig {
enabled: true,
max_age: Duration::from_mins(1),
max_connecting: 16,
maintenance_interval,
ping_interval,
ping_timeout: Duration::from_secs(1),
liveness_timeout,
stream_idle_timeout: None,
..CodexWebSocketPoolConfig::default()
}
}
@@ -5,12 +5,13 @@ use gateway_core::{
account::{CredentialRevision, OutboundProxy, ProviderAccount, ProviderAccountId},
routing::ProviderKind,
};
use gateway_core::{engine::response_control::ResponseControl, lifecycle::CancellationToken};
use provider_openai::transport::websocket::PreviousResponseUnavailableReason;
fn limited_pool() -> Arc<CodexWebSocketPool> {
Arc::new(CodexWebSocketPool::with_config(CodexWebSocketPoolConfig {
max_age: Duration::from_mins(55),
..websocket_pool_config_for_tests(None, None, None)
..websocket_pool_config_for_tests(None)
}))
}
@@ -138,6 +139,276 @@ async fn proxy_capacity_preserves_idle_reuse_beyond_two_minutes_without_pressure
pool.shutdown().await;
}
#[tokio::test]
async fn proxy_capacity_idle_timeout_restarts_when_a_connection_is_returned() {
let proxy = LimitedProxy::start(32).await;
let pool = limited_pool();
let client = proxy.client(pool.clone(), "acct_a", 1);
respond(&client, "idle-reset").await;
for _ in 0..2 {
tokio::time::pause();
tokio::time::advance(Duration::from_secs(200)).await;
tokio::time::resume();
pool.maintain_idle_connections().await;
respond(&client, "idle-reset").await;
}
assert_eq!(proxy.opened.load(Ordering::SeqCst), 1);
pool.shutdown().await;
}
#[tokio::test]
async fn proxy_capacity_releases_idle_sockets_after_five_minutes() {
let proxy = LimitedProxy::start(32).await;
let pool = limited_pool();
let client = proxy.client(pool.clone(), "acct_a", 1);
respond(&client, "idle-expired").await;
tokio::time::pause();
tokio::time::advance(Duration::from_secs(300)).await;
tokio::time::resume();
pool.maintain_idle_connections().await;
wait_until(|| proxy.active.load(Ordering::SeqCst) == 0).await;
timeout(
Duration::from_secs(3),
respond(&client, "after-idle-expiry"),
)
.await
.unwrap();
assert_eq!(proxy.opened.load(Ordering::SeqCst), 2);
pool.shutdown().await;
}
#[tokio::test]
async fn proxy_capacity_preserves_bound_sockets_beyond_idle_timeout() {
let control = ResponseControl::default().with_connection_cancellation(CancellationToken::new());
let proxy = LimitedProxy::start(32).await;
let pool = limited_pool();
let client = proxy.client(pool.clone(), "acct_a", 1);
let mut request = pooled_websocket_request("bound-idle");
request.downstream_websocket_connection_id = Some("downstream-bound-idle".to_owned());
client
.create_response(
&request,
request_context("bound-first", Some("test-upstream"))
.with_response_control(Some(&control)),
)
.await
.unwrap();
tokio::time::pause();
tokio::time::advance(Duration::from_secs(301)).await;
tokio::time::resume();
pool.maintain_idle_connections().await;
client
.create_response(
&request,
request_context("bound-next", Some("test-upstream"))
.with_response_control(Some(&control)),
)
.await
.unwrap();
assert_eq!(proxy.opened.load(Ordering::SeqCst), 1);
pool.shutdown().await;
}
#[tokio::test]
async fn downstream_close_releases_idle_socket_without_waiting_for_maintenance() {
let proxy = LimitedProxy::start(32).await;
let pool = limited_pool();
let client = proxy.client(pool.clone(), "acct_a", 1);
let cancellation = CancellationToken::new();
let control = ResponseControl::default().with_connection_cancellation(cancellation.clone());
let mut request = pooled_websocket_request("closed-client");
request.downstream_websocket_connection_id = Some("downstream-closed".to_owned());
client
.create_response(
&request,
request_context("bound", Some("test-upstream")).with_response_control(Some(&control)),
)
.await
.unwrap();
cancellation.cancel();
wait_until(|| proxy.active.load(Ordering::SeqCst) == 0).await;
timeout(Duration::from_secs(3), respond(&client, "next-client"))
.await
.unwrap();
assert_eq!(proxy.opened.load(Ordering::SeqCst), 2);
pool.shutdown().await;
}
#[tokio::test]
async fn downstream_close_cancels_active_response_and_prevents_return_to_pool() {
let gate = Arc::new(tokio::sync::Semaphore::new(0));
let proxy = LimitedProxy::with_gates(32, None, Some(gate.clone())).await;
let pool = limited_pool();
let client = proxy.client(pool.clone(), "acct_a", 1);
let cancellation = CancellationToken::new();
let control = ResponseControl::default().with_connection_cancellation(cancellation.clone());
let mut request = pooled_websocket_request("active-client");
request.downstream_websocket_connection_id = Some("downstream-active".to_owned());
let task = tokio::spawn(async move {
client
.create_response(
&request,
request_context("active", Some("test-upstream"))
.with_response_control(Some(&control)),
)
.await
});
wait_until(|| !proxy.requests.lock().unwrap().is_empty()).await;
cancellation.cancel();
let response = timeout(Duration::from_secs(3), task)
.await
.unwrap()
.unwrap()
.unwrap();
assert!(response.body.is_empty());
wait_until(|| proxy.active.load(Ordering::SeqCst) == 0).await;
gate.add_permits(2);
let client = proxy.client(pool.clone(), "acct_a", 1);
timeout(
Duration::from_secs(3),
respond(&client, "after-active-cancel"),
)
.await
.unwrap();
assert_eq!(proxy.opened.load(Ordering::SeqCst), 2);
pool.shutdown().await;
}
#[tokio::test]
async fn downstream_close_cancels_connecting_socket_and_releases_proxy_capacity() {
let gate = Arc::new(tokio::sync::Semaphore::new(0));
let proxy = LimitedProxy::with_gates(32, Some(gate.clone()), None).await;
let pool = limited_pool();
let client = proxy.client(pool.clone(), "acct_a", 1);
let cancellation = CancellationToken::new();
let control = ResponseControl::default().with_connection_cancellation(cancellation.clone());
let mut request = pooled_websocket_request("opening-client");
request.downstream_websocket_connection_id = Some("downstream-opening".to_owned());
request.set_previous_response_id(None);
request.previous_response_scope = None;
let task = tokio::spawn(async move {
client
.create_response(
&request,
request_context("opening", Some("test-upstream"))
.with_response_control(Some(&control)),
)
.await
});
wait_until(|| proxy.active.load(Ordering::SeqCst) == 1).await;
cancellation.cancel();
assert!(
timeout(Duration::from_secs(3), task)
.await
.unwrap()
.unwrap()
.is_err()
);
wait_until(|| proxy.active.load(Ordering::SeqCst) == 0).await;
gate.add_permits(2);
let client = proxy.client(pool.clone(), "acct_a", 1);
timeout(
Duration::from_secs(3),
respond(&client, "after-opening-cancel"),
)
.await
.unwrap();
pool.shutdown().await;
}
#[tokio::test]
async fn old_downstream_close_does_not_destroy_a_transferred_continuation() {
let proxy = LimitedProxy::start(32).await;
let pool = limited_pool();
let client = proxy.client(pool.clone(), "acct_a", 1);
let old_cancellation = CancellationToken::new();
let new_cancellation = CancellationToken::new();
let old_control =
ResponseControl::default().with_connection_cancellation(old_cancellation.clone());
let new_control =
ResponseControl::default().with_connection_cancellation(new_cancellation.clone());
let mut request = pooled_websocket_request("transferred-chain");
request.downstream_websocket_connection_id = Some("downstream-old".to_owned());
client
.create_response(
&request,
request_context("old", Some("test-upstream")).with_response_control(Some(&old_control)),
)
.await
.unwrap();
request.downstream_websocket_connection_id = Some("downstream-new".to_owned());
let waiting_control = old_control.clone();
let old_reader = tokio::spawn(async move { waiting_control.receive().await });
tokio::task::yield_now().await;
request.set_previous_response_id(Some("resp_proxy_0".to_owned()));
request.previous_response_scope = Some(PreviousResponseScope::ConnectionLocal);
client
.create_response(
&request,
request_context("transferred", Some("test-upstream"))
.with_response_control(Some(&new_control)),
)
.await
.unwrap();
assert!(
timeout(Duration::from_secs(3), old_reader)
.await
.unwrap()
.unwrap()
.is_err()
);
assert!(
old_control
.send(r#"{"type":"response.interrupt"}"#)
.await
.is_err()
);
old_cancellation.cancel();
tokio::task::yield_now().await;
request.set_previous_response_id(Some("resp_proxy_0".to_owned()));
client
.create_response(
&request,
request_context("continued", Some("test-upstream"))
.with_response_control(Some(&new_control)),
)
.await
.unwrap();
assert_eq!(proxy.opened.load(Ordering::SeqCst), 1);
new_cancellation.cancel();
wait_until(|| proxy.active.load(Ordering::SeqCst) == 0).await;
pool.shutdown().await;
}
#[tokio::test]
async fn rebinding_downstream_retires_old_socket_before_waiting_for_capacity() {
let proxy = LimitedProxy::start(32).await;
let pool = limited_pool();
let client = proxy.client(pool.clone(), "acct_a", 1);
let cancellation = CancellationToken::new();
let control = ResponseControl::default().with_connection_cancellation(cancellation.clone());
for conversation in ["old-chain", "replacement-chain"] {
let mut request = pooled_websocket_request(conversation);
request.downstream_websocket_connection_id = Some("same-downstream".to_owned());
timeout(
Duration::from_secs(3),
client.create_response(
&request,
request_context("rebind", Some("test-upstream"))
.with_response_control(Some(&control)),
),
)
.await
.unwrap()
.unwrap();
}
assert_eq!(proxy.opened.load(Ordering::SeqCst), 2);
assert!(proxy.peak.load(Ordering::SeqCst) <= 1);
cancellation.cancel();
wait_until(|| proxy.active.load(Ordering::SeqCst) == 0).await;
pool.shutdown().await;
}
#[tokio::test]
async fn proxy_capacity_does_not_limit_direct_connections() {
let pool = limited_pool();
@@ -208,13 +479,17 @@ async fn proxy_capacity_releases_failed_handshake_before_the_next_connection() {
#[tokio::test]
async fn proxy_capacity_protects_downstream_bindings_and_blocks_unpooled_bypass() {
let control = ResponseControl::default().with_connection_cancellation(CancellationToken::new());
let proxy = LimitedProxy::start(32).await;
let pool = limited_pool();
let client = proxy.client(pool.clone(), "acct_a", 1);
let mut pinned = pooled_websocket_request("pinned");
pinned.downstream_websocket_connection_id = Some("downstream-a".to_owned());
client
.create_response(&pinned, request_context("pinned", Some("test-upstream")))
.create_response(
&pinned,
request_context("pinned", Some("test-upstream")).with_response_control(Some(&control)),
)
.await
.unwrap();
let unpooled = codex_request("gpt-test", "be brief", Vec::new());
@@ -233,7 +508,8 @@ async fn proxy_capacity_protects_downstream_bindings_and_blocks_unpooled_bypass(
client
.create_response(
&pinned,
request_context("pinned-reuse", Some("test-upstream")),
request_context("pinned-reuse", Some("test-upstream"))
.with_response_control(Some(&control)),
)
.await
.unwrap();
@@ -433,7 +709,16 @@ impl LimitedProxy {
stream.write_all(b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\n\r\n").await.unwrap();
return;
}
if let Some(gate) = connect_gate { gate.acquire().await.unwrap().forget(); }
if let Some(gate) = connect_gate {
let mut byte = [0];
tokio::select! {
permit = gate.acquire() => permit.unwrap().forget(),
result = stream.peek(&mut byte) => {
assert_eq!(result.unwrap(), 0, "cancelled opening must close TCP");
return;
}
}
}
stream.write_all(b"HTTP/1.1 200 Connection Established\r\n\r\n").await.unwrap();
if request.contains("X-Capacity-Probe: health") {
let health = read_http_request(&mut stream).await;
@@ -449,7 +734,16 @@ impl LimitedProxy {
match message {
Message::Text(payload) => {
requests.lock().unwrap().push(serde_json::from_str(&payload).unwrap());
if let Some(gate) = &response_gate { gate.acquire().await.unwrap().forget(); }
if let Some(gate) = &response_gate {
tokio::select! {
permit = gate.acquire() => permit.unwrap().forget(),
message = ws.next() => {
assert!(matches!(message, None | Some(Err(_)) | Some(Ok(Message::Close(_)))));
let _ = ws.flush().await;
break;
}
}
}
if ws.send(Message::Text(completed_websocket_response(&format!("resp_proxy_{id}"), 1, 1).into())).await.is_err() { break; }
}
Message::Ping(payload) => { if ws.send(Message::Pong(payload)).await.is_err() { break; } }
@@ -573,14 +867,22 @@ async fn proxy_capacity_enabling_and_lowering_limits_keeps_existing_socket_count
#[tokio::test]
async fn proxy_capacity_enabling_limit_preserves_existing_downstream_bindings() {
let controls = [
ResponseControl::default().with_connection_cancellation(CancellationToken::new()),
ResponseControl::default().with_connection_cancellation(CancellationToken::new()),
];
let proxy = LimitedProxy::start(32).await;
let pool = limited_pool();
let unlimited = proxy.client(pool.clone(), "acct_a", 0);
for index in 0..2 {
for (index, control) in controls.iter().enumerate() {
let mut request = pooled_websocket_request(&format!("pinned-{index}"));
request.downstream_websocket_connection_id = Some(format!("downstream-{index}"));
unlimited
.create_response(&request, request_context("pinned", Some("test-upstream")))
.create_response(
&request,
request_context("pinned", Some("test-upstream"))
.with_response_control(Some(control)),
)
.await
.unwrap();
}
@@ -590,7 +892,8 @@ async fn proxy_capacity_enabling_limit_preserves_existing_downstream_bindings()
limited
.create_response(
&retained,
request_context("pinned-reused", Some("test-upstream")),
request_context("pinned-reused", Some("test-upstream"))
.with_response_control(Some(&controls[0])),
)
.await
.unwrap();
@@ -509,6 +509,7 @@ async fn fetch_should_use_wham_usage_headers_only() {
let usage = client(&server.uri())
.fetch_usage(CodexRequestContext {
response_control: None,
trace: None,
authorization: "Bearer oauth-access",
account_id: Some("acct_123"),
@@ -908,6 +909,7 @@ fn profile() -> CodexWireProfileState {
fn context() -> CodexRequestContext<'static> {
CodexRequestContext {
response_control: None,
trace: None,
authorization: "Bearer oauth-access",
account_id: Some("acct_123"),
@@ -38,8 +38,6 @@ async fn send_timeout_should_abort_stalled_pump_and_allow_next_request() {
.unwrap();
});
let pool = Arc::new(CodexWebSocketPool::with_config(CodexWebSocketPoolConfig {
ping_interval: None,
liveness_timeout: None,
maintenance_interval: None,
..CodexWebSocketPoolConfig::default()
}));
@@ -91,6 +91,7 @@ async fn official_websocket_wire_fingerprint_should_remain_aligned() {
.create_response(
&request,
CodexRequestContext {
response_control: None,
trace: None,
turn_metadata: request.turn_metadata.as_deref(),
beta_features: request.beta_features.as_deref(),
@@ -1855,9 +1855,7 @@ async fn codex_backend_client_stream_should_preserve_burst_during_downstream_bac
let mut websocket = accept_codex_test_websocket(stream).await;
let _message = websocket.next().await.unwrap().unwrap();
let (mut sink, mut source) = websocket.split();
// 服务端持续读取控制帧,让 tungstenite 正常回 Pong
// 若客户端在本地入站背压
// 期间仍执行 Pong deadline,它仍会因为暂时读不到已返回的 Pong 而误杀连接
// 服务端持续读取关闭帧,业务背压不应丢失已发送的输出
let control_frames = tokio::spawn(async move {
while let Some(message) = source.next().await {
match message.unwrap() {
@@ -1900,12 +1898,9 @@ async fn codex_backend_client_stream_should_preserve_burst_during_downstream_bac
let _ = sink.send(Message::Close(None)).await;
control_frames.abort();
});
let pool = Arc::new(CodexWebSocketPool::with_config(CodexWebSocketPoolConfig {
ping_interval: Some(Duration::from_millis(5)),
ping_timeout: Duration::from_millis(20),
liveness_timeout: None,
..websocket_pool_config_for_tests(None, None, None)
}));
let pool = Arc::new(CodexWebSocketPool::with_config(
websocket_pool_config_for_tests(None),
));
let backend = CodexBackendClient::new(
reqwest::Client::builder().no_proxy().build().unwrap(),
format!("http://{addr}"),
@@ -2223,6 +2218,7 @@ async fn codex_backend_client_should_use_websocket_when_previous_response_id_is_
.create_response(
&request,
CodexRequestContext {
response_control: None,
trace: None,
authorization: "Bearer access-token",
account_id: Some("chatgpt-account"),
@@ -367,7 +367,7 @@ async fn codex_backend_client_should_open_fresh_socket_when_idle_pooled_websocke
});
// 无 maintenance、无主动 ping:完全依赖 pump 后台读取感知连接死亡
let pool = Arc::new(CodexWebSocketPool::with_config(
websocket_pool_config_for_tests(None, None, None),
websocket_pool_config_for_tests(None),
));
let backend = CodexBackendClient::new(
reqwest::Client::builder().no_proxy().build().unwrap(),
@@ -460,7 +460,7 @@ async fn codex_backend_client_stream_should_keep_fresh_socket_after_structural_a
});
let pool = Arc::new(CodexWebSocketPool::with_config(CodexWebSocketPoolConfig {
stream_idle_timeout: Some(Duration::from_millis(200)),
..websocket_pool_config_for_tests(None, None, None)
..websocket_pool_config_for_tests(None)
}));
let backend = CodexBackendClient::new(
reqwest::Client::builder().no_proxy().build().unwrap(),
@@ -549,7 +549,7 @@ async fn codex_backend_client_stream_should_keep_reused_socket_after_structural_
});
let pool = Arc::new(CodexWebSocketPool::with_config(CodexWebSocketPoolConfig {
stream_idle_timeout: Some(Duration::from_millis(200)),
..websocket_pool_config_for_tests(None, None, None)
..websocket_pool_config_for_tests(None)
}));
let backend = CodexBackendClient::new(
reqwest::Client::builder().no_proxy().build().unwrap(),
@@ -607,7 +607,7 @@ async fn codex_backend_client_stream_should_use_http_when_connecting_limit_is_ex
let pool = Arc::new(CodexWebSocketPool::with_config(CodexWebSocketPoolConfig {
max_connecting: 0,
stream_idle_timeout: Some(Duration::from_millis(200)),
..websocket_pool_config_for_tests(None, None, None)
..websocket_pool_config_for_tests(None)
}));
let backend = CodexBackendClient::new(
reqwest::Client::builder().no_proxy().build().unwrap(),
@@ -1063,7 +1063,7 @@ async fn websocket_pool_should_keep_sixty_fifth_conversation_on_same_account_on_
});
let pool = Arc::new(CodexWebSocketPool::with_config(CodexWebSocketPoolConfig {
max_connecting: 1,
..websocket_pool_config_for_tests(None, None, None)
..websocket_pool_config_for_tests(None)
}));
let backend = CodexBackendClient::new(
reqwest::Client::builder().no_proxy().build().unwrap(),
@@ -1168,7 +1168,7 @@ async fn concurrent_pool_transport_counts(max_connecting: usize) -> (usize, usiz
});
let pool = Arc::new(CodexWebSocketPool::with_config(CodexWebSocketPoolConfig {
max_connecting,
..websocket_pool_config_for_tests(None, None, None)
..websocket_pool_config_for_tests(None)
}));
let backend = CodexBackendClient::new(
reqwest::Client::builder().no_proxy().build().unwrap(),
@@ -1210,70 +1210,46 @@ async fn concurrent_pool_transport_counts(max_connecting: usize) -> (usize, usiz
}
#[tokio::test]
async fn websocket_pool_should_replace_idle_connection_after_pong_deadline() {
const PING_INTERVAL: Duration = Duration::from_secs(30);
const PONG_TIMEOUT: Duration = Duration::from_secs(30);
// 服务端读取 Ping 后暂不继续 poll,避免 tungstenite 自动回 Pong;pump 必须在独立
// deadline 到期时主动关闭连接,acquire 随后只读 closed 状态并直接新建连接
async fn websocket_pool_should_reject_expired_idle_continuation_despite_upstream_ping() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let accepted_connections = Arc::new(AtomicUsize::new(0));
let accepted_connections_for_server = Arc::clone(&accepted_connections);
let (ping_seen_tx, ping_seen_rx) = tokio::sync::oneshot::channel();
let (inspect_close_tx, inspect_close_rx) = tokio::sync::oneshot::channel();
let (pong_timeout_closed_tx, pong_timeout_closed_rx) = tokio::sync::oneshot::channel();
let (send_ping_tx, send_ping_rx) = tokio::sync::oneshot::channel();
let (pong_seen_tx, pong_seen_rx) = tokio::sync::oneshot::channel();
let server = tokio::spawn(async move {
let (first_stream, _) = listener.accept().await.unwrap();
accepted_connections_for_server.fetch_add(1, Ordering::SeqCst);
let mut first_websocket = accept_codex_test_websocket(first_stream).await;
let _first_message = first_websocket.next().await.unwrap().unwrap();
first_websocket
let (stream, _) = listener.accept().await.unwrap();
let mut first = accept_codex_test_websocket(stream).await;
first.next().await.unwrap().unwrap();
first
.send(Message::Text(
completed_websocket_response("resp_no_pong_first", 2, 1).into(),
completed_websocket_response("resp_idle_first", 2, 1).into(),
))
.await
.unwrap();
let ping = timeout(Duration::from_secs(120), first_websocket.next())
send_ping_rx.await.unwrap();
let payload = b"upstream-idle-probe".to_vec();
first
.send(Message::Ping(payload.clone().into()))
.await
.expect("pump should send a keepalive ping")
.expect("keepalive ping should be present")
.expect("keepalive ping should be valid");
let Message::Ping(payload) = ping else {
panic!("expected keepalive ping, got {ping:?}");
};
assert_eq!(payload.len(), 8);
ping_seen_tx.send(()).unwrap();
.unwrap();
let pong = first.next().await.unwrap().unwrap();
assert_eq!(pong, Message::Pong(payload.into()));
pong_seen_tx.send(()).unwrap();
std::assert_matches!(first.next().await.unwrap().unwrap(), Message::Close(_));
inspect_close_rx
.await
.expect("test should release the server after the Pong deadline");
let close = timeout(Duration::from_secs(120), first_websocket.next())
.await
.expect("Pong deadline should close the idle websocket")
.expect("Pong deadline should send a close frame")
.expect("close frame should be valid");
std::assert_matches!(close, Message::Close(_));
pong_timeout_closed_tx.send(()).unwrap();
let (second_stream, _) = listener.accept().await.unwrap();
accepted_connections_for_server.fetch_add(1, Ordering::SeqCst);
let mut second_websocket = accept_codex_test_websocket(second_stream).await;
let _second_message = second_websocket.next().await.unwrap().unwrap();
second_websocket
let (stream, _) = listener.accept().await.unwrap();
let mut second = accept_codex_test_websocket(stream).await;
second.next().await.unwrap().unwrap();
second
.send(Message::Text(
completed_websocket_response("resp_no_pong_second", 2, 1).into(),
completed_websocket_response("resp_idle_second", 2, 1).into(),
))
.await
.unwrap();
second_websocket.close(None).await.unwrap();
second.close(None).await.unwrap();
});
let pool = Arc::new(CodexWebSocketPool::with_config(CodexWebSocketPoolConfig {
ping_interval: Some(PING_INTERVAL),
ping_timeout: PONG_TIMEOUT,
liveness_timeout: None,
maintenance_interval: None,
..websocket_pool_config_for_tests(None, None, None)
..CodexWebSocketPoolConfig::default()
}));
let backend = CodexBackendClient::new(
reqwest::Client::builder().no_proxy().build().unwrap(),
@@ -1281,66 +1257,60 @@ async fn websocket_pool_should_replace_idle_connection_after_pong_deadline() {
test_wire_profile(),
)
.with_websocket_pool(Arc::clone(&pool));
let request = pooled_websocket_request("conversation-no-pong");
let first = backend
let request = pooled_websocket_request("conversation-idle-expiry");
backend
.create_response(
&request,
request_context("req_no_pong_first", Some("chatgpt-account")),
request_context("req_idle_first", Some("chatgpt-account")),
)
.await
.expect("first websocket response should succeed");
.unwrap();
tokio::time::pause();
tokio::time::advance(PING_INTERVAL).await;
tokio::task::yield_now().await;
ping_seen_rx
.await
.expect("server should report the keepalive ping");
tokio::time::advance(PONG_TIMEOUT + Duration::from_secs(1)).await;
tokio::task::yield_now().await;
tokio::task::yield_now().await;
inspect_close_tx.send(()).unwrap();
pong_timeout_closed_rx
.await
.expect("server should report the Pong-timeout close");
tokio::time::advance(Duration::from_secs(299)).await;
tokio::time::resume();
send_ping_tx.send(()).unwrap();
timeout(Duration::from_secs(2), pong_seen_rx)
.await
.unwrap()
.unwrap();
tokio::time::pause();
tokio::time::advance(Duration::from_secs(1)).await;
tokio::time::resume();
let mut continuation = request.clone();
continuation.set_previous_response_id(Some("resp_no_pong_first".to_owned()));
continuation.set_previous_response_id(Some("resp_idle_first".to_owned()));
continuation.previous_response_scope = Some(PreviousResponseScope::ConnectionLocal);
let error = backend
.create_response(
&continuation,
request_context("req_no_pong_continuation", Some("chatgpt-account")),
request_context("req_idle_continuation", Some("chatgpt-account")),
)
.await
.expect_err("Pong-timeout connection cannot satisfy an exact continuation");
.expect_err("heartbeat must not extend the cached connection's idle lifetime");
let CodexClientError::WebSocket(error) = error else {
panic!("Pong-timeout continuation should remain a typed WebSocket error");
panic!("expired continuation should remain a typed WebSocket error");
};
assert_eq!(
error.continuation_unavailable_reason(),
Some(PreviousResponseUnavailableReason::ReusedConnectionLost)
);
assert_eq!(
error
.connection_observation()
.expect("Pong-timeout tombstone should retain its lifecycle observation")
.exit_reason(),
"pong_timeout"
error.connection_observation().unwrap().exit_reason(),
"idle_timeout_expired"
);
let second = backend
let response = backend
.create_response(
&request,
request_context("req_no_pong_second", Some("chatgpt-account")),
request_context("req_idle_second", Some("chatgpt-account")),
)
.await
.expect("second websocket response should use a fresh connection");
.unwrap();
assert!(response.body.contains("resp_idle_second"));
assert_eq!(response.websocket_pool_decision.unwrap().kind(), "new");
server.await.unwrap();
assert!(first.body.contains("resp_no_pong_first"));
assert!(second.body.contains("resp_no_pong_second"));
assert_eq!(second.websocket_pool_decision.unwrap().kind(), "new");
assert_eq!(accepted_connections.load(Ordering::SeqCst), 2);
pool.shutdown().await;
}
#[tokio::test]
@@ -1384,8 +1354,6 @@ async fn websocket_pool_should_gc_expired_idle_connections() {
let pool = Arc::new(CodexWebSocketPool::with_config(CodexWebSocketPoolConfig {
max_age: MAX_AGE,
maintenance_interval: None,
ping_interval: None,
liveness_timeout: None,
..CodexWebSocketPoolConfig::default()
}));
let backend = CodexBackendClient::new(
@@ -1428,8 +1396,6 @@ fn websocket_pool_constructed_outside_runtime_should_start_maintenance_on_first_
let pool = Arc::new(CodexWebSocketPool::with_config(CodexWebSocketPoolConfig {
max_age: MAX_AGE,
maintenance_interval: Some(Duration::from_millis(20)),
ping_interval: None,
liveness_timeout: None,
..CodexWebSocketPoolConfig::default()
}));
let runtime = tokio::runtime::Builder::new_current_thread()
@@ -1490,90 +1456,44 @@ async fn websocket_pool_zero_maintenance_interval_should_not_start_a_panicking_t
}
#[tokio::test]
async fn codex_backend_client_should_keep_idle_pooled_websocket_alive_across_repeated_pings() {
async fn codex_backend_client_should_reply_to_upstream_pings_and_reuse_idle_socket() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let ping_count = Arc::new(AtomicUsize::new(0));
let ping_count_for_server = Arc::clone(&ping_count);
let (pings_observed_tx, pings_observed_rx) = tokio::sync::oneshot::channel();
let (pongs_seen_tx, pongs_seen_rx) = tokio::sync::oneshot::channel();
let server = tokio::spawn(async move {
let (stream, _) = listener.accept().await.unwrap();
let mut websocket = accept_codex_test_websocket(stream).await;
let _first_message = websocket.next().await.unwrap().unwrap();
websocket.next().await.unwrap().unwrap();
websocket
.send(Message::Text(
json!({
"type": "response.completed",
"response": {
"id": "resp_pool_background_first",
"object": "response",
"output": [],
"usage": {
"input_tokens": 3,
"output_tokens": 1,
"total_tokens": 4
}
}
})
.to_string()
.into(),
completed_websocket_response("resp_passive_first", 3, 1).into(),
))
.await
.unwrap();
// pump 会在 idle 期间反复发送 keepalive ping;服务端计数并回 pong,
// 直到下一个业务请求(response.create)到达为止
let mut previous_ping = None;
let mut pings_observed_tx = Some(pings_observed_tx);
loop {
let message = timeout(Duration::from_secs(1), websocket.next())
for payload in [b"probe-a".to_vec(), b"probe-b".to_vec()] {
websocket
.send(Message::Ping(payload.clone().into()))
.await
.expect("pump keepalive / second request should arrive")
.expect("frame should be present")
.expect("frame should be valid");
match message {
Message::Ping(payload) => {
assert_eq!(payload.len(), 8);
if let Some(previous) = &previous_ping {
assert_ne!(previous, &payload, "keepalive ping sequence must be unique");
}
previous_ping = Some(payload.clone());
let ping_count = ping_count_for_server.fetch_add(1, Ordering::SeqCst) + 1;
if ping_count == 2
&& let Some(pings_observed_tx) = pings_observed_tx.take()
{
pings_observed_tx.send(()).unwrap();
}
websocket.send(Message::Pong(payload)).await.unwrap();
}
Message::Text(_) => break,
other => panic!("unexpected frame while idle: {other:?}"),
}
.unwrap();
let frame = timeout(Duration::from_secs(2), websocket.next())
.await
.unwrap()
.unwrap()
.unwrap();
assert_eq!(frame, Message::Pong(payload.into()));
}
pongs_seen_tx.send(()).unwrap();
std::assert_matches!(websocket.next().await.unwrap().unwrap(), Message::Text(_));
websocket
.send(Message::Text(
json!({
"type": "response.completed",
"response": {
"id": "resp_pool_background_second",
"object": "response",
"output": [],
"usage": {
"input_tokens": 3,
"output_tokens": 1,
"total_tokens": 4
}
}
})
.to_string()
.into(),
completed_websocket_response("resp_passive_second", 3, 1).into(),
))
.await
.unwrap();
websocket.close(None).await.unwrap();
});
let pool = Arc::new(CodexWebSocketPool::with_config(
websocket_pool_config_for_tests(None, Some(Duration::from_millis(10)), None),
websocket_pool_config_for_tests(None),
));
let backend = CodexBackendClient::new(
reqwest::Client::builder().no_proxy().build().unwrap(),
@@ -1581,43 +1501,33 @@ async fn codex_backend_client_should_keep_idle_pooled_websocket_alive_across_rep
test_wire_profile(),
)
.with_websocket_pool(Arc::clone(&pool));
let mut request = codex_request_with_prompt_cache_key(
"gpt-5.5",
"be brief",
Vec::new(),
"conversation-pool-background",
);
request.set_previous_response_id(Some("resp_previous".to_string()));
request.previous_response_scope = Some(PreviousResponseScope::Persisted);
let first = backend
let request = pooled_websocket_request("conversation-passive-heartbeat");
backend
.create_response(
&request,
request_context("req_pool_background_first", Some("chatgpt-account")),
request_context("req_passive_first", Some("chatgpt-account")),
)
.await
.expect("first pooled websocket response should succeed");
timeout(Duration::from_secs(2), pings_observed_rx)
.unwrap();
timeout(Duration::from_secs(2), pongs_seen_rx)
.await
.expect("pump should emit repeated keepalive pings")
.expect("server should observe repeated keepalive pings");
let second = backend
.unwrap()
.unwrap();
let response = backend
.create_response(
&request,
request_context("req_pool_background_second", Some("chatgpt-account")),
request_context("req_passive_second", Some("chatgpt-account")),
)
.await
.expect("second pooled websocket response should reuse the kept-alive socket");
.unwrap();
assert!(response.body.contains("resp_passive_second"));
assert_eq!(response.websocket_pool_decision.unwrap().kind(), "reuse");
server.await.unwrap();
pool.shutdown().await;
assert!(first.body.contains("resp_pool_background_first"));
assert!(second.body.contains("resp_pool_background_second"));
assert!(ping_count.load(Ordering::SeqCst) >= 2);
}
#[tokio::test]
async fn codex_backend_client_should_treat_active_business_frames_as_ping_liveness() {
async fn websocket_pool_should_not_apply_idle_timeout_to_active_stream() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let server = tokio::spawn(async move {
@@ -1626,7 +1536,7 @@ async fn codex_backend_client_should_treat_active_business_frames_as_ping_livene
let _request = websocket.next().await.unwrap().unwrap();
for index in 0..8 {
tokio::time::sleep(Duration::from_secs(1)).await;
tokio::time::sleep(Duration::from_millis(30)).await;
websocket
.send(Message::Text(
json!({
@@ -1648,10 +1558,8 @@ async fn codex_backend_client_should_treat_active_business_frames_as_ping_livene
websocket.close(None).await.unwrap();
});
let pool = Arc::new(CodexWebSocketPool::with_config(CodexWebSocketPoolConfig {
ping_interval: Some(Duration::from_millis(100)),
ping_timeout: Duration::from_secs(5),
liveness_timeout: None,
..websocket_pool_config_for_tests(None, None, None)
idle_timeout: Duration::from_millis(100),
..websocket_pool_config_for_tests(Some(Duration::from_millis(20)))
}));
let backend = CodexBackendClient::new(
reqwest::Client::builder().no_proxy().build().unwrap(),
@@ -1667,7 +1575,7 @@ async fn codex_backend_client_should_treat_active_business_frames_as_ping_livene
request_context("req_active_without_pong", Some("chatgpt-account")),
)
.await
.expect("active business frames should satisfy the Ping liveness probe");
.expect("active stream should not be evicted by the idle pool timeout");
server.await.unwrap();
assert!(response.body.contains("active-0;"));
@@ -1740,7 +1648,7 @@ async fn codex_backend_client_should_close_idle_pooled_websocket_when_account_is
second_websocket.close(None).await.unwrap();
});
let pool = Arc::new(CodexWebSocketPool::with_config(
websocket_pool_config_for_tests(None, None, None),
websocket_pool_config_for_tests(None),
));
let backend = CodexBackendClient::new(
reqwest::Client::builder().no_proxy().build().unwrap(),
@@ -1824,7 +1732,7 @@ async fn codex_backend_client_should_stop_reusing_pooled_websockets_after_shutdo
write_completed_sse_response(&mut second_stream).await;
});
let pool = Arc::new(CodexWebSocketPool::with_config(
websocket_pool_config_for_tests(None, None, None),
websocket_pool_config_for_tests(None),
));
let backend = CodexBackendClient::new(
reqwest::Client::builder().no_proxy().build().unwrap(),
@@ -1869,118 +1777,66 @@ async fn codex_backend_client_should_stop_reusing_pooled_websockets_after_shutdo
}
#[tokio::test]
async fn codex_backend_client_should_close_idle_pooled_websocket_after_liveness_timeout() {
const LIVENESS_TIMEOUT: Duration = Duration::from_secs(30);
async fn websocket_pool_should_reclaim_idle_connections_on_maintenance_ticks() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let accepted_connections = Arc::new(AtomicUsize::new(0));
let accepted_connections_for_server = Arc::clone(&accepted_connections);
let (liveness_closed_tx, liveness_closed_rx) = tokio::sync::oneshot::channel();
let (closed_tx, closed_rx) = tokio::sync::oneshot::channel();
let server = tokio::spawn(async move {
let (first_stream, _) = listener.accept().await.unwrap();
accepted_connections_for_server.fetch_add(1, Ordering::SeqCst);
let mut first_websocket = accept_codex_test_websocket(first_stream).await;
let _first_message = first_websocket.next().await.unwrap().unwrap();
first_websocket
let (stream, _) = listener.accept().await.unwrap();
let mut websocket = accept_codex_test_websocket(stream).await;
websocket.next().await.unwrap().unwrap();
websocket
.send(Message::Text(
json!({
"type": "response.completed",
"response": {
"id": "resp_pool_liveness_first",
"object": "response",
"output": [],
"usage": {
"input_tokens": 3,
"output_tokens": 1,
"total_tokens": 4
}
}
})
.to_string()
.into(),
completed_websocket_response("resp_maintenance_idle", 3, 1).into(),
))
.await
.unwrap();
let close = timeout(Duration::from_secs(60), first_websocket.next())
.await
.expect("liveness timeout should close the idle websocket")
.expect("liveness timeout should send a close frame")
.expect("close frame should be valid");
std::assert_matches!(close, Message::Close(_));
liveness_closed_tx.send(()).unwrap();
let (second_stream, _) = listener.accept().await.unwrap();
accepted_connections_for_server.fetch_add(1, Ordering::SeqCst);
let mut second_websocket = accept_codex_test_websocket(second_stream).await;
let _second_message = second_websocket.next().await.unwrap().unwrap();
second_websocket
.send(Message::Text(
json!({
"type": "response.completed",
"response": {
"id": "resp_pool_liveness_second",
"object": "response",
"output": [],
"usage": {
"input_tokens": 4,
"output_tokens": 1,
"total_tokens": 5
}
}
})
.to_string()
.into(),
))
.await
.unwrap();
second_websocket.close(None).await.unwrap();
let message = websocket.next().await.unwrap().unwrap();
std::assert_matches!(message, Message::Close(_));
closed_tx.send(()).unwrap();
});
let pool = Arc::new(CodexWebSocketPool::with_config(
websocket_pool_config_for_tests(None, None, Some(LIVENESS_TIMEOUT)),
));
let pool = Arc::new(CodexWebSocketPool::default());
let backend = CodexBackendClient::new(
reqwest::Client::builder().no_proxy().build().unwrap(),
format!("http://{addr}"),
test_wire_profile(),
)
.with_websocket_pool(Arc::clone(&pool));
let mut request = codex_request_with_prompt_cache_key(
"gpt-5.5",
"be brief",
Vec::new(),
"conversation-pool-liveness",
);
request.set_previous_response_id(Some("resp_previous".to_string()));
request.previous_response_scope = Some(PreviousResponseScope::Persisted);
let first = backend
let request = pooled_websocket_request("conversation-maintenance-idle");
backend
.create_response(
&request,
request_context("req_pool_liveness_first", Some("chatgpt-account")),
request_context("req_maintenance_idle", Some("chatgpt-account")),
)
.await
.expect("first pooled websocket response should succeed");
.unwrap();
tokio::time::pause();
tokio::time::advance(LIVENESS_TIMEOUT).await;
tokio::task::yield_now().await;
liveness_closed_rx
.await
.expect("liveness watchdog should close the idle connection");
pool.maintain_idle_connections().await;
tokio::time::advance(Duration::from_secs(305)).await;
tokio::time::resume();
let second = backend
timeout(Duration::from_secs(2), closed_rx)
.await
.unwrap()
.unwrap();
let mut continuation = request;
continuation.set_previous_response_id(Some("resp_maintenance_idle".to_owned()));
continuation.previous_response_scope = Some(PreviousResponseScope::ConnectionLocal);
let error = backend
.create_response(
&request,
request_context("req_pool_liveness_second", Some("chatgpt-account")),
&continuation,
request_context("req_maintenance_continuation", Some("chatgpt-account")),
)
.await
.expect("second websocket response should open a fresh socket after liveness close");
.expect_err("maintenance should preserve the expired continuation's loss reason");
let CodexClientError::WebSocket(error) = error else {
panic!("expired continuation should remain a typed WebSocket error");
};
assert_eq!(
error.connection_observation().unwrap().exit_reason(),
"idle_timeout_expired"
);
server.await.unwrap();
assert!(first.body.contains("resp_pool_liveness_first"));
assert!(second.body.contains("resp_pool_liveness_second"));
assert_eq!(accepted_connections.load(Ordering::SeqCst), 2);
pool.shutdown().await;
}
#[tokio::test]
+9 -4
View File
@@ -305,9 +305,10 @@ Responses 流式请求完成准入后,每 15 秒无输出会发送一次 SSE
`api.request_timeout_seconds` 默认 `null`,只控制 HTTP 路由返回响应前的等待,不是流式正文或 WebSocket 每轮执行的总时限。
排队、插件显式执行期限与客户端断开的边界见 [请求期限](../docs/api.md#请求期限)
OpenAI 上游池化 WebSocket 默认每 25 秒发送一次 Ping,发出后允许等待 30 秒;
收到 Pong 或其他入站帧即解除本次心跳截止,持续无响应则以 `pong_timeout` 关闭连接。
此策略也覆盖正在生成的请求,与等待下一条上游消息的 `stream_idle_timeout_ms` 分别计时
OpenAI 上游 WebSocket 被动回应 Ping,不主动发送心跳;Pong 写入最多等待 5 秒。
下游 Responses WebSocket 每 25 秒发送 Ping,第一条未响应的 Ping 发出后 30 秒没有 Pong 或其他有效入站帧即断开;
后续 Ping 不延长这次截止,心跳入队及写入最多等待 5 秒。两段心跳各自处理,不跨 WS 转发或转换为 SSE 事件。
心跳不重置上游业务消息的 `openai.ws_pool.stream_idle_timeout_ms`
每条已保存代理可在管理端配置“上游 WS 连接上限”,默认留空、不限制连接数。
预算限制单个 CPR 进程使用该代理的 OpenAI Responses 上游 WS 总数,关联账号共享,包含建连中、在途、空闲及非池化连接;直连不受影响。
@@ -316,7 +317,11 @@ OpenAI 上游池化 WebSocket 默认每 25 秒发送一次 Ping,发出后允
修改上限对后续请求生效,已存在的连接继续计数;调低上限不会强制中断在途或绑定下游 WS 的连接。
预算仅覆盖宿主原生 OpenAI Responses 连接,插件自建连接及其他 HTTP/WS 流量仍需预留容量
有容量时继续复用,连接总寿命仍由 `max_age_ms` 控制,默认 55 分钟,不设置固定空闲回收时间。
未绑定活跃客户端 WS 的池化连接,归池后按 `openai.ws_pool.idle_timeout_ms` 计时,默认 `300000`(5 分钟,必须大于零)。
Ping/Pong 不延长空闲期限;默认每 5 秒扫描,通常在空闲 300~305 秒开始回收,复用前也检查是否到期。
关闭先停止复用,最多等待 1 秒发送 Close,随后终止并等待 socket 释放,容量在物理连接释放后才归还。
仍绑定活跃客户端 WS 的连接不受空闲期限约束;客户端退出会释放仍归它所有的上游连接,重新绑定或切到 HTTP 时释放失去用途的旧连接。
连接总寿命由 `openai.ws_pool.max_age_ms` 控制,默认 55 分钟;到期退出复用,不为清池打断正在生成的响应。
满额时优先关闭同代理下最久未使用的普通空闲 WS;在途、建连中和绑定下游 WS 的连接不参与容量淘汰。
没有可回收连接时等待容量,受原请求期限、客户端取消和连接池关闭约束;无显式请求期限时不会额外设置排队超时。
被回收连接上的精确续接沿用既有客户端完整回放合同,可能多一次恢复往返
+7
View File
@@ -82,6 +82,13 @@ openai:
residency: us
# 账号路由参考值:us、us_cr;填写值原样发送,省略或 null 时删除对应请求头
account_routing_override: null
# 可选:完整的 Responses 上游 WS 池配置;省略整段使用以下默认值。
# ws_pool:
# enabled: true
# max_age_ms: 3300000
# max_connecting: 8
# idle_timeout_ms: 300000 # 仅回收未绑定活跃客户端 WS 的空闲连接,必须大于零。
# stream_idle_timeout_ms: 300000
# Compose 启动桥接区;密码仍只维护在 store 中。
services:
+13
View File
@@ -466,6 +466,19 @@ Responses WebSocket 接受文本消息,`response.create` 在同一连接串行
留在有界接收队列中,待当前响应完成终结和写出后再逐条校验、准入与执行,不因请求提前到达而断开。
接收队列容量为 32 个事件,超载仍关闭连接;Ping/Pong、客户端关闭和服务关闭不等待队列中的请求执行
网关分别维护两段链路的心跳:上游 WS 被动回复 Pong,下游 WS 独立探活;Ping/Pong 不跨段转发,也不转换为 SSE 事件。
客户端断开时取消当前执行,并释放仍归该客户端所有的上游连接。`response.completed` 只结束本轮响应,保留客户端 WS 等待下一轮请求
| 入站 → 上游 | 正常回答结束后的连接处理 |
| --- | --- |
| SSE → SSE | 结束本次响应流,底层 HTTP 连接由 HTTP 连接池管理 |
| SSE → WS | 结束 SSE,上游 WS 归池,按无客户端绑定的空闲期限回收 |
| WS → SSE | 上游本次响应结束,下游 WS 保留 |
| WS → WS | 保留当前绑定的上游 WS,支持下一轮续接 |
活跃客户端 WS 的绑定不受普通空闲回收期限约束;上游关闭、总寿命到期或连接被回收仍可能使连接内续接失效,
客户端按既有 `previous_response_not_found` 合同完整重放。具体心跳和回收参数见[部署与运维](../deploy/README.md#公网访问)
其余文本消息发送到当前执行绑定的上游 WebSocket,未知类型、模式和扩展字段由上游校验。
`response.interrupt` 的 `discard_partial_items` 模式若指向当前逻辑请求已交付的响应 ID,
网关会保留中断意图;重试交付新的 `response.created` 后,以新 ID 转发中断,避免换号空档丢失控制消息。
+8
View File
@@ -485,6 +485,14 @@ API 暂存的下一轮请求和连接泵接收队列共用容量名额,控制
Core 只理解 `Operation`、能力要求、Provider 候选、稳定错误和 canonical event,不读取 Provider SDK
类型。Provider 独占 credential schema、OAuth、账号选择、模型目录、额度投影和上游 transport
Responses 下游连接的心跳和退出由 API 拥有,每条 WS 使用独立取消信号,沿 Core 的 `ResponseControl` 传到 Provider 请求上下文。
单轮执行结束不取消连接生命周期,连接退出则取消当前执行及仍归它所有的上游资源;API 不直接调用具体 Provider 清池。
OpenAI 的连接池监督该取消信号,按当前归属保护绑定连接;精确续接接手 socket 时撤销旧控制收发并转移归属,旧客户端不能消费新响应或清理新 owner 的连接。
重新绑定或切到 HTTP 时移除失去用途的旧连接,失效 reservation 不能重新归池。
普通空闲期限复用池槽位的 `idle_since`,只在归池时重置;心跳不刷新业务空闲,建连与在途不参与空闲回收。
回收先移出可复用集合,再在锁外有界关闭并等待 pump/socket 释放,物理容量随 socket 归还;批量关闭并行执行。
协议合同与运行参数分别见 [Responses 传输](api.md#websocket-与上游传输)和[部署与运维](../deploy/README.md#公网访问)
OpenAI 的 OAuth 与 API Key 共用现有账号和事务。API Key 的 Base URL、密钥和传输策略属于 Provider 凭据 JSON,
随 credential revision 更新;普通详情只投影非敏感连接设置。API Key 客户端目录按账号、凭据版本和客户端版本隔离,
标准 `data` 模型列表通过通用画像输出,完整 Codex `models` 目录保留原生对象。通用账号层按 Provider 提交的