diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 63c3551a0..bb90288de 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -1743,6 +1743,8 @@ dependencies = [ "http", "http-body", "http-body-util", + "hyper", + "hyper-util", "regex", "secrecy", "serde", diff --git a/backend/crates/gateway-api/Cargo.toml b/backend/crates/gateway-api/Cargo.toml index 616da1dfa..8071590c8 100644 --- a/backend/crates/gateway-api/Cargo.toml +++ b/backend/crates/gateway-api/Cargo.toml @@ -25,13 +25,16 @@ gateway-admin = { path = "../gateway-admin" } gateway-core = { path = "../gateway-core" } gateway-protocol = { path = "../gateway-protocol" } headers = "0.4.1" +hyper = { version = "=1.11.1", default-features = false } +hyper-util = { version = "=0.1.21", default-features = false, features = ["tokio"] } regex = "1.13.1" secrecy = "0.10" serde = { version = "1.0.228", features = ["derive"] } serde_json = { version = "1.0.150", features = ["arbitrary_precision", "preserve_order"] } sha2 = "0.11.0" thiserror = "2.0.18" -tokio = { version = "1.52.3", default-features = false, features = ["macros", "rt", "sync", "time"] } +tokio = { version = "1.52.3", default-features = false, features = ["io-util", "macros", "rt", "sync", "time"] } +tokio-tungstenite = { version = "0.29.0", default-features = false, features = ["handshake"] } tower = { version = "0.5.3", features = ["util"] } tower-http = { version = "0.7.0", features = ["compression-gzip", "cors", "fs", "request-id", "timeout", "trace"] } tracing = "0.1.44" 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 45ddaf0ba..93866c499 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 @@ -4,8 +4,9 @@ use super::{ frame::*, pump::run_pump, state::{ConnectionCommand, ConnectionStats}, + transport::Socket, }; -use axum::extract::ws::{CloseFrame, Message, Utf8Bytes, WebSocket, close_code}; +use axum::extract::ws::{CloseFrame, Message, Utf8Bytes, close_code}; use futures::{Sink, Stream}; use gateway_core::{ engine::middleware::{FrozenMiddlewarePlan, MiddlewareHeader}, @@ -21,7 +22,7 @@ use std::{ time::Duration, }; use tokio::{ - sync::{mpsc, oneshot}, + sync::{mpsc, oneshot, watch}, task::JoinHandle, time::{Instant, timeout}, }; @@ -38,19 +39,21 @@ pub struct ResponsesWebSocketConnection { exited: oneshot::Receiver, pump_task: Option>, stats: Arc, + activity: watch::Sender, config: ConnectionConfig, exit_reason: Option, } impl ResponsesWebSocketConnection { pub(in super::super) fn new( - socket: WebSocket, + socket: Socket, connection_id: String, cancellation: CancellationToken, middleware: Option, headers: Arc<[MiddlewareHeader]>, ) -> Self { spawn_connection( + socket.activity.clone(), socket, Arc::::from(connection_id), cancellation, @@ -278,8 +281,7 @@ impl ResponsesWebSocketConnection { pub(in super::super) fn log_summary(&self, request_count: u64) { let connection_age_ms = self.age().as_millis(); - let read_idle_ms = connection_age_ms - .saturating_sub(u128::from(self.stats.last_read_ms.load(Ordering::Relaxed))); + let read_idle_ms = self.activity.borrow().elapsed().as_millis(); let write_idle_ms = connection_age_ms .saturating_sub(u128::from(self.stats.last_write_ms.load(Ordering::Relaxed))); tracing::info!( @@ -319,7 +321,9 @@ impl Drop for ResponsesWebSocketConnection { /// 为 WebSocket transport 启动包含冻结中间件计划的单 owner pump /// cancellation 必须属于本连接，连接退出或句柄释放时会取消它 +/// activity 仅由 transport 的实际入站字节或 pump 的完整消息更新，不能由写入刷新 pub fn spawn_connection( + activity: watch::Sender, socket: S, connection_id: Arc, cancellation: CancellationToken, @@ -342,6 +346,7 @@ where .unwrap_or_default(); let pump_task = tokio::spawn(run_pump( socket, + activity.clone(), command_rx, incoming_tx, exit_tx, @@ -366,6 +371,7 @@ where exited: exit_rx, pump_task: Some(pump_task), stats, + activity, config, exit_reason: None, } diff --git a/backend/crates/gateway-api/src/openai/responses/websocket/connection/mod.rs b/backend/crates/gateway-api/src/openai/responses/websocket/connection/mod.rs index 981bd2117..1c4823628 100644 --- a/backend/crates/gateway-api/src/openai/responses/websocket/connection/mod.rs +++ b/backend/crates/gateway-api/src/openai/responses/websocket/connection/mod.rs @@ -5,9 +5,12 @@ mod handle; mod middleware; mod pump; mod state; +mod transport; pub use frame::{ ConnectionConfig, ConnectionEvent, ConnectionWriteError, FramePhase, PumpExitReason, WriteContext, WriteOutcome, }; pub use handle::{ResponsesWebSocketConnection, spawn_connection}; +pub(super) use transport::Socket; +pub(crate) use transport::Upgrade; 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 6d6df01af..19c3abc8b 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 @@ -26,6 +26,7 @@ use tokio::{ #[expect(clippy::too_many_arguments)] pub(super) async fn run_pump( socket: S, + activity: watch::Sender, commands: mpsc::Receiver, incoming: mpsc::Sender, exited: oneshot::Sender, @@ -58,7 +59,7 @@ 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(()); + let received_activity = activity.subscribe(); // 心跳独立于业务写入等待，但共享唯一 writer，避免长写入或排队阻止失联检测 let reason = tokio::select! { biased; @@ -81,7 +82,7 @@ async fn read_connection( socket: &mut S, incoming: &mpsc::Sender, sender: Arc, - activity: &watch::Sender<()>, + activity: &watch::Sender, context: &PumpContext, ) -> PumpExitReason where @@ -106,10 +107,9 @@ 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(_))) { + if matches!(&message, Some(Ok(_))) { // 链路活动以真实入站为准，插件丢弃消息不应抹掉对端已经响应的事实 - activity.send_replace(()); + activity.send_replace(Instant::now()); } let message = match message { Some(Ok(message)) => match middleware::transform(message, contract::Direction::Incoming, sender.clone(), context).await { @@ -201,7 +201,7 @@ where async fn heartbeat_connection( writer: Arc>, - mut activity: watch::Receiver<()>, + mut activity: watch::Receiver, context: &PumpContext, ) -> PumpExitReason where diff --git a/backend/crates/gateway-api/src/openai/responses/websocket/connection/state.rs b/backend/crates/gateway-api/src/openai/responses/websocket/connection/state.rs index 5ac7e348d..cf494877c 100644 --- a/backend/crates/gateway-api/src/openai/responses/websocket/connection/state.rs +++ b/backend/crates/gateway-api/src/openai/responses/websocket/connection/state.rs @@ -30,18 +30,10 @@ pub(super) struct ConnectionStats { pub(super) ping_received_count: AtomicU64, pub(super) ping_written_count: AtomicU64, pub(super) pong_received_count: AtomicU64, - pub(super) last_read_ms: AtomicU64, pub(super) last_write_ms: AtomicU64, } impl ConnectionStats { - pub(super) fn record_read(&self, opened_at: Instant) { - self.last_read_ms.store( - u64::try_from(opened_at.elapsed().as_millis()).unwrap_or(u64::MAX), - Ordering::Relaxed, - ); - } - pub(super) fn record_write(&self, opened_at: Instant) { self.last_write_ms.store( u64::try_from(opened_at.elapsed().as_millis()).unwrap_or(u64::MAX), diff --git a/backend/crates/gateway-api/src/openai/responses/websocket/connection/transport.rs b/backend/crates/gateway-api/src/openai/responses/websocket/connection/transport.rs new file mode 100644 index 000000000..63a025069 --- /dev/null +++ b/backend/crates/gateway-api/src/openai/responses/websocket/connection/transport.rs @@ -0,0 +1,204 @@ +//! 保留 Axum 握手校验，在 WS 消息解码前记录下游真实接收进度 + +use std::{ + io, + pin::Pin, + task::{Context, Poll}, +}; + +use axum::{ + extract::{ + FromRequestParts, + ws::{self, Message}, + }, + http::{HeaderValue, StatusCode, Version, header, request::Parts}, + response::{IntoResponse, Response}, +}; +use futures::{Sink, Stream, ready}; +use hyper::upgrade::{OnUpgrade, Upgraded}; +use hyper_util::rt::TokioIo; +use tokio::{ + io::{AsyncRead, AsyncWrite, ReadBuf}, + sync::watch, + time::Instant, +}; +use tokio_tungstenite::{ + WebSocketStream, + tungstenite::{ + self, + protocol::{Role, WebSocketConfig}, + }, +}; + +pub(crate) struct Upgrade { + pending: OnUpgrade, + key: Option, +} + +impl FromRequestParts for Upgrade { + type Rejection = ws::rejection::WebSocketUpgradeRejection; + + async fn from_request_parts(parts: &mut Parts, state: &S) -> Result { + let pending = parts.extensions.get::().cloned(); + // Axum 的 WebSocket 不暴露底层 I/O；只复用其握手校验，由本层接管唯一升级流 + let _validated = ws::WebSocketUpgrade::from_request_parts(parts, state).await?; + let pending = pending.expect("Axum validated the presence of OnUpgrade"); + let key = (parts.version <= Version::HTTP_11) + .then(|| parts.headers.get(header::SEC_WEBSOCKET_KEY).cloned()) + .flatten(); + Ok(Self { pending, key }) + } +} + +impl Upgrade { + pub(in super::super) fn on_upgrade(self, callback: F) -> Response + where + F: FnOnce(Socket) -> Fut + Send + 'static, + Fut: Future + Send + 'static, + { + tokio::spawn(async move { + let Ok(upgraded) = self.pending.await else { + return; + }; + let (activity, _) = watch::channel(Instant::now()); + let io = Receiving { + inner: TokioIo::new(upgraded), + activity: activity.clone(), + }; + // 与原生 Responses 升级保持一致，不引入消息或帧长度上限 + let config = WebSocketConfig::default() + .max_message_size(None) + .max_frame_size(None); + let inner = WebSocketStream::from_raw_socket(io, Role::Server, Some(config)).await; + callback(Socket { inner, activity }).await; + }); + if let Some(key) = self.key { + ( + StatusCode::SWITCHING_PROTOCOLS, + [ + (header::CONNECTION, "upgrade".to_owned()), + (header::UPGRADE, "websocket".to_owned()), + ( + header::SEC_WEBSOCKET_ACCEPT, + tungstenite::handshake::derive_accept_key(key.as_bytes()), + ), + ], + ) + .into_response() + } else { + StatusCode::OK.into_response() + } + } +} + +struct Receiving { + inner: TokioIo, + activity: watch::Sender, +} + +impl AsyncRead for Receiving { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffer: &mut ReadBuf<'_>, + ) -> Poll> { + let before = buffer.filled().len(); + let result = Pin::new(&mut self.inner).poll_read(cx, buffer); + if buffer.filled().len() > before { + // 完整消息尚未收齐时，客户端可能仍在发送、暂时无法读取 Ping + self.activity.send_replace(Instant::now()); + } + result + } +} + +impl AsyncWrite for Receiving { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffer: &[u8], + ) -> Poll> { + Pin::new(&mut self.inner).poll_write(cx, buffer) + } + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_flush(cx) + } + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_shutdown(cx) + } +} + +pub(in super::super) struct Socket { + inner: WebSocketStream, + pub(super) activity: watch::Sender, +} + +impl Stream for Socket { + type Item = Result; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + loop { + let message = match ready!(Pin::new(&mut self.inner).poll_next(cx)) { + Some(Ok(message)) => message, + Some(Err(error)) => return Poll::Ready(Some(Err(axum::Error::new(error)))), + None => return Poll::Ready(None), + }; + let message = match message { + tungstenite::Message::Text(text) => Message::Text( + ws::Utf8Bytes::try_from(bytes::Bytes::from(text)).map_err(axum::Error::new)?, + ), + tungstenite::Message::Binary(data) => Message::Binary(data), + tungstenite::Message::Ping(data) => Message::Ping(data), + tungstenite::Message::Pong(data) => Message::Pong(data), + tungstenite::Message::Close(frame) => { + Message::Close(frame.map(|frame| ws::CloseFrame { + code: frame.code.into(), + reason: frame.reason.to_string().into(), + })) + } + tungstenite::Message::Frame(_) => continue, + }; + return Poll::Ready(Some(Ok(message))); + } + } +} + +impl Sink for Socket { + type Error = axum::Error; + + fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner) + .poll_ready(cx) + .map_err(axum::Error::new) + } + fn start_send(mut self: Pin<&mut Self>, message: Message) -> Result<(), Self::Error> { + let message = match message { + Message::Text(text) => tungstenite::Message::Text( + tungstenite::Utf8Bytes::try_from(bytes::Bytes::from(text)) + .map_err(axum::Error::new)?, + ), + Message::Binary(data) => tungstenite::Message::Binary(data), + Message::Ping(data) => tungstenite::Message::Ping(data), + Message::Pong(data) => tungstenite::Message::Pong(data), + Message::Close(frame) => { + tungstenite::Message::Close(frame.map(|frame| tungstenite::protocol::CloseFrame { + code: frame.code.into(), + reason: frame.reason.to_string().into(), + })) + } + }; + Pin::new(&mut self.inner) + .start_send(message) + .map_err(axum::Error::new) + } + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner) + .poll_flush(cx) + .map_err(axum::Error::new) + } + fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner) + .poll_close(cx) + .map_err(axum::Error::new) + } +} 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 f8b242892..735b05464 100644 --- a/backend/crates/gateway-api/src/openai/responses/websocket/mod.rs +++ b/backend/crates/gateway-api/src/openai/responses/websocket/mod.rs @@ -11,11 +11,7 @@ use std::{ }; use axum::{ - extract::{ - Extension, State, - connect_info::ConnectInfo, - ws::{WebSocket, WebSocketUpgrade}, - }, + extract::{Extension, State, connect_info::ConnectInfo}, http::{HeaderMap, StatusCode}, response::{IntoResponse, Response}, }; @@ -45,7 +41,9 @@ use crate::{ use super::{ request::OpenAiRequestHeaders, request_client_context, validation::ResponseValidationFacts, }; -use connection::{ConnectionEvent, FramePhase, ResponsesWebSocketConnection, WriteContext}; +use connection::{ + ConnectionEvent, FramePhase, ResponsesWebSocketConnection, Socket, Upgrade, WriteContext, +}; use forward::{ ConnectionReplaySnapshot, ForwardOutcome, execution_response, forward_control_event, forward_response, new_replay_capture, send_control, send_gateway_error, send_middleware_error, @@ -64,7 +62,7 @@ pub(crate) async fn responses_websocket( middleware: Option>, lifetime: Option>>, headers: HeaderMap, - websocket: WebSocketUpgrade, + websocket: Upgrade, ) -> Response { let service = state.openai().clone(); let client = match authenticate_client(&service, &headers).await { @@ -104,7 +102,7 @@ impl ResponsesWebSocketAdapter { #[expect(clippy::too_many_arguments)] fn upgrade_with_client_context( &self, - websocket: WebSocketUpgrade, + websocket: Upgrade, client: AuthenticatedClient, client_ip: Option, user_agent: Option, @@ -138,15 +136,10 @@ impl ResponsesWebSocketAdapter { middleware, }; let origin = crate::middleware::current(); - websocket - // 覆盖 axum/tungstenite 的私有 64 MiB message 与 16 MiB frame 默认值 - // Responses JSON 的协议可接受性由上游决定，代理不另设 wire 长度上限 - .max_message_size(usize::MAX) - .max_frame_size(usize::MAX) - .on_upgrade(move |socket| async move { - let _lifetime = lifetime; - crate::middleware::scope(origin, serve_responses_websocket(socket, session)).await; - }) + websocket.on_upgrade(move |socket| async move { + let _lifetime = lifetime; + crate::middleware::scope(origin, serve_responses_websocket(socket, session)).await; + }) } } @@ -163,7 +156,7 @@ struct ResponsesWebSocketSession { middleware: Option, } -async fn serve_responses_websocket(socket: WebSocket, session: ResponsesWebSocketSession) { +async fn serve_responses_websocket(socket: Socket, session: ResponsesWebSocketSession) { let ResponsesWebSocketSession { service, client, diff --git a/backend/crates/gateway-api/tests/openai/responses/websocket/connection/middleware.rs b/backend/crates/gateway-api/tests/openai/responses/websocket/connection/middleware.rs index d5d2ebfc4..4719750b9 100644 --- a/backend/crates/gateway-api/tests/openai/responses/websocket/connection/middleware.rs +++ b/backend/crates/gateway-api/tests/openai/responses/websocket/connection/middleware.rs @@ -66,6 +66,7 @@ async fn client_close_does_not_flush_a_retained_sender_with_an_inflight_write() dropped: Arc::new(AtomicBool::new(false)), }; let mut connection = spawn_connection( + tokio::sync::watch::channel(tokio::time::Instant::now()).0, socket, Arc::from("retained-writer"), CancellationToken::new(), 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 1d7609b9f..8246e4fbd 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 @@ -1,6 +1,7 @@ //! Responses WebSocket 连接生命周期与多轮请求边界测试 mod middleware; +mod transport; use std::{ fmt, @@ -311,6 +312,7 @@ fn test_connection(stall_writes: bool) -> PumpHarness { dropped: Arc::clone(&dropped), }; let connection = spawn_connection( + tokio::sync::watch::channel(tokio::time::Instant::now()).0, socket, Arc::from("ws_test"), cancellation.clone(), diff --git a/backend/crates/gateway-api/tests/openai/responses/websocket/connection/transport.rs b/backend/crates/gateway-api/tests/openai/responses/websocket/connection/transport.rs new file mode 100644 index 000000000..4156dd0cd --- /dev/null +++ b/backend/crates/gateway-api/tests/openai/responses/websocket/connection/transport.rs @@ -0,0 +1,181 @@ +//! 实际 HTTP 升级及 WS 编解码下，慢速上传与半帧停传的心跳边界 + +use std::{ + io, + pin::Pin, + sync::{Arc, atomic::Ordering}, + task::{Context, Poll}, + time::Duration, +}; + +use futures::{SinkExt, StreamExt, ready}; +use hyper_util::rt::TokioIo; +use serde_json::{Value, json}; +use tokio::{ + io::{AsyncRead, AsyncWrite, AsyncWriteExt, DuplexStream, ReadBuf}, + time::{Instant, Sleep}, +}; +use tokio_tungstenite::{ + WebSocketStream, client_async, + tungstenite::{Message, client::IntoClientRequest}, +}; +use tower::ServiceExt; + +use super::{CrossingLimitExecution, CrossingLimitTrace, api_router, authenticated_client}; + +struct SlowUpload { + inner: DuplexStream, + slow: bool, + next_write: Option>>, +} + +impl AsyncRead for SlowUpload { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffer: &mut ReadBuf<'_>, + ) -> Poll> { + Pin::new(&mut self.inner).poll_read(cx, buffer) + } +} + +impl AsyncWrite for SlowUpload { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffer: &[u8], + ) -> Poll> { + if self.slow { + if let Some(delay) = self.next_write.as_mut() { + ready!(delay.as_mut().poll(cx)); + } + let size = ready!( + Pin::new(&mut self.inner).poll_write(cx, &buffer[..buffer.len().min(4096)]) + )?; + self.next_write = Some(Box::pin(tokio::time::sleep(Duration::from_secs(1)))); + Poll::Ready(Ok(size)) + } else { + Pin::new(&mut self.inner).poll_write(cx, buffer) + } + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_flush(cx) + } + + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_shutdown(cx) + } +} + +async fn connect() -> (WebSocketStream, Arc) { + let trace = Arc::new(CrossingLimitTrace::default()); + let app = api_router(Arc::new(CrossingLimitExecution { + client: authenticated_client("sk_ws_crossing_limit"), + trace: trace.clone(), + })) + .await; + let (client, server) = tokio::io::duplex(65536); + tokio::spawn(async move { + hyper::server::conn::http1::Builder::new() + .serve_connection( + TokioIo::new(server), + hyper::service::service_fn(move |request| app.clone().oneshot(request)), + ) + .with_upgrades() + .await + .expect("serve Responses upgrade"); + }); + let mut request = "ws://localhost/v1/responses".into_client_request().unwrap(); + request.headers_mut().insert( + http::header::AUTHORIZATION, + "Bearer sk_ws_crossing_limit".parse().unwrap(), + ); + let (socket, response) = client_async( + request, + SlowUpload { + inner: client, + slow: false, + next_write: None, + }, + ) + .await + .unwrap(); + assert_eq!(response.status(), http::StatusCode::SWITCHING_PROTOCOLS); + (socket, trace) +} + +async fn receive_event(socket: &mut WebSocketStream, expected: &str) { + tokio::time::timeout(Duration::from_secs(2), async { + loop { + let message = socket + .next() + .await + .expect("connection stays open") + .expect("valid frame"); + if let Message::Text(text) = message { + let event: Value = serde_json::from_str(&text).unwrap(); + if event["type"] == expected { + break; + } + assert_eq!(event["type"], "response.metadata"); + } else { + assert!(matches!(message, Message::Ping(_))); + } + } + }) + .await + .expect("receive response after upload"); +} + +#[tokio::test(start_paused = true)] +async fn slow_unfragmented_uploads_survive_heartbeats_on_first_and_later_requests() { + let (mut socket, trace) = connect().await; + let input = "x".repeat(320 * 1024); + for round in 1..=2 { + let started = Instant::now(); + socket.get_mut().slow = true; + // 对齐 Codex 的发送语义：整个 send 完成前不读 Ping，也不主动发送 Pong + tokio::time::timeout( + Duration::from_secs(100), + socket.send(Message::Text( + json!({"type": "response.create", "model": "model-a", "input": input}) + .to_string() + .into(), + )), + ) + .await + .expect("upload completes") + .expect("heartbeat must not interrupt an active upload"); + assert!(started.elapsed() >= Duration::from_secs(80)); + socket.get_mut().slow = false; + receive_event(&mut socket, "response.created").await; + assert_eq!(trace.starts.load(Ordering::Acquire), round); + trace.release_terminal.notify_one(); + receive_event(&mut socket, "response.completed").await; + } + assert_eq!( + *trace.inputs.lock().unwrap(), + vec![json!(input), json!(input)] + ); + socket.close(None).await.unwrap(); +} + +#[tokio::test(start_paused = true)] +async fn stalled_partial_frame_without_pongs_still_releases_the_connection() { + let (mut socket, trace) = connect().await; + // 已掩码的文本帧声明 4096 字节，只交付一个字节，之后不再收发 + socket + .get_mut() + .write_all(&[0x81, 0xfe, 0x10, 0x00, 0, 0, 0, 0, b'x']) + .await + .unwrap(); + tokio::time::sleep(Duration::from_secs(56)).await; + let error = socket + .get_mut() + .write_all(b"x") + .await + .expect_err("idle partial frame must be closed"); + assert_eq!(error.kind(), io::ErrorKind::BrokenPipe); + assert_eq!(trace.starts.load(Ordering::Acquire), 0); +} diff --git a/deploy/README.md b/deploy/README.md index 20be23dca..a325e05db 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -306,7 +306,8 @@ Responses 流式请求完成准入后，每 15 秒无输出会发送一次 SSE 排队、插件显式执行期限与客户端断开的边界见 [请求期限](../docs/api.md#请求期限) OpenAI 上游 WebSocket 被动回应 Ping，不主动发送心跳；Pong 写入最多等待 5 秒。 -下游 Responses WebSocket 每 25 秒发送 Ping，第一条未响应的 Ping 发出后 30 秒没有 Pong 或其他有效入站帧即断开； +下游 Responses WebSocket 每 25 秒发送 Ping，第一条未响应的 Ping 发出后 30 秒没有入站活动即断开； +接收中的消息字节、Pong 和其他有效入站消息均算活动，慢速上传不必等整帧收齐才刷新探活状态。 后续 Ping 不延长这次截止，心跳入队及写入最多等待 5 秒。两段心跳各自处理，不跨 WS 转发或转换为 SSE 事件。 心跳不重置上游业务消息的 `openai.ws_pool.stream_idle_timeout_ms` diff --git a/docs/api.md b/docs/api.md index 0ba505a96..1b9c96a5b 100644 --- a/docs/api.md +++ b/docs/api.md @@ -468,6 +468,7 @@ Responses WebSocket 接受文本消息，`response.create` 在同一连接串行 接收队列容量为 32 个事件，超载仍关闭连接；Ping/Pong、客户端关闭和服务关闭不等待队列中的请求执行 网关分别维护两段链路的心跳：上游 WS 被动回复 Pong，下游 WS 独立探活；Ping/Pong 不跨段转发，也不转换为 SSE 事件。 +下游探活包含消息接收中的字节进度，完整消息收齐后才进入中间件与请求处理。 客户端断开时取消当前执行，并释放仍归该客户端所有的上游连接。`response.completed` 只结束本轮响应，保留客户端 WS 等待下一轮请求 | 入站 → 上游 | 正常回答结束后的连接处理 |