[Rust Frontend] Support `--sse-keep-alive-interval` (#58306)
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com> Signed-off-by: Bugen Zhao <i@bugenzhao.com>
Bugen Zhao committed
Sep 24, 2026 at 14:19 UTC
e028554a1f23f5034b5475847956b22934354ddd
7 files changed
+135
-10
rust/src/cmd/src/cli.rs
+10
@@ -363,6 +363,14 @@ pub struct SharedRuntimeArgs {
363
#[serde(default)]
364
pub enable_scale_out: bool,
365
366
+ /// Send an SSE keep-alive comment line every this many seconds when a
367
+ /// streaming response is idle (queued, prefill, or between tokens), to
368
+ /// prevent reverse proxies/tunnels with read timeouts from closing the
369
+ /// connection. Defaults to 0, which disables keep-alive comments entirely.
370
+ #[arg(long, default_value_t = 0)]
371
+ #[serde(default)]
372
+ pub sse_keep_alive_interval: u64,
373
+
374
/// If provided, the server will require one of these keys to be presented
375
/// in the Authorization header.
376
#[educe(Debug(ignore))]
@@ -616,6 +624,8 @@ impl SharedRuntimeArgs {
624
enable_prompt_tokens_details: self.enable_prompt_tokens_details,
625
enable_request_id_headers: self.enable_request_id_headers,
626
enable_scale_out: self.enable_scale_out,
627
+ sse_keep_alive_interval: (self.sse_keep_alive_interval > 0)
628
+ .then(|| Duration::from_secs(self.sse_keep_alive_interval)),
629
}
630
}
631
rust/src/cmd/src/cli/tests.rs
+6
@@ -200,6 +200,7 @@ fn serve_args_forward_python_flags_with_separator() {
200
enable_prompt_tokens_details: false,
201
enable_request_id_headers: false,
202
enable_scale_out: false,
203
+ sse_keep_alive_interval: 0,
204
disable_log_stats: false,
205
served_model_name: [],
206
allowed_origins: JsonStringList(
@@ -1040,6 +1041,7 @@ fn frontend_args_accept_json() {
1041
enable_prompt_tokens_details: false,
1042
enable_request_id_headers: false,
1043
enable_scale_out: false,
1044
+ sse_keep_alive_interval: 0,
1045
disable_log_stats: false,
1046
served_model_name: [],
1047
allowed_origins: JsonStringList(
@@ -1711,6 +1713,7 @@ fn serve_args_accept_handshake_aliases() {
1713
enable_prompt_tokens_details: false,
1714
enable_request_id_headers: false,
1715
enable_scale_out: false,
1716
+ sse_keep_alive_interval: 0,
1717
disable_log_stats: false,
1718
served_model_name: [],
1719
allowed_origins: JsonStringList(
@@ -1867,6 +1870,7 @@ fn serve_frontend_config_uses_dp_address_as_advertised_host() {
1870
enable_prompt_tokens_details: false,
1871
enable_request_id_headers: false,
1872
enable_scale_out: false,
1873
+ sse_keep_alive_interval: None,
1874
},
1875
cors: CorsConfig {
1876
allow_origins: [
@@ -1960,6 +1964,7 @@ fn serve_frontend_config_keeps_tcp_transport_for_non_local_only_topology() {
1964
enable_prompt_tokens_details: false,
1965
enable_request_id_headers: false,
1966
enable_scale_out: false,
1967
+ sse_keep_alive_interval: None,
1968
},
1969
cors: CorsConfig {
1970
allow_origins: [
@@ -2074,6 +2079,7 @@ fn frontend_config_uses_external_coordinator_when_coordinator_address_is_present
2079
enable_prompt_tokens_details: false,
2080
enable_request_id_headers: false,
2081
enable_scale_out: false,
2082
+ sse_keep_alive_interval: None,
2083
},
2084
cors: CorsConfig {
2085
allow_origins: [
rust/src/server/src/config.rs
+3
@@ -59,6 +59,9 @@ pub struct ApiServerOptions {
59
pub enable_request_id_headers: bool,
60
/// When `true`, register the scale-out `/inference/v1/generate` route.
61
pub enable_scale_out: bool,
62
+ /// Idle interval after which streaming SSE responses send a keep-alive
63
+ /// comment. `None` disables keep-alive comments.
64
+ pub sse_keep_alive_interval: Option<Duration>,
65
}
66
67
/// CORS settings mirroring Python's `CORSMiddleware`; the default is permissive.
rust/src/server/src/routes/inference/generate.rs
+3
-3
@@ -14,7 +14,7 @@ use asynk_strim_attr::{TryYielder, try_stream};
14
use axum::Json;
15
use axum::extract::State;
16
use axum::http::HeaderMap;
17
-use axum::response::sse::{Event, Sse};
17
+use axum::response::sse::Event;
18
use axum::response::{IntoResponse, Response};
19
use futures::{Stream, StreamExt as _, pin_mut};
20
use thiserror_ext::AsReport as _;
@@ -40,7 +40,7 @@ use crate::routes::openai::utils::logprobs::clamp_logprob;
40
use crate::routes::openai::utils::types::{ChatLogProbs, ChatLogProbsContent, TopLogProb, Usage};
41
use crate::routes::openai::utils::validated_json::ValidatedJson;
42
use crate::state::AppState;
43
-use crate::utils::resolve_request_context;
43
+use crate::utils::{resolve_request_context, sse_response};
44
45
/// Validate one token-in/token-out request and proxy it into the shared
46
/// `vllm-text` stack.
@@ -114,7 +114,7 @@ pub async fn generate(
114
);
115
let sse_stream = generate_sse_stream(chunk_stream).instrument(request_span);
116
117
- return Sse::new(sse_stream).into_response();
117
+ return sse_response(sse_stream, api_server_options.sse_keep_alive_interval);
118
}
119
120
let collected = match raw_stream.collect_output().instrument(request_span.clone()).await {
rust/src/server/src/routes/openai/chat_completions.rs
+3
-3
@@ -13,7 +13,7 @@ use asynk_strim_attr::{TryYielder, try_stream};
13
use axum::Json;
14
use axum::extract::State;
15
use axum::http::HeaderMap;
16
-use axum::response::sse::{Event, Sse};
16
+use axum::response::sse::Event;
17
use axum::response::{IntoResponse, Response};
18
use futures::{Stream, StreamExt as _, pin_mut};
19
use serde_json::Value;
@@ -45,7 +45,7 @@ use crate::routes::openai::utils::types::{
45
use crate::routes::openai::utils::usage::ContinuousUsage;
46
use crate::routes::openai::utils::validated_json::ValidatedJson;
47
use crate::state::AppState;
48
-use crate::utils::{ResolvedRequestContext, resolve_request_context, unix_timestamp};
48
+use crate::utils::{ResolvedRequestContext, resolve_request_context, sse_response, unix_timestamp};
49
50
pub(crate) fn lower_chat_request(
51
request: ChatCompletionRequest,
@@ -99,7 +99,7 @@ pub async fn chat_completions(
99
);
100
let sse_stream = chat_completion_sse_stream(chunk_stream).instrument(request_span);
101
102
- Sse::new(sse_stream).into_response()
102
+ sse_response(sse_stream, api_server_options.sse_keep_alive_interval)
103
} else {
104
let response = match collect_chat_completion(
105
chat_stream,
rust/src/server/src/routes/openai/completions.rs
+3
-3
@@ -13,7 +13,7 @@ use asynk_strim_attr::{TryYielder, try_stream};
13
use axum::Json;
14
use axum::extract::State;
15
use axum::http::HeaderMap;
16
-use axum::response::sse::{Event, Sse};
16
+use axum::response::sse::Event;
17
use axum::response::{IntoResponse, Response};
18
use futures::{Stream, StreamExt as _, pin_mut};
19
use serde_json::Value;
@@ -45,7 +45,7 @@ use crate::routes::openai::utils::types::LogProbs;
45
use crate::routes::openai::utils::usage::ContinuousUsage;
46
use crate::routes::openai::utils::validated_json::ValidatedJson;
47
use crate::state::AppState;
48
-use crate::utils::{ResolvedRequestContext, resolve_request_context, unix_timestamp};
48
+use crate::utils::{ResolvedRequestContext, resolve_request_context, sse_response, unix_timestamp};
49
50
pub(crate) fn lower_completion_request(
51
request: CompletionRequest,
@@ -111,7 +111,7 @@ pub async fn completions(
111
);
112
let sse_stream = completion_sse_stream(chunk_stream).instrument(request_span);
113
114
- Sse::new(sse_stream).into_response()
114
+ sse_response(sse_stream, api_server_options.sse_keep_alive_interval)
115
} else {
116
let response = match collect_completion(
117
text_stream,
rust/src/server/src/utils.rs
+107
-1
@@ -2,9 +2,13 @@
2
// SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
4
use std::collections::HashMap;
5
-use std::time::{SystemTime, UNIX_EPOCH};
5
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
6
7
+use axum::BoxError;
8
use axum::http::HeaderMap;
9
+use axum::response::sse::{Event, KeepAlive, Sse};
10
+use axum::response::{IntoResponse, Response};
11
+use futures::Stream;
12
use serde_json::Value;
13
use thiserror_ext::AsReport;
14
use uuid::Uuid;
@@ -27,6 +31,27 @@ pub fn unix_timestamp() -> u64 {
31
.unwrap_or_default()
32
}
33
34
+/// Build a streaming SSE response, optionally sending keep-alive comments
35
+/// while the stream is idle.
36
+///
37
+/// With `keep_alive_interval` set, a `: keep-alive` comment is sent whenever no
38
+/// event has been produced for that long (queued, prefill, between tokens, or
39
+/// while a parser buffers a long tool call). SSE parsers ignore comments, but
40
+/// they still count as bytes for read timeouts in reverse proxies and clients.
41
+pub fn sse_response<S, E>(stream: S, keep_alive_interval: Option<Duration>) -> Response
42
+where
43
+ S: Stream<Item = Result<Event, E>> + Send + 'static,
44
+ E: Into<BoxError>,
45
+{
46
+ let sse = Sse::new(stream);
47
+ match keep_alive_interval {
48
+ Some(interval) => sse
49
+ .keep_alive(KeepAlive::new().interval(interval).text("keep-alive"))
50
+ .into_response(),
51
+ None => sse.into_response(),
52
+ }
53
+}
54
+
55
/// Construct an API error for a failed utility call to the engine core.
56
pub fn utility_call_error(method: &str, error: impl AsReport) -> ApiError {
57
ApiError::server_error(format!("failed to call {method}: {}", error.as_report()))
@@ -157,3 +182,84 @@ pub fn resolve_base_request_id(
182
id
183
})
184
}
185
+
186
+#[cfg(test)]
187
+mod tests {
188
+ use std::convert::Infallible;
189
+
190
+ use futures::{StreamExt as _, stream};
191
+ use tokio::time::{Instant, sleep};
192
+
193
+ use super::*;
194
+
195
+ /// Collect `(elapsed seconds, frame)` pairs from a response whose event
196
+ /// stream idles for 20 seconds between its two events.
197
+ async fn idle_stream_frames(keep_alive_interval: Option<Duration>) -> Vec<(u64, String)> {
198
+ let events = stream::iter([Ok::<_, Infallible>(Event::default().data("first"))]).chain(
199
+ stream::once(async {
200
+ sleep(Duration::from_secs(20)).await;
201
+ Ok(Event::default().data("second"))
202
+ }),
203
+ );
204
+ let start = Instant::now();
205
+ sse_response(events, keep_alive_interval)
206
+ .into_body()
207
+ .into_data_stream()
208
+ .map(|frame| {
209
+ let frame = frame.expect("read frame");
210
+ (
211
+ start.elapsed().as_secs(),
212
+ String::from_utf8(frame.to_vec()).expect("utf-8 frame"),
213
+ )
214
+ })
215
+ .collect()
216
+ .await
217
+ }
218
+
219
+ /// An idle gap is filled with `: keep-alive` comments at the configured
220
+ /// interval, and real events still pass through in order.
221
+ #[tokio::test(start_paused = true)]
222
+ async fn sse_response_sends_keep_alive_comments_while_idle() {
223
+ let frames = idle_stream_frames(Some(Duration::from_secs(8))).await;
224
+ expect_test::expect![[r#"
225
+ [
226
+ (
227
+ 0,
228
+ "data: first\n\n",
229
+ ),
230
+ (
231
+ 8,
232
+ ": keep-alive\n\n",
233
+ ),
234
+ (
235
+ 16,
236
+ ": keep-alive\n\n",
237
+ ),
238
+ (
239
+ 20,
240
+ "data: second\n\n",
241
+ ),
242
+ ]
243
+ "#]]
244
+ .assert_debug_eq(&frames);
245
+ }
246
+
247
+ /// Without an interval, an idle gap produces no extra bytes.
248
+ #[tokio::test(start_paused = true)]
249
+ async fn sse_response_without_keep_alive_interval_sends_no_comments() {
250
+ let frames = idle_stream_frames(None).await;
251
+ expect_test::expect![[r#"
252
+ [
253
+ (
254
+ 0,
255
+ "data: first\n\n",
256
+ ),
257
+ (
258
+ 20,
259
+ "data: second\n\n",
260
+ ),
261
+ ]
262
+ "#]]
263
+ .assert_debug_eq(&frames);
264
+ }
265
+}