[Router] Abort the engine when a client disconnects mid-request

experimental/sgl-router/src/proxy/abort.rs

@@ -0,0 +1,47 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
+// SPDX-License-Identifier: Apache-2.0
+
+use axum::http::{header::AUTHORIZATION, HeaderMap};
+use reqwest::{Client, RequestBuilder, Url};
+use std::time::Duration;
+
+/// Cancels unfinished engine work without delaying request cleanup.
+pub(super) struct AbortOnDrop(Option<RequestBuilder>);
+
+impl AbortOnDrop {
+    // Only router-minted IDs are safe: the engine aborts by prefix.
+    pub(super) fn new(
+        client: &Client,
+        worker: &Url,
+        headers: &HeaderMap,
+        rid: Option<&str>,
+    ) -> Self {
+        Self(rid.filter(|rid| !rid.is_empty()).map(|rid| {
+            let mut request = client
+                .post(worker.join("/abort_request").expect("validated worker URL"))
+                .json(&serde_json::json!({"rid": rid, "abort_all": false}))
+                .timeout(Duration::from_secs(5));
+            if let Some(auth) = headers.get(AUTHORIZATION) {
+                request = request.header(AUTHORIZATION, auth);
+            }
+            request
+        }))
+    }
+
+    pub(super) fn disarm(&mut self) {
+        self.0 = None;
+    }
+}
+
+impl Drop for AbortOnDrop {
+    fn drop(&mut self) {
+        let Some(request) = self.0.take() else { return };
+        if let Ok(runtime) = tokio::runtime::Handle::try_current() {
+            runtime.spawn(async move {
+                if let Err(error) = request.send().await.and_then(|r| r.error_for_status()) {
+                    tracing::warn!(%error, "engine abort failed");
+                }
+            });
+        }
+    }
+}

experimental/sgl-router/src/proxy/mod.rs

@@ -3,8 +3,11 @@
 
 //! HTTP proxy — forwards requests to the upstream SGLang worker.
 
+mod abort;
 pub mod sse;
 
+use abort::AbortOnDrop;
+
 use crate::health::circuit_breaker::CircuitBreaker;
 use crate::server::error::ApiError;
 use crate::server::header_utils::should_forward_request_header;
@@ -203,6 +206,7 @@ impl Proxy {
     /// path concatenation (no double-slash) and pass a typed URL to the
     /// split error variants (`UpstreamUnreachable` / `UpstreamTimeout` /
     /// `UpstreamStatus`).
+    #[allow(clippy::too_many_arguments)]
     pub async fn forward_json_to(
         &self,
         worker_url: &str,
@@ -211,6 +215,7 @@ impl Proxy {
         path: &str,
         headers: &HeaderMap,
         body: Bytes,
+        abort_rid: Option<&str>,
     ) -> Result<Response<Body>, ApiError> {
         let permit = breaker.acquire().ok_or_else(|| ApiError::BreakerOpen {
             worker: worker_url.to_string(),
@@ -228,6 +233,8 @@ impl Proxy {
         req = req
             .header("content-type", "application/json")
             .timeout(self.request_timeout);
+        let mut abort =
+            AbortOnDrop::new(self.client_for(protocol), &worker_url, headers, abort_rid);
         let resp = req.send().await.map_err(|e| {
             breaker.record_failure();
             Self::classify_reqwest_error_for(worker_url.clone(), e, path)
@@ -257,6 +264,7 @@ impl Proxy {
                 return Err(ApiError::UpstreamStatus { status });
             }
         };
+        abort.disarm();
         match breaker_outcome(status) {
             BreakerOutcome::Failure => breaker.record_failure(),
             BreakerOutcome::Success => breaker.record_success(),
@@ -301,6 +309,7 @@ impl Proxy {
         path: &str,
         headers: &HeaderMap,
         body: Bytes,
+        abort_rid: Option<&str>,
         stream_guards: Option<Box<dyn Send + 'static>>,
         on_first_byte: Option<Box<dyn FnOnce() + Send + 'static>>,
         on_stream_end: Option<Box<dyn FnOnce(sse::StreamEnd) + Send + 'static>>,
@@ -322,11 +331,16 @@ impl Proxy {
         req = req
             .header("content-type", "application/json")
             .header("accept", "text/event-stream");
+        let mut abort =
+            AbortOnDrop::new(self.client_for(protocol), &worker_url, headers, abort_rid);
         let resp = req.send().await.map_err(|e| {
             breaker.record_failure();
             Self::classify_reqwest_error_for(worker_url.clone(), e, path)
         })?;
         let status = resp.status();
+        if !status.is_success() {
+            abort.disarm();
+        }
         let upstream_ct = resp
             .headers()
             .get(reqwest::header::CONTENT_TYPE)
@@ -366,6 +380,9 @@ impl Proxy {
                 BreakerOutcome::Success => {
                     let breaker_for_hook = Arc::clone(breaker);
                     Some(Box::new(move |end| {
+                        if end.reason == sse::StreamEndReason::Completed {
+                            abort.disarm();
+                        }
                         match stream_breaker_outcome(end) {
                             BreakerOutcome::Success => breaker_for_hook.record_success(),
                             BreakerOutcome::Failure => breaker_for_hook.record_failure(),
@@ -465,6 +482,7 @@ mod tests {
                 "/chat",
                 &headers,
                 Bytes::new(),
+                None,
             )
             .now_or_never()
             .is_none());
@@ -481,6 +499,7 @@ mod tests {
                 None,
                 None,
                 None,
+                None,
             )
             .now_or_never()
             .is_none());
@@ -583,6 +602,7 @@ mod tests {
                 None,
                 None,
                 None,
+                None,
                 expiration,
             )
             .await
@@ -694,6 +714,7 @@ mod tests {
                     "/v1/chat/completions",
                     &headers,
                     Bytes::from_static(b"{}"),
+                    None,
                 )
                 .await
                 .expect("dispatch should reach the worker (breaker must stay closed)");
@@ -737,6 +758,7 @@ mod tests {
                     "/v1/chat/completions",
                     &headers,
                     Bytes::from_static(b"{}"),
+                    None,
                 )
                 .await;
         }
@@ -780,6 +802,7 @@ mod tests {
                 "/v1/chat/completions",
                 &headers,
                 Bytes::from_static(b"{}"),
+                None,
             )
             .await
             .expect("the half-open probe must be admitted and reach the worker");
@@ -819,6 +842,7 @@ mod tests {
                     None,
                     None,
                     None,
+                    None,
                 )
                 .await
                 .expect("streaming dispatch should reach the worker");

experimental/sgl-router/src/server/app.rs

@@ -150,6 +150,9 @@ async fn access_log_and_record(
                 .unwrap_or_else(|| outcome_from_status(status.as_u16()))
                 .as_str(),
             worker = log_ctx.map(|c| c.worker_url.as_str()).unwrap_or(""),
+            engine_rid = log_ctx
+                .and_then(|c| c.engine_rid.as_deref())
+                .unwrap_or(""),
             model = log_ctx.map(|c| c.model_id.as_str()).unwrap_or(""),
             stream = log_ctx.is_some_and(|c| c.streaming),
             latency_ms,
@@ -446,6 +449,7 @@ mod tests {
                         model_id: "tiny".into(),
                         streaming: false,
                         outcome: RequestOutcome::Cancelled,
+                        engine_rid: Some("1f0c2b7a4e9d4f3ab6c5d8e7f0a1b2c3".into()),
                     });
                     resp
                 }),
@@ -468,6 +472,11 @@ mod tests {
             logs.contains("worker=\"http://worker-a:30000\"") && logs.contains("model=\"tiny\""),
             "a routed request must be logged with its worker and model; captured:\n{logs}",
         );
+        assert!(
+            logs.contains("engine_rid=\"1f0c2b7a4e9d4f3ab6c5d8e7f0a1b2c3\"")
+                && logs.contains("request_id="),
+            "the minted rid must be logged beside the caller's request id; captured:\n{logs}",
+        );
         // The handler's outcome must win over the status-derived fallback —
         // otherwise the log and `worker_requests_total` can disagree about a
         // request the handler classified itself (here, a cancellation served

experimental/sgl-router/src/server/metrics.rs

@@ -223,6 +223,8 @@ pub struct RequestLogContext {
     /// line and `worker_requests_total` cannot disagree — the middleware can
     /// only see the status, which cannot express a router-side cancellation.
     pub outcome: RequestOutcome,
+    /// Router-minted engine ID, logged beside the caller's correlation ID.
+    pub engine_rid: Option<String>,
 }
 
 /// Final outcome of a 2xx SSE stream.

experimental/sgl-router/src/server/routes/chat/forward.rs

@@ -81,7 +81,12 @@ pub(super) async fn forward_chat_request(
         };
         (decode, bootstrap)
     });
-    let body = request.into_outgoing_body(ctx, pd.as_ref().map(|(_, bootstrap)| bootstrap))?;
+    let engine_rid = request.engine_rid(pd.is_some());
+    let body = request.into_outgoing_body(
+        ctx,
+        pd.as_ref().map(|(_, bootstrap)| bootstrap),
+        engine_rid.as_deref(),
+    )?;
     let prefill_load_guards = (worker_load_guard, active_request_guard);
 
     // In PD mode, prefill runs independently and decode supplies the client response.
@@ -112,6 +117,7 @@ pub(super) async fn forward_chat_request(
         &response_worker,
         &headers,
         body,
+        engine_rid.as_deref(),
         response_load_guards,
         &metrics,
         expiration_token.clone(),
@@ -124,7 +130,7 @@ pub(super) async fn forward_chat_request(
             model: metrics.model.clone(),
         }),
     };
-    let log_context = metrics.record_dispatch_result(&result);
+    let log_context = metrics.record_dispatch_result(&result, engine_rid);
     // Materialize dispatch errors here so the access log retains the selected worker.
     let mut response = match result {
         Ok(mut response) => {
@@ -171,6 +177,7 @@ fn spawn_prefill_request(
                 CHAT_PATH,
                 &headers,
                 body,
+                None,
             )
             .await
         {
@@ -188,11 +195,13 @@ fn spawn_prefill_request(
     });
 }
 
+#[allow(clippy::too_many_arguments)]
 async fn forward_to_response_worker(
     ctx: &AppContext,
     worker: &Worker,
     headers: &HeaderMap,
     body: Bytes,
+    engine_rid: Option<&str>,
     load_guards: LoadGuards,
     metrics: &DispatchMetrics,
     expiration: CancellationToken,
@@ -209,6 +218,7 @@ async fn forward_to_response_worker(
                 CHAT_PATH,
                 headers,
                 body,
+                engine_rid,
                 Some(stream_guards),
                 Some(metrics.first_byte_callback()),
                 Some(metrics.stream_end_callback(worker.url.clone())),
@@ -226,6 +236,7 @@ async fn forward_to_response_worker(
                 CHAT_PATH,
                 headers,
                 body,
+                engine_rid,
             )
             .await
     }
@@ -295,6 +306,7 @@ impl DispatchMetrics {
     fn record_dispatch_result(
         &self,
         result: &Result<Response<Body>, ApiError>,
+        engine_rid: Option<String>,
     ) -> RequestLogContext {
         let http_status = match result {
             Ok(response) => response.status().as_u16(),
@@ -326,6 +338,7 @@ impl DispatchMetrics {
             model_id: self.model.clone(),
             streaming: self.streaming,
             outcome,
+            engine_rid,
         }
     }
 }