From 254134946b11f1e6f56d2c68456a8a60cd4ebb8c Mon Sep 17 00:00:00 2001 From: DKalien <1063027053@qq.com> Date: Sun, 19 Jul 2026 10:09:10 +0800 Subject: [PATCH] fix(streaming): prevent SSE idle disconnects and total timeouts Keep downstream SSE connections alive during upstream idle periods, use a read-timeout-only client for streaming requests, and align generated provider idle settings with the longer stream budget. Add regression coverage for both behaviors. Upstream-Ref: HisenWeb/codex-opencode-adapter --- config.toml.example | 2 +- scripts/install-user-provider.ps1 | 2 +- src/config.rs | 1 + src/init.rs | 26 ++++++++++++++-- src/server.rs | 38 +++++++++++++++++++++++- src/upstream.rs | 49 +++++++++++++++++++++++++++++-- 6 files changed, 110 insertions(+), 8 deletions(-) diff --git a/config.toml.example b/config.toml.example index 72648ad..9432d9c 100644 --- a/config.toml.example +++ b/config.toml.example @@ -14,7 +14,7 @@ base_url = "http://127.0.0.1:4010/v1" wire_api = "responses" request_max_retries = 0 stream_max_retries = 0 -stream_idle_timeout_ms = 120000 +stream_idle_timeout_ms = 360000 [model_providers.opencode_go_adapter.auth] command = "codex-opencode-adapter" diff --git a/scripts/install-user-provider.ps1 b/scripts/install-user-provider.ps1 index a74784a..3abed1a 100644 --- a/scripts/install-user-provider.ps1 +++ b/scripts/install-user-provider.ps1 @@ -17,7 +17,7 @@ base_url = "http://127.0.0.1:4010/v1" wire_api = "responses" request_max_retries = 0 stream_max_retries = 0 -stream_idle_timeout_ms = 120000 +stream_idle_timeout_ms = 360000 [model_providers.opencode_go_adapter.auth] command = "codex-opencode-adapter" diff --git a/src/config.rs b/src/config.rs index 19db391..641a7f5 100644 --- a/src/config.rs +++ b/src/config.rs @@ -8,6 +8,7 @@ pub const DEFAULT_UPSTREAM_BASE: &str = "https://opencode.ai/zen/go/v1"; pub const DEFAULT_STATE_DB: &str = ".codex-opencode/state.sqlite"; pub const DEFAULT_STATE_TTL_SECONDS: i64 = 21_600; pub const DEFAULT_TIMEOUT_SECONDS: u64 = 300; +pub const DEFAULT_STREAM_IDLE_TIMEOUT_MS: i64 = 360_000; pub const DEFAULT_MAX_REQUEST_BYTES: usize = 8 * 1024 * 1024; pub const DEFAULT_MAX_CONCURRENCY: usize = 8; diff --git a/src/init.rs b/src/init.rs index ead303a..a3a4357 100644 --- a/src/init.rs +++ b/src/init.rs @@ -1,7 +1,7 @@ use crate::cli::InitArgs; use crate::config::{ DEFAULT_HOST, DEFAULT_MAX_CONCURRENCY, DEFAULT_MAX_REQUEST_BYTES, DEFAULT_STATE_DB, - DEFAULT_STATE_TTL_SECONDS, DEFAULT_TIMEOUT_SECONDS, + DEFAULT_STATE_TTL_SECONDS, DEFAULT_STREAM_IDLE_TIMEOUT_MS, DEFAULT_TIMEOUT_SECONDS, }; use crate::project::{generate_project_id, project_key_from_id, ProjectPaths, ProjectRegistry}; use anyhow::{anyhow, Context}; @@ -166,7 +166,7 @@ fn build_global_codex_config(path: &Path, port: u16) -> anyhow::Result { provider["wire_api"] = value("responses"); provider["request_max_retries"] = value(0); provider["stream_max_retries"] = value(0); - provider["stream_idle_timeout_ms"] = value(120000); + provider["stream_idle_timeout_ms"] = value(DEFAULT_STREAM_IDLE_TIMEOUT_MS); let auth = ensure_subtable(provider, "auth")?; auth["command"] = value("codex-opencode-adapter"); @@ -197,6 +197,28 @@ fn ensure_subtable<'a>(table: &'a mut Table, key: &str) -> anyhow::Result<&'a mu .ok_or_else(|| anyhow!("{key} must be a TOML table")) } +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn generated_provider_idle_timeout_exceeds_upstream_request_timeout() { + let path = std::env::temp_dir().join(format!( + "codex-opencode-config-{}.toml", + Uuid::new_v4().simple() + )); + let rendered = build_global_codex_config(&path, crate::config::DEFAULT_PORT).unwrap(); + let document = rendered.parse::().unwrap(); + let idle_timeout = document["model_providers"]["opencode_go_adapter"] + ["stream_idle_timeout_ms"] + .as_integer() + .unwrap(); + + assert_eq!(idle_timeout, DEFAULT_STREAM_IDLE_TIMEOUT_MS); + assert!(idle_timeout > (DEFAULT_TIMEOUT_SECONDS * 1_000) as i64); + } +} + fn create_backup_if_exists(path: &Path) -> anyhow::Result> { if !path.exists() { return Ok(None); diff --git a/src/server.rs b/src/server.rs index 6a60e46..b626b18 100644 --- a/src/server.rs +++ b/src/server.rs @@ -1,5 +1,6 @@ use axum::extract::State; use axum::http::{HeaderMap, StatusCode}; +use axum::response::sse::KeepAlive; use axum::response::{IntoResponse, Response, Sse}; use axum::routing::{get, post}; use axum::{Json, Router}; @@ -9,6 +10,7 @@ use std::collections::HashMap; use std::convert::Infallible; use std::path::PathBuf; use std::sync::{Arc, RwLock}; +use std::time::Duration; use tokio::sync::Semaphore; use uuid::Uuid; @@ -31,6 +33,12 @@ use crate::upstream::{ OpenCodeGoClient, UpstreamError, }; +const DOWNSTREAM_SSE_KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(15); + +fn downstream_sse_keep_alive(interval: Duration) -> KeepAlive { + KeepAlive::new().interval(interval).text("keep-alive") +} + #[derive(Clone)] pub struct ProjectRuntime { pub config: Config, @@ -541,7 +549,11 @@ async fn stream_response( }); let stream = tokio_stream::wrappers::UnboundedReceiverStream::new(rx); - Sse::new(stream).into_response() + Sse::new(stream) + .keep_alive(downstream_sse_keep_alive( + DOWNSTREAM_SSE_KEEP_ALIVE_INTERVAL, + )) + .into_response() } fn previous_response( @@ -932,3 +944,27 @@ fn upstream_error(error: UpstreamError) -> Response { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn downstream_sse_emits_keep_alive_while_response_stream_is_idle() { + let stream = futures::stream::pending::< + Result, + >(); + let response = Sse::new(stream) + .keep_alive(downstream_sse_keep_alive(Duration::from_millis(5))) + .into_response(); + let mut body = response.into_body().into_data_stream(); + + let chunk = tokio::time::timeout(Duration::from_millis(100), body.next()) + .await + .expect("keep-alive should arrive before the test timeout") + .expect("SSE body should still be open") + .expect("keep-alive body chunk should be readable"); + + assert_eq!(std::str::from_utf8(&chunk).unwrap(), ": keep-alive\n\n"); + } +} diff --git a/src/upstream.rs b/src/upstream.rs index 2c12997..85f0691 100644 --- a/src/upstream.rs +++ b/src/upstream.rs @@ -21,6 +21,7 @@ pub struct OpenCodeGoClient { base_url: String, api_key: String, client: reqwest::Client, + stream_client: reqwest::Client, } impl OpenCodeGoClient { @@ -29,11 +30,14 @@ impl OpenCodeGoClient { api_key: impl Into, timeout_seconds: u64, ) -> anyhow::Result { + let timeout = Duration::from_secs(timeout_seconds); Ok(Self { base_url: base_url.into().trim_end_matches('/').to_string(), api_key: api_key.into(), - client: reqwest::Client::builder() - .timeout(Duration::from_secs(timeout_seconds)) + client: reqwest::Client::builder().timeout(timeout).build()?, + stream_client: reqwest::Client::builder() + .connect_timeout(timeout) + .read_timeout(timeout) .build()?, }) } @@ -55,7 +59,7 @@ impl OpenCodeGoClient { ) -> Result>, UpstreamError> { payload["stream"] = Value::Bool(true); let response = self - .client + .stream_client .post(format!("{}/chat/completions", self.base_url)) .headers(self.headers("text/event-stream")?) .json(&payload) @@ -213,6 +217,7 @@ pub fn sse_data_from_block(block: &str) -> Option { #[cfg(test)] mod tests { use super::*; + use futures::StreamExt as _; use serde_json::json; // ── extract_error_message ── @@ -383,4 +388,42 @@ mod tests { let data = sse_data_from_block(block).unwrap(); assert_eq!(data, "{\"key\":\"value\"}"); } + + #[tokio::test] + async fn streaming_request_can_outlive_nonstream_total_timeout() { + async fn slow_stream() -> axum::response::Sse< + impl futures::Stream>, + > { + let stream = async_stream::stream! { + for index in 0..4 { + tokio::time::sleep(Duration::from_millis(350)).await; + yield Ok(axum::response::sse::Event::default() + .data(json!({"chunk": index}).to_string())); + } + }; + axum::response::Sse::new(stream) + } + + let app = axum::Router::new().route("/chat/completions", axum::routing::post(slow_stream)); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + + let client = OpenCodeGoClient::new(format!("http://{address}"), "test-key", 1).unwrap(); + let started = std::time::Instant::now(); + let mut stream = client + .chat_stream(json!({"model": "test-model"})) + .await + .unwrap(); + let mut chunks = 0; + while let Some(chunk) = stream.next().await { + chunk.expect("each SSE body chunk should arrive before the read timeout"); + chunks += 1; + } + + assert!(started.elapsed() > Duration::from_secs(1)); + assert!(chunks >= 4); + } }