From c21633655e2e6003dbded5a9839eefc537d8560e Mon Sep 17 00:00:00 2001 From: Zyy Date: Sun, 6 Sep 2026 16:49:35 +0800 Subject: [PATCH] fix(api): keep SSE responses alive during upstream silence Send SSE keepalive comments after 15 seconds without output while preserving the pending execution future. Disable response buffering and transformation so long compaction requests can maintain the downstream connection. Cover four minutes of upstream silence, cancellation, first-event commitment, and conflicting upstream headers. Document transport errors and the separate Codex event idle timeout. Validation: 259 gateway-api tests and Clippy passed with serde/rc enabled; cargo fmt and git diff --check passed. Co-authored-by: Codex --- .../gateway-api/src/openai/responses/http.rs | 29 +++- .../tests/openai/responses/http.rs | 157 ++++++++++++++++++ deploy/README.md | 10 ++ 3 files changed, 191 insertions(+), 5 deletions(-) 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`，