diff --git a/backend/crates/gateway-api/src/openai/responses/http.rs b/backend/crates/gateway-api/src/openai/responses/http.rs index d91344256..d70089233 100644 --- a/backend/crates/gateway-api/src/openai/responses/http.rs +++ b/backend/crates/gateway-api/src/openai/responses/http.rs @@ -3,6 +3,7 @@ use std::collections::VecDeque; use std::convert::Infallible; use std::net::{IpAddr, SocketAddr}; +use std::time::Duration; use axum::{ body::{Body, Bytes}, @@ -13,7 +14,7 @@ use axum::{ }, response::{IntoResponse, Response}, }; -use futures::stream; +use futures::{StreamExt, stream}; use gateway_core::diagnostics::TraceContext; use gateway_core::engine::execution::{ClientTransport, ExecutionSession, StartedExecution}; use gateway_core::engine::{CommitRequirement, EngineError}; @@ -36,6 +37,8 @@ use super::{ request::decode_request_with_headers, }; +const SSE_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(15); + /// `POST /v1/responses`。 pub(crate) async fn responses( State(state): State, @@ -335,7 +338,7 @@ pub async fn stream_execution_response( if state.encoder.is_completed() { state.finish_completed(Vec::new()).await; } - let body = Body::from_stream(stream::unfold(state, |mut state| async move { + let output = Box::pin(stream::unfold(state, |mut state| async move { loop { if let Some(chunk) = state.pending.pop_front() { state.trace.dump("downstream.chunk", &chunk); @@ -348,19 +351,35 @@ pub async fn stream_execution_response( state.advance().await; } })); + // 保活只轮询已固定的输出 stream;不能取消并重建 advance/next_event future, + // 否则一次心跳就可能丢失正在等待的上游事件或执行终态清理。 + let body = Body::from_stream(stream::unfold(output, |mut output| async move { + let chunk = tokio::select! { + biased; + chunk = output.next() => chunk?, + () = tokio::time::sleep(SSE_KEEPALIVE_INTERVAL) => { + Ok(Bytes::from_static(b": keep-alive\n\n")) + } + }; + Some((chunk, output)) + })); event_stream_response(body, &response_headers) } fn event_stream_response(body: Body, response_headers: &[ProviderResponseHeader]) -> Response { - let mut response = Response::new(body); + let mut response = apply_response_headers(Response::new(body), response_headers); *response.status_mut() = StatusCode::OK; response .headers_mut() .insert(CONTENT_TYPE, HeaderValue::from_static("text/event-stream")); + response.headers_mut().insert( + CACHE_CONTROL, + HeaderValue::from_static("no-cache, no-transform"), + ); response .headers_mut() - .insert(CACHE_CONTROL, HeaderValue::from_static("no-cache")); - apply_response_headers(response, response_headers) + .insert("x-accel-buffering", HeaderValue::from_static("no")); + response } fn internal_gateway_response(message: &'static str) -> Response { diff --git a/backend/crates/gateway-api/tests/openai/responses/http.rs b/backend/crates/gateway-api/tests/openai/responses/http.rs index ef12d5aeb..775cef024 100644 --- a/backend/crates/gateway-api/tests/openai/responses/http.rs +++ b/backend/crates/gateway-api/tests/openai/responses/http.rs @@ -4,6 +4,7 @@ use std::sync::{ Arc, Mutex, atomic::{AtomicBool, Ordering}, }; +use std::time::Duration; use axum::{ body::{Body, to_bytes}, @@ -14,6 +15,7 @@ use axum::{ }, }; use bytes::Bytes; +use futures::StreamExt; use futures::future::{BoxFuture, pending}; use gateway_core::engine::execution::{ AuthenticatedClient, ClientAuthenticationError, ExecutionService, ExecutionSession, @@ -79,6 +81,7 @@ impl Trace { enum NextStep { Event(CoordinatedEvent), + DelayedEvent(Duration, CoordinatedEvent), Error(EngineError), FinalizeCancelled, FinalizeSuccess, @@ -280,6 +283,12 @@ impl ExecutionSession for FakeSession { self.trace.push("next_event"); Ok(Some(event)) } + NextStep::DelayedEvent(delay, event) => { + self.trace.push("wait_event"); + tokio::time::sleep(delay).await; + self.trace.push("next_event"); + Ok(Some(event)) + } NextStep::Error(error) => { self.trace.push("next_error"); self.finalized = true; @@ -709,6 +718,154 @@ async fn streaming_encodes_first_frame_before_commit_and_http_delivery() { std::mem::forget(response); } +#[tokio::test(start_paused = true)] +async fn streaming_should_keep_alive_during_silent_compaction_without_restarting_execution() { + let trace = Arc::new(Trace::default()); + let session = FakeSession::streaming( + Arc::clone(&trace), + vec![ + NextStep::Event(delivery(started(), CommitRequirement::CommitBeforeDelivery)), + NextStep::DelayedEvent( + Duration::from_secs(240), + delivery(completed(), CommitRequirement::AlreadyCommitted), + ), + NextStep::FinalizeSuccess, + ], + ); + let response = stream_execution_response(Box::new(session), None).await; + let mut body = response.into_body().into_data_stream(); + let first = body.next().await.expect("first frame").expect("body bytes"); + assert!(String::from_utf8_lossy(&first).contains("response.created")); + let start = tokio::time::Instant::now(); + + for heartbeat in 1..16 { + let chunk = tokio::time::timeout(Duration::from_secs(20), body.next()) + .await + .expect("silent upstream must not leave the downstream idle") + .expect("heartbeat frame") + .expect("heartbeat bytes"); + assert_eq!(chunk, Bytes::from_static(b": keep-alive\n\n")); + assert_eq!(start.elapsed(), Duration::from_secs(15 * heartbeat)); + } + + let terminal = body + .next() + .await + .expect("terminal frame") + .expect("body bytes"); + assert!(String::from_utf8_lossy(&terminal).contains("response.completed")); + assert_eq!(start.elapsed(), Duration::from_secs(240)); + assert_eq!( + body.next().await.expect("done frame").expect("body bytes"), + Bytes::from_static(b"data: [DONE]\n\n") + ); + assert!(body.next().await.is_none()); + assert_eq!( + trace.snapshot(), + vec![ + "next_event", + "commit", + "wait_event", + "next_event", + "next_end" + ] + ); + assert!(!trace.is_cancelled()); +} + +#[tokio::test(start_paused = true)] +async fn dropping_stream_during_keepalive_should_cancel_and_finalize_execution() { + let trace = Arc::new(Trace::default()); + let session = FakeSession::streaming( + Arc::clone(&trace), + vec![ + NextStep::Event(delivery(started(), CommitRequirement::CommitBeforeDelivery)), + NextStep::DelayedEvent( + Duration::from_secs(240), + delivery(completed(), CommitRequirement::AlreadyCommitted), + ), + NextStep::FinalizeCancelled, + ], + ); + let response = stream_execution_response(Box::new(session), None).await; + let mut body = response.into_body().into_data_stream(); + let _ = body.next().await.expect("first frame").expect("body bytes"); + let heartbeat = tokio::time::timeout(Duration::from_secs(20), body.next()) + .await + .expect("heartbeat before client disconnect") + .expect("heartbeat frame") + .expect("heartbeat bytes"); + assert_eq!(heartbeat, Bytes::from_static(b": keep-alive\n\n")); + + drop(body); + tokio::task::yield_now().await; + + assert!(trace.is_cancelled()); + assert_eq!( + trace.snapshot(), + vec!["next_event", "commit", "wait_event", "cancel_finalize"] + ); +} + +#[tokio::test(start_paused = true)] +async fn streaming_keepalive_should_wait_for_first_event_before_committing_http_status() { + let trace = Arc::new(Trace::default()); + let session = FakeSession::streaming( + Arc::clone(&trace), + vec![ + NextStep::DelayedEvent( + Duration::from_secs(45), + delivery(started(), CommitRequirement::CommitBeforeDelivery), + ), + NextStep::FinalizeCancelled, + ], + ); + let response = tokio::spawn(stream_execution_response(Box::new(session), None)); + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_secs(30)).await; + + assert!(!response.is_finished()); + assert!(trace.client_statuses().is_empty()); + let response = response.await.expect("response task"); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(trace.snapshot(), vec!["wait_event", "next_event", "commit"]); +} + +#[tokio::test] +async fn streaming_response_should_disable_proxy_buffering_and_cache_transformation() { + let session = FakeSession::streaming( + Arc::new(Trace::default()), + vec![ + NextStep::Event(delivery(started(), CommitRequirement::CommitBeforeDelivery)), + NextStep::FinalizeCancelled, + ], + ) + .with_response_headers(vec![ + ProviderResponseHeader::new("x-accel-buffering", Bytes::from_static(b"yes")), + ProviderResponseHeader::new("cache-control", Bytes::from_static(b"public, max-age=60")), + ]); + + let response = stream_execution_response(Box::new(session), None).await; + + assert_eq!(response.headers()["x-accel-buffering"], "no"); + assert_eq!( + response.headers()["cache-control"], + "no-cache, no-transform" + ); + assert_eq!( + response + .headers() + .get_all("x-accel-buffering") + .iter() + .count(), + 1 + ); + assert_eq!( + response.headers().get_all("cache-control").iter().count(), + 1 + ); +} + #[tokio::test] async fn streaming_response_should_append_ordinary_headers_and_skip_unrepresentable_or_sensitive_ones() { diff --git a/deploy/README.md b/deploy/README.md index e294a3e4b..d4d382483 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -85,6 +85,16 @@ Compose 默认只绑定 `127.0.0.1`。从其他设备访问时,在应用前配 客户端使用 `https://你的域名/v1`,不要使用前端开发服务的 `5173/dev/v1`。 当前应用只支持单副本,不能通过复制容器扩容。 +流式响应在首个上游事件提交后,每 15 秒无输出会发送一次 SSE 注释保活, +并设置 `X-Accel-Buffering: no` 和 `Cache-Control: no-cache, no-transform`。 +反向代理仍需允许这些响应头生效;首个事件到达前的等待也需要足够的读取超时。 + +若 Codex 在压缩或长时间生成时出现 `error decoding response body`,这表示 +客户端读取 HTTP 响应体失败。请结合网关请求诊断中的 `upstream.read.failed`、 +`downstream.body.closed` 和反向代理日志判断断开位置,不能仅凭此消息认定是 JSON 格式错误。 +SSE 注释保活用于防止传输链路空闲断开,不会重置 Codex 等待完整 SSE 事件的超时; +若报错为 `idle timeout waiting for SSE`,再检查客户端的 `stream_idle_timeout_ms`。 + ## 客户端配置 在管理端创建客户端密钥,打开「使用密钥」,按操作系统复制 `config.toml` 和 `auth.json`,