diff --git a/backend/crates/gateway-api/src/openai/responses/websocket/connection/frame.rs b/backend/crates/gateway-api/src/openai/responses/websocket/connection/frame.rs index 4e9bd4a74..24a12040d 100644 --- a/backend/crates/gateway-api/src/openai/responses/websocket/connection/frame.rs +++ b/backend/crates/gateway-api/src/openai/responses/websocket/connection/frame.rs @@ -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", diff --git a/backend/crates/gateway-api/src/openai/responses/websocket/connection/handle.rs b/backend/crates/gateway-api/src/openai/responses/websocket/connection/handle.rs index dbc68e834..45ddaf0ba 100644 --- a/backend/crates/gateway-api/src/openai/responses/websocket/connection/handle.rs +++ b/backend/crates/gateway-api/src/openai/responses/websocket/connection/handle.rs @@ -30,6 +30,7 @@ use tokio::{ pub struct ResponsesWebSocketConnection { connection_id: Arc, opened_at: Instant, + cancellation: CancellationToken, expired: Arc, commands: Option>, incoming: mpsc::Receiver, @@ -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( socket: S, connection_id: Arc, @@ -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, diff --git a/backend/crates/gateway-api/src/openai/responses/websocket/connection/pump.rs b/backend/crates/gateway-api/src/openai/responses/websocket/connection/pump.rs index 8de82f5e5..6d6df01af 100644 --- a/backend/crates/gateway-api/src/openai/responses/websocket/connection/pump.rs +++ b/backend/crates/gateway-api/src/openai/responses/websocket/connection/pump.rs @@ -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: Stream> + Sink + 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( let (writer, mut reader) = socket.split(); let writer = Arc::new(middleware::Writer::new(writer, context.clone())); let sender: Arc = 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( socket: &mut S, incoming: &mpsc::Sender, sender: Arc, + 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 + 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( + writer: Arc>, + mut activity: watch::Receiver<()>, + context: &PumpContext, +) -> PumpExitReason +where + S: Sink + 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( + writer: &Arc>, + sequence: u64, + context: &PumpContext, +) -> Result +where + S: Sink + 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, diff --git a/backend/crates/gateway-api/src/openai/responses/websocket/mod.rs b/backend/crates/gateway-api/src/openai/responses/websocket/mod.rs index 1ce93e93b..b4cc7c941 100644 --- a/backend/crates/gateway-api/src/openai/responses/websocket/mod.rs +++ b/backend/crates/gateway-api/src/openai/responses/websocket/mod.rs @@ -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::::from(prepared.request_id().to_string()); response_control = prepared.response_control(); let capture = new_replay_capture(); diff --git a/backend/crates/gateway-api/tests/openai/responses/websocket/connection/mod.rs b/backend/crates/gateway-api/tests/openai/responses/websocket/connection/mod.rs index 4d6bc7c47..1d7609b9f 100644 --- a/backend/crates/gateway-api/tests/openai/responses/websocket/connection/mod.rs +++ b/backend/crates/gateway-api/tests/openai/responses/websocket/connection/mod.rs @@ -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::::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)); } diff --git a/backend/crates/gateway-core/src/engine/execution/contract.rs b/backend/crates/gateway-core/src/engine/execution/contract.rs index 0a234884d..8cb7ffffa 100644 --- a/backend/crates/gateway-core/src/engine/execution/contract.rs +++ b/backend/crates/gateway-core/src/engine/execution/contract.rs @@ -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 { let started_at = SystemTime::now(); let deadline_at = Deadline::from_timeout(started_at, client.execution_timeout()) diff --git a/backend/crates/gateway-core/src/engine/response_control.rs b/backend/crates/gateway-core/src/engine/response_control.rs index 880c419d7..3cbb15d60 100644 --- a/backend/crates/gateway-core/src/engine/response_control.rs +++ b/backend/crates/gateway-core/src/engine/response_control.rs @@ -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>>>, + connection_cancellation: Option, } /// 某次已交付响应所属的弱连接快照,重绑后不能把旧消息发送到新连接 @@ -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 { diff --git a/backend/crates/providers/openai/src/config.rs b/backend/crates/providers/openai/src/config.rs index 790ccd2a0..999e700c1 100644 --- a/backend/crates/providers/openai/src/config.rs +++ b/backend/crates/providers/openai/src/config.rs @@ -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() diff --git a/backend/crates/providers/openai/src/provider/execution.rs b/backend/crates/providers/openai/src/provider/execution.rs index 21422032f..cf21b514a 100644 --- a/backend/crates/providers/openai/src/provider/execution.rs +++ b/backend/crates/providers/openai/src/provider/execution.rs @@ -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, diff --git a/backend/crates/providers/openai/src/provider/failure.rs b/backend/crates/providers/openai/src/provider/failure.rs index 836e1ffee..9c7635804 100644 --- a/backend/crates/providers/openai/src/provider/failure.rs +++ b/backend/crates/providers/openai/src/provider/failure.rs @@ -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: "); diff --git a/backend/crates/providers/openai/src/provider/mod.rs b/backend/crates/providers/openai/src/provider/mod.rs index e07f87586..d033e059e 100644 --- a/backend/crates/providers/openai/src/provider/mod.rs +++ b/backend/crates/providers/openai/src/provider/mod.rs @@ -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()), diff --git a/backend/crates/providers/openai/src/provider/observation.rs b/backend/crates/providers/openai/src/provider/observation.rs index 38865c5ac..bbe36ac33 100644 --- a/backend/crates/providers/openai/src/provider/observation.rs +++ b/backend/crates/providers/openai/src/provider/observation.rs @@ -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(), diff --git a/backend/crates/providers/openai/src/transport/client.rs b/backend/crates/providers/openai/src/transport/client.rs index 0f8026165..280caeb66 100644 --- a/backend/crates/providers/openai/src/transport/client.rs +++ b/backend/crates/providers/openai/src/transport/client.rs @@ -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, pub(super) connection_budget: Option, 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, - ) -> Self { - self.response_control = control; - self - } - pub(crate) fn with_connection_budget( mut self, budget: gateway_core::engine::connection::ConnectionBudget, diff --git a/backend/crates/providers/openai/src/transport/client_sse.rs b/backend/crates/providers/openai/src/transport/client_sse.rs index bc5a42ad7..34229240b 100644 --- a/backend/crates/providers/openai/src/transport/client_sse.rs +++ b/backend/crates/providers/openai/src/transport/client_sse.rs @@ -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 { + 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() diff --git a/backend/crates/providers/openai/src/transport/websocket/control.rs b/backend/crates/providers/openai/src/transport/websocket/control.rs index eec55bf90..1300ce5f5 100644 --- a/backend/crates/providers/openai/src/transport/websocket/control.rs +++ b/backend/crates/providers/openai/src/transport/websocket/control.rs @@ -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, 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 { - 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), } diff --git a/backend/crates/providers/openai/src/transport/websocket/coordinator.rs b/backend/crates/providers/openai/src/transport/websocket/coordinator.rs index b3979fa3b..7d750db3b 100644 --- a/backend/crates/providers/openai/src/transport/websocket/coordinator.rs +++ b/backend/crates/providers/openai/src/transport/websocket/coordinator.rs @@ -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)>, + pool: Option<( + &CodexWebSocketPool, + Option, + Option<&gateway_core::lifecycle::CancellationToken>, + )>, breaker: &WebSocketOriginBreaker, origin_key: &str, fast_path_budget: Option, @@ -120,7 +124,7 @@ pub(crate) async fn prepare_response_create_request_with_pool( fallback_stream_idle_timeout: Option, ) -> Result, 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, context: PumpLogContext, proxy_permit: Option, @@ -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, ) -> 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, diff --git a/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs b/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs index 6cf896ac9..486a5fe81 100644 --- a/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs +++ b/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs @@ -69,6 +69,10 @@ pub(in crate::transport::websocket) fn stream_websocket_response( response_control: Option, ) -> 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, shutdown: CancellationToken, + downstream_cancellation: Option, rate_limit_updates: CodexWebSocketRateLimitUpdates, response_metadata_updates: CodexWebSocketResponseMetadataUpdates, tx: mpsc::Sender>, @@ -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, ) { let Some(pool_return) = pool_return else { - websocket.close().await; + websocket.close_and_wait().await; return; }; pool_return diff --git a/backend/crates/providers/openai/src/transport/websocket/handshake.rs b/backend/crates/providers/openai/src/transport/websocket/handshake.rs index 180d503df..a2084dbce 100644 --- a/backend/crates/providers/openai/src/transport/websocket/handshake.rs +++ b/backend/crates/providers/openai/src/transport/websocket/handshake.rs @@ -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, ) -> Result<(PumpedWebSocket, WsResponse>>), 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( diff --git a/backend/crates/providers/openai/src/transport/websocket/pool/capacity.rs b/backend/crates/providers/openai/src/transport/websocket/pool/capacity.rs index 6581bb939..6bce41f17 100644 --- a/backend/crates/providers/openai/src/transport/websocket/pool/capacity.rs +++ b/backend/crates/providers/openai/src/transport/websocket/pool/capacity.rs @@ -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)) } diff --git a/backend/crates/providers/openai/src/transport/websocket/pool/downstream.rs b/backend/crates/providers/openai/src/transport/websocket/pool/downstream.rs new file mode 100644 index 000000000..41f318855 --- /dev/null +++ b/backend/crates/providers/openai/src/transport/websocket/pool/downstream.rs @@ -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) { + 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 { + 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::>(); + 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 + } +} diff --git a/backend/crates/providers/openai/src/transport/websocket/pool/mod.rs b/backend/crates/providers/openai/src/transport/websocket/pool/mod.rs index d2471e504..de58655f5 100644 --- a/backend/crates/providers/openai/src/transport/websocket/pool/mod.rs +++ b/backend/crates/providers/openai/src/transport/websocket/pool/mod.rs @@ -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, - /// 池化连接的探活 ping 间隔(包括正在生成的连接);`None` 表示不主动 ping - pub ping_interval: Option, - /// 发送 ping 后等待任意入站帧的超时时间;零值表示不校验 Pong deadline - pub ping_timeout: Duration, - /// idle socket 无活动多久后视为失活 - pub liveness_timeout: Option, + /// 未绑定下游 WS 的连接归池后可空闲的时间,不由 Ping/Pong 刷新 + pub idle_timeout: Duration, /// 等待下一条上游消息的空闲超时;`None` 或零值使用默认 300 秒 pub stream_idle_timeout: Option, } @@ -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 { 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), } diff --git a/backend/crates/providers/openai/src/transport/websocket/pool/state.rs b/backend/crates/providers/openai/src/transport/websocket/pool/state.rs index b5e439250..981aeeffc 100644 --- a/backend/crates/providers/openai/src/transport/websocket/pool/state.rs +++ b/backend/crates/providers/openai/src/transport/websocket/pool/state.rs @@ -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, + pub(super) downstream_connections: HashMap, continuation_tombstones: VecDeque, 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) { - for connection in connections { - close_pooled_connection(connection).await; - } + // 每条连接独立执行有界关闭,避免批量回收累计握手等待并拖延后续扫描 + let mut closing = connections + .into_iter() + .map(close_pooled_connection) + .collect::>(); + 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 { + 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) -> String { diff --git a/backend/crates/providers/openai/src/transport/websocket/pool/supervisor.rs b/backend/crates/providers/openai/src/transport/websocket/pool/supervisor.rs index 0b53b189c..ae69f6de3 100644 --- a/backend/crates/providers/openai/src/transport/websocket/pool/supervisor.rs +++ b/backend/crates/providers/openai/src/transport/websocket/pool/supervisor.rs @@ -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::>(); - 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)) => { diff --git a/backend/crates/providers/openai/src/transport/websocket/pump.rs b/backend/crates/providers/openai/src/transport/websocket/pump.rs index 44a2a5e58..68288e2aa 100644 --- a/backend/crates/providers/openai/src/transport/websocket/pump.rs +++ b/backend/crates/providers/openai/src/transport/websocket/pump.rs @@ -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 WebSocketIo for T where pub(crate) type RawWsStream = Box; 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, - /// 主动 `Ping` 发出后等待任意入站帧的 deadline;`None` 表示不校验响应 - pub(crate) ping_timeout: Option, - /// 无入站活动多久后判定失活;`None` 表示只靠显式 close/传输错误发现死亡 - pub(crate) liveness_timeout: Option, -} - -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, } -#[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, rx_message: SharedMessages, - control: Option>, + control: Option>, closed: Arc, lifecycle: Arc>, pump: Option>, @@ -308,7 +258,6 @@ impl PumpedWebSocket { /// 用底层流启动一个 pump 任务并返回句柄 pub(crate) fn new( inner: RawWsStream, - keepalive: PumpKeepalive, context: PumpLogContext, proxy_permit: Option, ) -> 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 = 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 = 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, tx_message: mpsc::Sender>, - keepalive: PumpKeepalive, context: PumpLogContext, closed: Arc, lifecycle: Arc>, ) { - 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 = None; - let mut pending_ping: Option = 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>, - pending_inbound: &mut Option, + pending_inbound: &mut Option>, 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, 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, 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, -) -> Result, 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) { - match ticker { - Some(ticker) => { - ticker.tick().await; - } - None => std::future::pending::<()>().await, - } -} - -/// 等待可选 deadline;`None` 时永远挂起,让 `select!` 分支实际禁用 -async fn wait_until(deadline: Option) { - match deadline { - Some(deadline) => tokio::time::sleep_until(deadline).await, - None => std::future::pending::<()>().await, - } -} diff --git a/backend/crates/providers/openai/tests/config.rs b/backend/crates/providers/openai/tests/config.rs index 6c9cce75f..ac34f1751 100644 --- a/backend/crates/providers/openai/tests/config.rs +++ b/backend/crates/providers/openai/tests/config.rs @@ -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() } diff --git a/backend/crates/providers/openai/tests/provider/contract/mod.rs b/backend/crates/providers/openai/tests/provider/contract/mod.rs index 7c15f85a0..7237cd394 100644 --- a/backend/crates/providers/openai/tests/provider/contract/mod.rs +++ b/backend/crates/providers/openai/tests/provider/contract/mod.rs @@ -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"; diff --git a/backend/crates/providers/openai/tests/provider/contract/response_interrupt/cleanup.rs b/backend/crates/providers/openai/tests/provider/contract/response_interrupt/cleanup.rs index 4392e8913..85ea4d641 100644 --- a/backend/crates/providers/openai/tests/provider/contract/response_interrupt/cleanup.rs +++ b/backend/crates/providers/openai/tests/provider/contract/response_interrupt/cleanup.rs @@ -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; diff --git a/backend/crates/providers/openai/tests/provider/contract/response_interrupt/mod.rs b/backend/crates/providers/openai/tests/provider/contract/response_interrupt/mod.rs index d45e898cf..3a3280494 100644 --- a/backend/crates/providers/openai/tests/provider/contract/response_interrupt/mod.rs +++ b/backend/crates/providers/openai/tests/provider/contract/response_interrupt/mod.rs @@ -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), diff --git a/backend/crates/providers/openai/tests/transport/catalog.rs b/backend/crates/providers/openai/tests/transport/catalog.rs index 8ac62109b..94446e5a3 100644 --- a/backend/crates/providers/openai/tests/transport/catalog.rs +++ b/backend/crates/providers/openai/tests/transport/catalog.rs @@ -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"), diff --git a/backend/crates/providers/openai/tests/transport/client.rs b/backend/crates/providers/openai/tests/transport/client.rs index 2d89298fb..722408c0e 100644 --- a/backend/crates/providers/openai/tests/transport/client.rs +++ b/backend/crates/providers/openai/tests/transport/client.rs @@ -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"), diff --git a/backend/crates/providers/openai/tests/transport/downstream/headers.rs b/backend/crates/providers/openai/tests/transport/downstream/headers.rs index b868767df..a631b90d0 100644 --- a/backend/crates/providers/openai/tests/transport/downstream/headers.rs +++ b/backend/crates/providers/openai/tests/transport/downstream/headers.rs @@ -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"), diff --git a/backend/crates/providers/openai/tests/transport/headers.rs b/backend/crates/providers/openai/tests/transport/headers.rs index bad262ed6..10988fbb5 100644 --- a/backend/crates/providers/openai/tests/transport/headers.rs +++ b/backend/crates/providers/openai/tests/transport/headers.rs @@ -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"), diff --git a/backend/crates/providers/openai/tests/transport/http_client.rs b/backend/crates/providers/openai/tests/transport/http_client.rs index fa8cc4f0b..eea8e4149 100644 --- a/backend/crates/providers/openai/tests/transport/http_client.rs +++ b/backend/crates/providers/openai/tests/transport/http_client.rs @@ -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"), diff --git a/backend/crates/providers/openai/tests/transport/mod.rs b/backend/crates/providers/openai/tests/transport/mod.rs index 8b19e0421..2c49a7c2b 100644 --- a/backend/crates/providers/openai/tests/transport/mod.rs +++ b/backend/crates/providers/openai/tests/transport/mod.rs @@ -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, - ping_interval: Option, - liveness_timeout: Option, ) -> 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() } } diff --git a/backend/crates/providers/openai/tests/transport/proxy_capacity.rs b/backend/crates/providers/openai/tests/transport/proxy_capacity.rs index 3b7da3f9c..a75cf6433 100644 --- a/backend/crates/providers/openai/tests/transport/proxy_capacity.rs +++ b/backend/crates/providers/openai/tests/transport/proxy_capacity.rs @@ -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 { 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(); diff --git a/backend/crates/providers/openai/tests/transport/usage.rs b/backend/crates/providers/openai/tests/transport/usage.rs index 44ee71053..7d962b9cd 100644 --- a/backend/crates/providers/openai/tests/transport/usage.rs +++ b/backend/crates/providers/openai/tests/transport/usage.rs @@ -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"), diff --git a/backend/crates/providers/openai/tests/transport/websocket/cleanup.rs b/backend/crates/providers/openai/tests/transport/websocket/cleanup.rs index 1a7d6ec1a..aa4028b5a 100644 --- a/backend/crates/providers/openai/tests/transport/websocket/cleanup.rs +++ b/backend/crates/providers/openai/tests/transport/websocket/cleanup.rs @@ -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() })); diff --git a/backend/crates/providers/openai/tests/transport/websocket/fingerprint.rs b/backend/crates/providers/openai/tests/transport/websocket/fingerprint.rs index 08439fd92..15814ff4f 100644 --- a/backend/crates/providers/openai/tests/transport/websocket/fingerprint.rs +++ b/backend/crates/providers/openai/tests/transport/websocket/fingerprint.rs @@ -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(), diff --git a/backend/crates/providers/openai/tests/transport/websocket/mod.rs b/backend/crates/providers/openai/tests/transport/websocket/mod.rs index ae5b89fcb..e5c69dcac 100644 --- a/backend/crates/providers/openai/tests/transport/websocket/mod.rs +++ b/backend/crates/providers/openai/tests/transport/websocket/mod.rs @@ -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"), diff --git a/backend/crates/providers/openai/tests/transport/websocket_pool.rs b/backend/crates/providers/openai/tests/transport/websocket_pool.rs index 943fd738b..504681f6e 100644 --- a/backend/crates/providers/openai/tests/transport/websocket_pool.rs +++ b/backend/crates/providers/openai/tests/transport/websocket_pool.rs @@ -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] diff --git a/deploy/README.md b/deploy/README.md index 85689e93c..78625fec5 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -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 的连接不参与容量淘汰。 没有可回收连接时等待容量,受原请求期限、客户端取消和连接池关闭约束;无显式请求期限时不会额外设置排队超时。 被回收连接上的精确续接沿用既有客户端完整回放合同,可能多一次恢复往返 diff --git a/deploy/config.example.yaml b/deploy/config.example.yaml index 83534a9c7..7ae7ae9c8 100644 --- a/deploy/config.example.yaml +++ b/deploy/config.example.yaml @@ -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: diff --git a/docs/api.md b/docs/api.md index 0539fcdbb..83ec01e10 100644 --- a/docs/api.md +++ b/docs/api.md @@ -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 转发中断,避免换号空档丢失控制消息。 diff --git a/docs/architecture.md b/docs/architecture.md index 5108ac141..65b37278e 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -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 提交的